diff --git a/apps/api/prisma/migrations/20260617270000_add_content_drafts/migration.sql b/apps/api/prisma/migrations/20260617270000_add_content_drafts/migration.sql new file mode 100644 index 0000000..28ed8d3 --- /dev/null +++ b/apps/api/prisma/migrations/20260617270000_add_content_drafts/migration.sql @@ -0,0 +1,38 @@ +-- CreateEnum +CREATE TYPE "DraftType" AS ENUM ('EVENT', 'SEVA_TASK', 'FAQ_ITEM'); + +-- CreateEnum +CREATE TYPE "DraftStatus" AS ENUM ('PENDING_REVIEW', 'APPROVED', 'REJECTED'); + +-- CreateTable +CREATE TABLE "ContentDraft" ( + "id" TEXT NOT NULL, + "tenantId" TEXT NOT NULL, + "type" "DraftType" NOT NULL, + "status" "DraftStatus" NOT NULL DEFAULT 'PENDING_REVIEW', + "sourceMessageId" TEXT NOT NULL, + "proposedJson" JSONB NOT NULL, + "confidence" DOUBLE PRECISION, + "reviewedBy" TEXT, + "reviewedAt" TIMESTAMP(3), + "publishedId" TEXT, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "ContentDraft_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE INDEX "ContentDraft_tenantId_status_idx" ON "ContentDraft"("tenantId", "status"); + +-- CreateIndex +CREATE INDEX "ContentDraft_tenantId_type_status_idx" ON "ContentDraft"("tenantId", "type", "status"); + +-- CreateIndex +CREATE INDEX "ContentDraft_sourceMessageId_idx" ON "ContentDraft"("sourceMessageId"); + +-- AddForeignKey +ALTER TABLE "ContentDraft" ADD CONSTRAINT "ContentDraft_tenantId_fkey" FOREIGN KEY ("tenantId") REFERENCES "Tenant"("id") ON DELETE RESTRICT ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "ContentDraft" ADD CONSTRAINT "ContentDraft_sourceMessageId_fkey" FOREIGN KEY ("sourceMessageId") REFERENCES "Message"("id") ON DELETE RESTRICT ON UPDATE CASCADE; diff --git a/apps/api/prisma/schema.prisma b/apps/api/prisma/schema.prisma index 8b04ca9..f0386c9 100644 --- a/apps/api/prisma/schema.prisma +++ b/apps/api/prisma/schema.prisma @@ -100,6 +100,7 @@ model Tenant { knowledgeItems KnowledgeItem[] galleryAlbums GalleryAlbum[] galleryItems GalleryItem[] + contentDrafts ContentDraft[] } enum AdminRole { @@ -279,7 +280,8 @@ model Message { createdAt DateTime @default(now()) updatedAt DateTime @updatedAt - approval Approval? + approval Approval? + contentDrafts ContentDraft[] @@unique([platform, platformMsgId]) @@index([tenantId]) @@ -837,3 +839,40 @@ model GalleryItem { @@index([albumId]) @@index([tenantId]) } + +// ============================================================================ +// Content Drafts — AI-proposed Event/Seva/FAQ pending admin review +// ============================================================================ + +enum DraftType { + EVENT + SEVA_TASK + FAQ_ITEM +} + +enum DraftStatus { + PENDING_REVIEW + APPROVED + REJECTED +} + +model ContentDraft { + id String @id @default(cuid()) + tenantId String + tenant Tenant @relation(fields: [tenantId], references: [id]) + type DraftType + status DraftStatus @default(PENDING_REVIEW) + sourceMessageId String + sourceMessage Message @relation(fields: [sourceMessageId], references: [id]) + proposedJson Json + confidence Float? + reviewedBy String? + reviewedAt DateTime? + publishedId String? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + @@index([tenantId, status]) + @@index([tenantId, type, status]) + @@index([sourceMessageId]) +} diff --git a/apps/api/src/app.module.ts b/apps/api/src/app.module.ts index 2f6db05..853ba04 100644 --- a/apps/api/src/app.module.ts +++ b/apps/api/src/app.module.ts @@ -23,6 +23,7 @@ import { SevaModule } from './modules/seva/seva.module'; import { CirclesModule } from './modules/circles/circles.module'; import { KnowledgeModule } from './modules/knowledge/knowledge.module'; import { GalleryModule } from './modules/gallery/gallery.module'; +import { DraftsModule } from './modules/drafts/drafts.module'; @Module({ imports: [ @@ -50,6 +51,7 @@ import { GalleryModule } from './modules/gallery/gallery.module'; CirclesModule, KnowledgeModule, GalleryModule, + DraftsModule, ], }) export class AppModule {} diff --git a/apps/api/src/modules/audit/audit.types.ts b/apps/api/src/modules/audit/audit.types.ts index 3ab7f29..771c636 100644 --- a/apps/api/src/modules/audit/audit.types.ts +++ b/apps/api/src/modules/audit/audit.types.ts @@ -46,6 +46,8 @@ export const AuditAction = { SEVA_COMPLETED: 'SEVA_COMPLETED', CIRCLE_CREATED: 'CIRCLE_CREATED', CIRCLE_DELETED: 'CIRCLE_DELETED', + DRAFT_APPROVED: 'DRAFT_APPROVED', + DRAFT_REJECTED: 'DRAFT_REJECTED', } as const; export type AuditActionValue = (typeof AuditAction)[keyof typeof AuditAction]; diff --git a/apps/api/src/modules/drafts/drafts.controller.ts b/apps/api/src/modules/drafts/drafts.controller.ts new file mode 100644 index 0000000..052ae19 --- /dev/null +++ b/apps/api/src/modules/drafts/drafts.controller.ts @@ -0,0 +1,29 @@ +import { Controller, Get, Post, Param, Query, UseGuards, Request } from '@nestjs/common'; +import { DraftsService } from './drafts.service'; +import { JwtAuthGuard } from '../auth/jwt-auth.guard'; +import { DraftStatus, DraftType } from '@prisma/client'; + +@Controller('admin/drafts') +@UseGuards(JwtAuthGuard) +export class DraftsController { + constructor(private readonly service: DraftsService) {} + + @Get() + list( + @Request() req: any, + @Query('type') type?: DraftType, + @Query('status') status?: DraftStatus, + ) { + return this.service.list(req.user.tenantId, type, status ?? 'PENDING_REVIEW'); + } + + @Post(':id/approve') + approve(@Param('id') id: string, @Request() req: any) { + return this.service.approve(id, req.user.tenantId, req.user.sub); + } + + @Post(':id/reject') + reject(@Param('id') id: string, @Request() req: any) { + return this.service.reject(id, req.user.tenantId, req.user.sub); + } +} diff --git a/apps/api/src/modules/drafts/drafts.module.ts b/apps/api/src/modules/drafts/drafts.module.ts new file mode 100644 index 0000000..2c3bdef --- /dev/null +++ b/apps/api/src/modules/drafts/drafts.module.ts @@ -0,0 +1,12 @@ +import { Module } from '@nestjs/common'; +import { DraftsController } from './drafts.controller'; +import { DraftsService } from './drafts.service'; +import { AuthModule } from '../auth/auth.module'; +import { AuditModule } from '../audit/audit.module'; + +@Module({ + imports: [AuthModule, AuditModule], + controllers: [DraftsController], + providers: [DraftsService], +}) +export class DraftsModule {} diff --git a/apps/api/src/modules/drafts/drafts.service.ts b/apps/api/src/modules/drafts/drafts.service.ts new file mode 100644 index 0000000..3199d85 --- /dev/null +++ b/apps/api/src/modules/drafts/drafts.service.ts @@ -0,0 +1,151 @@ +import { Injectable, NotFoundException, BadRequestException } from '@nestjs/common'; +import { PrismaService } from '../../prisma/prisma.service'; +import { DraftStatus, DraftType } from '@prisma/client'; +import { AuditService } from '../audit/audit.service'; +import { AuditAction } from '../audit/audit.types'; + +export interface ApproveDraftResult { + publishedId: string; + type: DraftType; +} + +@Injectable() +export class DraftsService { + constructor( + private readonly prisma: PrismaService, + private readonly audit: AuditService, + ) {} + + list(tenantId: string, type?: DraftType, status: DraftStatus = 'PENDING_REVIEW') { + return this.prisma.contentDraft.findMany({ + where: { tenantId, ...(type ? { type } : {}), status }, + include: { + sourceMessage: { + select: { id: true, content: true, senderName: true, createdAt: true }, + }, + }, + orderBy: { createdAt: 'desc' }, + take: 100, + }); + } + + async approve(draftId: string, tenantId: string, adminId: string): Promise { + const draft = await this.prisma.contentDraft.findUnique({ where: { id: draftId } }); + if (!draft || draft.tenantId !== tenantId) throw new NotFoundException('Draft not found'); + if (draft.status !== 'PENDING_REVIEW') { + throw new BadRequestException('Draft has already been reviewed'); + } + + const proposed = draft.proposedJson as Record; + let publishedId: string; + + if (draft.type === 'EVENT') { + const startsAt = proposed.date + ? new Date(`${proposed.date}T${proposed.time ?? '00:00'}:00`) + : new Date(); + + const event = await this.prisma.event.create({ + data: { + tenantId, + title: proposed.title ?? 'Untitled Event', + description: proposed.description ?? null, + location: proposed.location ?? null, + startsAt, + createdBy: adminId, + isPublished: true, + }, + }); + publishedId = event.id; + + } else if (draft.type === 'SEVA_TASK') { + const startsAt = proposed.date ? new Date(`${proposed.date}T00:00:00`) : undefined; + const seva = await this.prisma.sevaOpportunity.create({ + data: { + tenantId, + title: proposed.title ?? 'Seva Opportunity', + description: proposed.parentEventTitle + ? `Part of: ${proposed.parentEventTitle}` + : (proposed.description ?? null), + slots: proposed.slots ?? null, + startsAt: startsAt ?? null, + createdBy: adminId, + isPublished: true, + }, + }); + publishedId = seva.id; + + } else { + // FAQ_ITEM — find or create a default board + let board = await this.prisma.knowledgeBoard.findFirst({ + where: { tenantId }, + orderBy: { sortOrder: 'asc' }, + }); + if (!board) { + board = await this.prisma.knowledgeBoard.create({ + data: { + tenantId, + name: 'Community FAQs', + slug: 'community-faqs', + }, + }); + } + const item = await this.prisma.knowledgeItem.create({ + data: { + boardId: board.id, + tenantId, + question: proposed.question ?? 'Untitled question', + answer: proposed.answer ?? '', + tags: proposed.tags ?? [], + status: 'PUBLISHED', + createdBy: adminId, + }, + }); + publishedId = item.id; + } + + await this.prisma.contentDraft.update({ + where: { id: draftId }, + data: { + status: 'APPROVED', + reviewedBy: adminId, + reviewedAt: new Date(), + publishedId, + }, + }); + + await this.audit.log({ + tenantId, + actorType: 'ADMIN', + actorId: adminId, + action: AuditAction.DRAFT_APPROVED, + resourceType: 'ContentDraft', + resourceId: draftId, + payload: { type: draft.type, publishedId }, + }); + + return { publishedId, type: draft.type }; + } + + async reject(draftId: string, tenantId: string, adminId: string): Promise { + const draft = await this.prisma.contentDraft.findUnique({ where: { id: draftId } }); + if (!draft || draft.tenantId !== tenantId) throw new NotFoundException('Draft not found'); + if (draft.status !== 'PENDING_REVIEW') { + throw new BadRequestException('Draft has already been reviewed'); + } + + await this.prisma.contentDraft.update({ + where: { id: draftId }, + data: { status: 'REJECTED', reviewedBy: adminId, reviewedAt: new Date() }, + }); + + await this.audit.log({ + tenantId, + actorType: 'ADMIN', + actorId: adminId, + action: AuditAction.DRAFT_REJECTED, + resourceType: 'ContentDraft', + resourceId: draftId, + payload: { type: draft.type }, + }); + } +} diff --git a/apps/web/app/_lib/sidebar.tsx b/apps/web/app/_lib/sidebar.tsx index 14f45a5..8c55f3a 100644 --- a/apps/web/app/_lib/sidebar.tsx +++ b/apps/web/app/_lib/sidebar.tsx @@ -10,6 +10,7 @@ const NAV_LINKS = [ { href: '/search', label: 'Search' }, { href: '/groups', label: 'Groups & Routes' }, { href: '/messages/pending', label: 'Pending messages' }, + { href: '/drafts', label: 'AI Drafts' }, { href: '/settings/rules', label: 'Rules' }, { href: '/settings/bot', label: 'Bot' }, ]; @@ -18,6 +19,7 @@ const SUPER_ADMIN_LINKS = [ { href: '/admin', label: 'Dashboard' }, { href: '/admin/tenants', label: 'Tenants' }, { href: '/admin/bots', label: 'Bot Pool' }, + { href: '/admin/drafts', label: 'AI Drafts' }, ]; const PUBLIC_PATHS = ['/login', '/signup', '/onboard']; diff --git a/apps/web/app/admin/drafts/DraftCard.tsx b/apps/web/app/admin/drafts/DraftCard.tsx new file mode 100644 index 0000000..06201a4 --- /dev/null +++ b/apps/web/app/admin/drafts/DraftCard.tsx @@ -0,0 +1,137 @@ +'use client'; + +import { useState } from 'react'; +import { useRouter } from 'next/navigation'; + +interface SourceMessage { + id: string; + content: string; + senderName: string | null; + createdAt: string; +} + +interface Draft { + id: string; + type: 'EVENT' | 'SEVA_TASK' | 'FAQ_ITEM'; + status: string; + proposedJson: Record; + confidence: number | null; + createdAt: string; + sourceMessage: SourceMessage; +} + +const TYPE_LABEL: Record = { + EVENT: 'Event', + SEVA_TASK: 'Seva Task', + FAQ_ITEM: 'FAQ', +}; + +const TYPE_COLOR: Record = { + EVENT: 'bg-indigo-50 text-indigo-700 border-indigo-200', + SEVA_TASK: 'bg-green-50 text-green-700 border-green-200', + FAQ_ITEM: 'bg-amber-50 text-amber-700 border-amber-200', +}; + +function ProposedPreview({ type, data }: { type: Draft['type']; data: Record }) { + if (type === 'EVENT') { + return ( +
+

Title: {data.title ?? '—'}

+ {data.date &&

Date: {data.date} {data.time ?? ''}

} + {data.location &&

Location: {data.location}

} + {data.description &&

Description: {data.description}

} + {data.rsvpNeeded &&

RSVP required

} + {data.sevaActionItems?.length > 0 && ( +

+{data.sevaActionItems.length} seva task(s) will also be created

+ )} +
+ ); + } + + if (type === 'SEVA_TASK') { + return ( +
+

Title: {data.title ?? '—'}

+ {data.parentEventTitle &&

Event: {data.parentEventTitle}

} + {data.slots &&

Slots: {data.slots}

} + {data.skills?.length > 0 &&

Skills: {data.skills.join(', ')}

} +
+ ); + } + + return ( +
+

Q: {data.question ?? '—'}

+

A: {data.answer ?? '—'}

+ {data.tags?.length > 0 && ( +
+ {data.tags.map((t: string) => ( + {t} + ))} +
+ )} +
+ ); +} + +export function DraftCard({ draft }: { draft: Draft }) { + const router = useRouter(); + const [loading, setLoading] = useState<'approve' | 'reject' | null>(null); + + async function act(action: 'approve' | 'reject') { + setLoading(action); + try { + await fetch(`/api/admin/drafts/${draft.id}/${action}`, { method: 'POST' }); + router.refresh(); + } finally { + setLoading(null); + } + } + + const confidence = draft.confidence != null ? Math.round(draft.confidence * 100) : null; + + return ( +
+
+ + {TYPE_LABEL[draft.type]} + + {confidence != null && ( + {confidence}% confidence + )} +
+ + {/* Source message */} +
+

+ Source message{draft.sourceMessage.senderName ? ` — ${draft.sourceMessage.senderName}` : ''} +

+

{draft.sourceMessage.content}

+
+ + {/* AI proposed content */} +
+

AI proposed

+ +
+ + {/* Actions */} +
+ + +
+
+ ); +} diff --git a/apps/web/app/admin/drafts/page.tsx b/apps/web/app/admin/drafts/page.tsx new file mode 100644 index 0000000..20a484e --- /dev/null +++ b/apps/web/app/admin/drafts/page.tsx @@ -0,0 +1,100 @@ +import { apiFetch } from '../../_lib/api'; +import { redirect } from 'next/navigation'; +import { getToken } from '../../_lib/api'; +import { DraftCard } from './DraftCard'; +import Link from 'next/link'; + +type DraftType = 'EVENT' | 'SEVA_TASK' | 'FAQ_ITEM'; + +interface Draft { + id: string; + type: DraftType; + status: string; + proposedJson: Record; + confidence: number | null; + createdAt: string; + sourceMessage: { + id: string; + content: string; + senderName: string | null; + createdAt: string; + }; +} + +const TABS: { label: string; value: DraftType | 'ALL' }[] = [ + { label: 'All', value: 'ALL' }, + { label: 'Events', value: 'EVENT' }, + { label: 'Seva Tasks', value: 'SEVA_TASK' }, + { label: 'FAQs', value: 'FAQ_ITEM' }, +]; + +async function fetchDrafts(type?: DraftType): Promise { + const query = new URLSearchParams({ status: 'PENDING_REVIEW' }); + if (type) query.set('type', type); + const res = await apiFetch(`/admin/drafts?${query}`); + if (!res.ok) return []; + return (await res.json()) as Draft[]; +} + +export default async function DraftsPage({ + searchParams, +}: { + searchParams: Promise<{ type?: string }>; +}) { + const token = await getToken(); + if (!token) redirect('/login'); + + const { type } = await searchParams; + const activeType = (type as DraftType) ?? undefined; + const drafts = await fetchDrafts(activeType); + + return ( +
+
+
+

AI Content Drafts

+

+ Events, seva tasks, and FAQs auto-detected from WhatsApp messages +

+
+ ← Admin +
+ + {/* Type filter tabs */} +
+ {TABS.map((tab) => { + const href = tab.value === 'ALL' ? '/admin/drafts' : `/admin/drafts?type=${tab.value}`; + const isActive = (tab.value === 'ALL' && !activeType) || tab.value === activeType; + return ( + + {tab.label} + + ); + })} +
+ + {drafts.length === 0 ? ( +
+

No pending drafts

+

+ Drafts appear here when the AI detects events, volunteer asks, or FAQ patterns in group messages. +

+
+ ) : ( +
+ {drafts.map((d) => ( + + ))} +
+ )} +
+ ); +} diff --git a/apps/web/app/admin/page.tsx b/apps/web/app/admin/page.tsx index aa63bce..647d42b 100644 --- a/apps/web/app/admin/page.tsx +++ b/apps/web/app/admin/page.tsx @@ -49,9 +49,10 @@ export default function AdminDashboard() {
{totalBots ? Math.round(totalTenants / totalBots) : 0}
-
+
Manage Tenants Manage Bots + AI Content Drafts
); diff --git a/apps/web/app/api/admin/drafts/[id]/approve/route.ts b/apps/web/app/api/admin/drafts/[id]/approve/route.ts new file mode 100644 index 0000000..37ec049 --- /dev/null +++ b/apps/web/app/api/admin/drafts/[id]/approve/route.ts @@ -0,0 +1,13 @@ +import { jsonResponse, apiFetch } from '../../../../../_lib/api'; + +export const dynamic = 'force-dynamic'; + +export async function POST( + _request: Request, + { params }: { params: Promise<{ id: string }> }, +): Promise { + const { id } = await params; + const res = await apiFetch(`/admin/drafts/${id}/approve`, { method: 'POST' }); + const body = await res.json(); + return jsonResponse(body, res.status); +} diff --git a/apps/web/app/api/admin/drafts/[id]/reject/route.ts b/apps/web/app/api/admin/drafts/[id]/reject/route.ts new file mode 100644 index 0000000..43eeb28 --- /dev/null +++ b/apps/web/app/api/admin/drafts/[id]/reject/route.ts @@ -0,0 +1,13 @@ +import { jsonResponse, apiFetch } from '../../../../../_lib/api'; + +export const dynamic = 'force-dynamic'; + +export async function POST( + _request: Request, + { params }: { params: Promise<{ id: string }> }, +): Promise { + const { id } = await params; + const res = await apiFetch(`/admin/drafts/${id}/reject`, { method: 'POST' }); + const body = await res.json(); + return jsonResponse(body, res.status); +} diff --git a/apps/web/app/api/admin/drafts/route.ts b/apps/web/app/api/admin/drafts/route.ts new file mode 100644 index 0000000..22e720e --- /dev/null +++ b/apps/web/app/api/admin/drafts/route.ts @@ -0,0 +1,14 @@ +import { jsonResponse, apiFetch } from '../../../_lib/api'; + +export const dynamic = 'force-dynamic'; + +export async function GET(request: Request): Promise { + const { searchParams } = new URL(request.url); + const type = searchParams.get('type'); + const status = searchParams.get('status') ?? 'PENDING_REVIEW'; + const query = new URLSearchParams({ status }); + if (type) query.set('type', type); + const res = await apiFetch(`/admin/drafts?${query}`); + const body = await res.json(); + return jsonResponse(body, res.status); +} diff --git a/apps/web/app/drafts/page.tsx b/apps/web/app/drafts/page.tsx new file mode 100644 index 0000000..0a510a2 --- /dev/null +++ b/apps/web/app/drafts/page.tsx @@ -0,0 +1,98 @@ +import { apiFetch } from '../_lib/api'; +import { redirect } from 'next/navigation'; +import { getToken } from '../_lib/api'; +import { DraftCard } from '../admin/drafts/DraftCard'; +import Link from 'next/link'; + +type DraftType = 'EVENT' | 'SEVA_TASK' | 'FAQ_ITEM'; + +interface Draft { + id: string; + type: DraftType; + status: string; + proposedJson: Record; + confidence: number | null; + createdAt: string; + sourceMessage: { + id: string; + content: string; + senderName: string | null; + createdAt: string; + }; +} + +const TABS: { label: string; value: DraftType | 'ALL' }[] = [ + { label: 'All', value: 'ALL' }, + { label: 'Events', value: 'EVENT' }, + { label: 'Seva Tasks', value: 'SEVA_TASK' }, + { label: 'FAQs', value: 'FAQ_ITEM' }, +]; + +async function fetchDrafts(type?: DraftType): Promise { + const query = new URLSearchParams({ status: 'PENDING_REVIEW' }); + if (type) query.set('type', type); + const res = await apiFetch(`/admin/drafts?${query}`); + if (!res.ok) return []; + return (await res.json()) as Draft[]; +} + +export default async function TenantDraftsPage({ + searchParams, +}: { + searchParams: Promise<{ type?: string }>; +}) { + const token = await getToken(); + if (!token) redirect('/login'); + + const { type } = await searchParams; + const activeType = (type as DraftType) ?? undefined; + const drafts = await fetchDrafts(activeType); + + return ( +
+
+
+

AI Content Drafts

+

+ Events, seva tasks, and FAQs auto-detected from your group messages +

+
+
+ +
+ {TABS.map((tab) => { + const href = tab.value === 'ALL' ? '/drafts' : `/drafts?type=${tab.value}`; + const isActive = (tab.value === 'ALL' && !activeType) || tab.value === activeType; + return ( + + {tab.label} + + ); + })} +
+ + {drafts.length === 0 ? ( +
+

No pending drafts

+

+ Drafts appear here when the AI detects events, volunteer asks, or FAQ patterns in your group messages. +

+
+ ) : ( +
+ {drafts.map((d) => ( + + ))} +
+ )} +
+ ); +} diff --git a/apps/worker/src/expire/expire-scheduler.ts b/apps/worker/src/expire/expire-scheduler.ts index 9124081..329f133 100644 --- a/apps/worker/src/expire/expire-scheduler.ts +++ b/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 = createLogger('expire-scheduler')): void { let running = false; const tick = async () => { diff --git a/apps/worker/src/main.ts b/apps/worker/src/main.ts index dc8352b..11441b0 100644 --- a/apps/worker/src/main.ts +++ b/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>(); @@ -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); }; diff --git a/apps/worker/src/queues/classify.processor.ts b/apps/worker/src/queues/classify.processor.ts index cd9397e..f6a46c7 100644 --- a/apps/worker/src/queues/classify.processor.ts +++ b/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, dncQueue: Queue, spamQueue: Queue, + eventExtractQueue: Queue, + faqCandidateQueue: Queue, ): Promise { 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, + faqCandidateQueue: Queue, +): Promise { + 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, dncQueue: Queue, spamQueue: Queue, + eventExtractQueue: Queue, + faqCandidateQueue: Queue, ): Worker { return new Worker( '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 }, ); diff --git a/apps/worker/src/queues/event-extract.processor.ts b/apps/worker/src/queues/event-extract.processor.ts new file mode 100644 index 0000000..a9b6ed7 --- /dev/null +++ b/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 { + 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 { + return new Worker( + '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 }, + ); +} diff --git a/apps/worker/src/queues/event-extract.queue.ts b/apps/worker/src/queues/event-extract.queue.ts new file mode 100644 index 0000000..091a365 --- /dev/null +++ b/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 { + return new Queue('tower.event.extract', { connection: parseRedisUrl(redisUrl) }); +} diff --git a/apps/worker/src/queues/faq-candidate.queue.ts b/apps/worker/src/queues/faq-candidate.queue.ts new file mode 100644 index 0000000..7c57456 --- /dev/null +++ b/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 { + return new Queue('tower.faq.candidate', { connection: parseRedisUrl(redisUrl) }); +} diff --git a/apps/worker/src/queues/faq-curator.processor.ts b/apps/worker/src/queues/faq-curator.processor.ts new file mode 100644 index 0000000..9b2e189 --- /dev/null +++ b/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 { + 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 { + return new Worker( + '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 }, + ); +} diff --git a/apps/worker/src/whatsapp/match-rules.ts b/apps/worker/src/whatsapp/match-rules.ts index 8866ca4..d4a3577 100644 --- a/apps/worker/src/whatsapp/match-rules.ts +++ b/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 = { - 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); diff --git a/packages/types/src/message.ts b/packages/types/src/message.ts index 9b2c999..bf0f103 100644 --- a/packages/types/src/message.ts +++ b/packages/types/src/message.ts @@ -17,6 +17,7 @@ export interface NormalizedMessage { senderName?: string; content: string; accountId: string; // which bot account received this message + quotedPlatformMsgId?: string; } // Platform-neutral normalized reaction (e.g. WhatsApp emoji reaction) @@ -68,7 +69,7 @@ export interface IngestJobData { senderName?: string; content: string; tags: string[]; - effectiveAction?: 'FLAG' | 'AUTO_APPROVE' | 'SKIP' | 'REJECT'; + effectiveAction?: 'FLAG' | 'AUTO_APPROVE' | 'SKIP' | 'REJECT' | 'P1'; } export interface ForwardJobData { @@ -160,3 +161,17 @@ export interface SpamJobData { contentHash: string; traceId?: string; } + +export interface EventExtractJobData { + tenantId: string; + messageId: string; + content: string; + traceId?: string; +} + +export interface FaqCandidateJobData { + tenantId: string; + messageId: string; + content: string; + traceId?: string; +}