fix: comments

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

Author: themarolt

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

@@ -70,17 +70,9 @@ const job: IJobDefinition = {
     const emitter = new IntegrationStreamWorkerEmitter(queueService, ctx.log)
     await emitter.init()
 
-    let triggered = 0
-    for (const webhook of webhooks) {
-      await emitter.triggerWebhookProcessing(webhook.platform, webhook.id)
-      triggered++
-
-      if (triggered % 100 === 0) {
-        ctx.log.info(`Re-triggered ${triggered} webhooks!`)
-      }
-    }
+    await emitter.triggerWebhookProcessingBatch(webhooks.map((w) => w.id))
 
-    ctx.log.info(`Re-triggered ${triggered} stuck pending webhooks in total!`)
+    ctx.log.info(`Re-triggered ${webhooks.length} stuck pending webhooks in total!`)
   },
 }
 

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

@@ -1,4 +1,4 @@
-import { generateUUIDv1 } from '@crowd/common'
+import { generateUUIDv1, partition } from '@crowd/common'
 import { Logger } from '@crowd/logging'
 import { CrowdQueue, IQueue } from '@crowd/queue'
 import {
@@ -59,4 +59,16 @@ export class IntegrationStreamWorkerEmitter extends QueuePriorityService {
       { onboarding: true },
     )
   }
+
+  public async triggerWebhookProcessingBatch(webhookIds: string[]): Promise<void> {
+    for (const batch of partition(webhookIds, 10)) {
+      await this.sendMessages(
+        batch.map((webhookId) => ({
+          payload: new ProcessWebhookStreamQueueMessage(webhookId),
+          groupId: generateUUIDv1(),
+          deduplicationId: webhookId,
+        })),
+      )
+    }
+  }
 }

services/libs/data-access-layer/src/old/apps/data_sink_worker/repo/requestedForErasureMemberIdentities.repo.ts

@@ -74,10 +74,11 @@ export default class RequestedForErasureMemberIdentitiesRepository extends Repos
       }
 
       if (identity.type === MemberIdentityType.EMAIL) {
+        // SQL matches case-insensitively (lower(value) = lower(?)), so mirror that here.
         const row = singleOrDefault(data, (r) => {
           return (
             r.type === identity.type &&
-            r.value === identity.value &&
+            r.value.toLowerCase() === identity.value.toLowerCase() &&
             r.platform === identity.platform
           )
         })
@@ -93,11 +94,12 @@ export default class RequestedForErasureMemberIdentitiesRepository extends Repos
         // (type, value) but on different platforms would both match. The previous
         // singleOrDefault call threw "Array contains more than one matching element!" in that
         // case — a deterministic crash that never self-heals.
+        // SQL matches case-insensitively (lower(value) = lower(?)), so mirror that here.
         const row =
           data.find(
             (r) =>
               r.type === identity.type &&
-              r.value === identity.value &&
+              r.value.toLowerCase() === identity.value.toLowerCase() &&
               r.platform === identity.platform,
           ) ?? null