diff --git a/apps/worker/src/main.ts b/apps/worker/src/main.ts index dca01f0..dc8352b 100644 --- a/apps/worker/src/main.ts +++ b/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>(); @@ -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); }; diff --git a/apps/worker/src/queues/classify.processor.ts b/apps/worker/src/queues/classify.processor.ts index 574df9c..cd9397e 100644 --- a/apps/worker/src/queues/classify.processor.ts +++ b/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, p1Queue: Queue, dncQueue: Queue, + spamQueue: Queue, ): Promise { 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, p1Queue: Queue, dncQueue: Queue, + spamQueue: Queue, ): Worker { return new Worker( '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 }, ); diff --git a/apps/worker/src/queues/dnc.processor.ts b/apps/worker/src/queues/dnc.processor.ts index 029add1..df50bc5 100644 --- a/apps/worker/src/queues/dnc.processor.ts +++ b/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 = { + 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 { - // 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 { diff --git a/apps/worker/src/queues/spam.processor.ts b/apps/worker/src/queues/spam.processor.ts new file mode 100644 index 0000000..45573ab --- /dev/null +++ b/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 { + 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 { + return new Worker( + 'tower.classify.spam', + async (job) => processSpamJob(job.data, prisma), + { connection: parseRedisUrl(redisUrl) }, + ); +} diff --git a/apps/worker/src/queues/spam.queue.ts b/apps/worker/src/queues/spam.queue.ts new file mode 100644 index 0000000..534519b --- /dev/null +++ b/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 { + return new Queue('tower.classify.spam', { connection: parseRedisUrl(redisUrl) }); +} diff --git a/apps/worker/src/spam/spam-detector.ts b/apps/worker/src/spam/spam-detector.ts new file mode 100644 index 0000000..71d6ad1 --- /dev/null +++ b/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 { + 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); +} diff --git a/packages/types/src/message.ts b/packages/types/src/message.ts index 254e04b..9b2c999 100644 --- a/packages/types/src/message.ts +++ b/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; }