Telemetry fixes

course-mw/course-actors-common/src/main/java/org/sunbird/keys/SunbirdKey.java

@@ -57,6 +57,7 @@ public final class SunbirdKey {
     public static final String COLLECTION = "collection";
     public static final String EVENT_SET = "eventSet";
     public static final String OBJECT_TYPE = "objectType";
+    public static final String ROOT_ORG_ID = "rootOrgId";
 
     private SunbirdKey() {}
 }

course-mw/course-actors-common/src/main/java/org/sunbird/learner/actors/certificate/service/CourseBatchCertificateActor.java

@@ -14,8 +14,6 @@
 import org.sunbird.common.exception.ProjectCommonException;
 import org.sunbird.common.models.response.Response;
 import org.sunbird.common.models.util.JsonKey;
-import org.sunbird.common.models.util.LoggerUtil;
-import org.sunbird.common.models.util.ProjectLogger;
 import org.sunbird.common.models.util.TelemetryEnvKey;
 import org.sunbird.common.request.Request;
 import org.sunbird.common.request.RequestContext;
@@ -148,6 +146,19 @@ private Map<String, Object> mapESFieldsToObject(Map<String, Object> courseBatch)
                                       cert_template.getKey(), mapToObject(cert_template.getValue())));
       courseBatch.put(CourseJsonKey.CERTIFICATE_TEMPLATES_COLUMN, certificateTemplates);
     }
+    Map<String, Map<String, Object>> batchAttributes =
+            (Map<String, Map<String, Object>>)
+                    courseBatch.getOrDefault(CourseJsonKey.BATCH_ATTRIBUTES, null);
+    if(MapUtils.isNotEmpty(batchAttributes)){
+      batchAttributes
+              .entrySet()
+              .stream()
+              .forEach(
+                      batchAttribute ->
+                              batchAttributes.put(
+                                      batchAttribute.getKey(), mapToObject(batchAttribute.getValue())));
+      courseBatch.put(CourseJsonKey.BATCH_ATTRIBUTES, batchAttributes);
+    }
     return courseBatch;
   }
 

course-mw/course-actors-common/src/main/java/org/sunbird/learner/constants/CourseJsonKey.java

@@ -33,5 +33,6 @@ public abstract class CourseJsonKey {
       "sunbird_send_email_notifictaion_api";
   public static final String NOTIFY_TEMPLATE="notifyTemplate";
   public static final String CERT_TEMPLATES = "certTemplates";
-    public static final String ADDITIONAL_PROPS = "additionalProps";
+  public static final String ADDITIONAL_PROPS = "additionalProps";
+  public static final String BATCH_ATTRIBUTES = "batchAttributes";
 }

course-mw/course-actors-common/src/main/java/org/sunbird/models/course/batch/CourseBatch.java

@@ -43,6 +43,7 @@ public class CourseBatch implements Serializable {
   private Integer status;
 
   private Map<String, Object> certTemplates;
+  private Map<String, Object> batchAttributes;
 
   public String getCourseCreator() {
     return courseCreator;
@@ -213,4 +214,13 @@ public String getOldUpdatedDate() {
   public void setOldUpdatedDate(String updatedDate) {
     this.oldUpdatedDate = updatedDate;
   }
+
+  public Map<String, Object> getBatchAttributes() {
+    return batchAttributes;
+  }
+
+  public void setBatchAttributes(Map<String, Object> batchAttributes) {
+    this.batchAttributes = batchAttributes;
+  }
+
 }

course-mw/course-actors/src/main/java/org/sunbird/learner/actors/coursebatch/CourseBatchManagementActor.java

@@ -97,55 +97,58 @@ private void createCourseBatch(Request actorMessage) throws Throwable {
     List<Map<String, Object>> correlatedObject = new ArrayList<>();
     String courseBatchId = ProjectUtil.getUniqueIdFromTimestamp(actorMessage.getEnv());
     Map<String, String> headers =
-        (Map<String, String>) actorMessage.getContext().get(JsonKey.HEADER);
+            (Map<String, String>) actorMessage.getContext().get(JsonKey.HEADER);
     String requestedBy = (String) actorMessage.getContext().get(JsonKey.REQUESTED_BY);
 
     if (Util.isNotNull(request.get(JsonKey.PARTICIPANTS))) {
       ProjectCommonException.throwClientErrorException(
-          ResponseCode.invalidRequestParameter,
-          ProjectUtil.formatMessage(
-              ResponseCode.invalidRequestParameter.getErrorMessage(), PARTICIPANTS));
+              ResponseCode.invalidRequestParameter,
+              ProjectUtil.formatMessage(
+                      ResponseCode.invalidRequestParameter.getErrorMessage(), PARTICIPANTS));
     }
     CourseBatch courseBatch = JsonUtil.convert(request, CourseBatch.class);
     courseBatch.setStatus(setCourseBatchStatus(actorMessage.getRequestContext(), (String) request.get(JsonKey.START_DATE)));
     String courseId = (String) request.get(JsonKey.COURSE_ID);
-    Map<String, Object> contentDetails = getContentDetails(actorMessage.getRequestContext(),courseId, headers);
+    Map<String, Object> contentDetails = getContentDetails(actorMessage.getRequestContext(), courseId, headers);
     courseBatch.setCreatedDate(ProjectUtil.getTimeStamp());
-    if(StringUtils.isBlank(courseBatch.getCreatedBy()))
-    	courseBatch.setCreatedBy(requestedBy);
+    String primaryCategory = (String) contentDetails.getOrDefault(JsonKey.PRIMARYCATEGORY, "");
+    if (StringUtils.isBlank(courseBatch.getCreatedBy()))
+      courseBatch.setCreatedBy(requestedBy);
     validateContentOrg(actorMessage.getRequestContext(), courseBatch.getCreatedFor());
     validateMentors(courseBatch, (String) actorMessage.getContext().getOrDefault(JsonKey.X_AUTH_TOKEN, ""), actorMessage.getRequestContext());
     courseBatch.setBatchId(courseBatchId);
+    if (primaryCategory.equalsIgnoreCase(JsonKey.PRIMARY_CATEGORY_BLENDED_PROGRAM) && courseBatch.getBatchAttributes().get(JsonKey.CURRENT_BATCH_SIZE) == null) {
+      ProjectCommonException.throwClientErrorException(
+              ResponseCode.currentBatchSizeMissing, ResponseCode.currentBatchSizeMissing.getErrorMessage());
+    }
     Response result = courseBatchDao.create(actorMessage.getRequestContext(), courseBatch);
     result.put(JsonKey.BATCH_ID, courseBatchId);
 
     Map<String, Object> esCourseMap = CourseBatchUtil.esCourseMapping(courseBatch, dateFormat);
     CourseBatchUtil.syncCourseBatchForeground(actorMessage.getRequestContext(),
-        courseBatchId, esCourseMap);
+            courseBatchId, esCourseMap);
     sender().tell(result, self());
-
     targetObject =
-        TelemetryUtil.generateTargetObject(
-            courseBatchId, TelemetryEnvKey.BATCH, JsonKey.CREATE, null);
+            TelemetryUtil.generateTargetObject(
+                    courseBatchId, TelemetryEnvKey.BATCH, JsonKey.CREATE, null);
     TelemetryUtil.generateCorrelatedObject(
-        (String) request.get(JsonKey.COURSE_ID), JsonKey.COURSE, null, correlatedObject);
+            (String) request.get(JsonKey.COURSE_ID), JsonKey.COURSE, null, correlatedObject);
 
     Map<String, String> rollUp = new HashMap<>();
     rollUp.put("l1", (String) request.get(JsonKey.COURSE_ID));
     TelemetryUtil.addTargetObjectRollUp(rollUp, targetObject);
     TelemetryUtil.telemetryProcessingCall(request, targetObject, correlatedObject, actorMessage.getContext());
-
-  //  updateBatchCount(courseBatch);
-      updateCollection(actorMessage.getRequestContext(), esCourseMap, contentDetails);
+    //  updateBatchCount(courseBatch);
+    updateCollection(actorMessage.getRequestContext(), esCourseMap, contentDetails);
     if (courseNotificationActive()) {
       batchOperationNotifier(actorMessage, courseBatch, null);
     }
   }
 
   private boolean courseNotificationActive() {
     return Boolean.parseBoolean(
-        PropertiesCache.getInstance()
-            .getProperty(JsonKey.SUNBIRD_COURSE_BATCH_NOTIFICATIONS_ENABLED));
+            PropertiesCache.getInstance()
+                    .getProperty(JsonKey.SUNBIRD_COURSE_BATCH_NOTIFICATIONS_ENABLED));
   }
 
   private void batchOperationNotifier(Request actorMessage, CourseBatch courseBatch, Map<String, Object> participantMentorMap) {
@@ -157,13 +160,13 @@ private void batchOperationNotifier(Request actorMessage, CourseBatch courseBatc
     if (participantMentorMap != null) {
       batchNotificationMap.put(JsonKey.UPDATE, true);
       batchNotificationMap.put(
-          JsonKey.ADDED_MENTORS, participantMentorMap.get(JsonKey.ADDED_MENTORS));
+              JsonKey.ADDED_MENTORS, participantMentorMap.get(JsonKey.ADDED_MENTORS));
       batchNotificationMap.put(
-          JsonKey.REMOVED_MENTORS, participantMentorMap.get(JsonKey.REMOVED_MENTORS));
+              JsonKey.REMOVED_MENTORS, participantMentorMap.get(JsonKey.REMOVED_MENTORS));
       batchNotificationMap.put(
-          JsonKey.ADDED_PARTICIPANTS, participantMentorMap.get(JsonKey.ADDED_PARTICIPANTS));
+              JsonKey.ADDED_PARTICIPANTS, participantMentorMap.get(JsonKey.ADDED_PARTICIPANTS));
       batchNotificationMap.put(
-          JsonKey.REMOVED_PARTICIPANTS, participantMentorMap.get(JsonKey.REMOVED_PARTICIPANTS));
+              JsonKey.REMOVED_PARTICIPANTS, participantMentorMap.get(JsonKey.REMOVED_PARTICIPANTS));
 
     } else {
       batchNotificationMap.put(JsonKey.OPERATION_TYPE, JsonKey.ADD);
@@ -187,34 +190,35 @@ private void updateCourseBatch(Request actorMessage) throws Exception {
     Map<String, Object> request = actorMessage.getRequest();
     if (Util.isNotNull(request.get(JsonKey.PARTICIPANTS))) {
       ProjectCommonException.throwClientErrorException(
-          ResponseCode.invalidRequestParameter,
-          ProjectUtil.formatMessage(
-              ResponseCode.invalidRequestParameter.getErrorMessage(), PARTICIPANTS));
+              ResponseCode.invalidRequestParameter,
+              ProjectUtil.formatMessage(
+                      ResponseCode.invalidRequestParameter.getErrorMessage(), PARTICIPANTS));
     }
     String batchId =
-        request.containsKey(JsonKey.BATCH_ID)
-            ? (String) request.get(JsonKey.BATCH_ID)
-            : (String) request.get(JsonKey.ID);
+            request.containsKey(JsonKey.BATCH_ID)
+                    ? (String) request.get(JsonKey.BATCH_ID)
+                    : (String) request.get(JsonKey.ID);
     CourseBatch oldBatch =
-        courseBatchDao.readById((String) request.get(JsonKey.COURSE_ID), batchId, actorMessage.getRequestContext());
+            courseBatchDao.readById((String) request.get(JsonKey.COURSE_ID), batchId, actorMessage.getRequestContext());
     CourseBatch courseBatch = getUpdateCourseBatch(actorMessage.getRequestContext(), request, oldBatch);
     courseBatch.setUpdatedDate(ProjectUtil.getTimeStamp());
-    Map<String, Object> contentDetails = getContentDetails(actorMessage.getRequestContext(),courseBatch.getCourseId(), headers);
+    Map<String, Object> contentDetails = getContentDetails(actorMessage.getRequestContext(), courseBatch.getCourseId(), headers);
     validateUserPermission(courseBatch, requestedBy);
     validateContentOrg(actorMessage.getRequestContext(), courseBatch.getCreatedFor());
     validateMentors(courseBatch, (String) actorMessage.getContext().getOrDefault(JsonKey.X_AUTH_TOKEN, ""), actorMessage.getRequestContext());
+    String primaryCategory = (String) contentDetails.getOrDefault(JsonKey.PRIMARYCATEGORY, "");
     participantsMap = getMentorLists(participantsMap, oldBatch, courseBatch);
     Map<String, Object> courseBatchMap = CourseBatchUtil.cassandraCourseMapping(courseBatch, dateFormat);
     Response result =
-        courseBatchDao.update(actorMessage.getRequestContext(), (String) request.get(JsonKey.COURSE_ID), batchId, courseBatchMap);
+            courseBatchDao.update(actorMessage.getRequestContext(), (String) request.get(JsonKey.COURSE_ID), batchId, courseBatchMap);
     CourseBatch updatedCourseObject = mapESFieldsToObject(courseBatch);
     sender().tell(result, self());
     Map<String, Object> esCourseMap = CourseBatchUtil.esCourseMapping(updatedCourseObject, dateFormat);
 
     CourseBatchUtil.syncCourseBatchForeground(actorMessage.getRequestContext(), batchId, esCourseMap);
 
     targetObject =
-        TelemetryUtil.generateTargetObject(batchId, TelemetryEnvKey.BATCH, JsonKey.UPDATE, null);
+            TelemetryUtil.generateTargetObject(batchId, TelemetryEnvKey.BATCH, JsonKey.UPDATE, null);
 
     Map<String, String> rollUp = new HashMap<>();
     rollUp.put("l1", courseBatch.getCourseId());
@@ -592,7 +596,7 @@ private boolean isOrgValid(RequestContext requestContext, String orgId) {
   }
 
   private Map<String, Object> getContentDetails(RequestContext requestContext, String courseId, Map<String, String> headers) {
-    Map<String, Object> ekStepContent = ContentUtil.getContent(courseId, Arrays.asList("status", "batches", "leafNodesCount"));
+    Map<String, Object> ekStepContent = ContentUtil.getContent(courseId, Arrays.asList("status", "batches", "leafNodesCount", "primaryCategory"));
     logger.info(requestContext, "CourseBatchManagementActor:getEkStepContent: courseId: " + courseId, null,
             ekStepContent);
     String status = (String) ((Map<String, Object>)ekStepContent.getOrDefault("content", new HashMap<>())).getOrDefault("status", "");
@@ -652,6 +656,17 @@ private CourseBatch mapESFieldsToObject(CourseBatch courseBatch) {
                                       cert_template.getKey(), mapToObject((Map<String, Object>) cert_template.getValue())));
       courseBatch.setCertTemplates(certificateTemplates);
     }
+    Map<String, Object> batchAttributes = courseBatch.getBatchAttributes();
+    if(MapUtils.isNotEmpty(batchAttributes)) {
+      batchAttributes
+              .entrySet()
+              .stream()
+              .forEach(
+                      batchAttribute ->
+                              batchAttributes.put(
+                                      batchAttribute.getKey(), mapToObject((Map<String, Object>) batchAttribute.getValue())));
+      courseBatch.setBatchAttributes(batchAttributes);
+    }
     return courseBatch;
   }
 

course-mw/enrolment-actor/src/main/scala/org/sunbird/enrolments/ContentConsumptionActor.scala

@@ -422,6 +422,7 @@ class ContentConsumptionActor @Inject() extends BaseEnrolmentActor {
         val userId: String = request.getOrDefault(JsonKey.USER_ID, primaryUserId).asInstanceOf[String]
         val courseId: String = request.getOrDefault(JsonKey.COURSE_ID, "").asInstanceOf[String]
         val batchId: String = request.getOrDefault(JsonKey.BATCH_ID, "").asInstanceOf[String]
+        val rootOrgId: String = request.getOrDefault(JsonKey.ROOT_ORG_ID, "").asInstanceOf[String]
         val filters = Map[String, AnyRef]("userid"-> userId, "courseid"-> courseId, "batchid"-> batchId).asJava
         val result = cassandraOperation
           .getRecords(request.getRequestContext, enrolmentDBInfo.getKeySpace, enrolmentDBInfo.getTableName, filters,
@@ -431,7 +432,7 @@ class ContentConsumptionActor @Inject() extends BaseEnrolmentActor {
           .asInstanceOf[java.util.List[java.util.Map[String, AnyRef]]]
         val response = {
             if (CollectionUtils.isNotEmpty(resp)) {
-                pushEnrolmentSyncEvent(userId, courseId, batchId)
+                pushEnrolmentSyncEvent(userId, courseId, batchId, rootOrgId)
                 successResponse()
             } else {
                 new ProjectCommonException(ResponseCode.invalidRequestData.getErrorCode,
@@ -441,13 +442,13 @@ class ContentConsumptionActor @Inject() extends BaseEnrolmentActor {
         sender().tell(response, self)
     }
 
-    def pushEnrolmentSyncEvent(userId: String, courseId: String, batchId: String) = {
+    def pushEnrolmentSyncEvent(userId: String, courseId: String, batchId: String, rooOrgId: String) = {
         val now = System.currentTimeMillis()
         val event =
-            s"""{"eid":"BE_JOB_REQUEST","ets":$now,"mid":"LP.$now.${UUID.randomUUID()}"
+            s"""{"eid":"AUDIT","ets":$now,"mid":"AUDIT.$now.${UUID.randomUUID()}"
                |,"actor":{"type":"System","id":"Course Batch Updater"},"context":{"pdata":{"ver":"1.0","id":"org.sunbird.platform"}}
                |,"object":{"type":"CourseBatchEnrolment","id":"${batchId}_${userId}"},"edata":{"action":"user-enrolment-sync"
-               |,"iteration":1,"batchId":"$batchId","userId":"$userId","courseId":"$courseId"}}""".stripMargin
+               |,"iteration":1,"batchId":"$batchId","userId":"$userId","courseId":"$courseId","rootOrgId":"$rooOrgId"}}""".stripMargin
               .replaceAll("\n", "")
         if(pushTokafkaEnabled){
             val topic = ProjectUtil.getConfigValue("kafka_enrolment_sync_topic")

course-mw/sunbird-util/sunbird-platform-core/common-util/src/main/java/org/sunbird/common/models/util/JsonKey.java

@@ -1078,5 +1078,10 @@ public final class JsonKey {
   public static final String CURRENT_OFFSET = "currentOffSet";
   public static final String SECURE_SETTINGS = "secureSettings";
   public static final String X_AUTH_USER_ORG_ID = "x-authenticated-user-orgid";
-  private JsonKey() {}
+  public static final String CURRENT_BATCH_SIZE = "currentBatchSize";
+  public static final String PRIMARYCATEGORY = "primaryCategory";
+  public static final String PRIMARY_CATEGORY_BLENDED_PROGRAM = "Blended Program";
+
+
+    private JsonKey() {}
 }

course-mw/sunbird-util/sunbird-platform-core/common-util/src/main/java/org/sunbird/common/responsecode/ResponseCode.java

@@ -832,6 +832,10 @@ public enum ResponseCode {
   cannotUpdateEventSetHavingEnrollments(
           ResponseMessage.Key.CANNOT_UPDATE_EVENT_SET_HAVING_ENROLLMENTS,
           ResponseMessage.Message.CANNOT_UPDATE_EVENT_SET_HAVING_ENROLLMENTS),
+  currentBatchSizeMissing(
+          ResponseMessage.Key.CURRENT_BATCH_SIZE_MISSING,
+          ResponseMessage.Message.CURRENT_BATCH_SIZE_PARAMETER),
+
   accessDeniedToEnrolOrUnenrolCourse(ResponseMessage.Key.USER_DOES_NOT_HAVE_ACCESS,ResponseMessage.Message.USER_DOES_NOT_HAVE_ACCESS),
   OK(200),
   CLIENT_ERROR(400),

course-mw/sunbird-util/sunbird-platform-core/common-util/src/main/java/org/sunbird/common/responsecode/ResponseMessage.java

@@ -460,6 +460,7 @@ interface Message {
     String MISSING_FIXED_BATCH_ID = "Missing Fixed Batch Id.";
     String CANNOT_UPDATE_EVENT_SET_HAVING_ENROLLMENTS = "Cannot update event set having enrollments.";
     String USER_DOES_NOT_HAVE_ACCESS = "User doesn't have access to this Course Id";
+    String CURRENT_BATCH_SIZE_PARAMETER = "currentBatchSize param missing in request. Please check the incoming request";
   }
 
   interface Key {
@@ -848,5 +849,6 @@ interface Key {
     String MISSING_FIXED_BATCH_ID = "MISSING_FIXED_BATCH_ID";
     String CANNOT_UPDATE_EVENT_SET_HAVING_ENROLLMENTS = "CANNOT_UPDATE_EVENT_SET_HAVING_ENROLLMENTS";
     String USER_DOES_NOT_HAVE_ACCESS = "USER_DOES_NOT_HAVE_ACCESS";
+    String CURRENT_BATCH_SIZE_MISSING =  "CURRENT_BATCH_SIZE_MISSING";
   }
 }

course-mw/sunbird-util/sunbird-platform-core/common-util/src/main/resources/cassandratablecolumn.properties

@@ -261,4 +261,5 @@ enrolled_date=enrolledDate
 last_access_time=lastAccessTime
 last_completed_time=lastCompletedTime
 last_updated_time=lastUpdatedTime
-cert_templates=certTemplates
+cert_templates=certTemplates
+batchattributes=batchAttributes

service/app/controllers/LearnerController.java

@@ -88,7 +88,8 @@ public CompletionStage<Result> updateContentState(Http.Request httpRequest) {
     String loggingHeaders =  httpRequest.attrs().getOptional(Attrs.X_LOGGING_HEADERS).orElse(null);
     String requestedBy = httpRequest.attrs().getOptional(Attrs.USER_ID).orElse(null);
     String requestedFor = httpRequest.attrs().getOptional(Attrs.REQUESTED_FOR).orElse(null);
-    String apiDebugLog = "UpdateContentState Request: " + requestData.toString() + " RequestedBy: " + requestedBy + " RequestedFor: " + requestedFor + " ";
+    String rootOrgId = httpRequest.attrs().getOptional(Attrs.X_AUTH_USER_ORG_ID).orElse(null);
+    String apiDebugLog = "UpdateContentState Request: " + requestData.toString() + " RequestedBy: " + requestedBy + " RequestedFor: " + requestedFor + " " + " Root Org Id: " + rootOrgId + " ";
       try {
       Request reqObj = (Request) mapper.RequestMapper.mapRequest(requestData, Request.class);
       RequestValidator.validateUpdateContent(reqObj);
@@ -107,7 +108,9 @@ public CompletionStage<Result> updateContentState(Http.Request httpRequest) {
         innerMap.put(JsonKey.CONTENTS, reqObj.get(JsonKey.CONTENTS));
         innerMap.put(JsonKey.ASSESSMENT_EVENTS, reqObj.getRequest().get(JsonKey.ASSESSMENT_EVENTS));
       }
-      innerMap.put(JsonKey.USER_ID, reqObj.getRequest().get(JsonKey.USER_ID));
+        innerMap.put(JsonKey.USER_ID, reqObj.getRequest().get(JsonKey.USER_ID));
+        if (StringUtils.isNotBlank(rootOrgId))
+          innerMap.put(SunbirdKey.ROOT_ORG_ID, rootOrgId);
       reqObj.setRequest(innerMap);
       CompletionStage<Result> result = actorResponseHandler(contentConsumptionActor, reqObj, timeout, null, httpRequest);
       return result.thenApplyAsync(r -> {

service/app/util/Attrs.java

@@ -19,4 +19,5 @@ public class Attrs {
     public static final TypedKey<String> ACTOR_TYPE = TypedKey.<String>create(JsonKey.ACTOR_TYPE);
     public static final TypedKey<String> X_AUTH_TOKEN = TypedKey.<String>create(JsonKey.X_AUTH_TOKEN);
     public static final TypedKey<String> X_LOGGING_HEADERS = TypedKey.<String>create(JsonKey.X_LOGGING_HEADERS);
+    public static final TypedKey<String> X_AUTH_USER_ORG_ID = TypedKey.<String>create(JsonKey.X_AUTH_USER_ORG_ID);
 }