feat: report profiles and scenes updates to NATS

Author: aleortega

.env.default

@@ -17,6 +17,9 @@ REGISTRY_URL=<URL>
 ASSET_BUNDLE_ADMIN_URL=<URL>
 PROFILE_IMAGES_URL=<URL>
 
+# NATS
+NATS_URL=
+
 # Sync
 DISABLE_PROFILE_SYNC=true
 

package.json

@@ -34,6 +34,7 @@
     "@dcl/memory-queue-component": "^2.0.0",
     "@dcl/pg-component": "^0.1.0",
     "@dcl/platform-crypto-middleware": "^1.1.0",
+    "@dcl/protocol": "https://sdk-team-cdn.decentraland.org/@dcl/protocol/branch//dcl-protocol-1.0.0-22239310034.commit-1571733.tgz",
     "@dcl/schemas": "^24.0.0",
     "@dcl/snapshots-fetcher": "^9.2.2",
     "@dcl/sqs-component": "^2.0.4",
@@ -43,6 +44,7 @@
     "@well-known-components/interfaces": "^1.4.3",
     "@well-known-components/logger": "^3.1.3",
     "@well-known-components/metrics": "^2.1.0",
+    "@well-known-components/nats-component": "^2.0.0",
     "bloom-filters": "^3.0.4",
     "dcl-catalyst-client": "^21.10.3",
     "lru-cache": "9.1.2",

src/components.ts

@@ -37,6 +37,8 @@ import { createPointerChangesHandlerComponent } from './logic/sync/pointer-chang
 import { createSynchronizerComponent } from './logic/sync/synchronizer'
 import { createOwnershipValidatorJob } from './logic/sync/ownership-validator-job'
 import { createCoordinatesComponent } from './logic/coordinates'
+import { createConfigComponent } from '@well-known-components/env-config-provider'
+import { createNatsComponent } from '@well-known-components/nats-component'
 
 // Initialize all the components of the app
 export async function initComponents(): Promise<AppComponents> {
@@ -116,11 +118,14 @@ export async function initComponents(): Promise<AppComponents> {
     config
   })
   const catalyst = await createCatalystAdapter({ logs, fetch, config })
+  const natsLogs = await createLogComponent({ config: createConfigComponent({ LOG_LEVEL: 'WARN' }) })
+  const nats = await createNatsComponent({ config, logs: natsLogs })
   const entityPersister = createEntityPersisterComponent({
     logs,
     db,
     profilesCache,
-    entityDeploymentTracker
+    entityDeploymentTracker,
+    nats
   })
   const profileRetriever = createProfileRetrieverComponent({
     logs,
@@ -204,7 +209,8 @@ export async function initComponents(): Promise<AppComponents> {
     coordinates,
     db,
     logs,
-    config
+    config,
+    nats
   })
   const messageConsumer = createMessagesConsumerComponent({
     logs,
@@ -243,6 +249,7 @@ export async function initComponents(): Promise<AppComponents> {
     snapshotsHandler,
     pointerChangesHandler,
     synchronizer,
-    ownershipValidatorJob
+    ownershipValidatorJob,
+    nats
   }
 }

src/logic/handlers/deployment-handler.ts

@@ -2,16 +2,18 @@ import { Entity } from '@dcl/schemas'
 import { AppComponents, IEventHandlerComponent, EventHandlerName, EventHandlerResult, Registry } from '../../types'
 import { DeploymentToSqs } from '@dcl/schemas/dist/misc/deployments-to-sqs'
 import { Authenticator } from '@dcl/crypto'
+import { SceneUpdateNotification } from '@dcl/protocol/out-js/decentraland/kernel/comms/v3/archipelago.gen'
 
 export const createDeploymentEventHandler = ({
   registry,
   catalyst,
   worlds,
   db,
-  logs
+  logs,
+  nats
 }: Pick<
   AppComponents,
-  'catalyst' | 'worlds' | 'db' | 'logs' | 'registry'
+  'catalyst' | 'worlds' | 'db' | 'logs' | 'registry' | 'nats'
 >): IEventHandlerComponent<DeploymentToSqs> => {
   const HANDLER_NAME = EventHandlerName.DEPLOYMENT
   const logger = logs.getLogger('deployment-handler')
@@ -77,6 +79,25 @@ export const createDeploymentEventHandler = ({
           versions: defaultVersions
         })
 
+        // Publish scene update notification for non-world deployments
+        if (!worlds.isWorldDeployment(event) && entity.pointers?.length > 0) {
+          try {
+            nats.publish(
+              'service.scene_update',
+              SceneUpdateNotification.encode({
+                sceneId: entity.id,
+                parcels: entity.pointers,
+                timestamp: entity.timestamp
+              }).finish()
+            )
+          } catch (err: any) {
+            logger.warn('Failed to publish scene update notification', {
+              sceneId: entity.id,
+              error: err.message
+            })
+          }
+        }
+
         return { ok: true, handlerName: HANDLER_NAME }
       } catch (error: any) {
         logger.error('Failed to process', {

src/logic/message-processor.ts

@@ -30,10 +30,11 @@ export async function createMessageProcessorComponent({
   coordinates,
   db,
   logs,
-  config
+  config,
+  nats
 }: Pick<
   AppComponents,
-  'catalyst' | 'worlds' | 'registry' | 'queuesStatusManager' | 'db' | 'logs' | 'config' | 'coordinates'
+  'catalyst' | 'worlds' | 'registry' | 'queuesStatusManager' | 'db' | 'logs' | 'config' | 'coordinates' | 'nats'
 >): Promise<IMessageProcessorComponent> {
   const MAX_RETRIES: number = (await config.getNumber('MAX_RETRIES')) || 3
   const log = logs.getLogger('message-processor')
@@ -45,7 +46,7 @@ export async function createMessageProcessorComponent({
     | WorldUndeploymentEvent
     | WorldSpawnCoordinateSetEvent
   >[] = [
-    createDeploymentEventHandler({ catalyst, worlds, registry, db, logs }),
+    createDeploymentEventHandler({ catalyst, worlds, registry, db, logs, nats }),
     createTexturesEventHandler({
       db,
       logs,

src/logic/sync/entity-persister.ts

@@ -2,6 +2,7 @@ import PQueue from 'p-queue'
 import { AppComponents, IEntityPersisterComponent } from '../../types'
 import { Sync } from '../../types'
 import { Entity } from '@dcl/schemas'
+import { ProfileUpdateNotification } from '@dcl/protocol/out-js/decentraland/kernel/comms/v3/archipelago.gen'
 
 const DB_PERSISTENCE_CONCURRENCY = 30
 
@@ -20,8 +21,9 @@ export function createEntityPersisterComponent({
   logs,
   db,
   profilesCache,
-  entityDeploymentTracker
-}: Pick<AppComponents, 'logs' | 'db' | 'profilesCache' | 'entityDeploymentTracker'>): IEntityPersisterComponent {
+  entityDeploymentTracker,
+  nats
+}: Pick<AppComponents, 'logs' | 'db' | 'profilesCache' | 'entityDeploymentTracker' | 'nats'>): IEntityPersisterComponent {
   const logger = logs.getLogger('entity-persistent')
 
   let bootstrapComplete = false
@@ -43,6 +45,23 @@ export function createEntityPersisterComponent({
     // mark as processed in bloom filter (permanent tracking)
     entityDeploymentTracker.markAsProcessed(entity.id)
 
+    // publish profile update notification to NATS
+    try {
+      nats.publish(
+        'service.profile_update',
+        ProfileUpdateNotification.encode({
+          address: entity.pointers[0],
+          profileJson: JSON.stringify(entity),
+          timestamp: entity.timestamp
+        }).finish()
+      )
+    } catch (err: any) {
+      logger.warn('Failed to publish profile update notification', {
+        address: entity.pointers[0],
+        error: err.message
+      })
+    }
+
     const dbEntity: Sync.ProfileDbEntity = {
       ...entity,
       pointer: entity.pointers[0],

src/types/system.ts

@@ -30,6 +30,7 @@ import {
 import { metricDeclarations } from '../metrics'
 import { IContentStorageComponent } from '@dcl/catalyst-storage'
 import { ICoordinatesComponent, SpawnCoordinate } from '../logic/coordinates'
+import { INatsComponent } from '@well-known-components/nats-component/dist/types'
 
 export type GlobalContext = {
   components: BaseComponents
@@ -71,6 +72,7 @@ export type AppComponents = BaseComponents & {
   pointerChangesHandler: IProfilesSynchronizerComponent
   failedProfilesRetrier: IFailedProfilesRetrierComponent
   ownershipValidatorJob: IBaseComponent
+  nats: INatsComponent
 }
 
 // components used in tests

test/components.ts

@@ -19,6 +19,7 @@ import { createCoordinatesComponent } from '../src/logic/coordinates'
 import { createQueuesStatusManagerComponent } from '../src/logic/queues-status-manager'
 import { createInMemoryCacheComponent } from '../src/adapters/memory-cache'
 import { createWorldsMockComponent } from './unit/mocks/worlds'
+import { INatsComponent } from '@well-known-components/nats-component/dist/types'
 
 /**
  * Behaves like Jest "describe" function, used to describe a test for a
@@ -64,6 +65,14 @@ async function initComponents(): Promise<TestComponents> {
   const memoryStorage = createInMemoryCacheComponent()
   const queuesStatusManager = createQueuesStatusManagerComponent({ memoryStorage })
 
+  // Create a mock NATS component for tests
+  const nats: INatsComponent = {
+    publish: () => {},
+    subscribe: () => ({ unsubscribe: () => {} }),
+    start: async () => {},
+    stop: async () => {}
+  } as unknown as INatsComponent
+
   // Create message processor for integration tests
   // Uses the real catalyst from originalInitComponents() so that jest.spyOn works
   // consistently for both HTTP handler tests and message processor tests.
@@ -75,7 +84,8 @@ async function initComponents(): Promise<TestComponents> {
     coordinates,
     db,
     logs,
-    config
+    config,
+    nats
   })
 
   const messageConsumer = createMessageConsumerMockComponent()
@@ -88,6 +98,7 @@ async function initComponents(): Promise<TestComponents> {
     coordinates,
     registry,
     metrics,
+    nats,
     localFetch: await createLocalFetchCompoment(config),
     messageConsumer,
     messageProcessor,