feat: add BullMQ ingest queue and processor

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-27 15:15:08 +05:30
parent 7151e0dd6d
commit a4135fe983
3 changed files with 93 additions and 0 deletions
@@ -0,0 +1,33 @@
+import { Worker } from 'bullmq';
+import { IngestJobData } from '@tower/types';
+
+export async function processIngestJob(job: IngestJobData, prisma: any): Promise<void> {
+  await prisma.message.upsert({
+    where: {
+      platform_platformMsgId: {
+        platform: job.platform,
+        platformMsgId: job.platformMsgId,
+      },
+    },
+    create: {
+      platform: job.platform,
+      platformMsgId: job.platformMsgId,
+      sourceGroupId: job.sourceGroupId,
+      senderJid: job.senderJid,
+      senderName: job.senderName,
+      content: job.content,
+      tags: job.tags,
+      status: 'PENDING',
+    },
+    update: {},
+  });
+}
+
+export function createIngestWorker(redisUrl: string, prisma: any): Worker<IngestJobData> {
+  const connection = { host: 'localhost', port: 6379, maxRetriesPerRequest: null } as any;
+  return new Worker<IngestJobData>(
+    'tower:ingest',
+    async (job) => processIngestJob(job.data, prisma),
+    { connection },
+  );
+}