From ff727226b1e1df6edf5bd336c7cf81891fd157fe Mon Sep 17 00:00:00 2001 From: KrishnaSSH Date: Mon, 7 Sep 2026 02:14:42 +0530 Subject: [PATCH] bot ingest: R2 + KB pipeline for uploads, live-step /ingest/stream - /api/bots/ingest: a linked user's PDFs are fetched once, replied to synchronously as JSON, and pushed in the background through queueKbIngest (chunk -> embeddings -> Pinecone memories) + queueExtraction (PDF -> JSON -> draft invoice), each writing its own D1 rows. Response gains { kb: { queued } }. - /api/bots/ingest/stream: SSE 'step' events as the finance agent calls tools, then a 'done' event with the answer, so bots can show progress instead of a spinner. Writes a bot_events audit row. - rename 0014_bot_link_compat.sql -> 0016 (0014 collided with 0014_org_currency) --- ...nk_compat.sql => 0016_bot_link_compat.sql} | 0 workers/bot-compat.ts | 142 +++++++++++++++++- 2 files changed, 140 insertions(+), 2 deletions(-) rename migrations/{0014_bot_link_compat.sql => 0016_bot_link_compat.sql} (100%) diff --git a/migrations/0014_bot_link_compat.sql b/migrations/0016_bot_link_compat.sql similarity index 100% rename from migrations/0014_bot_link_compat.sql rename to migrations/0016_bot_link_compat.sql diff --git a/workers/bot-compat.ts b/workers/bot-compat.ts index 06628d0..46b79f9 100644 --- a/workers/bot-compat.ts +++ b/workers/bot-compat.ts @@ -10,7 +10,18 @@ */ import { Hono } from "hono"; import type { MiddlewareHandler } from "hono"; -import { handleInbound, type InboundFile } from "./bot-router"; +import { streamSSE } from "hono/streaming"; +import { askAgent, handleInbound, type InboundFile } from "./bot-router"; +import { createFinanceAgent } from "~/lib/agent.server"; +import { queueKbIngest } from "~/lib/kb.server"; +import { queueExtraction } from "~/lib/docs.server"; + +/** kb.server / docs.server need the R2 + queue bindings; they exist at runtime + * (wrangler.jsonc) but the checked-in worker-configuration.d.ts lags. */ +type PipelineEnv = Env & { KB_QUEUE: Queue; DOCS_BUCKET: R2Bucket }; + +const isPdf = (f: { name?: string; mime?: string }) => + /\.pdf$/i.test(f.name ?? "") || f.mime === "application/pdf"; const SOURCES = ["telegram", "slack", "discord"] as const; type Source = (typeof SOURCES)[number]; @@ -80,6 +91,38 @@ botCompatRoutes.post("/ingest", async (c) => { mime: f.mime, })); + // Documents from a linked user: fetch the bytes once, and in the background + // push them to R2 + the KB pipeline (chunk → embeddings → Pinecone memories, + // and PDF → structured JSON → draft invoice, each writing its own D1 rows). + // We reuse the fetched bytes for the synchronous pdf-to-json reply below. + const kbQueued: string[] = []; + if (link?.org_id && files.some(isPdf)) { + const env = c.env as PipelineEnv; + for (const f of files) { + if (!isPdf(f) || !f.url) continue; + try { + const res = await fetch(f.url); + if (!res.ok) continue; + const bytes = await res.arrayBuffer(); + f.blob = new Blob([bytes], { type: "application/pdf" }); // reuse for the sync reply + f.url = undefined; + kbQueued.push(f.name); + c.executionCtx.waitUntil( + Promise.allSettled([ + queueKbIngest(env, link.org_id, f.name, bytes), + queueExtraction(env, link.org_id, f.name, bytes), + ]).then((r) => { + for (const x of r) + if (x.status === "rejected") + console.error(`bot ingest pipeline (${f.name}): ${x.reason}`); + }), + ); + } catch (err) { + console.error(`bot ingest fetch (${f.name}): ${err}`); + } + } + } + const result = await handleInbound(c.env, { source: b.source, orgId: link?.org_id ?? null, @@ -89,7 +132,102 @@ botCompatRoutes.post("/ingest", async (c) => { files, }); - return c.json(result); + return c.json(kbQueued.length ? { ...result, kb: { queued: kbQueued } } : result); +}); + +/* ─────────────────────────────────────────────── streaming ask (live steps) */ + +type LinkRow = { user_id: string; org_id: string }; + +async function resolveLink( + db: D1Database, + source: Source, + externalUser: string | null | undefined, +): Promise { + if (!externalUser) return null; + return db + .prepare("SELECT user_id, org_id FROM bot_links WHERE source = ? AND external_user = ?") + .bind(source, externalUser) + .first(); +} + +/** + * Like /ingest but text-only and Server-Sent-Events: emits `step` events as the + * agent calls tools, then a final `done` event with the answer. Bots edit their + * message on each event so the user watches the work happen. + */ +botCompatRoutes.post("/ingest/stream", tokenGate, async (c) => { + const b = await c.req.json().catch(() => null); + if (!b || !isSource(b.source) || !b.text?.trim()) + return c.json({ error: "source and text are required" }, 400); + const source = b.source; + const text = b.text.trim(); + const link = await resolveLink(c.env.DB, source, b.externalUser); + + const eventId = crypto.randomUUID(); + await c.env.DB.prepare( + `INSERT INTO bot_events (id, org_id, source, external_user, kind, summary, route, status) + VALUES (?, ?, ?, ?, 'message', ?, 'ask', 'routed')`, + ) + .bind(eventId, link?.org_id ?? null, source, b.externalUser ?? null, text.slice(0, 140)) + .run(); + + const settle = (status: "done" | "error", detail: unknown) => + c.env.DB.prepare( + "UPDATE bot_events SET status = ?, detail = ?, updated_at = unixepoch() WHERE id = ?", + ) + .bind(status, JSON.stringify(detail), eventId) + .run(); + + return streamSSE(c, async (stream) => { + const send = (event: string, data: unknown) => + stream.writeSSE({ event, data: JSON.stringify(data) }); + + // Not linked → no org/user context, so just the lightweight fallback. + if (!link?.org_id || !link.user_id) { + const { reply } = await askAgent(c.env, { + source, + externalUser: b.externalUser ?? null, + text, + files: [], + }); + await send("done", { text: reply, linked: false }); + await settle("done", { reply }); + return; + } + + try { + await send("step", { kind: "start" }); + const agent = createFinanceAgent(c.env, { + userId: link.user_id, + orgId: link.org_id, + }); + const res = await agent.stream({ + prompt: `[Message from ${source}] ${text}`.slice(0, 8000), + }); + + let wroteWriting = false; + for await (const part of res.fullStream) { + if (part.type === "tool-call") { + await send("step", { kind: "tool", name: part.toolName }); + } else if (part.type === "text-delta" && !wroteWriting) { + wroteWriting = true; + await send("step", { kind: "writing" }); + } else if (part.type === "error") { + await send("step", { kind: "error", message: String(part.error) }); + } + } + + const answer = + (await res.text).trim() || "I couldn't produce an answer to that."; + await send("done", { text: answer, linked: true }); + await settle("done", { reply: answer }); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + await send("done", { text: `Sorry, that failed: ${message}`, linked: true }); + await settle("error", { error: message }); + } + }); }); botCompatRoutes.post("/link/start", async (c) => {