fix: small ones

Signed-off-by: Uroš Marolt <uros@marolt.me>

Author: themarolt

services/apps/cron_service/src/jobs/incomingWebhooksCheck.job.ts

@@ -99,7 +99,10 @@ const job: IJobDefinition = {
     const emitter = new IntegrationStreamWorkerEmitter(queueService, ctx.log)
     await emitter.init()
 
-    await emitter.triggerWebhookProcessingBatch(webhooks.map((w) => w.id))
+    await emitter.triggerWebhookProcessingBatch(
+      webhooks.map((w) => w.id),
+      true,
+    )
 
     ctx.log.info(`Re-triggered ${webhooks.length} stuck pending webhooks in total!`)
   },

services/libs/common_services/src/services/emitters/integrationStreamWorker.emitter.ts

@@ -60,14 +60,23 @@ export class IntegrationStreamWorkerEmitter extends QueuePriorityService {
     )
   }
 
-  public async triggerWebhookProcessingBatch(webhookIds: string[]): Promise<void> {
+  // Sends in parallel batches of 10. Uses concurrent sendMessage calls (not sendMessages)
+  // so the priority context (onboarding flag) is respected — sendMessages() has no way to
+  // pass a priority override and always routes to NORMAL.
+  public async triggerWebhookProcessingBatch(
+    webhookIds: string[],
+    onboarding: boolean,
+  ): Promise<void> {
     for (const batch of partition(webhookIds, 10)) {
-      await this.sendMessages(
-        batch.map((webhookId) => ({
-          payload: new ProcessWebhookStreamQueueMessage(webhookId),
-          groupId: generateUUIDv1(),
-          deduplicationId: webhookId,
-        })),
+      await Promise.all(
+        batch.map((webhookId) =>
+          this.sendMessage(
+            generateUUIDv1(),
+            new ProcessWebhookStreamQueueMessage(webhookId),
+            webhookId,
+            { onboarding },
+          ),
+        ),
       )
     }
   }