tower/apps/worker/src/main.ts

import { PrismaClient } from '@prisma/client';
import { createLogger } from '@tower/logger';
import { validateEnv } from '@tower/config';
import { createMeiliClient, configureIndex } from '@tower/search';
import { createIngestQueue } from './queues/ingest.queue';
import { createIngestWorker } from './queues/ingest.processor';
import { createForwardQueue } from './queues/forward.queue';
import { createForwardWorker } from './queues/forward.processor';
import { createIndexQueue } from './queues/index.queue';
import { createIndexWorker } from './queues/index.processor';
import { createSpamQueue } from './queues/spam.queue';
import { createSpamWorker } from './queues/spam.processor';
import { WhatsAppSessionPool } from './whatsapp/session-pool';
import { matchContentRules, TenantRuleRow } from './whatsapp/match-rules';
import { syncGroups } from './whatsapp/group-sync';
import { handleReaction } from './core/approval';
import { approveMessage } from './core/approve-message';
import { startOtpSenderLoop } from './whatsapp/otp-sender';
import { handleCommand } from './whatsapp/command-handler';
import { detectSpam, checkDuplicateHash } from './spam/spam-detector';

const logger = createLogger('tower-worker');

async function bootstrap() {
  const env = validateEnv();
  const prisma = new PrismaClient();
  await prisma.$connect();
  const meiliClient = createMeiliClient(env.MEILI_URL, env.MEILI_MASTER_KEY);
  await configureIndex(meiliClient).catch((err) =>
    logger.warn({ err }, 'Failed to configure Meilisearch index — search may be degraded'),
  );

  const ingestQueue = createIngestQueue(env.REDIS_URL);
  const forwardQueue = createForwardQueue(env.REDIS_URL);
  const indexQueue = createIndexQueue(env.REDIS_URL);
  const spamQueue = createSpamQueue(env.REDIS_URL);
  const pool = new WhatsAppSessionPool();

  const ingestWorker = createIngestWorker(env.REDIS_URL, prisma, pool, forwardQueue, indexQueue);
  const forwardWorker = createForwardWorker(env.REDIS_URL, pool);
  const indexWorker = createIndexWorker(env.REDIS_URL, meiliClient);
  const spamWorker = createSpamWorker(env.REDIS_URL, prisma);

  ingestWorker.on('completed', (job) => logger.info({ jobId: job.id }, 'Ingest job completed'));
  ingestWorker.on('failed', (job, err) => logger.error({ jobId: job?.id, err }, 'Ingest job failed'));
  forwardWorker.on('completed', (job) => logger.info({ jobId: job.id }, 'Forward job completed'));
  forwardWorker.on('failed', (job, err) => logger.error({ jobId: job?.id, err }, 'Forward job failed'));
  indexWorker.on('completed', (job) => logger.info({ jobId: job.id }, 'Index job completed'));
  indexWorker.on('failed', (job, err) => logger.error({ jobId: job?.id, err }, 'Index job failed'));
  spamWorker.on('failed', (job, err) => logger.warn({ jobId: job?.id, err }, 'Spam job failed'));

  const groupMaps = new Map<string, Map<string, string>>();

  async function startAccount(account: { id: string; sessionPath: string }) {
    groupMaps.set(account.id, new Map());
    try {
      await pool.add(
        account.id,
        account.sessionPath,
        async (msg, accountId) => {
          logger.info({ groupJid: msg.sourceGroupJid, senderJid: msg.senderJid, content: msg.content?.slice(0, 80) }, 'Message received');

          // Command handler intercepts STOP/START/PORTAL from non-bot members
          await handleCommand(msg, accountId, prisma, pool).catch((err) =>
            logger.error({ err }, 'Command handler error'),
          );

          const groupMap = groupMaps.get(accountId);
          if (!groupMap) {
            logger.error({ accountId }, 'No group map for account — message dropped');
            return;
          }
          const sourceGroupId = groupMap.get(msg.sourceGroupJid);
          if (!sourceGroupId) {
            logger.warn({ jid: msg.sourceGroupJid, accountId }, 'Unknown group — skipping message');
            return;
          }

          // Resolve tenant from the group; drop messages from unclaimed groups
          const group = await prisma.group.findUnique({
            where: { id: sourceGroupId },
            select: { tenantId: true },
          });
          if (!group || !group.tenantId) {
            logger.info({ sourceGroupId }, 'Group not yet claimed — message dropped');
            return;
          }

          // Check tenant state
          const tenant = await prisma.tenant.findUnique({
            where: { id: group.tenantId },
            select: { isActive: true, isForwardingPaused: true },
          });
          if (!tenant || !tenant.isActive) {
            logger.info({ tenantId: group.tenantId, sourceGroupId }, 'Tenant is inactive — message dropped');
            return;
          }
          if (tenant.isForwardingPaused) {
            logger.info({ tenantId: group.tenantId, sourceGroupId }, 'Forwarding paused for tenant — message dropped');
            return;
          }

          // Spam pre-filter — runs before rule engine; spam messages are never stored
          const spamResult = detectSpam(msg.content);
          const isDupe = spamResult.isSpam
            ? false
            : await checkDuplicateHash(spamResult.contentHash, group.tenantId, sourceGroupId, 60 * 60 * 1000, prisma);
          if (spamResult.isSpam || isDupe) {
            const reason = isDupe ? 'duplicate_hash' : (spamResult.reason ?? 'emoji_only');
            logger.debug({ senderJid: msg.senderJid, reason }, 'Spam filtered before ingest');
            await spamQueue.add('spam', {
              tenantId: group.tenantId,
              messageId: 'pre-ingest',
              sourceGroupId,
              senderJid: msg.senderJid,
              reason,
              contentHash: spamResult.contentHash,
            }, { attempts: 1 }).catch(() => void 0);
            return;
          }

          // Load active rules for this tenant
          const ruleRows: TenantRuleRow[] = await prisma.tenantRule.findMany({
            where: { tenantId: group.tenantId, isActive: true },
            select: { id: true, matchType: true, matchValue: true, action: true, priority: true },
            orderBy: { priority: 'asc' },
          });

          const { tags, effectiveAction } = matchContentRules(msg.content, ruleRows);
          logger.info({ tags, effectiveAction }, 'Rule match result');

          // No matching rules — drop the message
          if (tags.length === 0) return;

          // SKIP action — silently drop
          if (effectiveAction === 'SKIP') {
            logger.info({ platformMsgId: msg.platformMsgId }, 'Message skipped by rule');
            return;
          }

          // For AUTO_APPROVE, check if the sender is a group admin
          let finalAction = effectiveAction;
          if (effectiveAction === 'AUTO_APPROVE') {
            try {
              const sock = pool.get(accountId);
              if (sock) {
                const metadata = await sock.groupMetadata(msg.sourceGroupJid);
                const isAdmin = metadata.participants?.some(
                  (p: any) => p.id === msg.senderJid && (p.admin === 'admin' || p.admin === 'superadmin'),
                );
                if (isAdmin) {
                  finalAction = 'AUTO_APPROVE';
                } else {
                  logger.info({ senderJid: msg.senderJid }, 'Sender is not a group admin — downgrading AUTO_APPROVE to FLAG');
                  finalAction = 'FLAG';
                }
              }
            } catch (err) {
              logger.warn({ err, senderJid: msg.senderJid }, 'Failed to check group admin status — downgrading to FLAG');
              finalAction = 'FLAG';
            }
          }

          await ingestQueue.add(
            'ingest',
            {
              tenantId: group.tenantId,
              platformMsgId: msg.platformMsgId,
              platform: 'whatsapp',
              accountId,
              sourceGroupId,
              senderJid: msg.senderJid,
              senderName: msg.senderName,
              content: msg.content,
              tags,
              effectiveAction: finalAction ?? undefined,
            },
            { attempts: 3, backoff: { type: 'exponential', delay: 1000 } },
          );

          logger.info({ platformMsgId: msg.platformMsgId, tags, effectiveAction: finalAction }, 'Message enqueued');
        },
        async (reaction) => {
          const result = await handleReaction(reaction, prisma, pool, forwardQueue, indexQueue);
          if (!result) return;

          const { forwardJobs, indexDoc } = result;

          await indexQueue.add('index', indexDoc, {
            attempts: 3,
            backoff: { type: 'exponential', delay: 1000 },
          });

          for (const job of forwardJobs) {
            await forwardQueue.add('forward', job, {
              attempts: 3,
              backoff: { type: 'exponential', delay: 2000 },
            });
          }

          logger.info(
            { messageId: indexDoc.messageId, forwardCount: forwardJobs.length },
            'Message approved — indexed and forwarded',
          );
        },
        async (groups, accountId) => {
          logger.info({ count: Object.keys(groups).length, accountId }, 'Syncing groups');
          const map = await syncGroups(groups, accountId, prisma, pool);
          groupMaps.set(accountId, map);
        },
        async (qr, accountId) => {
          await prisma.account.update({
            where: { id: accountId },
            data: { qrCode: qr, status: 'PAIRING' },
          }).catch((err) => logger.error({ accountId, err }, 'Failed to store QR in DB'));
          logger.info({ accountId }, 'QR code updated');
        },
        async (status, accountId, jid?) => {
          if (status === 'connected') {
            await prisma.account.update({
              where: { id: accountId },
              data: { qrCode: null, status: 'ACTIVE', ...(jid ? { jid } : {}) },
            }).catch((err) => logger.error({ accountId, err }, 'Failed to update account status'));
            logger.info({ accountId, jid }, 'Account connected — QR cleared');
          } else if (status === 'logged_out') {
            await prisma.account.update({
              where: { id: accountId },
              data: { status: 'DISCONNECTED' },
            }).catch((err) => logger.error({ accountId, err }, 'Failed to update account status'));
            logger.info({ accountId }, 'Account logged out — awaiting QR scan');
          } else if (status === 'disconnected') {
            await prisma.account.update({
              where: { id: accountId },
              data: { status: 'DISCONNECTED' },
            }).catch((err) => logger.error({ accountId, err }, 'Failed to update account status'));
            logger.info({ accountId }, 'Account disconnected');
          }
        },
      );
    } catch (err) {
      logger.error({ accountId: account.id, err }, 'Failed to start session — skipping account');
    }
  }

  // Load bot accounts at startup: ACTIVE, DISCONNECTED (need re-auth), PAIRING (mid-pairing)
  const accounts = await prisma.account.findMany({
    where: { isBot: true, status: { in: ['ACTIVE', 'DISCONNECTED', 'PAIRING'] }, platform: 'whatsapp' },
    select: { id: true, sessionPath: true },
  });

  if (accounts.length === 0) {
    logger.warn('No bot accounts found — pair one via /settings/bot');
  }

  for (const account of accounts) {
    await startAccount(account);
  }

  logger.info({ accountCount: accounts.length }, 'Tower worker ready');

  // Poll every 30s for accounts added via the dashboard while worker is running.
  // Every 60s, also re-fetch groups for each active session so newly-added groups
  // are detected without requiring a worker restart.
  setInterval(async () => {
    try {
      const all = await prisma.account.findMany({
        where: { isBot: true, status: { in: ['ACTIVE', 'DISCONNECTED', 'PAIRING'] }, platform: 'whatsapp' },
        select: { id: true, sessionPath: true },
      });
      for (const account of all) {
        if (!pool.get(account.id)) {
          logger.info({ accountId: account.id }, 'New bot account detected — starting session');
          await startAccount(account);
        }
      }
    } catch (err) {
      logger.error({ err }, 'Error polling for new accounts');
    }
  }, 30_000);

  setInterval(async () => {
    try {
      for (const [accountId, sock] of pool.getAll()) {
        try {
          const groups = await sock.groupFetchAllParticipating();
          logger.info({ count: Object.keys(groups).length, accountId }, 'Periodic group re-sync');
          const map = await syncGroups(groups, accountId, prisma, pool);
          groupMaps.set(accountId, map);
        } catch (err) {
          logger.warn({ accountId, err }, 'Periodic group re-sync failed');
        }
      }
    } catch (err) {
      logger.error({ err }, 'Error in periodic group re-sync loop');
    }
  }, 60_000);

  // OTP-sender loop: pick up unsent OtpChallenges and DM them via the bot
  startOtpSenderLoop(prisma, pool, logger);

  const shutdown = async () => {
    logger.info('Shutting down...');
    await pool.closeAll();
    await ingestWorker.close();
    await forwardWorker.close();
    await indexWorker.close();
    await spamWorker.close();
    await ingestQueue.close();
    await forwardQueue.close();
    await indexQueue.close();
    await spamQueue.close();
    await prisma.$disconnect();
    process.exit(0);
  };

  process.on('SIGTERM', shutdown);
  process.on('SIGINT', shutdown);
}

bootstrap().catch((err) => {
  console.error('Worker failed to start', err);
  process.exit(1);
});