diff --git a/core/sunbird-platform-common/src/main/java/org/sunbird/operations/userorg/ActorOperations.java b/core/sunbird-platform-common/src/main/java/org/sunbird/operations/userorg/ActorOperations.java index cba00581..e90acb62 100644 --- a/core/sunbird-platform-common/src/main/java/org/sunbird/operations/userorg/ActorOperations.java +++ b/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; diff --git a/core/sunbird-platform-common/src/main/resources/externalresource.properties b/core/sunbird-platform-common/src/main/resources/externalresource.properties index ae0f95b0..b762f2da 100644 --- a/core/sunbird-platform-common/src/main/resources/externalresource.properties +++ b/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 diff --git a/modules/lern/service/app/util/LernServiceRequestInterceptor.java b/modules/lern/service/app/util/LernServiceRequestInterceptor.java index 221366b8..d4b74272 100644 --- a/modules/lern/service/app/util/LernServiceRequestInterceptor.java +++ b/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); } /** diff --git a/modules/lern/service/conf/application.conf b/modules/lern/service/conf/application.conf index bab4b64e..d4146a1e 100644 --- a/modules/lern/service/conf/application.conf +++ b/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 } diff --git a/modules/lern/service/conf/routes b/modules/lern/service/conf/routes index 415877cd..8860b0ff 100644 --- a/modules/lern/service/conf/routes +++ b/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) diff --git a/modules/lern/service/test/validators/DeviceRegisterRequestValidatorTest.java b/modules/lern/service/test/validators/DeviceRegisterRequestValidatorTest.java new file mode 100644 index 00000000..b814dce7 --- /dev/null +++ b/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 + } +} diff --git a/modules/userorg/controller/app/controllers/device/DeviceController.java b/modules/userorg/controller/app/controllers/device/DeviceController.java new file mode 100644 index 00000000..e28c6289 --- /dev/null +++ b/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 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("")); + } +} diff --git a/modules/userorg/controller/app/util/ACTORS.java b/modules/userorg/controller/app/util/ACTORS.java index 07cd94a5..526175fe 100644 --- a/modules/userorg/controller/app/util/ACTORS.java +++ b/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; diff --git a/modules/userorg/controller/app/validators/DeviceRegisterRequestValidator.java b/modules/userorg/controller/app/validators/DeviceRegisterRequestValidator.java new file mode 100644 index 00000000..390dc88a --- /dev/null +++ b/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 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()); + } + } +} diff --git a/modules/userorg/controller/conf/routes b/modules/userorg/controller/conf/routes index acf54cc3..a17bd5b5 100644 --- a/modules/userorg/controller/conf/routes +++ b/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) diff --git a/modules/userorg/service/src/main/java/org/sunbird/actor/device/DeviceRegisterActor.java b/modules/userorg/service/src/main/java/org/sunbird/actor/device/DeviceRegisterActor.java new file mode 100644 index 00000000..d682c7a7 --- /dev/null +++ b/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. + * + *

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. + * + *

Supported operations: + *

+ * + * @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. + * + *

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. + * + *

Delegates to the service layer to: + *

+ * + * 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()); + } +} diff --git a/modules/userorg/service/src/main/java/org/sunbird/dao/device/DeviceProfileDao.java b/modules/userorg/service/src/main/java/org/sunbird/dao/device/DeviceProfileDao.java new file mode 100644 index 00000000..d3f4c46b --- /dev/null +++ b/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 profile) throws Exception; +} diff --git a/modules/userorg/service/src/main/java/org/sunbird/dao/device/impl/DeviceProfileDaoImpl.java b/modules/userorg/service/src/main/java/org/sunbird/dao/device/impl/DeviceProfileDaoImpl.java new file mode 100644 index 00000000..de7acab0 --- /dev/null +++ b/modules/userorg/service/src/main/java/org/sunbird/dao/device/impl/DeviceProfileDaoImpl.java @@ -0,0 +1,79 @@ +package org.sunbird.dao.device.impl; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.SQLException; +import java.util.Map; +import org.sunbird.dao.device.DeviceProfileDao; +import org.sunbird.db.PostgreSQLConnectionManager; +import org.sunbird.logging.LoggerUtil; + +/** + * Implementation of DeviceProfileDao. + * Handles UPSERT of device profiles to lern_device_profile table in Sunbird YugabyteSQL. + */ +public class DeviceProfileDaoImpl implements DeviceProfileDao { + + private static final LoggerUtil log = new LoggerUtil(DeviceProfileDaoImpl.class); + + /** + * UPSERT SQL: INSERT with ON CONFLICT (device_id) DO UPDATE SET. + * Preserves first_access on conflict (not in UPDATE clause). + * Uses COALESCE for user_declared_* fields to only overwrite if new value is non-null. + * Converts epoch-ms timestamps to PostgreSQL timestamptz with to_timestamp(?/1000.0). + */ + private static final String UPSERT_SQL = + "INSERT INTO lern_device_profile " + + "(device_id, fcm_token, producer_id, api_last_updated_on, first_access, last_access, " + + " device_spec, uaspec, user_declared_state, user_declared_district, user_declared_on, updated_date) " + + "VALUES (?, ?, ?, to_timestamp(? / 1000.0), to_timestamp(? / 1000.0), to_timestamp(? / 1000.0), " + + " ?::json, ?::json, ?, ?, to_timestamp(? / 1000.0), now()) " + + "ON CONFLICT (device_id) DO UPDATE SET " + + " fcm_token = EXCLUDED.fcm_token, " + + " producer_id = EXCLUDED.producer_id, " + + " api_last_updated_on = EXCLUDED.api_last_updated_on, " + + " last_access = EXCLUDED.last_access, " + + " device_spec = EXCLUDED.device_spec, " + + " uaspec = EXCLUDED.uaspec, " + + " user_declared_state = COALESCE(EXCLUDED.user_declared_state, lern_device_profile.user_declared_state), " + + " user_declared_district= COALESCE(EXCLUDED.user_declared_district, lern_device_profile.user_declared_district), " + + " user_declared_on = COALESCE(EXCLUDED.user_declared_on, lern_device_profile.user_declared_on), " + + " updated_date = now()"; + + @Override + public void upsert(Map profile) throws Exception { + try (Connection conn = PostgreSQLConnectionManager.getInstance().getConnection(); + PreparedStatement stmt = conn.prepareStatement(UPSERT_SQL)) { + + long now = System.currentTimeMillis(); + Long firstAccess = + profile.get("first_access") != null + ? ((Number) profile.get("first_access")).longValue() + : now; + Long lastAccess = now; + Long userDeclaredOn = + profile.get("user_declared_on") != null + ? ((Number) profile.get("user_declared_on")).longValue() + : null; + + stmt.setString(1, (String) profile.get("device_id")); + stmt.setString(2, (String) profile.get("fcm_token")); + stmt.setString(3, (String) profile.get("producer_id")); + stmt.setLong(4, now); // api_last_updated_on + stmt.setLong(5, firstAccess); // first_access + stmt.setLong(6, lastAccess); // last_access + stmt.setString(7, (String) profile.get("device_spec")); // JSON string + stmt.setString(8, (String) profile.get("uaspec")); // JSON string + stmt.setString(9, (String) profile.get("user_declared_state")); + stmt.setString(10, (String) profile.get("user_declared_district")); + stmt.setObject(11, userDeclaredOn); // nullable Long + + stmt.executeUpdate(); + log.info("DeviceProfileDaoImpl: UPSERT successful for deviceId=" + profile.get("device_id")); + + } catch (SQLException ex) { + log.error("DeviceProfileDaoImpl: UPSERT failed for deviceId=" + profile.get("device_id"), ex); + throw ex; + } + } +} diff --git a/modules/userorg/service/src/main/java/org/sunbird/service/device/DeviceRegisterService.java b/modules/userorg/service/src/main/java/org/sunbird/service/device/DeviceRegisterService.java new file mode 100644 index 00000000..f51a3f73 --- /dev/null +++ b/modules/userorg/service/src/main/java/org/sunbird/service/device/DeviceRegisterService.java @@ -0,0 +1,20 @@ +package org.sunbird.service.device; + +import org.sunbird.request.Request; +import org.sunbird.response.Response; + +/** + * Service interface for device registration. + * Handles business logic: profile building, DB write, Kafka publish. + */ +public interface DeviceRegisterService { + + /** + * Register or update a device profile. + * + * @param request The device registration request containing deviceId, fcmToken, dspec, etc. + * @return Response with success message + * @throws Exception if processing fails + */ + Response registerDevice(Request request) throws Exception; +} diff --git a/modules/userorg/service/src/main/java/org/sunbird/service/device/impl/DeviceRegisterServiceImpl.java b/modules/userorg/service/src/main/java/org/sunbird/service/device/impl/DeviceRegisterServiceImpl.java new file mode 100644 index 00000000..18e43b98 --- /dev/null +++ b/modules/userorg/service/src/main/java/org/sunbird/service/device/impl/DeviceRegisterServiceImpl.java @@ -0,0 +1,140 @@ +package org.sunbird.service.device.impl; + +import com.fasterxml.jackson.databind.ObjectMapper; +import java.util.HashMap; +import java.util.Map; +import org.sunbird.common.ProjectUtil; +import org.sunbird.dao.device.DeviceProfileDao; +import org.sunbird.dao.device.impl.DeviceProfileDaoImpl; +import org.sunbird.kafka.KafkaClient; +import org.sunbird.logging.LoggerUtil; +import org.sunbird.request.Request; +import org.sunbird.response.Response; +import org.sunbird.service.device.DeviceRegisterService; + +/** + * Implementation of DeviceRegisterService. + * Orchestrates device profile registration: validation, DB write, Kafka publish. + */ +public class DeviceRegisterServiceImpl implements DeviceRegisterService { + + private static final LoggerUtil log = new LoggerUtil(DeviceRegisterServiceImpl.class); + private static final ObjectMapper MAPPER = new ObjectMapper(); + private final DeviceProfileDao dao; + + public DeviceRegisterServiceImpl() { + this.dao = new DeviceProfileDaoImpl(); + } + + /** + * Package-private constructor for testing with mocked DAO. + */ + DeviceRegisterServiceImpl(DeviceProfileDao dao) { + this.dao = dao; + } + + @Override + public Response registerDevice(Request request) throws Exception { + Map req = request.getRequest(); + + String deviceId = (String) req.get("deviceId"); + String fcmToken = (String) req.get("fcmToken"); + String producerId = (String) req.get("producer"); + String ipAddr = (String) req.get("ip_addr"); + Long firstAccess = toLong(req.get("first_access")); + Object dspec = req.get("dspec"); + Object userDeclaredLocation = req.get("userDeclaredLocation"); + String userAgent = (String) req.get("user_agent"); + Object uaspec = parseUaSpec(userAgent); + + long now = System.currentTimeMillis(); + + // 1. Build DeviceProfile map for DB + Map profile = new HashMap<>(); + profile.put("device_id", deviceId); + profile.put("fcm_token", fcmToken); + profile.put("producer_id", producerId); + profile.put("api_last_updated_on", now); + profile.put("first_access", firstAccess != null ? firstAccess : now); + profile.put("last_access", now); + profile.put( + "device_spec", dspec != null ? MAPPER.writeValueAsString(dspec) : null); + profile.put("uaspec", uaspec != null ? MAPPER.writeValueAsString(uaspec) : null); + + // Geo fields — null for now (Option A: skip geo-resolution) + profile.put("country_code", null); + profile.put("country", null); + profile.put("state_code", null); + profile.put("state", null); + profile.put("city", null); + profile.put("state_custom", null); + profile.put("state_code_custom", null); + profile.put("district_custom", null); + + // user-declared location + if (userDeclaredLocation instanceof Map) { + Map loc = (Map) userDeclaredLocation; + profile.put("user_declared_state", loc.get("state")); + profile.put("user_declared_district", loc.get("district")); + profile.put("user_declared_on", now); + } + + // 2. UPSERT to DB (synchronous, in actor thread) + dao.upsert(profile); + log.info("DeviceRegisterServiceImpl: Device registered - deviceId=" + deviceId); + + // 3. Publish Kafka event (best-effort, non-blocking failure) + publishKafkaEvent(profile, now); + + // 4. Build response + Response response = new Response(); + response.put("message", "Device registered successfully"); + return response; + } + + /** + * Publish device profile event to Kafka. + * Failure is logged but does not fail the API response. + */ + private void publishKafkaEvent(Map profile, long now) { + String kafkaEnabled = ProjectUtil.getConfigValue("device_profile_kafka_enabled"); + if (!"true".equalsIgnoreCase(kafkaEnabled)) { + return; + } + + try { + // Build the flat JSON event (same format as Obsrv) + Map event = new HashMap<>(profile); + + String topic = ProjectUtil.getConfigValue("kafka_topics_device_events"); + String payload = MAPPER.writeValueAsString(event); + KafkaClient.send(payload, topic); + + log.info( + "DeviceRegisterServiceImpl: Kafka event published for deviceId=" + profile.get("device_id")); + } catch (Exception ex) { + // Kafka failure must NOT fail the API response + log.error("DeviceRegisterServiceImpl: Failed to publish Kafka event", ex); + } + } + + private Long toLong(Object val) { + if (val instanceof Number) { + return ((Number) val).longValue(); + } + return null; + } + + /** + * Minimal user agent parsing. + * Stores raw UA string. Can be enhanced with ua-parser library later. + */ + private Map parseUaSpec(String userAgent) { + if (userAgent == null || userAgent.isEmpty()) { + return null; + } + Map ua = new HashMap<>(); + ua.put("raw", userAgent); + return ua; + } +}