priority report

Author: kuruk-mm

src/adapters/monitoring-reporter.ts

@@ -14,6 +14,7 @@ export interface HeartbeatData {
   currentStep?: string
   progressPercent?: number
   startedAt?: string
+  isPriority?: boolean
 }
 
 export interface JobCompleteData {
@@ -23,6 +24,7 @@ export interface JobCompleteData {
   completedAt: string
   durationMs: number
   errorMessage?: string
+  isPriority?: boolean
 }
 
 interface MonitoringReporterComponents {

src/adapters/sqs.ts

@@ -6,6 +6,7 @@ import { AppComponents } from '../types'
 
 export interface TaskQueueMessage {
   id: string
+  isPriority: boolean
 }
 
 export interface ITaskQueue<T> {
@@ -52,9 +53,9 @@ export function createMemoryQueueAdapter<T>(
     async stop() {
       q.close()
     },
-    async publish(job) {
+    async publish(job, prioritize?: boolean) {
       const id = 'job-' + (++lastJobId).toString()
-      const message: TaskQueueMessage = { id }
+      const message: TaskQueueMessage = { id, isPriority: !!prioritize }
       q.enqueue({ job, message })
       logger.info(`Publishing job`, { id })
       components.metrics.increment('job_queue_enqueue_total', { queue_name: options.queueName })
@@ -139,7 +140,7 @@ export function createSqsAdapter<T>(
 
       const published = await sqs.send(command)
 
-      const m: TaskQueueMessage = { id: published.MessageId! }
+      const m: TaskQueueMessage = { id: published.MessageId!, isPriority: !!prioritize }
 
       logger.info(`Publishing job ${JSON.stringify(m)}`)
 
@@ -154,7 +155,8 @@ export function createSqsAdapter<T>(
 
           if (response && response.Messages && response.Messages.length > 0) {
             for (const it of response.Messages) {
-              const message: TaskQueueMessage = { id: it.MessageId! }
+              const isPriority = options.priorityQueueUrl ? queueUsed === options.priorityQueueUrl : false
+              const message: TaskQueueMessage = { id: it.MessageId!, isPriority }
               const { end } = components.metrics.startTimer('job_queue_duration_seconds', {
                 queue_name: options.queueUrl
               })

src/service.ts

@@ -52,14 +52,20 @@ export async function main(program: Lifecycle.EntryPointParameters<AppComponents
         const sceneId = job.entity.entityId
         const startedAt = new Date().toISOString()
         const startTime = Date.now()
+        const isPriority = message.isPriority
+
+        if (isPriority) {
+          logger.info('Processing PRIORITY job', { sceneId })
+        }
 
         // Report job start
         components.monitoringReporter.reportHeartbeat({
           status: 'processing',
           currentSceneId: sceneId,
           currentStep: 'Starting',
           progressPercent: 0,
-          startedAt
+          startedAt,
+          isPriority
         })
 
         try {
@@ -87,7 +93,8 @@ export async function main(program: Lifecycle.EntryPointParameters<AppComponents
             status: 'success',
             startedAt,
             completedAt: new Date().toISOString(),
-            durationMs: Date.now() - startTime
+            durationMs: Date.now() - startTime,
+            isPriority
           })
         } catch (error) {
           logger.error(`Error processing job ${job.entity.entityId}`)
@@ -100,7 +107,8 @@ export async function main(program: Lifecycle.EntryPointParameters<AppComponents
             startedAt,
             completedAt: new Date().toISOString(),
             durationMs: Date.now() - startTime,
-            errorMessage: error instanceof Error ? error.message : 'Unknown error'
+            errorMessage: error instanceof Error ? error.message : 'Unknown error',
+            isPriority
           })
         }
       })