feat: validate entities

src/components.ts

@@ -39,6 +39,7 @@ 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 { createEntityValidatorComponent } from './logic/entity-validator'
 
 // Initialize all the components of the app
 export async function initComponents(): Promise<AppComponents> {
@@ -117,6 +118,7 @@ export async function initComponents(): Promise<AppComponents> {
     logs,
     config
   })
+  const entityValidator = createEntityValidatorComponent({ logs })
   const catalyst = await createCatalystAdapter({ logs, fetch, config })
   const entityPersister = createEntityPersisterComponent({
     logs,
@@ -144,7 +146,8 @@ export async function initComponents(): Promise<AppComponents> {
     db,
     profileSanitizer,
     entityPersister,
-    snapshotContentStorage
+    snapshotContentStorage,
+    entityValidator
   })
   const features = await createFeaturesComponent({ config, logs, fetch }, await config.requireString('SERVICE_URL'))
   const refreshableFeatures = await createRefreshableFeaturesComponent({ features, logs })
@@ -156,7 +159,8 @@ export async function initComponents(): Promise<AppComponents> {
     profileSanitizer,
     entityPersister,
     entityDeploymentTracker,
-    refreshableFeatures
+    refreshableFeatures,
+    entityValidator
   })
   const failedProfilesRetrier = createFailedProfilesRetrierComponent({
     logs,
@@ -209,7 +213,8 @@ export async function initComponents(): Promise<AppComponents> {
     coordinates,
     db,
     logs,
-    config
+    config,
+    entityValidator
   })
   const messageConsumer = createMessagesConsumerComponent({
     logs,
@@ -238,6 +243,7 @@ export async function initComponents(): Promise<AppComponents> {
     memoryStorage,
     queuesStatusManager,
     coordinates,
+    entityValidator,
     profileRetriever,
     profileSanitizer,
     entityDeploymentTracker,

src/logic/entity-validator.ts

@@ -0,0 +1,64 @@
+import { Entity, EntityType, Scene, Wearable, Emote, Profile } from '@dcl/schemas'
+import { BaseComponents, IEntityValidatorComponent, EntityValidationResult } from '../types'
+
+const metadataValidators: Record<string, { validate: (data: any) => boolean }> = {
+  [EntityType.SCENE]: Scene,
+  [EntityType.WEARABLE]: Wearable,
+  [EntityType.EMOTE]: Emote,
+  [EntityType.PROFILE]: Profile
+}
+
+/**
+ * Worlds are scenes deployed to world content servers. The worlds adapter
+ * tags them with type 'world' (not part of EntityType enum), but their
+ * metadata follows the Scene schema.
+ */
+function getMetadataValidator(entityType: string): { validate: (data: any) => boolean } | undefined {
+  if (entityType === 'world') {
+    return metadataValidators[EntityType.SCENE]
+  }
+  return metadataValidators[entityType]
+}
+
+export function createEntityValidatorComponent({ logs }: Pick<BaseComponents, 'logs'>): IEntityValidatorComponent {
+  const logger = logs.getLogger('entity-validator')
+
+  function validate(entity: unknown): EntityValidationResult {
+    const data = entity as Record<string, any>
+    const entityId = data?.id || 'unknown'
+
+    // Step 1: Validate the Entity envelope (id, version, type, pointers, timestamp, content)
+    if (!Entity.validate(entity)) {
+      const envelopeErrors = (Entity.validate.errors || []).map((e) => `envelope${e.instancePath}: ${e.message}`)
+      logger.error('Entity failed envelope validation', {
+        entityId,
+        errors: JSON.stringify(envelopeErrors)
+      })
+      return { ok: false, errors: envelopeErrors }
+    }
+
+    // Step 2: Validate metadata against the type-specific schema
+    if (entity.metadata) {
+      const validator = getMetadataValidator(entity.type)
+      if (validator) {
+        if (!validator.validate(entity.metadata)) {
+          const metadataErrors = ((validator.validate as any).errors || []).map(
+            (e: any) => `metadata${e.instancePath}: ${e.message}`
+          )
+          logger.error('Entity failed metadata validation', {
+            entityId: entity.id,
+            entityType: entity.type,
+            errors: JSON.stringify(metadataErrors)
+          })
+          return { ok: false, errors: metadataErrors }
+        }
+      }
+    }
+
+    return { ok: true }
+  }
+
+  return {
+    validate
+  }
+}

src/logic/handlers/deployment-handler.ts

@@ -8,10 +8,11 @@ export const createDeploymentEventHandler = ({
   catalyst,
   worlds,
   db,
-  logs
+  logs,
+  entityValidator
 }: Pick<
   AppComponents,
-  'catalyst' | 'worlds' | 'db' | 'logs' | 'registry'
+  'catalyst' | 'worlds' | 'db' | 'logs' | 'registry' | 'entityValidator'
 >): IEventHandlerComponent<DeploymentToSqs> => {
   const HANDLER_NAME = EventHandlerName.DEPLOYMENT
   const logger = logs.getLogger('deployment-handler')
@@ -48,6 +49,15 @@ export const createDeploymentEventHandler = ({
           }
         }
 
+        const validationResult = entityValidator.validate(entity)
+        if (!validationResult.ok) {
+          return {
+            ok: false,
+            errors: validationResult.errors,
+            handlerName: HANDLER_NAME
+          }
+        }
+
         const defaultBundles: Registry.Bundles = {
           assets: {
             windows: Registry.SimplifiedStatus.PENDING,

src/logic/message-processor.ts

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

src/logic/sync/pointer-changes-handler.ts

@@ -10,7 +10,8 @@ export async function createPointerChangesHandlerComponent({
   profileSanitizer,
   entityPersister,
   entityDeploymentTracker,
-  refreshableFeatures
+  refreshableFeatures,
+  entityValidator
 }: Pick<
   AppComponents,
   | 'config'
@@ -21,6 +22,7 @@ export async function createPointerChangesHandlerComponent({
   | 'entityPersister'
   | 'entityDeploymentTracker'
   | 'refreshableFeatures'
+  | 'entityValidator'
 >): Promise<IProfilesSynchronizerComponent> {
   const logger = logs.getLogger('pointer-changes-handler')
   const CATALYST_LOAD_BALANCER = await config.requireString('CATALYST_LOADBALANCER_HOST')
@@ -91,6 +93,14 @@ export async function createPointerChangesHandlerComponent({
             continue
           }
 
+          const validationResult = entityValidator.validate(sanitizedProfile[0])
+          if (!validationResult.ok) {
+            logger.warn('Skipping invalid profile from pointer changes', {
+              entityId: sanitizedProfile[0].id,
+              errors: JSON.stringify(validationResult.errors)
+            })
+          }
+
           await entityPersister.persistEntity(sanitizedProfile[0])
         } else {
           logger.info('Skipping profile update because it is marked as malicious', {

src/logic/sync/snapshots-handler.ts

@@ -11,10 +11,18 @@ export async function createSnapshotsHandlerComponent({
   db,
   profileSanitizer,
   entityPersister,
-  snapshotContentStorage
+  snapshotContentStorage,
+  entityValidator
 }: Pick<
   AppComponents,
-  'config' | 'logs' | 'fetch' | 'db' | 'profileSanitizer' | 'entityPersister' | 'snapshotContentStorage'
+  | 'config'
+  | 'logs'
+  | 'fetch'
+  | 'db'
+  | 'profileSanitizer'
+  | 'entityPersister'
+  | 'snapshotContentStorage'
+  | 'entityValidator'
 >): Promise<IProfilesSynchronizerComponent> {
   const CATALYST_LOAD_BALANCER = await config.requireString('CATALYST_LOADBALANCER_HOST')
   const logger = logs.getLogger('snapshots-handler')
@@ -119,7 +127,17 @@ export async function createSnapshotsHandlerComponent({
             errorMessage: 'Profile not found in Catalyst response'
           })
         })
-        await Promise.all(profilesFetched.map((p) => entityPersister.persistEntity(p)))
+        const validProfiles = profilesFetched.filter((p) => {
+          const result = entityValidator.validate(p)
+          if (!result.ok) {
+            logger.warn('Skipping invalid profile from snapshot', {
+              entityId: p.id,
+              errors: JSON.stringify(result.errors)
+            })
+          }
+          return result.ok
+        })
+        await Promise.all(validProfiles.map((p) => entityPersister.persistEntity(p)))
       }
       await db.markSnapshotProcessed(snapshot.hash)
       lastProcessedTimestamp = Math.max(lastProcessedTimestamp, snapshot.timeRange.endTimestamp)

src/types/service.ts

@@ -282,3 +282,12 @@ export interface IFailedProfilesRetrierComponent {
 }
 
 export interface ISynchronizerComponent extends IBaseComponent {}
+
+export type EntityValidationResult = {
+  ok: boolean
+  errors?: string[]
+}
+
+export interface IEntityValidatorComponent {
+  validate(entity: unknown): EntityValidationResult
+}

src/types/system.ts

@@ -28,6 +28,7 @@ import {
   IProfilesSynchronizerComponent,
   ISynchronizerComponent,
   IRefreshableFeaturesComponent
+  IEntityValidatorComponent
 } from './service'
 import { metricDeclarations } from '../metrics'
 import { IContentStorageComponent } from '@dcl/catalyst-storage'
@@ -56,6 +57,7 @@ export type BaseComponents = {
   coordinates: ICoordinatesComponent
   features: IFeaturesComponent
   refreshableFeatures: IRefreshableFeaturesComponent
+  entityValidator: IEntityValidatorComponent
 }
 
 // components used in runtime

test/components.ts

@@ -76,7 +76,8 @@ async function initComponents(): Promise<TestComponents> {
     coordinates,
     db,
     logs,
-    config
+    config,
+    entityValidator: components.entityValidator
   })
 
   const messageConsumer = createMessageConsumerMockComponent()