feat: implement device registration API with kafka event publishing

core/sunbird-platform-common/src/main/java/org/sunbird/operations/userorg/ActorOperations.java

@@ -146,7 +146,8 @@ public enum ActorOperations {
   ADD_ENCRYPTION_KEY("addEncryptionKey", "ADENCKEY"),
   USER_CURRENT_LOGIN("userCurrentLogin", "USRLOG"),
   DELETE_USER("deleteUser", "USRDLT"),
-  USER_OWNERSHIP_TRANSFER("userOwnershipTransfer", "UOWNTRANS");
+  USER_OWNERSHIP_TRANSFER("userOwnershipTransfer", "UOWNTRANS"),
+  REGISTER_DEVICE("registerDevice", "DEVREG");
 
   private final String value;
   private final String operationCode;

core/sunbird-platform-common/src/main/resources/externalresource.properties

@@ -87,6 +87,9 @@ sunbird_cert_completion_img_url=https://sunbirddev.blob.core.windows.net/orgemai
 sunbird_subdomain_keycloak_base_url=https://merge.dev.sunbirded.org/auth/
 kafka_topics_certificate_instruction=local.issue.certificate.request
 kafka_linger_ms=5
+# Device Profile API (Kafka topic for device registration events)
+kafka_topics_device_events=local.events.device
+device_profile_kafka_enabled=true
 sunbird_cert_service_base_url=
 #{0} instancename , {1} toaccountemail or phone in mask , {2} from account email/phone in mask
 #kafka_assessment_topic=local.telemetry.assess

modules/lern/service/app/util/LernServiceRequestInterceptor.java

@@ -129,6 +129,9 @@ private LernServiceRequestInterceptor() {}
     apiHeaderIgnoreMap.put("/v1/notification/send/sync", var);
     apiHeaderIgnoreMap.put("/v2/notification/send", var);
     apiHeaderIgnoreMap.put("/v1/notification/send", var);
+
+    // From Device Management (public endpoint - mobile devices before login)
+    apiHeaderIgnoreMap.put("/v1/device/register", var);
   }
 
   /**

modules/lern/service/conf/application.conf

@@ -244,6 +244,9 @@ pekko {
       "/ssu_user_create_actor/*" { dispatcher = pekko.actor.most-used-one-dispatcher }
       "/sso_user_create_actor" { router = smallest-mailbox-pool, nr-of-instances = 15, dispatcher = most-used-one-dispatcher }
       "/sso_user_create_actor/*" { dispatcher = pekko.actor.most-used-one-dispatcher }
+      # --- DEVICE MANAGEMENT ACTORS ---
+      "/device_register_actor" { router = smallest-mailbox-pool, nr-of-instances = 4, dispatcher = rr-dispatcher }
+      "/device_register_actor/*" { dispatcher = pekko.actor.rr-dispatcher }
 
       # --- LMS ACTORS ---
       "/page-management-actor" { router = smallest-mailbox-pool, nr-of-instances = 10, dispatcher = page-mgr-actor-dispatcher }

modules/lern/service/conf/routes

@@ -107,6 +107,9 @@ POST    /v1/notification/email					@controllers.notificationservice.EmailService
 POST    /private/user/v1/notification/email	    @controllers.notificationservice.EmailServiceController.sendMail(request: play.mvc.Http.Request)
 POST    /v2/notification					    @controllers.notificationservice.EmailServiceController.sendNotification(request: play.mvc.Http.Request)
 
+# Device Registration API
+POST    /v1/device/register/:deviceId           @controllers.device.DeviceController.registerDevice(deviceId: String, request: play.mvc.Http.Request)
+
 # Organisation management APIs
 POST    /v1/org/create		                    @controllers.organisationmanagement.OrgController.createOrg(request: play.mvc.Http.Request)
 PATCH   /v1/org/update		                    @controllers.organisationmanagement.OrgController.updateOrg(request: play.mvc.Http.Request)

modules/lern/service/test/validators/DeviceRegisterRequestValidatorTest.java

@@ -0,0 +1,72 @@
+package validators;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.sunbird.exception.ProjectCommonException;
+import org.sunbird.message.ResponseCode;
+import org.sunbird.request.Request;
+
+/**
+ * Unit tests for DeviceRegisterRequestValidator.
+ */
+public class DeviceRegisterRequestValidatorTest {
+
+  private DeviceRegisterRequestValidator validator;
+  private Request request;
+
+  @Before
+  public void setUp() {
+    validator = new DeviceRegisterRequestValidator();
+    request = new Request();
+    request.setRequest(new java.util.HashMap<>());
+  }
+
+  @Test
+  public void testValidate_throwsWhenDeviceIdMissing() {
+    // Arrange
+    request.getRequest().put("deviceId", null);
+
+    // Act & Assert
+    try {
+      validator.validate(request);
+      Assert.fail("Expected ProjectCommonException");
+    } catch (ProjectCommonException ex) {
+      Assert.assertEquals(ResponseCode.mandatoryParameterMissing.getErrorCode(), ex.getErrorCode());
+    }
+  }
+
+  @Test
+  public void testValidate_throwsWhenDeviceIdBlank() {
+    // Arrange
+    request.getRequest().put("deviceId", "");
+
+    // Act & Assert
+    try {
+      validator.validate(request);
+      Assert.fail("Expected ProjectCommonException");
+    } catch (ProjectCommonException ex) {
+      Assert.assertEquals(ResponseCode.mandatoryParameterMissing.getErrorCode(), ex.getErrorCode());
+    }
+  }
+
+  @Test
+  public void testValidate_passesWithOnlyDeviceId() {
+    // Arrange
+    request.getRequest().put("deviceId", "test-device-001");
+
+    // Act & Assert
+    validator.validate(request); // Should not throw
+  }
+
+  @Test
+  public void testValidate_passesWithFullRequest() {
+    // Arrange
+    request.getRequest().put("deviceId", "test-device-001");
+    request.getRequest().put("fcmToken", "token123");
+    request.getRequest().put("producer", "test-app");
+
+    // Act & Assert
+    validator.validate(request); // Should not throw
+  }
+}

modules/userorg/controller/app/controllers/device/DeviceController.java

@@ -0,0 +1,59 @@
+package controllers.device;
+
+import controllers.BaseController;
+import java.util.concurrent.CompletionStage;
+import javax.inject.Inject;
+import javax.inject.Named;
+import org.apache.pekko.actor.ActorRef;
+import org.sunbird.request.Request;
+import play.mvc.Http;
+import play.mvc.Result;
+import validators.DeviceRegisterRequestValidator;
+
+/**
+ * REST controller for device registration endpoint.
+ * Handles POST /v1/device/register/:deviceId requests.
+ */
+public class DeviceController extends BaseController {
+
+  @Inject 
+  @Named("device_register_actor") private ActorRef deviceRegisterActorRef;
+
+  /**
+   * Register or update a device profile.
+   * Extracts path param (deviceId) and headers (IP, User-Agent).
+   * Delegates to Pekko actor via ask pattern.
+   *
+   * @param deviceId Unique device identifier (path param)
+   * @param httpRequest HTTP request with body, headers
+   * @return Async result (success or error response)
+   */
+  public CompletionStage<Result> registerDevice(String deviceId, Http.Request httpRequest) {
+    return handleRequest(
+        deviceRegisterActorRef,
+        "registerDevice",
+        httpRequest.body().asJson(),
+        req -> {
+          Request request = (Request) req;
+          // Inject deviceId and headers into request
+          request.getRequest().put("deviceId", deviceId);
+          request.getRequest().put("ip_addr", resolveIp(httpRequest));
+          request.getRequest().put("user_agent", httpRequest.header("User-Agent").orElse(""));
+          // Validate
+          new DeviceRegisterRequestValidator().validate(request);
+          return null;
+        },
+        httpRequest);
+  }
+
+  /**
+   * Resolve client IP from headers.
+   * X-Real-IP (set by load balancer) takes precedence over remote address.
+   *
+   * @param req HTTP request
+   * @return Client IP or empty string
+   */
+  private String resolveIp(Http.Request req) {
+    return req.header("X-Real-IP").orElse(req.header("X-Forwarded-For").orElse(""));
+  }
+}

modules/userorg/controller/app/util/ACTORS.java

@@ -25,6 +25,7 @@
 import org.sunbird.actor.tenantpreference.TenantPreferenceManagementActor;
 import org.sunbird.actor.user.*;
 import org.sunbird.actor.userconsent.UserConsentActor;
+import org.sunbird.actor.device.DeviceRegisterActor;
 import org.sunbird.util.search.SearchTelemetryGenerator;
 
 public enum ACTORS {
@@ -104,7 +105,9 @@ public enum ACTORS {
   BACKGROUND_JOB_MANAGER_ACTOR(BackgroundJobManager.class, "background_job_manager_actor"),
   USER_DELETION_BACKGROUND_JOB_ACTOR(
       UserDeletionBackgroundJobActor.class, "user_deletion_background_job_actor"),
-  USER_OWNERSHIP_TRANSFER_ACTOR(UserOwnershipTransferActor.class,"user_ownership_transfer_actor");
+  USER_OWNERSHIP_TRANSFER_ACTOR(UserOwnershipTransferActor.class,"user_ownership_transfer_actor"),
+  // Device Registration Actor
+  DEVICE_REGISTER_ACTOR(DeviceRegisterActor.class, "device_register_actor");
 
   ACTORS(Class clazz, String name) {
     actorClass = clazz;

modules/userorg/controller/app/validators/DeviceRegisterRequestValidator.java

@@ -0,0 +1,33 @@
+package validators;
+
+import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
+import org.sunbird.exception.ProjectCommonException;
+import org.sunbird.message.ResponseCode;
+import org.sunbird.request.Request;
+
+/**
+ * Validator for device registration requests.
+ */
+public class DeviceRegisterRequestValidator {
+
+  /**
+   * Validates device registration request.
+   * Required: deviceId (non-blank).
+   * Optional: fcmToken, producer, dspec, userDeclaredLocation, etc.
+   *
+   * @param request The request to validate
+   * @throws ProjectCommonException if validation fails
+   */
+  public void validate(Request request) {
+    Map<String, Object> req = request.getRequest();
+
+    String deviceId = (String) req.get("deviceId");
+    if (StringUtils.isBlank(deviceId)) {
+      throw new ProjectCommonException(
+          ResponseCode.mandatoryParameterMissing.getErrorCode(),
+          ResponseCode.mandatoryParameterMissing.getErrorMessage() + " deviceId",
+          ResponseCode.CLIENT_ERROR.getResponseCode());
+    }
+  }
+}

modules/userorg/controller/conf/routes

@@ -114,6 +114,9 @@ PATCH   /v1/org/update/encryptionkey            @controllers.organisationmanagem
 GET     /health                                 @controllers.healthmanager.HealthController.health(request: play.mvc.Http.Request)
 GET     /:service/health                        @controllers.healthmanager.HealthController.serviceHealth(service:String, request: play.mvc.Http.Request)
 
+#Device Registration API
+POST    /v1/device/register/:deviceId           @controllers.device.DeviceController.registerDevice(deviceId: String, request: play.mvc.Http.Request)
+
 #Notes API
 POST    /v1/note/create                         @controllers.notesmanagement.NotesController.createNote(request: play.mvc.Http.Request)
 GET     /v1/note/read/:noteId                   @controllers.notesmanagement.NotesController.getNote(noteId:String, request: play.mvc.Http.Request)

modules/userorg/service/src/main/java/org/sunbird/actor/device/DeviceRegisterActor.java

@@ -0,0 +1,91 @@
+package org.sunbird.actor.device;
+
+import org.sunbird.actor.core.BaseActor;
+import org.sunbird.request.Request;
+import org.sunbird.response.Response;
+import org.sunbird.service.device.DeviceRegisterService;
+import org.sunbird.service.device.impl.DeviceRegisterServiceImpl;
+import org.sunbird.telemetry.dto.TelemetryEnvKey;
+import org.sunbird.util.Util;
+
+/**
+ * Pekko actor for device registration and FCM token management.
+ *
+ * <p>Handles asynchronous device registration requests and delegates to the service layer for
+ * persistence and event publishing. Each device registration updates or creates a device profile
+ * in YugabyteSQL and publishes a notification event to Kafka for downstream processing.
+ *
+ * <p>Supported operations:
+ * <ul>
+ *   <li>{@code registerDevice} - Register or update a device profile with FCM token and metadata.
+ * </ul>
+ *
+ * @see DeviceRegisterService
+ * @see DeviceRegisterServiceImpl
+ */
+public class DeviceRegisterActor extends BaseActor {
+
+  private DeviceRegisterService deviceRegisterService;
+
+  /**
+   * Constructs a DeviceRegisterActor with default DeviceRegisterServiceImpl.
+   * Used in production when the actor is instantiated by the Pekko framework.
+   */
+  public DeviceRegisterActor() {
+    this.deviceRegisterService = new DeviceRegisterServiceImpl();
+  }
+
+  /**
+   * Constructs a DeviceRegisterActor with a provided service for dependency injection.
+   * Used in testing to mock the service layer.
+   *
+   * @param service the DeviceRegisterService implementation to use
+   */
+  DeviceRegisterActor(DeviceRegisterService service) {
+    this.deviceRegisterService = service;
+  }
+
+  /**
+   * Receives and routes incoming device registration requests.
+   *
+   * <p>Initializes request context with telemetry metadata, routes the operation to the
+   * appropriate handler, and sends the response back to the sender.
+   *
+   * @param request the incoming request containing operation type and device data
+   * @throws Throwable if an error occurs during request processing
+   */
+  @Override
+  public void onReceive(Request request) throws Throwable {
+    Util.initializeContext(request, TelemetryEnvKey.USER);
+    String operation = request.getOperation();
+
+    switch (operation) {
+      case "registerDevice":
+        registerDevice(request);
+        break;
+      default:
+        onReceiveUnsupportedOperation();
+    }
+  }
+
+  /**
+   * Registers or updates a device profile with FCM token.
+   *
+   * <p>Delegates to the service layer to:
+   * <ul>
+   *   <li>Validate device metadata (deviceId, ip_addr, user_agent, fcmToken)
+   *   <li>Upsert device profile in YugabyteSQL
+   *   <li>Publish device registration event to Kafka
+   * </ul>
+   *
+   * The response is sent back to the sender asynchronously.
+   *
+   * @param request the device registration request containing deviceId and profile data
+   * @throws Throwable if service layer processing fails
+   */
+  private void registerDevice(Request request) throws Throwable {
+    logger.info(request.getRequestContext(), "DeviceRegisterActor:registerDevice: method called.");
+    Response response = deviceRegisterService.registerDevice(request);
+    sender().tell(response, self());
+  }
+}

modules/userorg/service/src/main/java/org/sunbird/dao/device/DeviceProfileDao.java

@@ -0,0 +1,19 @@
+package org.sunbird.dao.device;
+
+import java.util.Map;
+
+/**
+ * Data Access Object interface for device profiles.
+ * Handles UPSERT operations to YugabyteSQL database.
+ */
+public interface DeviceProfileDao {
+
+  /**
+   * UPSERT device profile to database.
+   * On conflict (device_id exists): updates all fields except first_access (preserved).
+   *
+   * @param profile Map containing device profile fields
+   * @throws Exception if database operation fails
+   */
+  void upsert(Map<String, Object> profile) throws Exception;
+}