good forst commit

apps/worker/src/queues/ingest.processor.ts

@@ -1,9 +1,71 @@
-import { Worker } from 'bullmq';
-import { IngestJobData } from '@tower/types';
+import { Worker, Queue } from 'bullmq';
+import { IngestJobData, ForwardJobData, IndexJobData } from '@tower/types';
 import { parseRedisUrl } from './redis-connection';
+import { approveMessage } from '../core/approve-message';
+import { createLogger } from '@tower/logger';
 
-export async function processIngestJob(job: IngestJobData, prisma: any): Promise<void> {
-  await prisma.message.upsert({
+const logger = createLogger('ingest-processor');
+
+export async function processIngestJob(
+  job: IngestJobData,
+  prisma: any,
+  pool?: any,
+  forwardQueue?: Queue<ForwardJobData>,
+  indexQueue?: Queue<IndexJobData>,
+): Promise<void> {
+  // Defensive: drop messages from non-CLAIMED groups
+  const group = await prisma.group.findUnique({
+    where: { id: job.sourceGroupId },
+    select: { claimStatus: true },
+  });
+  if (!group || group.claimStatus !== 'CLAIMED') {
+    return;
+  }
+
+  // Safety net: drop if tenant is inactive or paused
+  const tenant = await prisma.tenant.findUnique({
+    where: { id: job.tenantId },
+    select: { isActive: true, isForwardingPaused: true },
+  });
+  if (!tenant || !tenant.isActive || tenant.isForwardingPaused) {
+    return;
+  }
+
+  // If the sender has opted out of this group, drop the message
+  const phoneHash = `jid:${job.senderJid}`;
+  const optOut = await prisma.memberOptOut.findFirst({
+    where: {
+      tenantId: job.tenantId,
+      groupId: job.sourceGroupId,
+      user: { jid: job.senderJid, phoneHash },
+    },
+    select: { id: true },
+  });
+  if (optOut) {
+    return;
+  }
+
+  // Resolve or create TowerUser for sender
+  const user = await prisma.towerUser.upsert({
+    where: { tenantId_phoneHash: { tenantId: job.tenantId, phoneHash } },
+    update: {},
+    create: {
+      tenantId: job.tenantId,
+      phoneHash,
+      jid: job.senderJid,
+      displayName: job.senderName ?? job.senderJid,
+    },
+  });
+
+  // Determine the initial status based on effectiveAction
+  let initialStatus: string;
+  if (job.effectiveAction === 'REJECT') {
+    initialStatus = 'REJECTED';
+  } else {
+    initialStatus = 'PENDING';
+  }
+
+  const msg = await prisma.message.upsert({
     where: {
       platform_platformMsgId: {
         platform: job.platform,
@@ -11,23 +73,70 @@ export async function processIngestJob(job: IngestJobData, prisma: any): Promise
       },
     },
     create: {
+      tenantId: job.tenantId,
       platform: job.platform,
       platformMsgId: job.platformMsgId,
       sourceGroupId: job.sourceGroupId,
       senderJid: job.senderJid,
       senderName: job.senderName,
+      senderTowerUserId: user.id,
       content: job.content,
       tags: job.tags,
-      status: 'PENDING',
+      status: initialStatus,
     },
     update: {},
   });
+
+  // For REJECT, create an approval record so it's searchable as rejected
+  if (job.effectiveAction === 'REJECT') {
+    await prisma.approval.upsert({
+      where: { messageId: msg.id },
+      update: {},
+      create: {
+        tenantId: job.tenantId,
+        messageId: msg.id,
+        adminId: 'system',
+        decision: 'REJECTED',
+      },
+    });
+    logger.info({ messageId: msg.id }, 'Message rejected by rule');
+    return;
+  }
+
+  // For AUTO_APPROVE, immediately approve via approveMessage helper
+  if (job.effectiveAction === 'AUTO_APPROVE') {
+    const result = await approveMessage(msg.id, job.tenantId, 'system', prisma);
+    if (result) {
+      if (forwardQueue) {
+        for (const fwd of result.forwardJobs) {
+          await forwardQueue.add('forward', fwd, {
+            attempts: 3,
+            backoff: { type: 'exponential', delay: 2000 },
+          });
+        }
+      }
+      if (indexQueue) {
+        await indexQueue.add('index', result.indexDoc, {
+          attempts: 3,
+          backoff: { type: 'exponential', delay: 1000 },
+        });
+      }
+      logger.info({ messageId: msg.id, forwardCount: result.forwardJobs.length }, 'Message auto-approved');
+    }
+    return;
+  }
 }
 
-export function createIngestWorker(redisUrl: string, prisma: any): Worker<IngestJobData> {
+export function createIngestWorker(
+  redisUrl: string,
+  prisma: any,
+  pool?: any,
+  forwardQueue?: Queue<ForwardJobData>,
+  indexQueue?: Queue<IndexJobData>,
+): Worker<IngestJobData> {
   return new Worker<IngestJobData>(
     'tower-ingest',
-    async (job) => processIngestJob(job.data, prisma),
+    async (job) => processIngestJob(job.data, prisma, pool, forwardQueue, indexQueue),
     { connection: parseRedisUrl(redisUrl) },
   );
 }