/** * Telegram webhook controller (POST /api/webhooks/telegram). * * Steps: * 1. Optionally verify X-Telegram-Bot-Api-Secret-Token header. * 2. Dedupe by provider_event_id in webhook_events (onConflictDoNothing). * 3. Normalize the Telegram update to NormalizedInboundMessage. * 4. Call ConversationsService.ingestInbound. * 5. Emit Inngest events (message.received + evidence.uploaded per item). * * Day 2 is INBOUND ONLY — no messages are sent back (CLAUDE.md #2, #3). */ import { Controller, Headers, HttpCode, HttpException, HttpStatus, Inject, Post, Req } from "@nestjs/common"; import type { FastifyRequest } from "fastify"; import { eq } from "drizzle-orm"; import { type Database, webhookEvents } from "@courtmitra/db"; import type { NormalizedInboundMessage } from "@courtmitra/contracts"; import { DB } from "../db/db.module.js"; import { ConversationsService } from "../conversations/conversations.service.js"; import { WorkflowService } from "../workflow/inngest.module.js"; // ---- Telegram update shape (minimal, only what we need) ------------------- interface TgPhotoSize { file_id: string; file_size?: number; width: number; height: number; } interface TgDocument { file_id: string; mime_type?: string; file_name?: string; } interface TgMessage { message_id: number; chat: { id: number }; from?: { id: number; username?: string; first_name?: string }; text?: string; photo?: TgPhotoSize[]; document?: TgDocument; } interface TgUpdate { update_id: number; message?: TgMessage; } function pickLargestPhoto(photos: TgPhotoSize[]): TgPhotoSize | undefined { return photos.reduce((best, p) => { if (!best) return p; return (p.file_size ?? 0) > (best.file_size ?? 0) ? p : best; }, undefined); } function normalizeTelegramUpdate(update: TgUpdate): NormalizedInboundMessage | null { const msg = update.message; if (!msg) return null; // e.g. edited_message, callback_query — ignore for now const chatId = String(msg.chat.id); const fromId = String(msg.from?.id ?? chatId); const media: NormalizedInboundMessage["media"] = []; if (msg.photo) { const largest = pickLargestPhoto(msg.photo); if (largest) { media.push({ kind: "photo", providerFileId: largest.file_id, mimeType: "image/jpeg", filename: null }); } } if (msg.document) { media.push({ kind: "document", providerFileId: msg.document.file_id, mimeType: msg.document.mime_type ?? null, filename: msg.document.file_name ?? null, }); } return { channel: "telegram", externalThreadId: chatId, providerMessageId: String(msg.message_id), fromSubject: fromId, text: msg.text ?? null, media, rawPayloadRef: update, }; } // --------------------------------------------------------------------------- @Controller("webhooks/telegram") export class TelegramWebhookController { constructor( @Inject(DB) private readonly db: Database, @Inject(ConversationsService) private readonly conversations: ConversationsService, @Inject(WorkflowService) private readonly workflow: WorkflowService, ) {} @Post() @HttpCode(200) async handleUpdate( @Req() req: FastifyRequest, @Headers("x-telegram-bot-api-secret-token") secretToken?: string, ) { // 1. Secret token verification. // Fix 13: in production, reject requests if the secret is unset. const expectedSecret = process.env["TELEGRAM_WEBHOOK_SECRET"]; if (!expectedSecret && process.env["NODE_ENV"] === "production") { throw new HttpException("Forbidden: TELEGRAM_WEBHOOK_SECRET not configured", HttpStatus.FORBIDDEN); } if (expectedSecret && secretToken !== expectedSecret) { throw new HttpException("Forbidden", HttpStatus.FORBIDDEN); } const update = req.body as TgUpdate; if (!update || typeof update.update_id !== "number") { return { ok: true, skipped: "no update_id" }; } const providerEventId = `telegram:${update.update_id}`; // 2. Dedupe insert — if already processed, return early const [inserted] = await this.db .insert(webhookEvents) .values({ provider: "telegram", providerEventId, payload: update as unknown as Record, }) .onConflictDoNothing() .returning(); if (!inserted) { return { ok: true, deduped: true }; } // 3. Normalize const normalized = normalizeTelegramUpdate(update); if (!normalized) { // Mark as processed but nothing to ingest await this.db .update(webhookEvents) .set({ processedAt: new Date() }) .where(eq(webhookEvents.id, inserted.id)); return { ok: true, skipped: "unsupported update type" }; } // 4. Ingest const result = await this.conversations.ingestInbound(normalized, {}); // 5. Emit Inngest events try { // Emit message.received unconditionally — photo-only messages must also trigger intake. await this.workflow.send("message.received", { missionId: result.missionId, conversationId: result.conversationId, messageId: result.messageId, }); for (const evidenceItemId of result.evidenceItemIds) { await this.workflow.send("evidence.uploaded", { missionId: result.missionId, evidenceItemId, }); } } catch (err) { // Inngest dev server may not be running — log but don't fail the webhook console.warn("[telegram webhook] Inngest send failed (non-fatal):", String(err)); } // Mark webhook event as processed await this.db .update(webhookEvents) .set({ processedAt: new Date() }) .where(eq(webhookEvents.id, inserted.id)); return { ok: true }; } }