From 530f81416a04e03db9124d29d094dedd2f119617 Mon Sep 17 00:00:00 2001 From: maaz519 Date: Sat, 20 Jun 2026 13:06:17 +0530 Subject: [PATCH] feat: AI dev stream over HTTPS (SSE read + draft writeback) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the Redis-over-Tailscale handoff with a far simpler HTTP surface on the existing public API — no Redis client, Tailscale, or firewall changes needed on the dev side. - GET /ai/stream — SSE feed of every ingested message, reads the internal Redis stream; resumes from Last-Event-ID on reconnect (no missed messages); heartbeats keep the connection warm through proxies - POST /ai/drafts — writeback: creates a ContentDraft (PENDING_REVIEW) that shows up in the admin /admin/drafts review UI - AiStreamGuard — static bearer token auth (AI_STREAM_TOKEN), accepts header or ?token= query for browser EventSource - Remove Tailscale sidecar + public Redis port; Redis stays internal-only Co-Authored-By: Claude Sonnet 4.6 --- apps/api/src/app.module.ts | 2 + .../modules/ai-stream/ai-stream.controller.ts | 40 +++++ .../src/modules/ai-stream/ai-stream.guard.ts | 43 +++++ .../src/modules/ai-stream/ai-stream.module.ts | 9 ++ .../modules/ai-stream/ai-stream.service.ts | 150 ++++++++++++++++++ apps/api/src/modules/ai-stream/dto.ts | 24 +++ docker-compose.vps.yml | 24 +-- packages/config/src/index.ts | 1 + 8 files changed, 270 insertions(+), 23 deletions(-) create mode 100644 apps/api/src/modules/ai-stream/ai-stream.controller.ts create mode 100644 apps/api/src/modules/ai-stream/ai-stream.guard.ts create mode 100644 apps/api/src/modules/ai-stream/ai-stream.module.ts create mode 100644 apps/api/src/modules/ai-stream/ai-stream.service.ts create mode 100644 apps/api/src/modules/ai-stream/dto.ts diff --git a/apps/api/src/app.module.ts b/apps/api/src/app.module.ts index 853ba04..2185d5c 100644 --- a/apps/api/src/app.module.ts +++ b/apps/api/src/app.module.ts @@ -24,6 +24,7 @@ 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'; +import { AiStreamModule } from './modules/ai-stream/ai-stream.module'; @Module({ imports: [ @@ -52,6 +53,7 @@ import { DraftsModule } from './modules/drafts/drafts.module'; KnowledgeModule, GalleryModule, DraftsModule, + AiStreamModule, ], }) export class AppModule {} diff --git a/apps/api/src/modules/ai-stream/ai-stream.controller.ts b/apps/api/src/modules/ai-stream/ai-stream.controller.ts new file mode 100644 index 0000000..7bfc454 --- /dev/null +++ b/apps/api/src/modules/ai-stream/ai-stream.controller.ts @@ -0,0 +1,40 @@ +import { Controller, Get, Post, Body, Query, Headers, Sse, UseGuards, MessageEvent } from '@nestjs/common'; +import { Observable } from 'rxjs'; +import { Public } from '../auth/public.decorator'; +import { AiStreamGuard } from './ai-stream.guard'; +import { AiStreamService } from './ai-stream.service'; +import { CreateDraftDto } from './dto'; + +/** + * External AI-developer integration surface. + * + * GET /ai/stream — SSE feed of every ingested message (read) + * POST /ai/drafts — write back extracted content drafts (write) + * + * All endpoints are @Public() to bypass the global JWT guard, then protected by + * AiStreamGuard (static bearer token). Runs over the existing public HTTPS API, + * so no Redis client, Tailscale, or firewall changes are needed on the dev side. + */ +@Controller('ai') +@Public() +@UseGuards(AiStreamGuard) +export class AiStreamController { + constructor(private readonly service: AiStreamService) {} + + @Sse('stream') + stream( + @Query('after') after?: 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); + } + + @Post('drafts') + createDraft(@Body() dto: CreateDraftDto): Promise<{ draftId: string }> { + return this.service.createDraft(dto); + } +} diff --git a/apps/api/src/modules/ai-stream/ai-stream.guard.ts b/apps/api/src/modules/ai-stream/ai-stream.guard.ts new file mode 100644 index 0000000..94c2e78 --- /dev/null +++ b/apps/api/src/modules/ai-stream/ai-stream.guard.ts @@ -0,0 +1,43 @@ +import { CanActivate, ExecutionContext, Injectable, UnauthorizedException } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { timingSafeEqual } from 'crypto'; + +/** + * Guard for the AI-dev stream + writeback endpoints. + * + * Authenticates with a single static bearer token (AI_STREAM_TOKEN), accepted + * either as `Authorization: Bearer ` (POST / header-capable clients) or + * as a `?token=` query param (browser EventSource cannot set headers). + * + * This is a separate auth realm from the admin/member JWT — the AI dev never + * gets a login, just one long-lived token you rotate by changing the env var. + */ +@Injectable() +export class AiStreamGuard implements CanActivate { + constructor(private readonly config: ConfigService) {} + + canActivate(context: ExecutionContext): boolean { + const expected = this.config.get('AI_STREAM_TOKEN'); + if (!expected) { + throw new UnauthorizedException('AI stream is not configured'); + } + + const req = context.switchToHttp().getRequest(); + const header: string | undefined = req.headers?.authorization; + const fromHeader = header?.startsWith('Bearer ') ? header.slice(7) : undefined; + const fromQuery: string | undefined = req.query?.token; + const provided = fromHeader ?? fromQuery; + + if (!provided || !safeEqual(provided, expected)) { + throw new UnauthorizedException('Invalid AI stream token'); + } + return true; + } +} + +function safeEqual(a: string, b: string): boolean { + const ab = Buffer.from(a); + const bb = Buffer.from(b); + if (ab.length !== bb.length) return false; + return timingSafeEqual(ab, bb); +} diff --git a/apps/api/src/modules/ai-stream/ai-stream.module.ts b/apps/api/src/modules/ai-stream/ai-stream.module.ts new file mode 100644 index 0000000..f6824de --- /dev/null +++ b/apps/api/src/modules/ai-stream/ai-stream.module.ts @@ -0,0 +1,9 @@ +import { Module } from '@nestjs/common'; +import { AiStreamController } from './ai-stream.controller'; +import { AiStreamService } from './ai-stream.service'; + +@Module({ + controllers: [AiStreamController], + providers: [AiStreamService], +}) +export class AiStreamModule {} diff --git a/apps/api/src/modules/ai-stream/ai-stream.service.ts b/apps/api/src/modules/ai-stream/ai-stream.service.ts new file mode 100644 index 0000000..7ed4b4d --- /dev/null +++ b/apps/api/src/modules/ai-stream/ai-stream.service.ts @@ -0,0 +1,150 @@ +import { Injectable, NotFoundException, BadRequestException, Logger } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { MessageEvent } from '@nestjs/common'; +import { Observable } from 'rxjs'; +import Redis from 'ioredis'; +import { PrismaService } from '../../prisma/prisma.service'; +import { CreateDraftDto } from './dto'; + +const STREAM_KEY = 'tower:stream:messages'; +// Block this long per XREAD; on timeout we emit a heartbeat to keep the +// connection warm through proxies, then loop again. +const BLOCK_MS = 15000; + +interface StreamRecord { + messageId: string; + tenantId: string; + content: string; + senderJid: string; + senderName: string | null; + sourceGroupId: string; + tags: string[]; + effectiveAction: string | null; + traceId: string | null; + timestamp: string; +} + +@Injectable() +export class AiStreamService { + private readonly logger = new Logger(AiStreamService.name); + private readonly redisUrl: string; + + constructor( + private readonly config: ConfigService, + private readonly prisma: PrismaService, + ) { + this.redisUrl = this.config.get('REDIS_URL', 'redis://localhost:6379'); + } + + /** + * SSE source. Reads the Redis stream from `afterId` and emits one MessageEvent + * per message, with the stream entry id set as the SSE event id so the client's + * Last-Event-ID resumes exactly where it dropped off. + * + * afterId semantics (Redis stream cursor): + * '$' → only messages that arrive after connect (default for a fresh client) + * '0' → replay the entire retained stream (~50k messages) then follow live + * '' → resume strictly after that entry (used on reconnect) + */ + streamMessages(afterId: string): Observable { + return new Observable((subscriber) => { + const redis = new Redis(this.redisUrl, { maxRetriesPerRequest: null }); + let active = true; + let cursor = afterId; + + const loop = async () => { + while (active) { + try { + const res = (await redis.xread( + 'COUNT', 100, + 'BLOCK', BLOCK_MS, + 'STREAMS', STREAM_KEY, cursor, + )) as [string, [string, string[]][]][] | null; + + if (!res) { + // Idle timeout — heartbeat keeps the pipe alive. EventSource.onmessage + // ignores typed events, so this won't reach the dev's message handler. + if (active) subscriber.next({ type: 'heartbeat', data: '' } as MessageEvent); + continue; + } + + for (const [, entries] of res) { + for (const [id, fields] of entries) { + cursor = id; + subscriber.next({ id, data: parseFields(fields) } as MessageEvent); + } + } + } catch (err) { + if (active) subscriber.error(err); + break; + } + } + }; + + loop(); + + return () => { + active = false; + redis.disconnect(); + }; + }); + } + + /** + * Writeback: the AI dev posts an extracted draft. Creates a ContentDraft in + * PENDING_REVIEW status — it then shows up in the admin /admin/drafts UI for + * approve/reject, which publishes the real Event/SevaOpportunity/KnowledgeItem. + */ + async createDraft(dto: CreateDraftDto): Promise<{ draftId: string }> { + const message = await this.prisma.message.findUnique({ + where: { id: dto.sourceMessageId }, + select: { id: true, tenantId: true }, + }); + if (!message) { + throw new NotFoundException('sourceMessageId does not exist'); + } + if (message.tenantId !== dto.tenantId) { + throw new BadRequestException('tenantId does not match the source message'); + } + + const draft = await this.prisma.contentDraft.create({ + data: { + tenantId: dto.tenantId, + type: dto.type, + status: 'PENDING_REVIEW', + sourceMessageId: dto.sourceMessageId, + proposedJson: dto.proposedJson as object, + confidence: dto.confidence ?? null, + }, + select: { id: true }, + }); + + this.logger.log(`AI draft created: ${draft.id} (${dto.type}) from message ${dto.sourceMessageId}`); + return { draftId: draft.id }; + } +} + +function parseFields(fields: string[]): StreamRecord { + const obj: Record = {}; + for (let i = 0; i < fields.length; i += 2) { + obj[fields[i]] = fields[i + 1]; + } + let tags: string[] = []; + try { + tags = obj.tags ? JSON.parse(obj.tags) : []; + } catch { + tags = []; + } + return { + messageId: obj.messageId ?? '', + tenantId: obj.tenantId ?? '', + content: obj.content ?? '', + senderJid: obj.senderJid ?? '', + senderName: obj.senderName || null, + sourceGroupId: obj.sourceGroupId ?? '', + tags, + effectiveAction: obj.effectiveAction || null, + traceId: obj.traceId || null, + timestamp: obj.timestamp ?? '', + }; +} diff --git a/apps/api/src/modules/ai-stream/dto.ts b/apps/api/src/modules/ai-stream/dto.ts new file mode 100644 index 0000000..cc643cc --- /dev/null +++ b/apps/api/src/modules/ai-stream/dto.ts @@ -0,0 +1,24 @@ +import { IsString, IsIn, IsOptional, IsNumber, IsObject, Min, Max } from 'class-validator'; +import { DraftType } from '@prisma/client'; + +const DRAFT_TYPES: DraftType[] = ['EVENT', 'SEVA_TASK', 'FAQ_ITEM']; + +export class CreateDraftDto { + @IsString() + tenantId!: string; + + @IsString() + sourceMessageId!: string; + + @IsIn(DRAFT_TYPES) + type!: DraftType; + + @IsObject() + proposedJson!: Record; + + @IsOptional() + @IsNumber() + @Min(0) + @Max(1) + confidence?: number; +} diff --git a/docker-compose.vps.yml b/docker-compose.vps.yml index fc7e3c3..13a6324 100644 --- a/docker-compose.vps.yml +++ b/docker-compose.vps.yml @@ -32,28 +32,6 @@ services: networks: - dokploy-network - # Tailscale sidecar — shares Redis's network namespace so the Redis port - # becomes reachable on the tailnet as `tower-redis:6379`. Uses only outbound - # connections, so no OCI Security List / host firewall changes are needed. - tailscale-redis: - image: tailscale/tailscale:latest - hostname: tower-redis - environment: - TS_AUTHKEY: ${TS_AUTHKEY} - TS_STATE_DIR: /var/lib/tailscale - TS_USERSPACE: 'false' - volumes: - - tailscale_redis_state:/var/lib/tailscale - devices: - - /dev/net/tun - cap_add: - - NET_ADMIN - depends_on: - redis: - condition: service_healthy - restart: unless-stopped - network_mode: service:redis - meilisearch: image: getmeili/meilisearch:v1.11 ports: @@ -91,6 +69,7 @@ services: NODE_ENV: production LOG_LEVEL: ${LOG_LEVEL:-info} TOWER_PORTAL_BASE_URL: ${TOWER_PORTAL_BASE_URL} + AI_STREAM_TOKEN: ${AI_STREAM_TOKEN} SMTP_HOST: ${SMTP_HOST:-} SMTP_PORT: ${SMTP_PORT:-587} SMTP_SECURE: ${SMTP_SECURE:-false} @@ -146,7 +125,6 @@ volumes: redis_data: meilisearch_data: sessions: - tailscale_redis_state: networks: dokploy-network: diff --git a/packages/config/src/index.ts b/packages/config/src/index.ts index 7276851..b69cdb4 100644 --- a/packages/config/src/index.ts +++ b/packages/config/src/index.ts @@ -15,6 +15,7 @@ const envSchema = z.object({ JWT_EXPIRES_IN: z.string().default('7d'), MEMBER_JWT_EXPIRES_IN: z.string().default('30d'), OPENROUTER_API_KEY: z.string().optional(), + AI_STREAM_TOKEN: z.string().optional(), }); export type Env = z.infer;