feat: auto-detect events, seva tasks, and FAQs from WhatsApp messages

Adds the AI-driven content pipeline so portal content is never manually
created — the system proposes it, admins review and approve.

Worker:
- `tower.event.extract` queue + Event Extractor processor: LLM extracts
  event title/date/time/location from messages with #event hashtag,
  date/time patterns, RSVP phrases, or location terms; seva action items
  within event messages become separate SEVA_TASK drafts
- `tower.faq.candidate` queue + FAQ Curator processor: LLM extracts Q&A
  pairs from messages with question patterns + meaningful answer length
- `classify.processor.ts`: `isEventCandidate()` and `isFaqCandidate()`
  run after rule matching on all non-spam, non-DNC messages; candidates
  are enqueued to the extraction lanes in parallel with normal routing
- `match-rules.ts`: add P1 to TenantRuleRow action union so P1 rules
  are fully typed end-to-end
- `NormalizedMessage`: add `quotedPlatformMsgId` field
- `IngestJobData.effectiveAction`: add `'P1'` to the union

Schema:
- `ContentDraft` model with `DraftType` (EVENT|SEVA_TASK|FAQ_ITEM) and
  `DraftStatus` (PENDING_REVIEW|APPROVED|REJECTED); carries `proposedJson`
  from the LLM, `confidence` score, and `publishedId` after approval
- Migration: `20260617270000_add_content_drafts`

API:
- `DraftsModule`: `GET /admin/drafts`, `POST /admin/drafts/:id/approve`,
  `POST /admin/drafts/:id/reject`; approve creates the actual
  Event/SevaOpportunity/KnowledgeItem and marks the draft APPROVED;
  reject marks it REJECTED — both audit-logged

Web:
- `/admin/drafts` and `/drafts` — tabbed view by type (All/Events/Seva/FAQ)
- `DraftCard` — shows source message excerpt, AI proposed content, and
  approve/reject buttons; approve/reject immediately refresh the list
- "AI Drafts" added to both super-admin and tenant-admin sidebars
- "AI Content Drafts" button added to admin dashboard

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

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

@@ -2,7 +2,7 @@ import { createLogger } from '@tower/logger';
 
 const SIX_HOURS_MS = 6 * 60 * 60 * 1000;
 
-export function startExpireScheduler(prisma: any, logger = createLogger('expire-scheduler')): void {
+export function startExpireScheduler(prisma: any, logger: ReturnType<typeof createLogger> = createLogger('expire-scheduler')): void {
   let running = false;
 
   const tick = async () => {

apps/worker/src/main.ts

@@ -10,6 +10,10 @@ 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 { createEventExtractQueue } from './queues/event-extract.queue';
+import { createEventExtractWorker } from './queues/event-extract.processor';
+import { createFaqCandidateQueue } from './queues/faq-candidate.queue';
+import { createFaqCuratorWorker } from './queues/faq-curator.processor';
 import { WhatsAppSessionPool } from './whatsapp/session-pool';
 import { matchContentRules, TenantRuleRow } from './whatsapp/match-rules';
 import { syncGroups } from './whatsapp/group-sync';
@@ -34,12 +38,16 @@ async function bootstrap() {
   const forwardQueue = createForwardQueue(env.REDIS_URL);
   const indexQueue = createIndexQueue(env.REDIS_URL);
   const spamQueue = createSpamQueue(env.REDIS_URL);
+  const eventExtractQueue = createEventExtractQueue(env.REDIS_URL);
+  const faqCandidateQueue = createFaqCandidateQueue(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);
+  const eventExtractWorker = createEventExtractWorker(env.REDIS_URL, prisma, env.OPENROUTER_API_KEY ?? '');
+  const faqCuratorWorker = createFaqCuratorWorker(env.REDIS_URL, prisma, env.OPENROUTER_API_KEY ?? '');
 
   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'));
@@ -48,6 +56,8 @@ async function bootstrap() {
   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'));
+  eventExtractWorker.on('failed', (job, err) => logger.warn({ jobId: job?.id, err }, 'Event extract job failed'));
+  faqCuratorWorker.on('failed', (job, err) => logger.warn({ jobId: job?.id, err }, 'FAQ curator job failed'));
 
   const groupMaps = new Map<string, Map<string, string>>();
 
@@ -305,10 +315,14 @@ async function bootstrap() {
     await forwardWorker.close();
     await indexWorker.close();
     await spamWorker.close();
+    await eventExtractWorker.close();
+    await faqCuratorWorker.close();
     await ingestQueue.close();
     await forwardQueue.close();
     await indexQueue.close();
     await spamQueue.close();
+    await eventExtractQueue.close();
+    await faqCandidateQueue.close();
     await prisma.$disconnect();
     process.exit(0);
   };

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

@@ -1,11 +1,35 @@
 import { Worker, Queue } from 'bullmq';
-import { ClassifyJobData, ForwardJobData, IndexJobData, ReviewJobData, P1JobData, DncJobData, SpamJobData } from '@tower/types';
+import { ClassifyJobData, ForwardJobData, IndexJobData, ReviewJobData, P1JobData, DncJobData, SpamJobData, EventExtractJobData, FaqCandidateJobData } 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';
 
+// ── Candidate detection signals ──────────────────────────────────────────────
+const EVENT_HASHTAGS = /\#(event|programme|program|meeting|webinar|workshop|gathering|satsang)\b/i;
+const DATE_PATTERN = /\b(\d{1,2}[\/\-]\d{1,2}(\[\/\-]\d{2,4})?|\d{1,2}(st|nd|rd|th)?\s+(jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)|monday|tuesday|wednesday|thursday|friday|saturday|sunday|tomorrow|next\s+\w+)\b/i;
+const TIME_PATTERN = /\b(\d{1,2}:\d{2}\s*(am|pm)?|\d{1,2}\s*(am|pm))\b/i;
+const LOCATION_TERMS = /\b(at\s+\w|venue|location|address|hall|temple|mosque|church|community\s+cent(re|er)|ground|maidan|auditorium)\b/i;
+const RSVP_TERMS = /\b(rsvp|confirm\s+attendance|will\s+you\s+(attend|join|come)|please\s+(join|attend|confirm)|you('re| are)\s+invited)\b/i;
+
+const QUESTION_PATTERN = /\?|(\b(how\s+(to|do|can|does)|what\s+(is|are|should|can)|when\s+(can|should|do|is|are)|where\s+(is|are|can)|why\s+(is|are|should)|can\s+(we|i|you|someone|members?))\b)/i;
+
+function isEventCandidate(content: string): boolean {
+  if (EVENT_HASHTAGS.test(content)) return true;
+  const hasDate = DATE_PATTERN.test(content);
+  const hasTime = TIME_PATTERN.test(content);
+  const hasLocation = LOCATION_TERMS.test(content);
+  const hasRsvp = RSVP_TERMS.test(content);
+  return hasDate && (hasTime || hasLocation || hasRsvp);
+}
+
+function isFaqCandidate(content: string): boolean {
+  if (!QUESTION_PATTERN.test(content)) return false;
+  // Require meaningful length — a lone question with no answer isn't a FAQ
+  return content.length > 80;
+}
+
 const logger = createLogger('classify-processor');
 
 // Duplicate-hash window: 1 hour
@@ -21,6 +45,8 @@ export async function processClassifyJob(
   p1Queue: Queue<P1JobData>,
   dncQueue: Queue<DncJobData>,
   spamQueue: Queue<SpamJobData>,
+  eventExtractQueue: Queue<EventExtractJobData>,
+  faqCandidateQueue: Queue<FaqCandidateJobData>,
 ): Promise<void> {
   const message = await prisma.message.findUnique({
     where: { id: job.messageId },
@@ -114,6 +140,7 @@ export async function processClassifyJob(
     }, { attempts: 3, backoff: { type: 'exponential', delay: 1000 } }).catch((err: unknown) =>
       logger.error({ err, messageId: job.messageId }, 'Failed to enqueue P1 job'),
     );
+    await routeToContentQueues(message, job, eventExtractQueue, faqCandidateQueue);
     logger.warn({ messageId: job.messageId, tags }, 'P1 urgent message — admin review required');
     return;
   }
@@ -130,6 +157,7 @@ export async function processClassifyJob(
     }, { attempts: 1 }).catch((err: unknown) =>
       logger.warn({ err, messageId: job.messageId }, 'Failed to enqueue review job'),
     );
+    await routeToContentQueues(message, job, eventExtractQueue, faqCandidateQueue);
     logger.info({ messageId: job.messageId, tags }, 'Message flagged by rule → PENDING');
     return;
   }
@@ -169,10 +197,12 @@ export async function processClassifyJob(
       }, { attempts: 1 }).catch((err: unknown) => logger.warn({ err }, 'Failed to enqueue review job'));
       logger.info({ messageId: job.messageId }, 'AUTO_APPROVE downgraded — sender not admin');
     }
+    await routeToContentQueues(message, job, eventExtractQueue, faqCandidateQueue);
     return;
   }
 
   // ── Step 2: No rule matched → DNC (expires naturally at 72h) ────────────
+  // Still check for event/FAQ candidates even on no-rule-match messages
   await prisma.message.update({ where: { id: job.messageId }, data: { status: 'DNC' } });
   await dncQueue.add('dnc', {
     tenantId: job.tenantId,
@@ -185,6 +215,37 @@ export async function processClassifyJob(
   logger.debug({ messageId: job.messageId }, 'No rule matched — DNC');
 }
 
+async function routeToContentQueues(
+  message: { id: string; content: string },
+  job: ClassifyJobData,
+  eventExtractQueue: Queue<EventExtractJobData>,
+  faqCandidateQueue: Queue<FaqCandidateJobData>,
+): Promise<void> {
+  if (isEventCandidate(message.content)) {
+    await eventExtractQueue.add('event-extract', {
+      tenantId: job.tenantId,
+      messageId: message.id,
+      content: message.content,
+      traceId: job.traceId,
+    }, { attempts: 2, backoff: { type: 'exponential', delay: 2000 } }).catch(
+      (err: unknown) => logger.warn({ err, messageId: message.id }, 'Failed to enqueue event-extract job'),
+    );
+    logger.info({ messageId: message.id }, 'Event candidate → tower.event.extract');
+  }
+
+  if (isFaqCandidate(message.content)) {
+    await faqCandidateQueue.add('faq-candidate', {
+      tenantId: job.tenantId,
+      messageId: message.id,
+      content: message.content,
+      traceId: job.traceId,
+    }, { attempts: 2, backoff: { type: 'exponential', delay: 2000 } }).catch(
+      (err: unknown) => logger.warn({ err, messageId: message.id }, 'Failed to enqueue faq-candidate job'),
+    );
+    logger.info({ messageId: message.id }, 'FAQ candidate → tower.faq.candidate');
+  }
+}
+
 export function createClassifyWorker(
   redisUrl: string,
   prisma: any,
@@ -195,6 +256,8 @@ export function createClassifyWorker(
   p1Queue: Queue<P1JobData>,
   dncQueue: Queue<DncJobData>,
   spamQueue: Queue<SpamJobData>,
+  eventExtractQueue: Queue<EventExtractJobData>,
+  faqCandidateQueue: Queue<FaqCandidateJobData>,
 ): Worker<ClassifyJobData> {
   return new Worker<ClassifyJobData>(
     'tower.classify.v1',
@@ -202,6 +265,7 @@ export function createClassifyWorker(
       processClassifyJob(
         job.data, prisma, pool,
         forwardQueue, indexQueue, reviewQueue, p1Queue, dncQueue, spamQueue,
+        eventExtractQueue, faqCandidateQueue,
       ),
     { connection: parseRedisUrl(redisUrl), concurrency: 5 },
   );

apps/worker/src/queues/event-extract.processor.ts

@@ -0,0 +1,127 @@
+import { Worker } from 'bullmq';
+import { EventExtractJobData } from '@tower/types';
+import { parseRedisUrl } from './redis-connection';
+import { callLLM } from '../ai/llm-client';
+import { createLogger } from '@tower/logger';
+
+const logger = createLogger('event-extract-processor');
+
+const EVENT_EXTRACT_PROMPT = (content: string) => `
+You are an event extraction agent for a community WhatsApp group.
+Extract event details from the following message.
+
+Message: """
+${content}
+"""
+
+Return ONLY valid JSON matching this schema, or null if no event can be reliably extracted:
+{
+  "title": "string — event name",
+  "date": "YYYY-MM-DD or null",
+  "time": "HH:MM (24h) or null",
+  "location": "string or null",
+  "description": "string — 1-2 sentence summary",
+  "rsvpNeeded": boolean,
+  "sevaActionItems": [
+    { "title": "string", "slots": number_or_null, "skills": ["string"] }
+  ]
+}
+
+Rules:
+- If you cannot confidently extract a date or title, return null
+- sevaActionItems is an empty array if no volunteer asks are present
+- Do not invent details not present in the message
+`.trim();
+
+interface EventDraft {
+  title: string;
+  date: string | null;
+  time: string | null;
+  location: string | null;
+  description: string;
+  rsvpNeeded: boolean;
+  sevaActionItems: Array<{ title: string; slots: number | null; skills: string[] }>;
+}
+
+async function extractEvent(content: string, apiKey: string): Promise<EventDraft | null> {
+  const raw = await callLLM(EVENT_EXTRACT_PROMPT(content), apiKey);
+  const cleaned = raw.replace(/```json|```/g, '').trim();
+  if (cleaned === 'null') return null;
+  try {
+    return JSON.parse(cleaned) as EventDraft;
+  } catch {
+    return null;
+  }
+}
+
+export function createEventExtractWorker(
+  redisUrl: string,
+  prisma: any,
+  apiKey: string,
+): Worker<EventExtractJobData> {
+  return new Worker<EventExtractJobData>(
+    'tower.event.extract',
+    async (job) => {
+      const { tenantId, messageId, content } = job.data;
+
+      if (!apiKey) {
+        logger.debug({ messageId }, 'No OPENROUTER_API_KEY — skipping event extraction');
+        return;
+      }
+
+      let draft: EventDraft | null = null;
+      try {
+        draft = await extractEvent(content, apiKey);
+      } catch (err) {
+        logger.warn({ err, messageId }, 'LLM call failed for event extraction');
+        return;
+      }
+
+      if (!draft) {
+        logger.debug({ messageId }, 'LLM returned null — not an event');
+        return;
+      }
+
+      // Calculate a simple confidence score based on how many fields were extracted
+      const fields = [draft.title, draft.date, draft.time, draft.location, draft.description];
+      const filled = fields.filter(Boolean).length;
+      const confidence = filled / fields.length;
+
+      await prisma.contentDraft.create({
+        data: {
+          tenantId,
+          type: 'EVENT',
+          status: 'PENDING_REVIEW',
+          sourceMessageId: messageId,
+          proposedJson: draft as any,
+          confidence,
+        },
+      });
+
+      logger.info({ messageId, confidence, title: draft.title }, 'Event draft created');
+
+      // If the extracted event has seva action items, create a separate SEVA_TASK draft per item
+      for (const item of draft.sevaActionItems ?? []) {
+        if (!item.title) continue;
+        await prisma.contentDraft.create({
+          data: {
+            tenantId,
+            type: 'SEVA_TASK',
+            status: 'PENDING_REVIEW',
+            sourceMessageId: messageId,
+            proposedJson: {
+              title: item.title,
+              slots: item.slots,
+              skills: item.skills,
+              date: draft.date,
+              parentEventTitle: draft.title,
+            } as any,
+            confidence: confidence * 0.9,
+          },
+        });
+        logger.info({ messageId, sevaTitle: item.title }, 'Seva task draft created from event');
+      }
+    },
+    { connection: parseRedisUrl(redisUrl), concurrency: 3 },
+  );
+}

apps/worker/src/queues/event-extract.queue.ts

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

apps/worker/src/queues/faq-candidate.queue.ts

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

apps/worker/src/queues/faq-curator.processor.ts

@@ -0,0 +1,92 @@
+import { Worker } from 'bullmq';
+import { FaqCandidateJobData } from '@tower/types';
+import { parseRedisUrl } from './redis-connection';
+import { callLLM } from '../ai/llm-client';
+import { createLogger } from '@tower/logger';
+
+const logger = createLogger('faq-curator-processor');
+
+const FAQ_EXTRACT_PROMPT = (content: string) => `
+You are an FAQ curation agent for a community WhatsApp group.
+Extract a clear question-answer pair from the following message.
+
+Message: """
+${content}
+"""
+
+Return ONLY valid JSON matching this schema, or null if no clear Q&A pair exists:
+{
+  "question": "string — the question being asked",
+  "answer": "string — the answer provided",
+  "tags": ["string"]
+}
+
+Rules:
+- Only extract if BOTH a question AND a meaningful answer are present in the message
+- The question must be a genuine community question (how-to, what-is, when, where, etc.)
+- tags should be 1-3 lowercase keywords describing the topic
+- Do not invent content not in the message
+- Return null if the message is only a question with no answer, or only an answer with no question
+`.trim();
+
+interface FaqDraft {
+  question: string;
+  answer: string;
+  tags: string[];
+}
+
+async function extractFaq(content: string, apiKey: string): Promise<FaqDraft | null> {
+  const raw = await callLLM(FAQ_EXTRACT_PROMPT(content), apiKey);
+  const cleaned = raw.replace(/```json|```/g, '').trim();
+  if (cleaned === 'null') return null;
+  try {
+    return JSON.parse(cleaned) as FaqDraft;
+  } catch {
+    return null;
+  }
+}
+
+export function createFaqCuratorWorker(
+  redisUrl: string,
+  prisma: any,
+  apiKey: string,
+): Worker<FaqCandidateJobData> {
+  return new Worker<FaqCandidateJobData>(
+    'tower.faq.candidate',
+    async (job) => {
+      const { tenantId, messageId, content } = job.data;
+
+      if (!apiKey) {
+        logger.debug({ messageId }, 'No OPENROUTER_API_KEY — skipping FAQ extraction');
+        return;
+      }
+
+      let draft: FaqDraft | null = null;
+      try {
+        draft = await extractFaq(content, apiKey);
+      } catch (err) {
+        logger.warn({ err, messageId }, 'LLM call failed for FAQ extraction');
+        return;
+      }
+
+      if (!draft) {
+        logger.debug({ messageId }, 'LLM returned null — not a FAQ candidate');
+        return;
+      }
+
+      await prisma.contentDraft.create({
+        data: {
+          tenantId,
+          type: 'FAQ_ITEM',
+          status: 'PENDING_REVIEW',
+          sourceMessageId: messageId,
+          proposedJson: draft as any,
+          confidence: 0.8,
+        },
+      });
+
+      logger.info({ messageId, question: draft.question.slice(0, 80) }, 'FAQ draft created');
+    },
+    { connection: parseRedisUrl(redisUrl), concurrency: 3 },
+  );
+}

apps/worker/src/whatsapp/match-rules.ts

@@ -8,13 +8,13 @@ export interface TenantRuleRow {
   id: string;
   matchType: 'HASHTAG' | 'PREFIX' | 'REACTION_EMOJI';
   matchValue: string;
-  action: 'FLAG' | 'AUTO_APPROVE' | 'SKIP' | 'REJECT';
+  action: 'FLAG' | 'AUTO_APPROVE' | 'SKIP' | 'REJECT' | 'P1';
   priority: number;
 }
 
 export interface MatchResult {
   tags: string[];
-  effectiveAction: 'FLAG' | 'AUTO_APPROVE' | 'SKIP' | 'REJECT' | null;
+  effectiveAction: 'FLAG' | 'AUTO_APPROVE' | 'SKIP' | 'REJECT' | 'P1' | null;
 }
 
 /**
@@ -34,10 +34,11 @@ export function matchContentRules(
 
   let effectiveAction: MatchResult['effectiveAction'] = null;
 
-  // Action precedence: SKIP > REJECT > AUTO_APPROVE > FLAG
+  // Action precedence: SKIP > REJECT > P1 > AUTO_APPROVE > FLAG
   const actionRank: Record<string, number> = {
-    SKIP: 4,
-    REJECT: 3,
+    SKIP: 5,
+    REJECT: 4,
+    P1: 3,
     AUTO_APPROVE: 2,
     FLAG: 1,
   };
@@ -91,7 +92,7 @@ function stripVariationSelectors(s: string): string {
 export function matchReactionRules(
   emoji: string,
   rules: TenantRuleRow[],
-): { action: 'FLAG' | 'AUTO_APPROVE' | 'SKIP' | 'REJECT' } | null {
+): { action: 'FLAG' | 'AUTO_APPROVE' | 'SKIP' | 'REJECT' | 'P1' } | null {
   const reactionRules = rules.filter((r) => r.matchType === 'REACTION_EMOJI');
   const cleanEmoji = stripVariationSelectors(emoji);