updated API call from partnerName to orgId

src/main/java/com/igot/cb/consumer/KafkaConsumer.java

@@ -1,8 +1,11 @@
 package com.igot.cb.consumer;
 
+import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.igot.cb.enrollment.entity.ContentPartnerEntity;
+import com.igot.cb.enrollment.service.EnrollmentService;
 import com.igot.cb.producer.Producer;
 import com.igot.cb.util.CbServerProperties;
 import com.igot.cb.util.cache.CacheService;
@@ -17,7 +20,6 @@
 import org.apache.commons.lang.WordUtils;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.http.HttpEntity;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
@@ -27,7 +29,6 @@
 import org.springframework.util.CollectionUtils;
 import org.springframework.core.io.Resource;
 import org.springframework.core.io.ResourceLoader;
-import org.springframework.util.ResourceUtils;
 import org.springframework.web.client.RestTemplate;
 
 import java.time.LocalDate;
@@ -45,9 +46,6 @@ public class KafkaConsumer {
     @Autowired
     private CassandraOperation cassandraOperation;
 
-    @Autowired
-    CacheService cacheService;
-
     @Autowired
     private Producer producer;
 
@@ -57,6 +55,9 @@ public class KafkaConsumer {
     @Autowired
     private CbServerProperties cbServerProperties;
 
+    @Autowired
+    private EnrollmentService enrollmentService;
+
     @KafkaListener(topics = "${spring.kafka.cornell.topic.name}", groupId = "${spring.kafka.consumer.group.id}")
     public void enrollUpdateConsumer(ConsumerRecord<String, String> data) {
         log.info("KafkaConsumer::enrollUpdateConsumer:topic name: {} and recievedData: {}", data.topic(), data.value());
@@ -65,49 +66,70 @@ public void enrollUpdateConsumer(ConsumerRecord<String, String> data) {
             Timestamp timestamp = new Timestamp(System.currentTimeMillis());
             timestamp.setTime(timestamp.getTime() + timeZone.getOffset(timestamp.getTime()));
             Map<String, Object> userCourseEnrollMap = mapper.readValue(data.value(), HashMap.class);
-            if (userCourseEnrollMap.containsKey("userid") && userCourseEnrollMap.get("userid") instanceof String && userCourseEnrollMap.containsKey("courseid") && userCourseEnrollMap.get("courseid") instanceof String) {
+            if (userCourseEnrollMap.containsKey(Constants.USER_ID) && userCourseEnrollMap.get(Constants.USER_ID) instanceof String && userCourseEnrollMap.containsKey(Constants.COURSE_ID) && userCourseEnrollMap.get(Constants.COURSE_ID) instanceof String) {
+                String courseId = null;
+                String contentPartnerName = null;
+                String courseName = null;
+                String coursePosterImage = null;
+                String orgId = userCourseEnrollMap.get("orgId").toString();
+                JsonNode entity=enrollmentService.fetchPartnerInfoUsingApi(orgId);
                 String extCourseId = userCourseEnrollMap.get("courseid").toString();
-                JsonNode result=callExtApi(extCourseId);
-                String courseId=result.path("content").get("contentId").asText();
-                log.info("KafkaConsumer :: enrollUpdateConsumer ::courseId from cios api {}", courseId);
-                String[] parts = ((String) userCourseEnrollMap.get("userid")).split("@");
-                userCourseEnrollMap.put("userid", parts[0]);
+                JsonNode result = fetchCiosContentByCourseIdAndPartnerId(extCourseId,entity.path("result").get("id").asText());
+                JsonNode contentNode = result.path("content");
+                if (!contentNode.isMissingNode()) {
+                    courseId = contentNode.path("contentId").asText(null);
+                    courseName = contentNode.path("name").asText(null);
+                    JsonNode contentPartnerNode = contentNode.path("contentPartner");
+                    if (!contentPartnerNode.isMissingNode()) {
+                        contentPartnerName = contentPartnerNode.path("contentPartnerName").asText(null);
+                        coursePosterImage = contentPartnerNode.path("thumbnailUrl").asText(null);
+                    }
+                }
+                log.info("KafkaConsumer :: enrollUpdateConsumer ::courseId from cios api {} userid {}", courseId, userCourseEnrollMap.get(Constants.USER_ID));
+                String[] parts = ((String) userCourseEnrollMap.get(Constants.USER_ID)).split("@");
+                userCourseEnrollMap.put(Constants.USER_ID, parts[0]);
                 Map<String, Object> propertyMap = new HashMap<>();
-                propertyMap.put("userid", userCourseEnrollMap.get("userid"));
-                propertyMap.put("courseid", courseId);
+                propertyMap.put(Constants.USER_ID, userCourseEnrollMap.get(Constants.USER_ID));
+                propertyMap.put(Constants.COURSE_ID, courseId);
                 List<Map<String, Object>> listOfMasterData = cassandraOperation.getRecordsByPropertiesWithoutFiltering(Constants.KEYSPACE_SUNBIRD_COURSES, Constants.TABLE_USER_EXTERNAL_ENROLMENTS, propertyMap, null, 1);
                 if (!CollectionUtils.isEmpty(listOfMasterData)) {
-                    Map<String, Object> updatedMap = new HashMap<>();
-                    updatedMap.put("progress",
-                            100);
-                    updatedMap.put("status",
-                            2);
-                    if (userCourseEnrollMap.containsKey("completedon") && userCourseEnrollMap.get("completedon") instanceof String) {
-                        updatedMap.put("completedon", convertToTimestamp(
-                                (String) userCourseEnrollMap.get("completedon")));
-                        updatedMap.put("completionpercentage",
-                                100);
-                        updatedMap.put("updatedon",timestamp);
+                    String Status = userCourseEnrollMap.get("status").toString();
+                    log.info("status {}", Status);
+                    if (Status.equalsIgnoreCase("complete")) {
+                        Map<String, Object> updatedMap = new HashMap<>();
+                        updatedMap.put(Constants.PROGRESS, 100);
+                        updatedMap.put(Constants.STATUS, 2);
+                        updatedMap.put(Constants.COMPLETED_ON, convertToTimestamp((String) userCourseEnrollMap.get("completedon")));
+                        updatedMap.put(Constants.COMPLETION_PERCENTAGE, 100);
+                        updatedMap.put(Constants.UPDATED_ON, timestamp);
+                        cassandraOperation.updateRecord(Constants.KEYSPACE_SUNBIRD_COURSES, Constants.TABLE_USER_EXTERNAL_ENROLMENTS, updatedMap, propertyMap);
+                        JsonNode jsonNode = mapper.convertValue(entity.path("result").path("trasformCertificateJson"), new TypeReference<JsonNode>() {
+                        });
+                        Map<String, Object> certificateRequest = new HashMap<>();
+                        certificateRequest.put(Constants.USER_ID, userCourseEnrollMap.get(Constants.USER_ID));
+                        certificateRequest.put(Constants.COURSE_ID, courseId);
+                        certificateRequest.put(Constants.COMPLETION_DATE, userCourseEnrollMap.get("completedon"));
+                        certificateRequest.put(Constants.PROVIDER_NAME,contentPartnerName );
+                        certificateRequest.put(Constants.COURSE_NAME, courseName);
+                        certificateRequest.put(Constants.COURSE_POSTER_IMAGE, coursePosterImage);
+                        certificateRequest.put(Constants.RECIPIENT_NAME, readUserName(userCourseEnrollMap.get(Constants.USER_ID).toString()));
+                        replacePlaceholders(jsonNode, certificateRequest);
+                        producer.push(cbServerProperties.getCertificateTopic(), jsonNode);
+                        log.info("KafkaConsumer::enrollUpdateConsumer:updated");
+                    } else {
+                        Map<String, Object> updatedMap = new HashMap<>();
+                        updatedMap.put(Constants.PROGRESS, userCourseEnrollMap.get("progress_percetage"));
+                        updatedMap.put(Constants.STATUS, 0);
+                        updatedMap.put(Constants.COMPLETION_PERCENTAGE, userCourseEnrollMap.get("progress_percetage"));
+                        updatedMap.put(Constants.UPDATED_ON, timestamp);
+                        cassandraOperation.updateRecord(Constants.KEYSPACE_SUNBIRD_COURSES, Constants.TABLE_USER_EXTERNAL_ENROLMENTS, updatedMap, propertyMap);
                     }
-                    cassandraOperation.updateRecord(Constants.KEYSPACE_SUNBIRD_COURSES, Constants.TABLE_USER_EXTERNAL_ENROLMENTS, updatedMap, propertyMap);
-//                    cacheService.deleteCache(userCourseEnrollMap.get("userid").toString() + courseId);
-//                    cacheService.deleteCache(userCourseEnrollMap.get("userid").toString());
-                    Resource resource = resourceLoader.getResource("classpath:certificateTemplate.json");
-                    InputStream inputStream = resource.getInputStream();
-                    JsonNode jsonNode = mapper.readTree(inputStream);
-                    Map<String, Object> certificateRequest = new HashMap<>();
-                    certificateRequest.put("userid", userCourseEnrollMap.get("userid"));
-                    certificateRequest.put("courseid", courseId);
-                    certificateRequest.put("completiondate", userCourseEnrollMap.get("completedon"));
-                    certificateRequest.put("providerName",result.path("content").path("contentPartner").get("contentPartnerName").asText());
-                    certificateRequest.put("courseName",result.path("content").get("name").asText());
-                    certificateRequest.put("coursePosterImage",result.path("content").path("contentPartner").get("link").asText());
-                    certificateRequest.put("recipientName",readUserName(userCourseEnrollMap.get("userid").toString()));
-                    replacePlaceholders(jsonNode, certificateRequest);
-                    producer.push(cbServerProperties.getCertificateTopic(), jsonNode);
-                    inputStream.close();
-                    log.info("KafkaConsumer::enrollUpdateConsumer:updated");
+                } else {
+                    log.error("Data not present in DB");
+                    //add not enrolled data to file
                 }
+            } else {
+                log.error("Unable to get userid and courseid from kafka consumer");
             }
 
         } catch (Exception e) {
@@ -116,7 +138,7 @@ public void enrollUpdateConsumer(ConsumerRecord<String, String> data) {
     }
 
     private String readUserName(String userid) {
-        List<String> fields = Arrays.asList("firstname","lastname"); // Assuming user_id is the column name in your table
+        List<String> fields = Arrays.asList("firstname", "lastname"); // Assuming user_id is the column name in your table
         Map<String, Object> propertyMap = new HashMap<>();
         propertyMap.put("id", userid);
         List<Map<String, Object>> userEnrollmentList = cassandraOperation.getRecordsByProperties(
@@ -125,18 +147,18 @@ private String readUserName(String userid) {
                 propertyMap,
                 fields
         );
-        String firstname= (String) userEnrollmentList.stream().findFirst().get().get("firstname");
-        String lastname= (String) userEnrollmentList.stream().findFirst().get().get("lastname");
-        String fullname=firstname;
-        if(lastname!=null){
-            fullname= fullname+" "+lastname;
+        String firstname = (String) userEnrollmentList.stream().findFirst().get().get("firstname");
+        String lastname = (String) userEnrollmentList.stream().findFirst().get().get("lastname");
+        String fullname = firstname;
+        if (lastname != null) {
+            fullname = fullname + " " + lastname;
         }
         return fullname;
     }
 
-    private JsonNode callExtApi(String extCourseId) {
+    private JsonNode fetchCiosContentByCourseIdAndPartnerId(String extCourseId,String partnerId) {
         log.info("KafkaConsumer :: callExtApi");
-        String url = cbServerProperties.getBaseUrl() + cbServerProperties.getFixedUrl() + extCourseId;
+        String url = cbServerProperties.getBaseUrl() + cbServerProperties.getFixedUrl() + extCourseId+"/"+partnerId;
         HttpHeaders headers = new HttpHeaders();
         headers.set("Authorization", cbServerProperties.getToken());
         HttpEntity<String> entity = new HttpEntity<>(headers);
@@ -195,7 +217,7 @@ private void replacePlaceholders(JsonNode jsonNode, Map<String, Object> certific
 
     private String getReplacementValue(String placeholder, Map<String, Object> certificateRequest) {
         log.debug("KafkaConsumer :: getReplacementValue");
-        String value=WordUtils.wrap((String) certificateRequest.get("courseName"), cbServerProperties.getCertificateCharLength(), "\n", false);
+        String value = WordUtils.wrap((String) certificateRequest.get("courseName"), cbServerProperties.getCertificateCharLength(), "\n", false);
         switch (placeholder) {
             case "user.id":
                 return (String) certificateRequest.get("userid");
@@ -218,15 +240,14 @@ private String getReplacementValue(String placeholder, Map<String, Object> certi
                 int firstNewLineIndexExtended = value.indexOf("\n");
                 if (firstNewLineIndexExtended != -1) {
                     String textAfterFirstNewLine = value.substring(firstNewLineIndexExtended + 1).trim();
-                    String secondValue=WordUtils.wrap(textAfterFirstNewLine, cbServerProperties.getCertificateCharLength(), "\n", false);
-                    int secondNewLineIndex=secondValue.indexOf("\n");
-                    if(secondNewLineIndex!=-1){
+                    String secondValue = WordUtils.wrap(textAfterFirstNewLine, cbServerProperties.getCertificateCharLength(), "\n", false);
+                    int secondNewLineIndex = secondValue.indexOf("\n");
+                    if (secondNewLineIndex != -1) {
                         return secondValue.substring(0, secondNewLineIndex).trim();
-                    }else{
+                    } else {
                         return secondValue;
                     }
-                }
-                else {
+                } else {
                     return "";
                 }
             case "provider.name":
@@ -246,8 +267,7 @@ private static String convertDateFormat(String originalDate) {
         DateTimeFormatter originalFormatter = DateTimeFormatter.ofPattern("MM/dd/yyyy");
         LocalDate date = LocalDate.parse(originalDate, originalFormatter);
         DateTimeFormatter outputFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
-        String formattedDate = date.format(outputFormatter);
-        return formattedDate;
+        return date.format(outputFormatter);
     }
 
 }

src/main/java/com/igot/cb/enrollment/entity/ContentPartnerEntity.java

@@ -0,0 +1,50 @@
+package com.igot.cb.enrollment.entity;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.vladmihalcea.hibernate.type.json.JsonBinaryType;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import org.hibernate.annotations.Type;
+import org.hibernate.annotations.TypeDef;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.Id;
+import javax.persistence.Table;
+import java.sql.Timestamp;
+
+@Getter
+@Setter
+@NoArgsConstructor
+@AllArgsConstructor
+@Table(name ="content_partner")
+@TypeDef(name = "jsonb", typeClass = JsonBinaryType.class)
+@Entity
+public class ContentPartnerEntity {
+    @Id
+    private String id;
+
+    @Type(type = "jsonb")
+    @Column(columnDefinition = "jsonb")
+    private JsonNode data;
+
+    private Timestamp createdOn;
+
+    private Timestamp updatedOn;
+
+    private Boolean isActive;
+
+    @Type(type = "jsonb")
+    @Column(columnDefinition = "jsonb")
+    private JsonNode trasformContentJson;
+
+    @Type(type = "jsonb")
+    @Column(columnDefinition = "jsonb")
+    private JsonNode transformProgressJson;
+
+    @Type(type = "jsonb")
+    @Column(columnDefinition = "jsonb")
+    private JsonNode trasformCertificateJson;
+}

src/main/java/com/igot/cb/enrollment/repository/ContentPartnerRepository.java

@@ -0,0 +1,19 @@
+package com.igot.cb.enrollment.repository;
+
+import com.igot.cb.enrollment.entity.ContentPartnerEntity;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
+import org.springframework.stereotype.Repository;
+
+import java.util.Optional;
+
+@Repository
+public interface ContentPartnerRepository extends JpaRepository<ContentPartnerEntity,String> {
+//    @Query(value = "SELECT * FROM content_partner WHERE data->>'contentPartnerName' = ?1 AND (data->>'isActive')::boolean = ?2", nativeQuery = true)
+//    Optional<ContentPartnerEntity> getContentDetailsByContentPartnerNameAndIsActive(@Param("contentPartnerName") String name,@Param("isActive") boolean isActive);
+//@Query(value = "SELECT * FROM content_partner WHERE data->>'contentPartnerName' = :contentPartnerName AND (data->>'isActive')::boolean = :isActive", nativeQuery = true)
+//Optional<ContentPartnerEntity> findByContentPartnerNameAndIsActive(@Param("contentPartnerName") String contentPartnerName,@Param("isActive") boolean isActive);
+@Query(value = "SELECT * FROM content_partner WHERE data->>'contentPartnerName' = :contentPartnerName", nativeQuery = true)
+Optional<ContentPartnerEntity> findByContentPartnerName(@Param("contentPartnerName") String contentPartnerName);
+}

src/main/java/com/igot/cb/enrollment/service/EnrollmentService.java

@@ -1,6 +1,7 @@
 package com.igot.cb.enrollment.service;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.igot.cb.enrollment.entity.ContentPartnerEntity;
 import com.igot.cb.util.dto.SBApiResponse;
 
 public interface EnrollmentService {
@@ -10,4 +11,6 @@ public interface EnrollmentService {
   SBApiResponse readByUserId(String token);
 
   SBApiResponse readByUserIdAndCourseId(String courseId,String token);
+
+  JsonNode fetchPartnerInfoUsingApi(String orgId);
 }