feat: add worker queue fabric — classify, dnc, p1, review, digest, thread-resolve, outbox, AI modules

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

apps/worker/src/ai/llm-client.ts

@@ -0,0 +1,33 @@
+const OPENROUTER_BASE = 'https://openrouter.ai/api/v1';
+const DEFAULT_MODEL = 'google/gemini-3.5-flash';
+
+export async function callLLM(
+  prompt: string,
+  apiKey: string,
+  model = DEFAULT_MODEL,
+): Promise<string> {
+  const res = await fetch(`${OPENROUTER_BASE}/chat/completions`, {
+    method: 'POST',
+    headers: {
+      'Content-Type': 'application/json',
+      Authorization: `Bearer ${apiKey}`,
+      'HTTP-Referer': 'https://tower.insignia.app',
+      'X-Title': 'TOWER Community Platform',
+    },
+    body: JSON.stringify({
+      model,
+      messages: [{ role: 'user', content: prompt }],
+      temperature: 0.1,
+    }),
+  });
+
+  if (!res.ok) {
+    const body = await res.text().catch(() => '');
+    throw new Error(`OpenRouter error ${res.status}: ${body}`);
+  }
+
+  const data = (await res.json()) as { choices: Array<{ message: { content: string } }> };
+  const content = data.choices?.[0]?.message?.content;
+  if (!content) throw new Error('OpenRouter returned empty content');
+  return content;
+}

apps/worker/src/digest/digest-generator.d.ts

@@ -0,0 +1,5 @@
+export interface DigestResult {
+    text: string;
+    messageIds: string[];
+}
+export declare function generateDigest(tenantId: string, tenantName: string, prisma: any, digestDate: Date): Promise<DigestResult | null>;

apps/worker/src/digest/digest-generator.js

@@ -0,0 +1,61 @@
+"use strict";
+Object.defineProperty(exports, "__esModule", { value: true });
+exports.generateDigest = generateDigest;
+const MAX_MESSAGES = 20;
+const PREVIEW_LENGTH = 120;
+function formatDate(date) {
+    return date.toLocaleDateString('en-US', {
+        year: 'numeric',
+        month: 'long',
+        day: 'numeric',
+        timeZone: 'UTC',
+    });
+}
+async function generateDigest(tenantId, tenantName, prisma, digestDate) {
+    const from = new Date(digestDate);
+    const to = new Date(digestDate.getTime() + 24 * 60 * 60 * 1000);
+    const messages = await prisma.message.findMany({
+        where: {
+            tenantId,
+            status: { in: ['APPROVED', 'DISTRIBUTED'] },
+            updatedAt: { gte: from, lt: to },
+        },
+        include: {
+            sourceGroup: { select: { name: true } },
+        },
+        orderBy: { updatedAt: 'asc' },
+        take: MAX_MESSAGES,
+    });
+    if (messages.length === 0)
+        return null;
+    const groups = new Map();
+    for (const msg of messages) {
+        const tag = msg.tags?.[0] ?? 'General';
+        if (!groups.has(tag))
+            groups.set(tag, []);
+        groups.get(tag).push(msg);
+    }
+    const dateLabel = formatDate(digestDate);
+    const lines = [
+        `📋 *Daily Digest — ${tenantName}*`,
+        `_${dateLabel}_`,
+        '',
+    ];
+    for (const [tag, msgs] of groups) {
+        const sectionHeader = tag === 'General' ? '*📌 General*' : `*🏷️ ${tag}*`;
+        lines.push(sectionHeader);
+        for (const msg of msgs) {
+            const preview = msg.content.slice(0, PREVIEW_LENGTH).replace(/\n+/g, ' ');
+            const groupName = msg.sourceGroup?.name ?? 'Unknown Group';
+            lines.push(`• ${preview} _(${groupName})_`);
+        }
+        lines.push('');
+    }
+    const groupCount = new Set(messages.map((m) => m.sourceGroupId)).size;
+    lines.push(`_${messages.length} message${messages.length !== 1 ? 's' : ''} from ${groupCount} group${groupCount !== 1 ? 's' : ''} | TOWER_`);
+    return {
+        text: lines.join('\n'),
+        messageIds: messages.map((m) => m.id),
+    };
+}
+//# sourceMappingURL=digest-generator.js.map

apps/worker/src/digest/digest-generator.js.map

@@ -0,0 +1 @@
+{"version":3,"file":"digest-generator.js","sourceRoot":"","sources":["digest-generator.ts"],"names":[],"mappings":";;AAiBA,wCAyDC;AArED,MAAM,YAAY,GAAG,EAAE,CAAC;AACxB,MAAM,cAAc,GAAG,GAAG,CAAC;AAE3B,SAAS,UAAU,CAAC,IAAU;IAC5B,OAAO,IAAI,CAAC,kBAAkB,CAAC,OAAO,EAAE;QACtC,IAAI,EAAE,SAAS;QACf,KAAK,EAAE,MAAM;QACb,GAAG,EAAE,SAAS;QACd,QAAQ,EAAE,KAAK;KAChB,CAAC,CAAC;AACL,CAAC;AAEM,KAAK,UAAU,cAAc,CAClC,QAAgB,EAChB,UAAkB,EAClB,MAAW,EACX,UAAgB;IAEhB,MAAM,IAAI,GAAG,IAAI,IAAI,CAAC,UAAU,CAAC,CAAC;IAClC,MAAM,EAAE,GAAG,IAAI,IAAI,CAAC,UAAU,CAAC,OAAO,EAAE,GAAG,EAAE,GAAG,EAAE,GAAG,EAAE,GAAG,IAAI,CAAC,CAAC;IAEhE,MAAM,QAAQ,GAAG,MAAM,MAAM,CAAC,OAAO,CAAC,QAAQ,CAAC;QAC7C,KAAK,EAAE;YACL,QAAQ;YACR,MAAM,EAAE,EAAE,EAAE,EAAE,CAAC,UAAU,EAAE,aAAa,CAAC,EAAE;YAC3C,SAAS,EAAE,EAAE,GAAG,EAAE,IAAI,EAAE,EAAE,EAAE,EAAE,EAAE;SACjC;QACD,OAAO,EAAE;YACP,WAAW,EAAE,EAAE,MAAM,EAAE,EAAE,IAAI,EAAE,IAAI,EAAE,EAAE;SACxC;QACD,OAAO,EAAE,EAAE,SAAS,EAAE,KAAK,EAAE;QAC7B,IAAI,EAAE,YAAY;KACnB,CAAC,CAAC;IAEH,IAAI,QAAQ,CAAC,MAAM,KAAK,CAAC;QAAE,OAAO,IAAI,CAAC;IAGvC,MAAM,MAAM,GAAG,IAAI,GAAG,EAA2B,CAAC;IAClD,KAAK,MAAM,GAAG,IAAI,QAAQ,EAAE,CAAC;QAC3B,MAAM,GAAG,GAAG,GAAG,CAAC,IAAI,EAAE,CAAC,CAAC,CAAC,IAAI,SAAS,CAAC;QACvC,IAAI,CAAC,MAAM,CAAC,GAAG,CAAC,GAAG,CAAC;YAAE,MAAM,CAAC,GAAG,CAAC,GAAG,EAAE,EAAE,CAAC,CAAC;QAC1C,MAAM,CAAC,GAAG,CAAC,GAAG,CAAE,CAAC,IAAI,CAAC,GAAG,CAAC,CAAC;IAC7B,CAAC;IAED,MAAM,SAAS,GAAG,UAAU,CAAC,UAAU,CAAC,CAAC;IACzC,MAAM,KAAK,GAAa;QACtB,sBAAsB,UAAU,GAAG;QACnC,IAAI,SAAS,GAAG;QAChB,EAAE;KACH,CAAC;IAEF,KAAK,MAAM,CAAC,GAAG,EAAE,IAAI,CAAC,IAAI,MAAM,EAAE,CAAC;QACjC,MAAM,aAAa,GAAG,GAAG,KAAK,SAAS,CAAC,CAAC,CAAC,cAAc,CAAC,CAAC,CAAC,QAAQ,GAAG,GAAG,CAAC;QAC1E,KAAK,CAAC,IAAI,CAAC,aAAa,CAAC,CAAC;QAC1B,KAAK,MAAM,GAAG,IAAI,IAAI,EAAE,CAAC;YACvB,MAAM,OAAO,GAAG,GAAG,CAAC,OAAO,CAAC,KAAK,CAAC,CAAC,EAAE,cAAc,CAAC,CAAC,OAAO,CAAC,MAAM,EAAE,GAAG,CAAC,CAAC;YAC1E,MAAM,SAAS,GAAG,GAAG,CAAC,WAAW,EAAE,IAAI,IAAI,eAAe,CAAC;YAC3D,KAAK,CAAC,IAAI,CAAC,KAAK,OAAO,MAAM,SAAS,IAAI,CAAC,CAAC;QAC9C,CAAC;QACD,KAAK,CAAC,IAAI,CAAC,EAAE,CAAC,CAAC;IACjB,CAAC;IAED,MAAM,UAAU,GAAG,IAAI,GAAG,CAAC,QAAQ,CAAC,GAAG,CAAC,CAAC,CAAM,EAAE,EAAE,CAAC,CAAC,CAAC,aAAa,CAAC,CAAC,CAAC,IAAI,CAAC;IAC3E,KAAK,CAAC,IAAI,CAAC,IAAI,QAAQ,CAAC,MAAM,WAAW,QAAQ,CAAC,MAAM,KAAK,CAAC,CAAC,CAAC,CAAC,GAAG,CAAC,CAAC,CAAC,EAAE,SAAS,UAAU,SAAS,UAAU,KAAK,CAAC,CAAC,CAAC,CAAC,GAAG,CAAC,CAAC,CAAC,EAAE,WAAW,CAAC,CAAC;IAE7I,OAAO;QACL,IAAI,EAAE,KAAK,CAAC,IAAI,CAAC,IAAI,CAAC;QACtB,UAAU,EAAE,QAAQ,CAAC,GAAG,CAAC,CAAC,CAAM,EAAE,EAAE,CAAC,CAAC,CAAC,EAAE,CAAC;KAC3C,CAAC;AACJ,CAAC"}

apps/worker/src/digest/digest-generator.ts

@@ -0,0 +1,75 @@
+export interface DigestResult {
+  text: string;
+  messageIds: string[];
+}
+
+const MAX_MESSAGES = 20;
+const PREVIEW_LENGTH = 120;
+
+function formatDate(date: Date): string {
+  return date.toLocaleDateString('en-US', {
+    year: 'numeric',
+    month: 'long',
+    day: 'numeric',
+    timeZone: 'UTC',
+  });
+}
+
+export async function buildTemplateDigest(
+  tenantId: string,
+  tenantName: string,
+  prisma: any,
+  digestDate: Date,
+): Promise<DigestResult | null> {
+  const from = new Date(digestDate);
+  const to = new Date(digestDate.getTime() + 24 * 60 * 60 * 1000);
+
+  const messages = await prisma.message.findMany({
+    where: {
+      tenantId,
+      status: { in: ['APPROVED', 'DISTRIBUTED'] },
+      updatedAt: { gte: from, lt: to },
+    },
+    include: {
+      sourceGroup: { select: { name: true } },
+    },
+    orderBy: { updatedAt: 'asc' },
+    take: MAX_MESSAGES,
+  });
+
+  if (messages.length === 0) return null;
+
+  // Group messages by first tag; untagged go under "General"
+  const groups = new Map<string, typeof messages>();
+  for (const msg of messages) {
+    const tag = msg.tags?.[0] ?? 'General';
+    if (!groups.has(tag)) groups.set(tag, []);
+    groups.get(tag)!.push(msg);
+  }
+
+  const dateLabel = formatDate(digestDate);
+  const lines: string[] = [
+    `📋 *Daily Digest — ${tenantName}*`,
+    `_${dateLabel}_`,
+    '',
+  ];
+
+  for (const [tag, msgs] of groups) {
+    const sectionHeader = tag === 'General' ? '*📌 General*' : `*🏷️ ${tag}*`;
+    lines.push(sectionHeader);
+    for (const msg of msgs) {
+      const preview = msg.content.slice(0, PREVIEW_LENGTH).replace(/\n+/g, ' ');
+      const groupName = msg.sourceGroup?.name ?? 'Unknown Group';
+      lines.push(`• ${preview} _(${groupName})_`);
+    }
+    lines.push('');
+  }
+
+  const groupCount = new Set(messages.map((m: any) => m.sourceGroupId)).size;
+  lines.push(`_${messages.length} message${messages.length !== 1 ? 's' : ''} from ${groupCount} group${groupCount !== 1 ? 's' : ''} | TOWER_`);
+
+  return {
+    text: lines.join('\n'),
+    messageIds: messages.map((m: any) => m.id),
+  };
+}

apps/worker/src/digest/digest-scheduler.ts

@@ -0,0 +1,70 @@
+import { Queue } from 'bullmq';
+import { DigestJobData } from '@tower/types';
+import { createLogger } from '@tower/logger';
+
+const TICK_INTERVAL_MS = 5 * 60 * 1000; // 5 minutes
+
+function todayMidnightUtc(): Date {
+  const now = new Date();
+  return new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate()));
+}
+
+function isScheduleReached(scheduleHour: number, scheduleMinute: number): boolean {
+  const now = new Date();
+  const currentMinutes = now.getUTCHours() * 60 + now.getUTCMinutes();
+  const scheduledMinutes = scheduleHour * 60 + scheduleMinute;
+  return currentMinutes >= scheduledMinutes;
+}
+
+function hasAlreadySentToday(lastSentAt: Date | null, todayMidnight: Date): boolean {
+  if (!lastSentAt) return false;
+  return lastSentAt >= todayMidnight;
+}
+
+export function startDigestScheduler(
+  prisma: any,
+  digestQueue: Queue<DigestJobData>,
+  logger: ReturnType<typeof createLogger>,
+): void {
+  let running = false;
+
+  const tick = async (): Promise<void> => {
+    if (running) return;
+    running = true;
+    try {
+      const configs = await prisma.digestConfig.findMany({
+        where: { isActive: true },
+      });
+
+      const todayMidnight = todayMidnightUtc();
+
+      for (const config of configs) {
+        try {
+          if (!isScheduleReached(config.scheduleHour, config.scheduleMinute)) continue;
+          if (hasAlreadySentToday(config.lastSentAt, todayMidnight)) continue;
+
+          await digestQueue.add('digest', {
+            tenantId: config.tenantId,
+            digestDate: todayMidnight.toISOString(),
+            triggeredBy: 'schedule',
+          }, {
+            attempts: 3,
+            backoff: { type: 'exponential', delay: 2000 },
+          });
+
+          logger.info({ tenantId: config.tenantId }, 'Digest job enqueued by scheduler');
+        } catch (err) {
+          logger.error({ err, tenantId: config.tenantId }, 'Failed to enqueue digest job for tenant');
+        }
+      }
+    } catch (err) {
+      logger.error({ err }, 'digest-scheduler tick failed');
+    } finally {
+      running = false;
+    }
+  };
+
+  setTimeout(() => void tick(), 10_000);
+  setInterval(() => void tick(), TICK_INTERVAL_MS);
+  logger.info(`digest-scheduler loop scheduled (every ${TICK_INTERVAL_MS / 60_000}m)`);
+}

apps/worker/src/expire/expire-scheduler.ts

@@ -0,0 +1,34 @@
+import { createLogger } from '@tower/logger';
+
+const SIX_HOURS_MS = 6 * 60 * 60 * 1000;
+
+export function startExpireScheduler(prisma: any, logger = createLogger('expire-scheduler')): void {
+  let running = false;
+
+  const tick = async () => {
+    if (running) return;
+    running = true;
+    try {
+      const now = new Date();
+      const result = await prisma.message.deleteMany({
+        where: {
+          expiresAt: { lt: now },
+          status: { in: ['RAW', 'DNC'] },
+        },
+      });
+      if (result.count > 0) {
+        logger.info({ deleted: result.count }, 'Expired RAW/DNC messages purged');
+      }
+    } catch (err) {
+      logger.error({ err }, 'Expire scheduler tick failed');
+    } finally {
+      running = false;
+    }
+  };
+
+  // Run once at startup to clear any messages that expired while the worker was down
+  tick().catch((err: unknown) => logger.error({ err }, 'Expire scheduler initial tick failed'));
+
+  setInterval(tick, SIX_HOURS_MS);
+  logger.info('Expire scheduler started (6h interval)');
+}

apps/worker/src/outbox/outbox-listener.ts

@@ -0,0 +1,100 @@
+import { Client } from 'pg';
+import { Queue } from 'bullmq';
+import { ClassifyJobData, ThreadResolveJobData } from '@tower/types';
+import { createLogger } from '@tower/logger';
+
+const logger = createLogger('outbox-listener');
+const MAX_ATTEMPTS = 5;
+
+async function processEvent(
+  event: { id: string; type: string; payload: unknown },
+  classifyQueue: Queue<ClassifyJobData>,
+  threadResolveQueue: Queue<ThreadResolveJobData>,
+  prisma: any,
+): Promise<void> {
+  if (event.type === 'classify') {
+    await classifyQueue.add('classify', event.payload as ClassifyJobData, {
+      attempts: 3,
+      backoff: { type: 'exponential', delay: 1000 },
+    });
+  } else if (event.type === 'thread_resolve') {
+    await threadResolveQueue.add('thread-resolve', event.payload as ThreadResolveJobData, {
+      attempts: 2,
+    });
+  } else {
+    logger.warn({ type: event.type, eventId: event.id }, 'Unknown outbox event type — skipping');
+  }
+  await prisma.outboxEvent.delete({ where: { id: event.id } });
+}
+
+// One-time sweep at startup: drain events that arrived while the worker was offline
+async function drainPending(
+  prisma: any,
+  classifyQueue: Queue<ClassifyJobData>,
+  threadResolveQueue: Queue<ThreadResolveJobData>,
+): Promise<void> {
+  const events = await prisma.outboxEvent.findMany({
+    where: { attempts: { lt: MAX_ATTEMPTS } },
+    orderBy: { createdAt: 'asc' },
+    take: 500,
+  });
+
+  if (events.length > 0) {
+    logger.info({ count: events.length }, 'Draining pending outbox events from downtime');
+  }
+
+  for (const event of events) {
+    try {
+      await processEvent(event, classifyQueue, threadResolveQueue, prisma);
+      logger.info({ eventId: event.id, type: event.type }, 'Startup drain: processed');
+    } catch (err) {
+      await prisma.outboxEvent.update({
+        where: { id: event.id },
+        data: { attempts: { increment: 1 }, lastError: String(err) },
+      });
+      logger.warn({ err, eventId: event.id }, 'Startup drain: event failed');
+    }
+  }
+}
+
+export async function startOutboxListener(
+  databaseUrl: string,
+  prisma: any,
+  classifyQueue: Queue<ClassifyJobData>,
+  threadResolveQueue: Queue<ThreadResolveJobData>,
+): Promise<Client> {
+  await drainPending(prisma, classifyQueue, threadResolveQueue);
+
+  // Dedicated pg connection for LISTEN — Prisma does not support LISTEN/NOTIFY
+  const pgClient = new Client({ connectionString: databaseUrl });
+  await pgClient.connect();
+  await pgClient.query('LISTEN outbox_event');
+
+  pgClient.on('notification', (msg) => {
+    const eventId = msg.payload;
+    if (!eventId) return;
+
+    // eslint-disable-next-line @typescript-eslint/no-unsafe-call
+    prisma.outboxEvent
+      .findUnique({ where: { id: eventId } })
+      .then(async (event: { id: string; type: string; payload: unknown; attempts: number } | null) => {
+        if (!event || event.attempts >= MAX_ATTEMPTS) return;
+        try {
+          await processEvent(event, classifyQueue, threadResolveQueue, prisma);
+          logger.info({ eventId, type: event.type }, 'Outbox event processed');
+        } catch (err) {
+          await prisma.outboxEvent.update({
+            where: { id: event.id },
+            data: { attempts: { increment: 1 }, lastError: String(err) },
+          });
+          logger.warn({ err, eventId }, 'Outbox event failed — incremented attempts');
+        }
+      })
+      .catch((err: unknown) => logger.error({ err, eventId }, 'Outbox notification handler error'));
+  });
+
+  pgClient.on('error', (err) => logger.error({ err }, 'Outbox pg client error'));
+
+  logger.info('Outbox listener ready');
+  return pgClient;
+}

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

@@ -0,0 +1,180 @@
+import { Worker, Queue } from 'bullmq';
+import { ClassifyJobData, ForwardJobData, IndexJobData, ReviewJobData, P1JobData, DncJobData } from '@tower/types';
+import { parseRedisUrl } from './redis-connection';
+import { matchContentRules, TenantRuleRow } from '../whatsapp/match-rules';
+import { approveMessage } from '../core/approve-message';
+import { createLogger } from '@tower/logger';
+
+const logger = createLogger('classify-processor');
+
+export async function processClassifyJob(
+  job: ClassifyJobData,
+  prisma: any,
+  pool: any,
+  forwardQueue: Queue<ForwardJobData>,
+  indexQueue: Queue<IndexJobData>,
+  reviewQueue: Queue<ReviewJobData>,
+  p1Queue: Queue<P1JobData>,
+  dncQueue: Queue<DncJobData>,
+): Promise<void> {
+  const message = await prisma.message.findUnique({
+    where: { id: job.messageId },
+    select: { id: true, content: true, tags: true, status: true, tenantId: true, sourceGroupId: true, senderJid: true, platform: true, senderName: true, policyVersion: true },
+  });
+
+  if (!message) {
+    logger.warn({ messageId: job.messageId }, 'Message not found — skipping classify');
+    return;
+  }
+
+  // Idempotent guard — only process RAW messages
+  if (message.status !== 'RAW') {
+    logger.debug({ messageId: job.messageId, status: message.status }, 'Message already classified — skipping');
+    return;
+  }
+
+  // ── Step 1: Hashtag/prefix rules — org rules first (authoritative), then tenant ──
+  const tenant = await prisma.tenant.findUnique({
+    where: { id: job.tenantId },
+    select: { organizationId: true },
+  });
+
+  const [orgRuleRows, tenantRuleRows] = await Promise.all([
+    tenant?.organizationId
+      ? (prisma.orgRule.findMany({
+          where: { organizationId: tenant.organizationId, isActive: true },
+          select: { id: true, matchType: true, matchValue: true, action: true, priority: true },
+          orderBy: { priority: 'asc' },
+        }) as Promise<TenantRuleRow[]>)
+      : Promise.resolve([] as TenantRuleRow[]),
+    prisma.tenantRule.findMany({
+      where: { tenantId: job.tenantId, isActive: true },
+      select: { id: true, matchType: true, matchValue: true, action: true, priority: true },
+      orderBy: { priority: 'asc' },
+    }) as Promise<TenantRuleRow[]>,
+  ]);
+
+  const orgResult = matchContentRules(message.content, orgRuleRows);
+  const { tags, effectiveAction } = orgResult.effectiveAction
+    ? orgResult
+    : matchContentRules(message.content, tenantRuleRows);
+
+  if (effectiveAction === 'SKIP' || effectiveAction === 'REJECT') {
+    await prisma.message.update({ where: { id: job.messageId }, data: { status: 'DNC', tags } });
+    await dncQueue.add('dnc', {
+      tenantId: job.tenantId,
+      sourceGroupId: job.sourceGroupId,
+      accountId: job.accountId,
+      reason: 'rule_skip',
+      timestamp: new Date().toISOString(),
+      policyVersion: 'v1',
+    }, { attempts: 1 }).catch((err: unknown) => logger.warn({ err }, 'Failed to enqueue DNC job'));
+    logger.info({ messageId: job.messageId, effectiveAction }, 'Message DNC by rule');
+    return;
+  }
+
+  if (effectiveAction === 'P1') {
+    await prisma.message.update({ where: { id: job.messageId }, data: { status: 'PENDING', expiresAt: null, tags } });
+    await p1Queue.add('p1', {
+      tenantId: job.tenantId,
+      messageId: job.messageId,
+      content: message.content,
+      sourceGroupId: job.sourceGroupId,
+      senderJid: job.senderJid,
+      senderName: message.senderName,
+      tags,
+      policyVersion: 'v1',
+    }, { attempts: 3, backoff: { type: 'exponential', delay: 1000 } }).catch((err: unknown) =>
+      logger.error({ err, messageId: job.messageId }, 'Failed to enqueue P1 job'),
+    );
+    logger.warn({ messageId: job.messageId, tags }, 'P1 urgent message — admin review required');
+    return;
+  }
+
+  if (effectiveAction === 'FLAG') {
+    await prisma.message.update({ where: { id: job.messageId }, data: { status: 'PENDING', expiresAt: null, tags } });
+    await reviewQueue.add('review', {
+      tenantId: job.tenantId,
+      messageId: job.messageId,
+      sourceGroupId: job.sourceGroupId,
+      tags,
+      reason: 'flagged_by_rule',
+      policyVersion: 'v1',
+    }, { attempts: 1 }).catch((err: unknown) =>
+      logger.warn({ err, messageId: job.messageId }, 'Failed to enqueue review job'),
+    );
+    logger.info({ messageId: job.messageId, tags }, 'Message flagged by rule → PENDING');
+    return;
+  }
+
+  if (effectiveAction === 'AUTO_APPROVE') {
+    // Downgrade to FLAG if sender is not a group admin
+    let isAdmin = false;
+    try {
+      const meta = await pool.getGroupMeta?.(job.sourceGroupJid, job.accountId) ??
+        (pool.get?.(job.accountId) ? await pool.get(job.accountId).groupMetadata(job.sourceGroupJid).catch(() => null) : null);
+      isAdmin = (meta?.participants ?? []).some(
+        (p: any) => p.jid === job.senderJid && (p.admin === 'admin' || p.admin === 'superadmin'),
+      );
+    } catch {
+      isAdmin = false;
+    }
+
+    await prisma.message.update({ where: { id: job.messageId }, data: { status: 'PENDING', expiresAt: null, tags } });
+
+    if (isAdmin) {
+      const result = await approveMessage(job.messageId, job.tenantId, 'system', prisma);
+      if (result) {
+        for (const fwd of result.forwardJobs) {
+          await forwardQueue.add('forward', fwd, { attempts: 3, backoff: { type: 'exponential', delay: 2000 } });
+        }
+        await indexQueue.add('index', result.indexDoc, { attempts: 3, backoff: { type: 'exponential', delay: 1000 } });
+        logger.info({ messageId: job.messageId, forwardCount: result.forwardJobs.length }, 'Message auto-approved');
+      }
+    } else {
+      await reviewQueue.add('review', {
+        tenantId: job.tenantId,
+        messageId: job.messageId,
+        sourceGroupId: job.sourceGroupId,
+        tags,
+        reason: 'auto_approve_downgraded',
+        policyVersion: 'v1',
+      }, { attempts: 1 }).catch((err: unknown) => logger.warn({ err }, 'Failed to enqueue review job'));
+      logger.info({ messageId: job.messageId }, 'AUTO_APPROVE downgraded — sender not admin');
+    }
+    return;
+  }
+
+  // ── Step 2: No rule matched → DNC (expires naturally at 72h) ────────────
+  await prisma.message.update({ where: { id: job.messageId }, data: { status: 'DNC' } });
+  await dncQueue.add('dnc', {
+    tenantId: job.tenantId,
+    sourceGroupId: job.sourceGroupId,
+    accountId: job.accountId,
+    reason: 'no_rule_match',
+    timestamp: new Date().toISOString(),
+    policyVersion: 'v1',
+  }, { attempts: 1 }).catch((err: unknown) => logger.warn({ err }, 'Failed to enqueue DNC job'));
+  logger.debug({ messageId: job.messageId }, 'No rule matched — DNC');
+}
+
+export function createClassifyWorker(
+  redisUrl: string,
+  prisma: any,
+  pool: any,
+  forwardQueue: Queue<ForwardJobData>,
+  indexQueue: Queue<IndexJobData>,
+  reviewQueue: Queue<ReviewJobData>,
+  p1Queue: Queue<P1JobData>,
+  dncQueue: Queue<DncJobData>,
+): Worker<ClassifyJobData> {
+  return new Worker<ClassifyJobData>(
+    'tower.classify.v1',
+    async (job) =>
+      processClassifyJob(
+        job.data, prisma, pool,
+        forwardQueue, indexQueue, reviewQueue, p1Queue, dncQueue,
+      ),
+    { connection: parseRedisUrl(redisUrl), concurrency: 5 },
+  );
+}

apps/worker/src/queues/classify.queue.ts

@@ -0,0 +1,7 @@
+import { Queue } from 'bullmq';
+import { ClassifyJobData } from '@tower/types';
+import { parseRedisUrl } from './redis-connection';
+
+export function createClassifyQueue(redisUrl: string): Queue<ClassifyJobData> {
+  return new Queue<ClassifyJobData>('tower.classify.v1', { connection: parseRedisUrl(redisUrl) });
+}

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

@@ -0,0 +1,120 @@
+import { Worker } from 'bullmq';
+import { DigestJobData } from '@tower/types';
+import { parseRedisUrl } from './redis-connection';
+import { callLLM } from '../ai/llm-client';
+import { buildTemplateDigest } from '../digest/digest-generator';
+import { createLogger } from '@tower/logger';
+
+const logger = createLogger('digest-processor');
+
+const WINDOW_MS = 24 * 60 * 60 * 1000;
+
+function buildLLMPrompt(messages: Array<{ content: string; tags: string[]; status: string }>, tenantName: string, dateLabel: string): string {
+  const lines = messages.map((m, i) => {
+    const label = m.status === 'APPROVED' || m.status === 'DISTRIBUTED' ? '[curated]' : '[community]';
+    return `${i + 1}. ${label} ${m.content.slice(0, 200)}`;
+  });
+  return `You are a community digest writer for "${tenantName}", a South Asian diaspora community.
+Write a friendly, concise WhatsApp digest summarizing these ${messages.length} community messages from ${dateLabel}.
+Use WhatsApp markdown (*bold*, _italic_, bullet points with •).
+Keep it under 800 characters. Group related items. Start with a greeting line.
+
+Messages:
+${lines.join('\n')}
+
+Digest:`;
+}
+
+export async function processDigestJob(
+  job: DigestJobData,
+  prisma: any,
+  pool: any,
+  openRouterApiKey: string | undefined,
+): Promise<void> {
+  const config = await prisma.digestConfig.findUnique({
+    where: { tenantId: job.tenantId },
+    include: { tenant: { select: { name: true } } },
+  });
+
+  if (!config || !config.isActive) {
+    logger.info({ tenantId: job.tenantId }, 'Digest skipped — no active config');
+    return;
+  }
+
+  const digestDate = new Date(job.digestDate);
+  const windowStart = digestDate;
+  const windowEnd = new Date(digestDate.getTime() + WINDOW_MS);
+
+  // Fetch APPROVED/DISTRIBUTED messages + AI-tagged RAW messages
+  const messages = await prisma.message.findMany({
+    where: {
+      tenantId: job.tenantId,
+      OR: [
+        { status: { in: ['APPROVED', 'DISTRIBUTED'] }, updatedAt: { gte: windowStart, lt: windowEnd } },
+        { status: 'RAW', createdAt: { gte: windowStart, lt: windowEnd } },
+      ],
+    },
+    include: { sourceGroup: { select: { name: true } } },
+    orderBy: { createdAt: 'asc' },
+    take: 30,
+  });
+
+  if (messages.length === 0) {
+    await prisma.digestConfig.update({ where: { id: config.id }, data: { lastSentAt: new Date() } });
+    logger.info({ tenantId: job.tenantId }, 'Digest skipped — no messages in window');
+    return;
+  }
+
+  let text: string;
+
+  if (openRouterApiKey) {
+    try {
+      const dateLabel = digestDate.toLocaleDateString('en-US', { year: 'numeric', month: 'long', day: 'numeric', timeZone: 'UTC' });
+      const prompt = buildLLMPrompt(messages, config.tenant.name, dateLabel);
+      text = await callLLM(prompt, openRouterApiKey);
+    } catch (err) {
+      logger.warn({ err, tenantId: job.tenantId }, 'LLM call failed — falling back to template');
+      const fallback = await buildTemplateDigest(job.tenantId, config.tenant.name, prisma, digestDate);
+      if (!fallback) {
+        await prisma.digestConfig.update({ where: { id: config.id }, data: { lastSentAt: new Date() } });
+        return;
+      }
+      text = fallback.text;
+    }
+  } else {
+    const fallback = await buildTemplateDigest(job.tenantId, config.tenant.name, prisma, digestDate);
+    if (!fallback) {
+      await prisma.digestConfig.update({ where: { id: config.id }, data: { lastSentAt: new Date() } });
+      return;
+    }
+    text = fallback.text;
+  }
+
+  await pool.sendMessage(config.targetAccountId, config.targetGroupJid, text);
+
+  const messageIds = messages.map((m: any) => m.id);
+
+  await prisma.$transaction([
+    prisma.digest.upsert({
+      where: { tenantId_digestDate: { tenantId: job.tenantId, digestDate } },
+      create: { tenantId: job.tenantId, content: text, messageIds, digestDate },
+      update: { content: text, messageIds, sentAt: new Date() },
+    }),
+    prisma.digestConfig.update({ where: { id: config.id }, data: { lastSentAt: new Date() } }),
+  ]);
+
+  logger.info({ tenantId: job.tenantId, messageCount: messageIds.length, triggeredBy: job.triggeredBy }, 'Digest sent');
+}
+
+export function createDigestWorker(
+  redisUrl: string,
+  prisma: any,
+  pool: any,
+  openRouterApiKey: string | undefined,
+): Worker<DigestJobData> {
+  return new Worker<DigestJobData>(
+    'tower.digest.generate.v1',
+    async (job) => processDigestJob(job.data, prisma, pool, openRouterApiKey),
+    { connection: parseRedisUrl(redisUrl) },
+  );
+}

apps/worker/src/queues/digest.queue.ts

@@ -0,0 +1,7 @@
+import { Queue } from 'bullmq';
+import { DigestJobData } from '@tower/types';
+import { parseRedisUrl } from './redis-connection';
+
+export function createDigestQueue(redisUrl: string): Queue<DigestJobData> {
+  return new Queue<DigestJobData>('tower.digest.generate.v1', { connection: parseRedisUrl(redisUrl) });
+}

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

@@ -0,0 +1,29 @@
+import { Worker } from 'bullmq';
+import { DncJobData } from '@tower/types';
+import { parseRedisUrl } from './redis-connection';
+import { createLogger } from '@tower/logger';
+
+const logger = createLogger('dnc-processor');
+
+export async function processDncJob(job: DncJobData): Promise<void> {
+  // No content is stored — DNC events carry only metadata (tenant, group, reason, timestamp).
+  // This keeps the archive high-signal while giving analytics visibility into noise volume.
+  logger.info(
+    {
+      tenantId: job.tenantId,
+      sourceGroupId: job.sourceGroupId,
+      reason: job.reason,
+      timestamp: job.timestamp,
+    },
+    'Message classified as DNC — discarded',
+  );
+  // Future (Phase 3): increment per-tenant DNC counters in Redis for rule-tuning dashboard.
+}
+
+export function createDncWorker(redisUrl: string): Worker<DncJobData> {
+  return new Worker<DncJobData>(
+    'tower.policy.dnc.v1',
+    async (job) => processDncJob(job.data),
+    { connection: parseRedisUrl(redisUrl) },
+  );
+}

apps/worker/src/queues/dnc.queue.ts

@@ -0,0 +1,7 @@
+import { Queue } from 'bullmq';
+import { DncJobData } from '@tower/types';
+import { parseRedisUrl } from './redis-connection';
+
+export function createDncQueue(redisUrl: string): Queue<DncJobData> {
+  return new Queue<DncJobData>('tower.policy.dnc.v1', { connection: parseRedisUrl(redisUrl) });
+}

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

@@ -0,0 +1,52 @@
+import { Worker } from 'bullmq';
+import { P1JobData } from '@tower/types';
+import { parseRedisUrl } from './redis-connection';
+import { createLogger } from '@tower/logger';
+
+const logger = createLogger('p1-processor');
+
+export async function processP1Job(job: P1JobData, prisma: any): Promise<void> {
+  // P1 urgent messages stay PENDING — admin must explicitly approve before multicast.
+  // Log at WARN so this is visible in any log aggregator without a dashboard.
+  logger.warn(
+    {
+      tenantId: job.tenantId,
+      messageId: job.messageId,
+      sourceGroupId: job.sourceGroupId,
+      tags: job.tags,
+      contentPreview: job.content.slice(0, 120),
+    },
+    'P1 URGENT — admin review required before multicast',
+  );
+
+  // Write a high-priority audit event so the admin portal can surface it distinctly.
+  await prisma.auditEvent
+    .create({
+      data: {
+        tenantId: job.tenantId,
+        actorType: 'SYSTEM',
+        action: 'P1_REVIEW_REQUESTED',
+        resourceType: 'Message',
+        resourceId: job.messageId,
+        payload: {
+          sourceGroupId: job.sourceGroupId,
+          senderJid: job.senderJid,
+          tags: job.tags,
+          contentPreview: job.content.slice(0, 200),
+        },
+      },
+    })
+    .catch((err: unknown) =>
+      logger.error({ err, messageId: job.messageId }, 'Failed to write P1_REVIEW_REQUESTED audit event'),
+    );
+
+  // Phase 2: send an admin DM alert via the WhatsApp bot and/or push notification.
+}
+
+export function createP1Worker(redisUrl: string, prisma: any): Worker<P1JobData> {
+  return new Worker<P1JobData>(
+    'tower.urgent.p1.review.v1',
+    async (job) => processP1Job(job.data, prisma),
+    { connection: parseRedisUrl(redisUrl) },
+  );
+}

apps/worker/src/queues/p1.queue.ts

@@ -0,0 +1,7 @@
+import { Queue } from 'bullmq';
+import { P1JobData } from '@tower/types';
+import { parseRedisUrl } from './redis-connection';
+
+export function createP1Queue(redisUrl: string): Queue<P1JobData> {
+  return new Queue<P1JobData>('tower.urgent.p1.review.v1', { connection: parseRedisUrl(redisUrl) });
+}

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

@@ -0,0 +1,30 @@
+import { Worker } from 'bullmq';
+import { ReviewJobData } from '@tower/types';
+import { parseRedisUrl } from './redis-connection';
+import { createLogger } from '@tower/logger';
+
+const logger = createLogger('review-processor');
+
+export async function processReviewJob(job: ReviewJobData): Promise<void> {
+  logger.info(
+    {
+      tenantId: job.tenantId,
+      messageId: job.messageId,
+      sourceGroupId: job.sourceGroupId,
+      tags: job.tags,
+      reason: job.reason,
+    },
+    'Message queued for admin review',
+  );
+  // Phase 1: admin discovers pending messages via portal polling (GET /messages/pending).
+  // Phase 2: replace this processor with a WebSocket push or admin DM notification
+  // so the admin is alerted in real-time without polling.
+}
+
+export function createReviewWorker(redisUrl: string): Worker<ReviewJobData> {
+  return new Worker<ReviewJobData>(
+    'tower.review.requested.v1',
+    async (job) => processReviewJob(job.data),
+    { connection: parseRedisUrl(redisUrl) },
+  );
+}

apps/worker/src/queues/review.queue.ts

@@ -0,0 +1,7 @@
+import { Queue } from 'bullmq';
+import { ReviewJobData } from '@tower/types';
+import { parseRedisUrl } from './redis-connection';
+
+export function createReviewQueue(redisUrl: string): Queue<ReviewJobData> {
+  return new Queue<ReviewJobData>('tower.review.requested.v1', { connection: parseRedisUrl(redisUrl) });
+}

apps/worker/src/queues/thread-resolve.processor.ts

@@ -0,0 +1,63 @@
+import { Worker } from 'bullmq';
+import { ThreadResolveJobData } from '@tower/types';
+import { parseRedisUrl } from './redis-connection';
+import { createLogger } from '@tower/logger';
+
+const logger = createLogger('thread-resolve-processor');
+
+export async function processThreadResolveJob(
+  job: ThreadResolveJobData,
+  prisma: any,
+): Promise<void> {
+  // Only quoted replies create/join threads to avoid thousands of 1-message threads
+  if (!job.quotedPlatformMsgId) return;
+
+  let threadId: string;
+
+  const quoted = await prisma.message.findFirst({
+    where: { platformMsgId: job.quotedPlatformMsgId, sourceGroupId: job.sourceGroupId },
+    select: { id: true, threadId: true },
+  });
+
+  if (quoted) {
+    if (quoted.threadId) {
+      threadId = quoted.threadId;
+    } else {
+      const thread = await prisma.thread.create({
+        data: { tenantId: job.tenantId, sourceGroupId: job.sourceGroupId },
+      });
+      threadId = thread.id;
+      // Backfill the quoted message into the thread
+      await prisma.message.update({ where: { id: quoted.id }, data: { threadId } });
+      await prisma.thread.update({
+        where: { id: threadId },
+        data: { messageCount: { increment: 1 }, lastActivityAt: new Date() },
+      });
+    }
+  } else {
+    // Quoted message not in our DB (e.g. it was sent before TOWER was set up)
+    const thread = await prisma.thread.create({
+      data: { tenantId: job.tenantId, sourceGroupId: job.sourceGroupId },
+    });
+    threadId = thread.id;
+  }
+
+  await prisma.message.update({ where: { id: job.messageId }, data: { threadId } });
+  await prisma.thread.update({
+    where: { id: threadId },
+    data: { messageCount: { increment: 1 }, lastActivityAt: new Date() },
+  });
+
+  logger.info({ messageId: job.messageId, threadId }, 'Thread resolved');
+}
+
+export function createThreadResolveWorker(
+  redisUrl: string,
+  prisma: any,
+): Worker<ThreadResolveJobData> {
+  return new Worker<ThreadResolveJobData>(
+    'tower.thread.resolve.v1',
+    async (job) => processThreadResolveJob(job.data, prisma),
+    { connection: parseRedisUrl(redisUrl), concurrency: 3 },
+  );
+}

apps/worker/src/queues/thread-resolve.queue.ts

@@ -0,0 +1,7 @@
+import { Queue } from 'bullmq';
+import { ThreadResolveJobData } from '@tower/types';
+import { parseRedisUrl } from './redis-connection';
+
+export function createThreadResolveQueue(redisUrl: string): Queue<ThreadResolveJobData> {
+  return new Queue<ThreadResolveJobData>('tower.thread.resolve.v1', { connection: parseRedisUrl(redisUrl) });
+}

apps/worker/src/whatsapp/baileys-adapter.ts

@@ -0,0 +1,202 @@
+import { ChannelAdapter, AdapterHandlers, AdapterGroupMeta, AdapterOutboundCommand } from '@tower/types';
+import { createLogger } from '@tower/logger';
+import { WhatsAppSessionPool } from './session-pool';
+import { syncGroups } from './group-sync';
+import { startOtpSenderLoop } from './otp-sender';
+import { handleCommand } from './command-handler';
+
+const logger = createLogger('baileys-adapter');
+
+export class BaileysAdapter implements ChannelAdapter {
+  readonly platform = 'whatsapp';
+
+  private groupMaps = new Map<string, Map<string, string>>();
+  private accountPollTimer: ReturnType<typeof setInterval> | null = null;
+  private groupSyncTimer: ReturnType<typeof setInterval> | null = null;
+
+  constructor(
+    private readonly prisma: any,
+    private readonly pool: WhatsAppSessionPool,
+  ) {}
+
+  async start(handlers: AdapterHandlers): Promise<void> {
+    const accounts = await this.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 this.startAccount(account, handlers);
+    }
+
+    this.startPollingLoops(handlers);
+    startOtpSenderLoop(this.prisma, this.pool, logger);
+    logger.info({ accountCount: accounts.length }, 'BaileysAdapter started');
+  }
+
+  private async startAccount(
+    account: { id: string; sessionPath: string },
+    handlers: AdapterHandlers,
+  ): Promise<void> {
+    this.groupMaps.set(account.id, new Map());
+    try {
+      await this.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',
+          );
+
+          // Commands (STOP/START/PORTAL) run before the groupMap check so they work
+          // even if the group hasn't been synced yet.
+          await handleCommand(msg, accountId, this.prisma, this.pool).catch((err: unknown) =>
+            logger.error({ err }, 'Command handler error'),
+          );
+
+          const groupMap = this.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;
+          }
+
+          await handlers.onMessage({
+            platformMsgId: msg.platformMsgId,
+            sourceGroupJid: msg.sourceGroupJid,
+            sourceGroupId,
+            senderJid: msg.senderJid,
+            senderName: msg.senderName ?? null,
+            content: msg.content,
+            accountId,
+            quotedPlatformMsgId: msg.quotedPlatformMsgId,
+          });
+        },
+        async (reaction) => {
+          if (handlers.onReaction) {
+            await handlers.onReaction(reaction).catch((err: unknown) =>
+              logger.error({ err }, 'Reaction handler error'),
+            );
+          }
+        },
+        async (groups, accountId) => {
+          logger.info({ count: Object.keys(groups).length, accountId }, 'Syncing groups');
+          const map = await syncGroups(groups, accountId, this.prisma, this.pool);
+          this.groupMaps.set(accountId, map);
+        },
+        async (qr, accountId) => {
+          await this.prisma.account
+            .update({ where: { id: accountId }, data: { qrCode: qr, status: 'PAIRING' } })
+            .catch((err: unknown) => logger.error({ accountId, err }, 'Failed to store QR in DB'));
+          logger.info({ accountId }, 'QR code updated');
+        },
+        async (status, accountId, jid?) => {
+          if (status === 'connected') {
+            await this.prisma.account
+              .update({
+                where: { id: accountId },
+                data: { qrCode: null, status: 'ACTIVE', ...(jid ? { jid } : {}) },
+              })
+              .catch((err: unknown) => logger.error({ accountId, err }, 'Failed to update account status'));
+            logger.info({ accountId, jid }, 'Account connected — QR cleared');
+          } else if (status === 'logged_out') {
+            await this.prisma.account
+              .update({ where: { id: accountId }, data: { status: 'DISCONNECTED' } })
+              .catch((err: unknown) => logger.error({ accountId, err }, 'Failed to update account status'));
+            logger.info({ accountId }, 'Account logged out — awaiting QR scan');
+          } else if (status === 'disconnected') {
+            await this.prisma.account
+              .update({ where: { id: accountId }, data: { status: 'DISCONNECTED' } })
+              .catch((err: unknown) => 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');
+    }
+  }
+
+  async send(command: AdapterOutboundCommand): Promise<void> {
+    await this.pool.sendMessage(command.accountId, command.toGroupJid, command.text);
+  }
+
+  async getGroupMeta(groupJid: string, accountId: string): Promise<AdapterGroupMeta | null> {
+    const sock = this.pool.get(accountId);
+    if (!sock) return null;
+    try {
+      const meta = await sock.groupMetadata(groupJid);
+      return {
+        jid: groupJid,
+        name: meta.subject ?? groupJid,
+        participants: (meta.participants ?? []).map((p: any) => ({
+          jid: p.id,
+          isAdmin: p.admin === 'admin' || p.admin === 'superadmin',
+        })),
+      };
+    } catch {
+      return null;
+    }
+  }
+
+  async stop(): Promise<void> {
+    if (this.accountPollTimer) clearInterval(this.accountPollTimer);
+    if (this.groupSyncTimer) clearInterval(this.groupSyncTimer);
+    this.accountPollTimer = null;
+    this.groupSyncTimer = null;
+    await this.pool.closeAll();
+  }
+
+  private startPollingLoops(handlers: AdapterHandlers): void {
+    this.accountPollTimer = setInterval(async () => {
+      try {
+        const all = await this.prisma.account.findMany({
+          where: {
+            isBot: true,
+            status: { in: ['ACTIVE', 'DISCONNECTED', 'PAIRING'] },
+            platform: 'whatsapp',
+          },
+          select: { id: true, sessionPath: true },
+        });
+        for (const account of all) {
+          if (!this.pool.get(account.id)) {
+            logger.info({ accountId: account.id }, 'New bot account detected — starting session');
+            await this.startAccount(account, handlers);
+          }
+        }
+      } catch (err) {
+        logger.error({ err }, 'Error polling for new accounts');
+      }
+    }, 30_000);
+
+    this.groupSyncTimer = setInterval(async () => {
+      try {
+        for (const [accountId, sock] of this.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, this.prisma, this.pool);
+            this.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);
+  }
+}