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

import { Worker, Queue } from 'bullmq';
import Redis from 'ioredis';
import { IngestJobData, ForwardJobData, IndexJobData } from '@tower/types';
import { parseRedisUrl } from './redis-connection';
import { publishToMessageStream } from '../streams/message-stream';
import { matchContentRules, TenantRuleRow } from '../whatsapp/match-rules';
import { createLogger } from '@tower/logger';

const logger = createLogger('ingest-processor');

export async function processIngestJob(
  job: IngestJobData,
  prisma: any,
  pool?: any,
  forwardQueue?: Queue<ForwardJobData>,
  indexQueue?: Queue<IndexJobData>,
  redis?: Redis,
): Promise<void> {
  // Defensive: drop messages from non-CLAIMED groups
  const group = await prisma.group.findUnique({
    where: { id: job.sourceGroupId },
    select: { claimStatus: true, name: 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, organizationId: 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,
        platformMsgId: job.platformMsgId,
      },
    },
    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: initialStatus,
      quotedPlatformMsgId: job.quotedPlatformMsgId,
    },
    update: {},
  });

  // Publish to AI-dev stream immediately — before any routing decision.
  // Admin approval controls forwarding only; it must never block the stream.
  if (redis) {
    publishToMessageStream(redis, {
      messageId: msg.id,
      tenantId: job.tenantId,
      content: job.content,
      senderJid: job.senderJid,
      senderName: job.senderName,
      sourceGroupId: job.sourceGroupId,
      sourceGroupName: group.name,
      organizationId: tenant.organizationId ?? undefined,
      replyToMessageId: job.quotedPlatformMsgId,
      tags: job.tags,
      effectiveAction: job.effectiveAction ?? null,
      timestamp: new Date().toISOString(),
    }).catch((err: unknown) => logger.warn({ err, messageId: msg.id }, 'Failed to publish to message stream'));
  }

  // Global gate: REJECT means this message is not acceptable for this tenant
  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 tenant rule');
    return;
  }

  // Per-route rule evaluation — determines which routes auto-forward vs stay pending
  const syncRoutes = await prisma.syncRoute.findMany({
    where: { sourceGroupId: job.sourceGroupId, isActive: true },
    include: {
      targetGroup: { select: { id: true, platformId: true, accountId: true, isActive: true } },
      assignedRules: {
        include: {
          tenantRule: {
            select: { matchType: true, matchValue: true, action: true, ruleIntent: true, priority: true, isActive: true },
          },
        },
      },
    },
  });

  const autoForwardRoutes = syncRoutes.filter((route: any) => {
    if (!route.targetGroup?.isActive) return false;

    const activeRules: TenantRuleRow[] = route.assignedRules
      .filter((m: any) => m.tenantRule.isActive)
      .map((m: any) => m.tenantRule);

    if (activeRules.length === 0) {
      return route.approvalMode === 'AUTO';
    }

    const negativeRules = activeRules.filter((r) => (r as any).ruleIntent === 'NEGATIVE');
    const positiveRules = activeRules.filter((r) => (r as any).ruleIntent === 'POSITIVE');

    if (negativeRules.length > 0) {
      const { effectiveAction } = matchContentRules(job.content, negativeRules);
      if (effectiveAction) return false;
    }

    if (positiveRules.length > 0) {
      const { effectiveAction } = matchContentRules(job.content, positiveRules);
      if (effectiveAction) return true;
    }

    return route.approvalMode === 'AUTO';
  });

  if (autoForwardRoutes.length > 0 && forwardQueue) {
    for (const route of autoForwardRoutes) {
      await forwardQueue.add('forward', {
        tenantId: job.tenantId,
        messageId: msg.id,
        content: job.content,
        sourceGroupName: group.name,
        senderName: job.senderName,
        toGroupJid: route.targetGroup.platformId,
        fromAccountId: route.targetGroup.accountId ?? '',
      } as ForwardJobData, { attempts: 3, backoff: { type: 'exponential', delay: 2000 } });
    }

    await prisma.message.update({ where: { id: msg.id }, data: { status: 'APPROVED' } });
    await prisma.approval.upsert({
      where: { messageId: msg.id },
      update: {},
      create: { tenantId: job.tenantId, messageId: msg.id, adminId: 'system', decision: 'APPROVED' },
    });

    if (indexQueue) {
      await indexQueue.add('index', {
        tenantId: job.tenantId,
        messageId: msg.id,
        content: job.content,
        senderName: job.senderName ?? null,
        sourceGroupId: job.sourceGroupId,
        sourceGroupName: group.name,
        tags: job.tags,
        platform: job.platform,
        approvedAt: new Date().toISOString(),
      }, { attempts: 3, backoff: { type: 'exponential', delay: 1000 } });
    }

    logger.info({ messageId: msg.id, routeCount: autoForwardRoutes.length }, 'Message auto-forwarded via route rules');
  }
}

export function createIngestWorker(
  redisUrl: string,
  prisma: any,
  pool?: any,
  forwardQueue?: Queue<ForwardJobData>,
  indexQueue?: Queue<IndexJobData>,
  redis?: Redis,
): Worker<IngestJobData> {
  return new Worker<IngestJobData>(
    'tower-ingest',
    async (job) => processIngestJob(job.data, prisma, pool, forwardQueue, indexQueue, redis),
    { connection: parseRedisUrl(redisUrl) },
  );
}