feat: add spam detection lane (tower.classify.spam) with content-aware filtering

- Add SpamJobData + sync missing job types (DncJobData, ClassifyJobData, P1JobData, etc.) from dist back to source in @tower/types
- spam-detector.ts: emoji-only, short-greeting, link-only detection + duplicate-hash (1h window)
- spam.queue.ts / spam.processor.ts: BullMQ tower.classify.spam lane
- classify.processor.ts: spam check (Step 0) before rule engine, accepts spamQueue param
- main.ts: wire spamQueue + spamWorker; spam pre-filter before ingestQueue.add
- dnc.processor.ts: labelled reason logging

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

apps/worker/src/main.ts

@@ -8,6 +8,8 @@ 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';
@@ -15,6 +17,7 @@ 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');
 
@@ -30,11 +33,13 @@ async function bootstrap() {
   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'));
@@ -42,6 +47,7 @@ async function bootstrap() {
   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>>();
 
@@ -94,6 +100,25 @@ async function bootstrap() {
             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 },
@@ -279,9 +304,11 @@ async function bootstrap() {
     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);
   };

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

@@ -1,12 +1,16 @@
 import { Worker, Queue } from 'bullmq';
-import { ClassifyJobData, ForwardJobData, IndexJobData, ReviewJobData, P1JobData, DncJobData } from '@tower/types';
+import { ClassifyJobData, ForwardJobData, IndexJobData, ReviewJobData, P1JobData, DncJobData, SpamJobData } from '@tower/types';
 import { parseRedisUrl } from './redis-connection';
 import { matchContentRules, TenantRuleRow } from '../whatsapp/match-rules';
 import { approveMessage } from '../core/approve-message';
+import { detectSpam, checkDuplicateHash } from '../spam/spam-detector';
 import { createLogger } from '@tower/logger';
 
 const logger = createLogger('classify-processor');
 
+// Duplicate-hash window: 1 hour
+const DUPE_WINDOW_MS = 60 * 60 * 1000;
+
 export async function processClassifyJob(
   job: ClassifyJobData,
   prisma: any,
@@ -16,6 +20,7 @@ export async function processClassifyJob(
   reviewQueue: Queue<ReviewJobData>,
   p1Queue: Queue<P1JobData>,
   dncQueue: Queue<DncJobData>,
+  spamQueue: Queue<SpamJobData>,
 ): Promise<void> {
   const message = await prisma.message.findUnique({
     where: { id: job.messageId },
@@ -33,6 +38,28 @@ export async function processClassifyJob(
     return;
   }
 
+  // ── Step 0: Spam detection (runs before rule engine) ─────────────────────
+  const spamResult = detectSpam(message.content);
+  const isDupe = spamResult.isSpam
+    ? false
+    : await checkDuplicateHash(spamResult.contentHash, job.tenantId, job.sourceGroupId, DUPE_WINDOW_MS, prisma);
+
+  if (spamResult.isSpam || isDupe) {
+    const reason = isDupe ? 'duplicate_hash' : spamResult.reason!;
+    await prisma.message.update({ where: { id: job.messageId }, data: { status: 'DNC' } });
+    await spamQueue.add('spam', {
+      tenantId: job.tenantId,
+      messageId: job.messageId,
+      sourceGroupId: job.sourceGroupId,
+      senderJid: job.senderJid,
+      reason,
+      contentHash: spamResult.contentHash,
+      traceId: job.traceId,
+    }, { attempts: 1 }).catch((err: unknown) => logger.warn({ err }, 'Failed to enqueue spam job'));
+    logger.info({ messageId: job.messageId, reason }, 'Message classified as spam — DNC');
+    return;
+  }
+
   // ── Step 1: Hashtag/prefix rules — org rules first (authoritative), then tenant ──
   const tenant = await prisma.tenant.findUnique({
     where: { id: job.tenantId },
@@ -167,13 +194,14 @@ export function createClassifyWorker(
   reviewQueue: Queue<ReviewJobData>,
   p1Queue: Queue<P1JobData>,
   dncQueue: Queue<DncJobData>,
+  spamQueue: Queue<SpamJobData>,
 ): Worker<ClassifyJobData> {
   return new Worker<ClassifyJobData>(
     'tower.classify.v1',
     async (job) =>
       processClassifyJob(
         job.data, prisma, pool,
-        forwardQueue, indexQueue, reviewQueue, p1Queue, dncQueue,
+        forwardQueue, indexQueue, reviewQueue, p1Queue, dncQueue, spamQueue,
       ),
     { connection: parseRedisUrl(redisUrl), concurrency: 5 },
   );

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

@@ -5,19 +5,24 @@ import { createLogger } from '@tower/logger';
 
 const logger = createLogger('dnc-processor');
 
+const DNC_REASON_LABELS: Record<string, string> = {
+  rule_skip: 'Rule SKIP/REJECT action',
+  no_rule_match: 'No rule matched',
+  spam_detected: 'Spam detector',
+  content_filtered: 'Content filter',
+};
+
 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,
+      label: DNC_REASON_LABELS[job.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> {

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

@@ -0,0 +1,32 @@
+import { Worker } from 'bullmq';
+import { SpamJobData } from '@tower/types';
+import { parseRedisUrl } from './redis-connection';
+import { createLogger } from '@tower/logger';
+
+const logger = createLogger('spam-processor');
+
+export async function processSpamJob(job: SpamJobData, prisma: any): Promise<void> {
+  await prisma.message.update({
+    where: { id: job.messageId },
+    data: { status: 'DNC' },
+  });
+
+  logger.info(
+    {
+      tenantId: job.tenantId,
+      messageId: job.messageId,
+      senderJid: job.senderJid,
+      reason: job.reason,
+      contentHash: job.contentHash,
+    },
+    'Message classified as spam — discarded',
+  );
+}
+
+export function createSpamWorker(redisUrl: string, prisma: any): Worker<SpamJobData> {
+  return new Worker<SpamJobData>(
+    'tower.classify.spam',
+    async (job) => processSpamJob(job.data, prisma),
+    { connection: parseRedisUrl(redisUrl) },
+  );
+}

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

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

apps/worker/src/spam/spam-detector.ts

@@ -0,0 +1,77 @@
+import { createHash } from 'crypto';
+
+export type SpamReason = 'emoji_only' | 'short_greeting' | 'link_only' | 'forward_flood' | 'duplicate_hash';
+
+export interface SpamResult {
+  isSpam: boolean;
+  reason?: SpamReason;
+  contentHash: string;
+}
+
+const GREETING_PHRASES = [
+  'good morning', 'good evening', 'good night', 'good afternoon',
+  'gm', 'gn', 'subah ki shayari', 'शुभ प्रभात', 'शुभ रात्रि',
+  'happy', 'jai hind', 'jai shree ram', 'jai mata di',
+  'forwarded as received', 'please share', 'must share',
+];
+
+const EMOJI_RE = /^[\p{Emoji}\p{Emoji_Presentation}\s]+$/u;
+const URL_RE = /https?:\/\/\S+/gi;
+const TOKEN_MIN = 4;
+
+export function hashContent(content: string): string {
+  return createHash('sha256').update(content.trim().toLowerCase()).digest('hex').slice(0, 16);
+}
+
+export function detectSpam(content: string): SpamResult {
+  const trimmed = content.trim();
+  const contentHash = hashContent(trimmed);
+
+  if (EMOJI_RE.test(trimmed)) {
+    return { isSpam: true, reason: 'emoji_only', contentHash };
+  }
+
+  const lower = trimmed.toLowerCase();
+  const withoutUrls = lower.replace(URL_RE, '').trim();
+  const tokens = withoutUrls.split(/\s+/).filter(Boolean);
+
+  if (tokens.length < TOKEN_MIN) {
+    if (GREETING_PHRASES.some((phrase) => lower.includes(phrase))) {
+      return { isSpam: true, reason: 'short_greeting', contentHash };
+    }
+  }
+
+  const urls = trimmed.match(URL_RE) ?? [];
+  if (urls.length > 0 && tokens.length === 0) {
+    return { isSpam: true, reason: 'link_only', contentHash };
+  }
+
+  return { isSpam: false, contentHash };
+}
+
+export async function checkDuplicateHash(
+  contentHash: string,
+  tenantId: string,
+  sourceGroupId: string,
+  windowMs: number,
+  prisma: any,
+): Promise<boolean> {
+  const since = new Date(Date.now() - windowMs);
+  const count = await prisma.message.count({
+    where: {
+      tenantId,
+      sourceGroupId,
+      content: { startsWith: '' },
+      createdAt: { gte: since },
+    },
+  });
+  if (count === 0) return false;
+
+  const recent = await prisma.message.findMany({
+    where: { tenantId, sourceGroupId, createdAt: { gte: since } },
+    select: { content: true },
+    take: 100,
+  });
+
+  return recent.some((m: { content: string }) => hashContent(m.content) === contentHash);
+}

packages/types/src/message.ts

@@ -91,4 +91,72 @@ export interface IndexJobData {
   tags: string[];
   platform: string;
   approvedAt: string;      // ISO 8601 — converted to Unix ms in the index processor
+  traceId?: string;
+}
+
+export interface DncJobData {
+  tenantId: string;
+  sourceGroupId: string;
+  accountId: string;
+  reason: 'rule_skip' | 'no_rule_match' | 'spam_detected' | 'content_filtered';
+  timestamp: string;
+  policyVersion: string;
+  traceId?: string;
+}
+
+export interface ClassifyJobData {
+  messageId: string;
+  tenantId: string;
+  sourceGroupId: string;
+  sourceGroupJid: string;
+  senderJid: string;
+  accountId: string;
+  traceId?: string;
+}
+
+export interface ThreadResolveJobData {
+  messageId: string;
+  tenantId: string;
+  sourceGroupId: string;
+  quotedPlatformMsgId?: string;
+  traceId?: string;
+}
+
+export interface DigestJobData {
+  tenantId: string;
+  digestDate: string;
+  triggeredBy: 'schedule' | 'manual';
+  traceId?: string;
+}
+
+export interface ReviewJobData {
+  tenantId: string;
+  messageId: string;
+  sourceGroupId: string;
+  tags: string[];
+  reason: string;
+  policyVersion: string;
+  traceId?: string;
+}
+
+export interface P1JobData {
+  tenantId: string;
+  messageId: string;
+  content: string;
+  sourceGroupId: string;
+  senderJid: string;
+  senderName?: string;
+  tags: string[];
+  policyVersion: string;
+  traceId?: string;
+}
+
+export interface SpamJobData {
+  tenantId: string;
+  messageId: string;
+  sourceGroupId: string;
+  senderJid: string;
+  reason: 'forward_flood' | 'duplicate_hash' | 'emoji_only' | 'short_greeting' | 'link_only';
+  contentHash: string;
+  traceId?: string;
 }