email notification changes (#418)

* email notification changes

* restored extracting user id implementation

* review comments implemented

* rectified application properties

* modified properties, removed which are not required

* restored an existing property

src/main/java/org/sunbird/cbp/service/CbPlanServiceImpl.java

@@ -30,6 +30,7 @@
 import org.sunbird.common.util.CbExtServerProperties;
 import org.sunbird.common.util.Constants;
 import org.sunbird.common.util.ProjectUtil;
+import org.sunbird.core.producer.Producer;
 import org.sunbird.user.service.UserUtilityService;
 
 import com.datastax.driver.core.utils.UUIDs;
@@ -59,6 +60,12 @@ public class CbPlanServiceImpl implements CbPlanService {
 
     ObjectMapper mapper = new ObjectMapper();
 
+    @Autowired
+    Producer kafkaProducer;
+
+    @Autowired
+    private CbExtServerProperties cbExtServerProperties;
+
     @Override
     public SBApiResponse createCbPlan(SunbirdApiRequest request, String userOrgId, String authUserToken) {
         SBApiResponse response = ProjectUtil.createDefaultResponse(Constants.API_CB_PLAN_CREATE);
@@ -555,9 +562,15 @@ public SBApiResponse requestCbplanContent(SunbirdApiRequest request, String toke
             propertiesMap.put(Constants.DESCRIPTION, description);
             propertiesMap.put(Constants.CREATED_AT, new Date());
             propertiesMap.put(Constants.CREATED_BY, userId);
-            cassandraOperation.insertRecord(Constants.SUNBIRD_KEY_SPACE_NAME, Constants.CB_CONTENT_REQUEST_TABLE, propertiesMap);
+            SBApiResponse dbResponse = cassandraOperation.insertRecord(Constants.SUNBIRD_KEY_SPACE_NAME, Constants.CB_CONTENT_REQUEST_TABLE, propertiesMap);
+            if(!Constants.SUCCESS.equalsIgnoreCase((String)dbResponse.get(Constants.RESPONSE))){
+                throw new RuntimeException("An error occurred while creating new content Request");
+            }
+            propertiesMap.put("mdoName", mdoInfo.get(Constants.CHANNEL));
+            propertiesMap.put(Constants.EMAIL, mdoInfo.get(Constants.EMAIL));
+            kafkaProducer.push(cbExtServerProperties.getCbplanContentRequestKafkaTopic(), propertiesMap);
         } catch (Exception e) {
-            logger.error("Failed to send request for a content for cbplam. Exception: " + e.getMessage(), e);
+            logger.error("Failed to send request for a content for cbplan Exception: " + e.getMessage(), e);
             response.getParams().setStatus(Constants.FAILED);
             response.getParams().setErrmsg(e.getMessage());
             response.setResponseCode(HttpStatus.INTERNAL_SERVER_ERROR);

src/main/java/org/sunbird/cbp/service/CbplanContentConsumer.java

@@ -0,0 +1,191 @@
+package org.sunbird.cbp.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.velocity.VelocityContext;
+import org.apache.velocity.app.VelocityEngine;
+import org.jboss.resteasy.spi.ApplicationException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+import org.springframework.util.CollectionUtils;
+import org.sunbird.cassandra.utils.CassandraOperation;
+import org.sunbird.common.model.Config;
+import org.sunbird.common.model.NotificationRequest;
+import org.sunbird.common.model.Template;
+import org.sunbird.common.service.OutboundRequestHandlerServiceImpl;
+import org.sunbird.common.util.Constants;
+import org.sunbird.core.config.PropertiesConfig;
+import org.sunbird.core.logger.CbExtLogger;
+
+import java.io.StringWriter;
+import java.util.*;
+
+@Component
+public class CbplanContentConsumer {
+
+    private CbExtLogger logger = new CbExtLogger(getClass().getName());
+
+    @Autowired
+    PropertiesConfig configuration;
+
+    @Autowired
+    private OutboundRequestHandlerServiceImpl outboundReqService;
+
+    private ObjectMapper mapper = new ObjectMapper();
+
+    @Autowired
+    CassandraOperation cassandraOperation;
+
+    @KafkaListener(topics = "${kafka.topic.cbplan.content.request}", groupId = "${kafka.topic.cbplan.content.request.group}")
+    public void cbplanContentRequestConsumer(ConsumerRecord<String, String> data){
+        HashMap<String, Object> cbplanContentRequest = null;
+        try {
+            cbplanContentRequest = mapper.readValue(data.value(), HashMap.class);
+            String competencyInfo = (String) cbplanContentRequest.get(Constants.COMPETENCY_INFO);
+            Map<String, Object> competencyInfoMap = mapper.readValue(competencyInfo, HashMap.class);
+
+            List<Map<String, Object>> competencyThemes = (List<Map<String, Object>>) competencyInfoMap.get(Constants.CHILDREN);
+            StringBuilder allThemes = new StringBuilder();
+            StringBuilder allSubThemes = new StringBuilder();
+            for(Map<String, Object> theme : competencyThemes){
+                allThemes.append(theme.get(Constants.NAME)).append(", ");
+                List<Map<String, Object>> competencySubThemes = (List<Map<String, Object>>) theme.get(Constants.CHILDREN);
+                for(Map<String, Object>  subTheme : competencySubThemes){
+                    allSubThemes.append(subTheme.get(Constants.NAME)).append(", ");
+                }
+            }
+            Set<String> providerRootOrgIds = new HashSet<>((List<String>) cbplanContentRequest.get(Constants.PROVIDER_ORG_ID));
+
+            Map<String, Object> mailNotificationDetails = new HashMap<>();
+            mailNotificationDetails.put(Constants.PROVIDER_EMAIL_ID_LIST, getCBPAdminDetails(providerRootOrgIds));
+            mailNotificationDetails.put(Constants.MDO_NAME, cbplanContentRequest.get(Constants.MDO_NAME));
+            mailNotificationDetails.put(Constants.COMPETENCY_AREA, competencyInfoMap.get(Constants.NAME));
+            mailNotificationDetails.put(Constants.COMPETENCY_THEMES, allThemes.replace(allThemes.length()-2, allThemes.length() - 1, "."));
+            mailNotificationDetails.put(Constants.COMPETENCY_SUB_THEMES, allSubThemes.replace(allSubThemes.length()-2, allSubThemes.length()-1, "."));
+            mailNotificationDetails.put(Constants.DESCRIPTION , cbplanContentRequest.get(Constants.DESCRIPTION));
+            sendNotificationToProviders(mailNotificationDetails);
+
+        } catch (Exception e) {
+            logger.error("Exception occurred while sending email : " + e.getMessage()  + "Content request received " + cbplanContentRequest,e);
+        }
+    }
+
+    public List<String> getCBPAdminDetails(Set<String> rootOrgIds){
+        List<Map<String, String>> userRecords = new ArrayList<>();
+        Map<String, Object> request = getSearchObject(rootOrgIds);
+        HashMap<String, String> headersValue = new HashMap<>();
+        headersValue.put("Content-Type", "application/json");
+        try {
+            List<String> providerIdEmails = new ArrayList<>();
+            StringBuilder url = new StringBuilder(configuration.getLmsServiceHost()).append(configuration.getLmsUserSearchEndPoint());
+            Map<String, Object> searchProfileApiResp = outboundReqService.fetchResultUsingPost(url.toString(), request, headersValue);
+            if (searchProfileApiResp != null
+                    && "OK".equalsIgnoreCase((String) searchProfileApiResp.get(Constants.RESPONSE_CODE))) {
+                Map<String, Object> map = (Map<String, Object>) searchProfileApiResp.get(Constants.RESULT);
+                Map<String, Object> response = (Map<String, Object>) map.get(Constants.RESPONSE);
+                List<Map<String, Object>> contents = (List<Map<String, Object>>) response.get(Constants.CONTENT);
+                if (!CollectionUtils.isEmpty(contents)) {
+                    for(Map<String, Object> content: contents){
+                        String rootOrgId = (String)content.get(Constants.ROOT_ORG_ID);
+                        HashMap<String, Object> profileDetails = (HashMap<String, Object>) content.get(Constants.PROFILE_DETAILS);
+                        if (!CollectionUtils.isEmpty(profileDetails)) {
+                            HashMap<String, Object> personalDetails = (HashMap<String, Object>) profileDetails.get(Constants.PERSONAL_DETAILS);
+                            if (!CollectionUtils.isEmpty(personalDetails) && personalDetails.get(Constants.PRIMARY_EMAIL)!= null ) {
+                                if(rootOrgIds.contains(rootOrgId))
+                                    providerIdEmails.add((String)personalDetails.get(Constants.PRIMARY_EMAIL));
+                            }
+                        }
+                    }
+                }
+            }
+            logger.info("CBP Admin emails fetched successfully: " + providerIdEmails);
+            return providerIdEmails;
+        } catch (Exception e) {
+            logger.error("Exception while fetching cbp admin details : " +e.getMessage() + " request : " + request,e);
+            throw new ApplicationException("Hub Service ERROR: ", e);
+        }
+    }
+
+    private Map<String, Object> getSearchObject(Set<String> rootOrgIds) {
+        Map<String, Object> requestObject = new HashMap<>();
+        Map<String, Object> request = new HashMap<>();
+        Map<String, Object> filters = new HashMap<>();
+        filters.put(Constants.ROOT_ORG_ID, rootOrgIds);
+        filters.put(Constants.ORGANIZATIONS_ROLES, Collections.singletonList(Constants.CBP_ADMIN));
+        request.put(Constants.LIMIT, 100);
+        request.put(Constants.OFFSET, 0);
+        request.put(Constants.FILTERS, filters);
+        request.put(Constants.FIELDS_CONSTANT, Arrays.asList("profileDetails.personalDetails.primaryEmail", Constants.ROOT_ORG_ID));
+        requestObject.put(Constants.REQUEST, request);
+        return requestObject;
+    }
+
+    private void sendNotificationToProviders( Map<String, Object> mailNotificationDetails) {
+        List<String> providerIdList = (List<String>) mailNotificationDetails.get(Constants.PROVIDER_EMAIL_ID_LIST);
+        String mdoName = (String) mailNotificationDetails.get(Constants.MDO_NAME);
+
+        Map<String, Object> params = new HashMap<>();
+        NotificationRequest notificationRequest = new NotificationRequest();
+        notificationRequest.setDeliveryType(Constants.MESSAGE);
+        notificationRequest.setIds(providerIdList);
+        notificationRequest.setMode(Constants.EMAIL);
+
+        params.put(Constants.MDO_NAME_PARAM, mdoName);
+        params.put(Constants.NAME, mdoName);
+        params.put(Constants.COMPETENCY_AREA_PARAM, mailNotificationDetails.get(Constants.COMPETENCY_AREA));
+        params.put(Constants.COMPETENCY_THEME_PARAM, mailNotificationDetails.get(Constants.COMPETENCY_THEMES));
+        params.put(Constants.COMPETENCY_SUB_THEME_PARAM, mailNotificationDetails.get(Constants.COMPETENCY_SUB_THEMES));
+        params.put(Constants.DESCRIPTION, mailNotificationDetails.get(Constants.DESCRIPTION));
+        params.put(Constants.FROM_EMAIL, configuration.getSupportEmail());
+        params.put(Constants.ORG_NAME, mdoName);
+        Template template = new Template(constructEmailTemplate(configuration.getCbplanContentRequestTemplate(), params),configuration.getCbplanContentRequestTemplate(), params);
+        template.setParams(params);
+        Config config = new Config();
+        config.setSubject(Constants.REQUEST_CONTENT_SUBJECT);
+        config.setSender(configuration.getSupportEmail());
+        Map<String, Object> req = new HashMap<>();
+        notificationRequest.setTemplate(template);
+        notificationRequest.setConfig(config);
+        Map<String, List<NotificationRequest>> notificationMap = new HashMap<>();
+        notificationMap.put(Constants.NOTIFICATIONS, Collections.singletonList(notificationRequest));
+        req.put(Constants.REQUEST, notificationMap);
+        sendNotification(req);
+    }
+
+    private void sendNotification(Map<String, Object> request) {
+        StringBuilder builder = new StringBuilder();
+        builder.append(configuration.getNotifyServiceHost()).append(configuration.getNotifyServicePath());
+        try {
+            Map<String, Object> response = outboundReqService.fetchResultUsingPost(builder.toString(), request, null);
+            logger.debug("The email notification is successfully sent, response is: " + response);
+        } catch (Exception e) {
+            logger.error("Exception while posting the data in notification service: ", e);
+        }
+    }
+
+    private String constructEmailTemplate(String templateName, Map<String, Object> params) {
+        String replacedHTML = new String();
+        try {
+            Map<String, Object> propertyMap = new HashMap<>();
+            propertyMap.put(Constants.NAME, templateName);
+            List<Map<String, Object>> templateMap = cassandraOperation.getRecordsByProperties(Constants.KEYSPACE_SUNBIRD, Constants.TABLE_EMAIL_TEMPLATE, propertyMap, Collections.singletonList(Constants.TEMPLATE));
+            String htmlTemplate = templateMap.stream()
+                    .findFirst()
+                    .map(template -> (String) template.get(Constants.TEMPLATE))
+                    .orElse(null);
+            VelocityEngine velocityEngine = new VelocityEngine();
+            velocityEngine.init();
+            VelocityContext context = new VelocityContext();
+            for (Map.Entry<String, Object> entry : params.entrySet()) {
+                context.put(entry.getKey(), entry.getValue());
+            }
+            StringWriter writer = new StringWriter();
+            velocityEngine.evaluate(context, writer, "HTMLTemplate", htmlTemplate);
+            replacedHTML = writer.toString();
+        } catch (Exception e) {
+            logger.error("Unable to create template ", e);
+        }
+        return replacedHTML;
+    }
+}

src/main/java/org/sunbird/common/util/CbExtServerProperties.java

@@ -198,6 +198,9 @@ public class CbExtServerProperties {
 	@Value("${cb-plan.update.publish.authorized.roles}")
 	private String cbPlanUpdatePublishAuthorizedRoles;
 
+	@Value("${kafka.topic.cbplan.content.request}")
+	private String cbplanContentRequestKafkaTopic;
+
 	public String getRedisDataHostName() {
 		return redisDataHostName;
 	}
@@ -2146,4 +2149,12 @@ public Map<String, String> getSpvReportSubFolderTypeMap() {
 	public void setSpvReportSubFolderTypeMap(Map<String, String> spvReportSubFolderTypeMap) {
 		this.spvReportSubFolderTypeMap = spvReportSubFolderTypeMap;
 	}
+
+	public String getCbplanContentRequestKafkaTopic() {
+		return cbplanContentRequestKafkaTopic;
+	}
+
+	public void setCbplanContentRequestKafkaTopic(String cbplanContentRequestKafkaTopic) {
+		this.cbplanContentRequestKafkaTopic = cbplanContentRequestKafkaTopic;
+	}
 }

src/main/java/org/sunbird/common/util/Constants.java

@@ -943,8 +943,25 @@ public class Constants {
 	public static final String COMPETENCY_DETAILS_MISSING = "Competency details missing for content request";
 
 	public static final String TABLE_USER_KARMA_POINTS_SUMMARY ="user_karma_points_summary";
-
-	private Constants() {
+    public static final String REQUEST_CONTENT_SUBJECT = "New Content Request for Capacity Building Plan Development";
+	public static final String COMPETENCY_THEMES = "competencyThemes";
+	public static final String COMPETENCY_SUB_THEMES = "competencySubThemes";
+	public static final String FOOTNOTE = "footnote";
+	public static final String FIRST_PARA = "firstPara";
+	public static final String SECOND_PARA = "secondPara";
+	public static final String MDO_NAME = "mdoName";
+	public static final String PROVIDER_EMAIL_ID_LIST = "providerEmailIdList";
+	public static final String CBP_ADMIN = "CBP_Admin";
+	public static final String ORGANIZATIONS_ROLES = "organisations.roles";
+	public static final String MDO_NAME_PARAM= "mdo_name";
+	public static final String FIRST_BODY_PARAM = "body_para1"; //"body_para2"
+	public static final String SECOND_BODY_PARAM = "body_para2";
+	public static final String COMPETENCY_AREA_PARAM = "competency_area";
+	public static final String COMPETENCY_THEME_PARAM = "competency_theme";
+	public static final String COMPETENCY_SUB_THEME_PARAM = "competency_subtheme";
+
+
+    private Constants() {
 		throw new IllegalStateException("Utility class");
 	}
 

src/main/java/org/sunbird/core/config/PropertiesConfig.java

@@ -35,6 +35,13 @@ public class PropertiesConfig {
     @Value("${hierarchy.store.keyspace.name}")
     private String hierarchyStoreKeyspaceName;
 
+    @Value("${cbplan.content.request.notification.template}")
+    private String cbplanContentRequestTemplate;  //notify.service.path.async
+
+    @Value("${notify.service.path.async}")
+    private String notificationAsyncPath;
+
+
     public String getLmsServiceHost() {
         return lmsServiceHost;
     }
@@ -106,4 +113,20 @@ public String getHierarchyStoreKeyspaceName() {
     public void setHierarchyStoreKeyspaceName(String hierarchyStoreKeyspaceName) {
         this.hierarchyStoreKeyspaceName = hierarchyStoreKeyspaceName;
     }
+
+    public String getCbplanContentRequestTemplate() {
+        return cbplanContentRequestTemplate;
+    }
+
+    public void setCbplanContentRequestTemplate(String cbplanContentRequestTemplate) {
+        this.cbplanContentRequestTemplate = cbplanContentRequestTemplate;
+    }
+
+    public String getNotificationAsyncPath() {
+        return notificationAsyncPath;
+    }
+
+    public void setNotificationAsyncPath(String notificationAsyncPath) {
+        this.notificationAsyncPath = notificationAsyncPath;
+    }
 }

src/main/resources/application.properties

@@ -358,6 +358,12 @@ karma.points.limit = 10
 kafka.topics.claim.acbp.karma.points=dev.user.claim.acbp.karma.points
 
 spv_admin_channel_name=Karmayogi Bharat
+
+notify.service.path.async=/v2/notification/send
+kafka.topic.cbplan.content.request=dev.cbplan.content.request
+kafka.topic.cbplan.content.request.group=cbplanContentRequestAsyncHandlerGroup
+cbplan.content.request.notification.template=cbplanContentRequestTemplate
+
 spv.report.property.map={"mdo_content_completion":"topMdoCompletionRatio.csv","user_recent_completions":"topNRecentCompletions.csv","user_without_enrollment":"usersWithoutEnrollments.csv","recent_user_without_enrollment":"recentUsersWithoutEnrollments.csv","all_prarambh_completed_users":"prarambhUsersAllCompletions.csv","prarambh_completed_users":"prarambhUsers6Completions.csv","acbp_user_summary_exhaust":"ACBPUserSummaryReport.csv","acbp_enrollment_exhaust":"ACBPEnrollmentReport.csv"}
 report.property.map={"user-report":"UserReport.csv","user-enrollment-report":"ConsumptionReport.csv","course-report":"CBPReport.csv","cba-report":"UserAssessmentReport.csv","user-assessment-report-cbp":"StandaloneAssessmentReport.csv","blended-program-report-mdo":"BlendedProgramReport.csv","blended-program-report-cbp":"BlendedProgramReport.csv","acbp-report-mdo-summary":"ACBPUserSummaryReport.csv","acbp-report-mdo-enrolment":"ACBPEnrollmentReport.csv"}
-spv.report.property.subFolder.map={"topMdoCompletionRatio.csv":"comms-console","topNRecentCompletions.csv":"comms-console","usersWithoutEnrollments.csv":"comms-console","recentUsersWithoutEnrollments.csv":"comms-console","prarambhUsersAllCompletions.csv":"comms-console","prarambhUsers6Completions.csv":"comms-console","ACBPUserSummaryReport.csv":"acbp-report","ACBPEnrollmentReport.csv":"acbp-report"}
+spv.report.property.subFolder.map={"topMdoCompletionRatio.csv":"comms-console","topNRecentCompletions.csv":"comms-console","usersWithoutEnrollments.csv":"comms-console","recentUsersWithoutEnrollments.csv":"comms-console","prarambhUsersAllCompletions.csv":"comms-console","prarambhUsers6Completions.csv":"comms-console","ACBPUserSummaryReport.csv":"acbp-report","ACBPEnrollmentReport.csv":"acbp-report"}