From e598073dc7e941626265d53e62107df36f9a97a8 Mon Sep 17 00:00:00 2001 From: maaz519 Date: Mon, 22 Jun 2026 01:07:42 +0530 Subject: [PATCH] feat: enrich AI stream with replyToMessageId, sourceGroupName, organizationId + tenant/source filters MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Extract quotedPlatformMsgId from Baileys contextInfo.stanzaId in normalizer - Thread quotedPlatformMsgId through IngestJobData → ingest processor → Message record - Widen ingest processor group/tenant selects to include name and organizationId - Add replyToMessageId, sourceGroupName, organizationId to StreamMessagePayload and XADD - Update StreamRecord, parseFields, and streamMessages in AI stream service with new fields - Add optional ?tenant= and ?source= SSE filters; cursor advances for filtered entries so Last-Event-ID resume remains accurate across reconnects Co-Authored-By: Claude Sonnet 4.6 --- .../modules/ai-stream/ai-stream.controller.ts | 4 ++- .../modules/ai-stream/ai-stream.service.ts | 13 ++++++++-- apps/worker/src/main.ts | 1 + apps/worker/src/queues/ingest.processor.ts | 8 ++++-- apps/worker/src/streams/message-stream.ts | 26 ++++++++++++------- apps/worker/src/whatsapp/normalizer.ts | 1 + packages/types/src/message.ts | 1 + 7 files changed, 39 insertions(+), 15 deletions(-) diff --git a/apps/api/src/modules/ai-stream/ai-stream.controller.ts b/apps/api/src/modules/ai-stream/ai-stream.controller.ts index 7bfc454..6714a6b 100644 --- a/apps/api/src/modules/ai-stream/ai-stream.controller.ts +++ b/apps/api/src/modules/ai-stream/ai-stream.controller.ts @@ -24,13 +24,15 @@ export class AiStreamController { @Sse('stream') stream( @Query('after') after?: string, + @Query('tenant') tenant?: string, + @Query('source') source?: string, @Headers('last-event-id') lastEventId?: string, ): Observable { // Last-Event-ID (set automatically by EventSource on reconnect) wins so the // client resumes exactly where it dropped. Otherwise honour ?after=, default // to '$' = only new messages from now on. const cursor = lastEventId || after || '$'; - return this.service.streamMessages(cursor); + return this.service.streamMessages(cursor, { tenantId: tenant, sourceGroupId: source }); } @Post('drafts') diff --git a/apps/api/src/modules/ai-stream/ai-stream.service.ts b/apps/api/src/modules/ai-stream/ai-stream.service.ts index 7ed4b4d..471a5e5 100644 --- a/apps/api/src/modules/ai-stream/ai-stream.service.ts +++ b/apps/api/src/modules/ai-stream/ai-stream.service.ts @@ -18,6 +18,9 @@ interface StreamRecord { senderJid: string; senderName: string | null; sourceGroupId: string; + sourceGroupName: string | null; + organizationId: string | null; + replyToMessageId: string | null; tags: string[]; effectiveAction: string | null; traceId: string | null; @@ -46,7 +49,7 @@ export class AiStreamService { * '0' → replay the entire retained stream (~50k messages) then follow live * '' → resume strictly after that entry (used on reconnect) */ - streamMessages(afterId: string): Observable { + streamMessages(afterId: string, filters?: { tenantId?: string; sourceGroupId?: string }): Observable { return new Observable((subscriber) => { const redis = new Redis(this.redisUrl, { maxRetriesPerRequest: null }); let active = true; @@ -70,8 +73,11 @@ export class AiStreamService { for (const [, entries] of res) { for (const [id, fields] of entries) { + const record = parseFields(fields); cursor = id; - subscriber.next({ id, data: parseFields(fields) } as MessageEvent); + if (filters?.tenantId && record.tenantId !== filters.tenantId) continue; + if (filters?.sourceGroupId && record.sourceGroupId !== filters.sourceGroupId) continue; + subscriber.next({ id, data: record } as MessageEvent); } } } catch (err) { @@ -142,6 +148,9 @@ function parseFields(fields: string[]): StreamRecord { senderJid: obj.senderJid ?? '', senderName: obj.senderName || null, sourceGroupId: obj.sourceGroupId ?? '', + sourceGroupName: obj.sourceGroupName || null, + organizationId: obj.organizationId || null, + replyToMessageId: obj.replyToMessageId || null, tags, effectiveAction: obj.effectiveAction || null, traceId: obj.traceId || null, diff --git a/apps/worker/src/main.ts b/apps/worker/src/main.ts index 11812c2..9bf0c50 100644 --- a/apps/worker/src/main.ts +++ b/apps/worker/src/main.ts @@ -203,6 +203,7 @@ async function bootstrap() { content: msg.content, tags, effectiveAction: finalAction ?? undefined, + quotedPlatformMsgId: msg.quotedPlatformMsgId, }, { attempts: 3, backoff: { type: 'exponential', delay: 1000 } }, ); diff --git a/apps/worker/src/queues/ingest.processor.ts b/apps/worker/src/queues/ingest.processor.ts index de5f25e..18e5732 100644 --- a/apps/worker/src/queues/ingest.processor.ts +++ b/apps/worker/src/queues/ingest.processor.ts @@ -19,7 +19,7 @@ export async function processIngestJob( // Defensive: drop messages from non-CLAIMED groups const group = await prisma.group.findUnique({ where: { id: job.sourceGroupId }, - select: { claimStatus: true }, + select: { claimStatus: true, name: true }, }); if (!group || group.claimStatus !== 'CLAIMED') { return; @@ -28,7 +28,7 @@ export async function processIngestJob( // Safety net: drop if tenant is inactive or paused const tenant = await prisma.tenant.findUnique({ where: { id: job.tenantId }, - select: { isActive: true, isForwardingPaused: true }, + select: { isActive: true, isForwardingPaused: true, organizationId: true }, }); if (!tenant || !tenant.isActive || tenant.isForwardingPaused) { return; @@ -86,6 +86,7 @@ export async function processIngestJob( content: job.content, tags: job.tags, status: initialStatus, + quotedPlatformMsgId: job.quotedPlatformMsgId, }, update: {}, }); @@ -100,6 +101,9 @@ export async function processIngestJob( senderJid: job.senderJid, senderName: job.senderName, sourceGroupId: job.sourceGroupId, + sourceGroupName: group.name, + organizationId: tenant.organizationId ?? undefined, + replyToMessageId: job.quotedPlatformMsgId, tags: job.tags, effectiveAction: job.effectiveAction ?? null, timestamp: new Date().toISOString(), diff --git a/apps/worker/src/streams/message-stream.ts b/apps/worker/src/streams/message-stream.ts index cd38a5c..a68a0e5 100644 --- a/apps/worker/src/streams/message-stream.ts +++ b/apps/worker/src/streams/message-stream.ts @@ -13,6 +13,9 @@ export interface StreamMessagePayload { senderJid: string; senderName?: string; sourceGroupId: string; + sourceGroupName?: string; + organizationId?: string; + replyToMessageId?: string; tags: string[]; effectiveAction: string | null; traceId?: string; @@ -34,15 +37,18 @@ export async function publishToMessageStream( STREAM_KEY, 'MAXLEN', '~', String(STREAM_MAX_LEN), '*', - 'messageId', payload.messageId, - 'tenantId', payload.tenantId, - 'content', payload.content, - 'senderJid', payload.senderJid, - 'senderName', payload.senderName ?? '', - 'sourceGroupId', payload.sourceGroupId, - 'tags', JSON.stringify(payload.tags), - 'effectiveAction', payload.effectiveAction ?? '', - 'traceId', payload.traceId ?? '', - 'timestamp', payload.timestamp, + 'messageId', payload.messageId, + 'tenantId', payload.tenantId, + 'content', payload.content, + 'senderJid', payload.senderJid, + 'senderName', payload.senderName ?? '', + 'sourceGroupId', payload.sourceGroupId, + 'sourceGroupName', payload.sourceGroupName ?? '', + 'organizationId', payload.organizationId ?? '', + 'replyToMessageId', payload.replyToMessageId ?? '', + 'tags', JSON.stringify(payload.tags), + 'effectiveAction', payload.effectiveAction ?? '', + 'traceId', payload.traceId ?? '', + 'timestamp', payload.timestamp, ); } diff --git a/apps/worker/src/whatsapp/normalizer.ts b/apps/worker/src/whatsapp/normalizer.ts index 57676b9..9503537 100644 --- a/apps/worker/src/whatsapp/normalizer.ts +++ b/apps/worker/src/whatsapp/normalizer.ts @@ -40,6 +40,7 @@ export function normalizeMessage( senderName: msg.pushName ?? undefined, content, accountId, + quotedPlatformMsgId: msg.message?.extendedTextMessage?.contextInfo?.stanzaId ?? undefined, }; } diff --git a/packages/types/src/message.ts b/packages/types/src/message.ts index bf0f103..013adf9 100644 --- a/packages/types/src/message.ts +++ b/packages/types/src/message.ts @@ -70,6 +70,7 @@ export interface IngestJobData { content: string; tags: string[]; effectiveAction?: 'FLAG' | 'AUTO_APPROVE' | 'SKIP' | 'REJECT' | 'P1'; + quotedPlatformMsgId?: string; } export interface ForwardJobData {