feat: broadcast event whenever a conversion is manually queued

decentraland · Jan 8, 2025 · c2b9f95 · c2b9f95
1 parent cfbbd63
commit c2b9f95
Show file tree

Hide file tree

Showing 6 changed files with 30 additions and 15 deletions.
diff --git a/consumer-server/package.json b/consumer-server/package.json
@@ -17,7 +17,7 @@
   "dependencies": {
     "@aws-sdk/client-sns": "^3.699.0",
     "@dcl/cdn-uploader": "^1.4.1-20230208155013.commit-f75a6ee",
-    "@dcl/schemas": "15.3.3",
+    "@dcl/schemas": "https://sdk-team-cdn.decentraland.org/@dcl/schemas/branch/feat/new-AB-event/dcl-schemas-15.4.1-12678910043.commit-731691e.tgz",
     "@sentry/node": "^8.27.0",
     "@well-known-components/env-config-provider": "^1.2.0",
     "@well-known-components/http-server": "^2.1.0",

diff --git a/consumer-server/src/adapters/sns.ts b/consumer-server/src/adapters/sns.ts
@@ -1,5 +1,6 @@
 import { PublishCommand, SNSClient } from '@aws-sdk/client-sns'
 import { AppComponents, PublisherComponent } from '../types'
+import { AssetBundleConversionFinishedEvent, AssetBundleConversionManuallyQueuedEvent } from '@dcl/schemas'
 
 export async function createSnsComponent({ config }: Pick<AppComponents, 'config'>): Promise<PublisherComponent> {
   const snsArn = await config.requireString('AWS_SNS_ARN')
@@ -9,18 +10,20 @@ export async function createSnsComponent({ config }: Pick<AppComponents, 'config
     endpoint: optionalEndpoint ? optionalEndpoint : undefined
   })
 
-  async function publishMessage(event: any, attributes: { type: string; subType: string }): Promise<void> {
+  async function publishMessage(
+    event: AssetBundleConversionFinishedEvent | AssetBundleConversionManuallyQueuedEvent
+  ): Promise<void> {
     const command = new PublishCommand({
       TopicArn: snsArn,
       Message: JSON.stringify(event),
       MessageAttributes: {
         type: {
           DataType: 'String',
-          StringValue: attributes.type
+          StringValue: event.type
         },
         subType: {
           DataType: 'String',
-          StringValue: attributes.subType
+          StringValue: event.subType
         }
       }
     })

diff --git a/consumer-server/src/controllers/handlers/queue-conversion-handle.ts b/consumer-server/src/controllers/handlers/queue-conversion-handle.ts
@@ -1,15 +1,18 @@
+import { Events } from '@dcl/schemas'
 import { HandlerContextWithPath } from '../../types'
 import { DeploymentToSqs } from '@dcl/schemas/dist/misc/deployments-to-sqs'
 import { IHttpServerComponent } from '@well-known-components/interfaces'
 
 export async function queueTaskHandler(
-  context: HandlerContextWithPath<'taskQueue' | 'config', '/queue-task'>
+  context: HandlerContextWithPath<'taskQueue' | 'config' | 'publisher', '/queue-task'>
 ): Promise<IHttpServerComponent.IResponse> {
   const {
-    components: { taskQueue, config },
+    components: { taskQueue, config, publisher },
     request
   } = context
 
+  const platform = await config.requireString('PLATFORM')
+
   if (request.headers.get('Authorization') !== (await config.requireString('TMP_SECRET')))
     return { status: 401, body: 'Unauthorized' }
 
@@ -20,6 +23,18 @@ export async function queueTaskHandler(
   const shouldPrioritize = !!(body as any)?.prioritize
   const message = await taskQueue.publish(body as DeploymentToSqs, shouldPrioritize)
 
+  await publisher.publishMessage({
+    type: Events.Type.ASSET_BUNDLE,
+    subType: Events.SubType.AssetBundle.MANUALLY_QUEUED,
+    key: `${body.entity.entityId}-${platform}`,
+    timestamp: Date.now(),
+    metadata: {
+      platform: platform.toLocaleLowerCase() as 'windows' | 'mac' | 'webgl',
+      entityId: body.entity.entityId,
+      isLods: !!body.lods
+    }
+  })
+
   return {
     status: 201,
     body: message

diff --git a/consumer-server/src/service.ts b/consumer-server/src/service.ts
@@ -67,10 +67,7 @@ export async function main(program: Lifecycle.EntryPointParameters<AppComponents
             }
           }
 
-          await components.publisher.publishMessage(eventToPublish, {
-            type: Events.Type.ASSET_BUNDLE,
-            subType: Events.SubType.AssetBundle.CONVERTED
-          })
+          await components.publisher.publishMessage(eventToPublish)
         } finally {
           components.metrics.decrement('ab_converter_running_conversion')
         }

diff --git a/consumer-server/src/types.ts b/consumer-server/src/types.ts
@@ -12,6 +12,7 @@ import { DeploymentToSqs } from '@dcl/schemas/dist/misc/deployments-to-sqs'
 import { S3 } from 'aws-sdk'
 import { IRunnerComponent } from './adapters/runner'
 import { SentryComponent } from './adapters/sentry'
+import { AssetBundleConversionFinishedEvent, AssetBundleConversionManuallyQueuedEvent } from '@dcl/schemas'
 
 export type GlobalContext = {
   components: BaseComponents
@@ -56,5 +57,5 @@ export type HandlerContextWithPath<
 export type Context<Path extends string = any> = IHttpServerComponent.PathAwareContext<GlobalContext, Path>
 
 export type PublisherComponent = {
-  publishMessage(event: any, attributes: { type: string; subType: string }): Promise<void>
+  publishMessage(event: AssetBundleConversionFinishedEvent | AssetBundleConversionManuallyQueuedEvent): Promise<void>
 }
diff --git a/consumer-server/yarn.lock b/consumer-server/yarn.lock
@@ -751,10 +751,9 @@
     eslint-plugin-prettier "^5.1.3"
     prettier "^3.3.2"
 
-"@dcl/schemas@15.3.3":
-  version "15.3.3"
-  resolved "https://registry.yarnpkg.com/@dcl/schemas/-/schemas-15.3.3.tgz#29b35d4eb1af0085cb09e79ecdc39a0c074f59a9"
-  integrity sha512-AID1yUW7gvaE5abZrW8ATK6CA941KoHn4eFWauVCQINqTSfTgp1kBlCO3fYs+26di1tcJHI8XWvR2fuf3Q084w==
+"@dcl/schemas@https://sdk-team-cdn.decentraland.org/@dcl/schemas/branch/feat/new-AB-event/dcl-schemas-15.4.1-12678910043.commit-731691e.tgz":
+  version "15.4.1-12678910043.commit-731691e"
+  resolved "https://sdk-team-cdn.decentraland.org/@dcl/schemas/branch/feat/new-AB-event/dcl-schemas-15.4.1-12678910043.commit-731691e.tgz#84b0158e186d5aa2396d46d3e71159473b1c2a4a"
   dependencies:
     ajv "^8.11.0"
     ajv-errors "^3.0.0"