From db5c9a02a7499cfaafb40038117f302894dbdac5 Mon Sep 17 00:00:00 2001 From: siddharthd Date: Fri, 21 Aug 2026 12:10:05 +1000 Subject: [PATCH] =?UTF-8?q?intake:=20POST=20/api/intake=20router=20+=20PUT?= =?UTF-8?q?=20/api/transactions/[id]/line-items=20(plan=20=C2=A711=20C1,?= =?UTF-8?q?=20C2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One upload point, every applicable step. Persist the batch, classify each file, ONE LiteLLM call (cloud-budget, temperature 0, strict json_schema; PDFs as unpdf text, never pixels) with an order-item prompt that keeps names verbatim, header (merchant, date, total), merchant pattern (coles|woolworths|woolies|aldi|iga|foodworks|costco) -> grocery | food | ask, exact (platform, merchant, date, total) match against transactions + expense_metadata, then act: - food + unique match -> replace line_items via the C2 path, 'enriched' - food + miss/ambiguity -> 'candidates' (same merchant ±3d) with intake_use_candidate / intake_create buttons; never auto-creates - grocery -> original files to pantry POST /api/capture with finance_target_transaction_id + finance_action enrich|create (C3); 409 -> 'duplicate' - ask / needs_header / held / duplicate / unsupported as replies, not writes Reconciliation is against the printed item subtotal (a delivery total is items + fees - promotions: 18.94 of Coles lines settled at 5.74), or 2%, else 'held'. Originals are kept under INTAKE_FILE_DIR by sha256 so a later 'Groceries' tap can still forward them. C2 is an idempotent replace on transaction_id = :id OR matched_transaction_id = :id, accepts INTAKE_TOKEN or RECEIPT_INGEST_TOKEN, creates the expense_metadata row when absent, 404 when the transaction does not exist, returns {previous_line_items, line_items} so pantry's undo is the same call with the previous list. --- package-lock.json | 20 +- package.json | 3 +- src/app/api/intake/route.ts | 67 +++ .../api/transactions/[id]/line-items/route.ts | 58 ++ src/lib/intake-actions.ts | 118 ++++ src/lib/intake-blocks.ts | 143 +++++ src/lib/intake-extract.ts | 199 +++++++ src/lib/intake-files.ts | 51 ++ src/lib/intake-pantry.ts | 76 +++ src/lib/intake-router.ts | 505 ++++++++++++++++++ src/lib/intake-rules.ts | 228 ++++++++ src/lib/line-items.ts | 124 +++++ 12 files changed, 1590 insertions(+), 2 deletions(-) create mode 100644 src/app/api/intake/route.ts create mode 100644 src/app/api/transactions/[id]/line-items/route.ts create mode 100644 src/lib/intake-actions.ts create mode 100644 src/lib/intake-blocks.ts create mode 100644 src/lib/intake-extract.ts create mode 100644 src/lib/intake-files.ts create mode 100644 src/lib/intake-pantry.ts create mode 100644 src/lib/intake-router.ts create mode 100644 src/lib/intake-rules.ts create mode 100644 src/lib/line-items.ts diff --git a/package-lock.json b/package-lock.json index 78864e1..6aaf2a8 100644 --- a/package-lock.json +++ b/package-lock.json @@ -16,7 +16,8 @@ "prisma": "^7.4.2", "react": "19.2.3", "react-dom": "19.2.3", - "recharts": "^3.8.0" + "recharts": "^3.8.0", + "unpdf": "^1.8.1" }, "devDependencies": { "@tailwindcss/postcss": "^4", @@ -8494,6 +8495,23 @@ "dev": true, "license": "MIT" }, + "node_modules/unpdf": { + "version": "1.8.1", + "resolved": "https://registry.npmjs.org/unpdf/-/unpdf-1.8.1.tgz", + "integrity": "sha512-xkURhy2SoGpOIH0a1gLHNkASPIQYonadDJs2AQwPEfUakafeD9EA1WTWWsaR++gfTCXJpV27W7tU1nXuk82UKQ==", + "license": "MIT", + "engines": { + "node": ">=22" + }, + "peerDependencies": { + "@napi-rs/canvas": "^0.1.69 || ^1.0.0" + }, + "peerDependenciesMeta": { + "@napi-rs/canvas": { + "optional": true + } + } + }, "node_modules/unrs-resolver": { "version": "1.11.1", "resolved": "https://registry.npmjs.org/unrs-resolver/-/unrs-resolver-1.11.1.tgz", diff --git a/package.json b/package.json index 66546e0..3cc35ec 100644 --- a/package.json +++ b/package.json @@ -23,7 +23,8 @@ "prisma": "^7.4.2", "react": "19.2.3", "react-dom": "19.2.3", - "recharts": "^3.8.0" + "recharts": "^3.8.0", + "unpdf": "^1.8.1" }, "devDependencies": { "@tailwindcss/postcss": "^4", diff --git a/src/app/api/intake/route.ts b/src/app/api/intake/route.ts new file mode 100644 index 0000000..b40b2d4 --- /dev/null +++ b/src/app/api/intake/route.ts @@ -0,0 +1,67 @@ +import { NextRequest, NextResponse } from "next/server"; +import { sha256Of, type StoredFile } from "@/lib/intake-files"; +import { liteLLMExtractor } from "@/lib/intake-extract"; +import { pantryForwarder } from "@/lib/intake-pantry"; +import { isIntakeSource } from "@/lib/intake-rules"; +import { MAX_FILES, runIntake } from "@/lib/intake-router"; + +/** + * The one upload point (plan §11 C1). n8n forwards the iOS Shortcut's + * multipart POST here: `source`, `shared_by`, `batch_id`, and 1–6 files under + * `file`. The router reads what it is given and does every step that applies. + * + * Auth is a shared secret, like the two ingest siblings: this is machine-to- + * machine and there is no browser session to forward. Its own token so the + * Shortcut's credential rotates without touching the order or receipt lanes. + * Fails closed when unset. + */ +function authorised(req: NextRequest): boolean { + const expected = process.env.INTAKE_TOKEN; + if (!expected) return false; + const got = req.headers.get("x-ingest-token"); + return !!got && got === expected; +} + +const MAX_BYTES = 12_000_000; + +export async function POST(req: NextRequest) { + if (!authorised(req)) return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + + let form: FormData; + try { + form = await req.formData(); + } catch { + return NextResponse.json({ error: "multipart/form-data body required" }, { status: 400 }); + } + + const source = form.get("source"); + if (!isIntakeSource(source)) { + return NextResponse.json({ error: "source must be one of uber_eats|doordash|coles|woolworths|other" }, { status: 400 }); + } + const sharedBy = typeof form.get("shared_by") === "string" ? String(form.get("shared_by")).trim() || null : null; + const batchIdRaw = form.get("batch_id"); + const batchId = typeof batchIdRaw === "string" && /^[0-9a-f-]{36}$/i.test(batchIdRaw) ? batchIdRaw.toLowerCase() : null; + if (batchIdRaw && !batchId) return NextResponse.json({ error: "batch_id must be a uuid" }, { status: 400 }); + + const files: StoredFile[] = []; + for (const entry of form.getAll("file")) { + if (!(entry instanceof File) || !entry.size) continue; + if (entry.size > MAX_BYTES) return NextResponse.json({ error: `${entry.name} is larger than 12 MB` }, { status: 400 }); + const bytes = Buffer.from(await entry.arrayBuffer()); + files.push({ name: entry.name || `file-${files.length + 1}`, type: entry.type || "", bytes, sha256: sha256Of(bytes) }); + } + if (!files.length) return NextResponse.json({ error: "at least one file is required under field 'file'" }, { status: 400 }); + if (files.length > MAX_FILES) return NextResponse.json({ error: `at most ${MAX_FILES} files per batch` }, { status: 400 }); + + try { + const result = await runIntake( + { batchId, source, sharedBy, files }, + { extractor: liteLLMExtractor(), pantry: pantryForwarder() } + ); + const { http_status, ...body } = result; + return NextResponse.json(body, { status: http_status }); + } catch (e) { + const message = e instanceof Error ? e.message : String(e); + return NextResponse.json({ error: message }, { status: 500 }); + } +} diff --git a/src/app/api/transactions/[id]/line-items/route.ts b/src/app/api/transactions/[id]/line-items/route.ts new file mode 100644 index 0000000..90adb7c --- /dev/null +++ b/src/app/api/transactions/[id]/line-items/route.ts @@ -0,0 +1,58 @@ +import { NextRequest, NextResponse } from "next/server"; +import { LineItemsValidationError, replaceLineItems, TransactionNotFoundError, validateLineItems } from "@/lib/line-items"; + +/** + * Enrich endpoint (plan §11 C2): idempotent replace of the itemised receipt + * behind a transaction. + * + * Accepts either machine token. Pantry already holds RECEIPT_INGEST_TOKEN for + * the ING-10 lane and calls this on every confirm and on undo; the intake + * router holds INTAKE_TOKEN. No session auth — there is no browser here. + * Fails closed when neither is configured. + * + * Never creates a transaction: 404 when the id does not exist. Creates the + * `expense_metadata` row when the transaction has none. + */ +function authorised(req: NextRequest): boolean { + const got = req.headers.get("x-ingest-token"); + if (!got) return false; + const accepted = [process.env.INTAKE_TOKEN, process.env.RECEIPT_INGEST_TOKEN].filter(Boolean); + return accepted.includes(got); +} + +export async function PUT(req: NextRequest, { params }: { params: Promise<{ id: string }> }) { + if (!authorised(req)) return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + + const { id } = await params; + const transactionId = Number(id); + if (!Number.isInteger(transactionId) || transactionId <= 0) return NextResponse.json({ error: "invalid transaction id" }, { status: 400 }); + + let body: { line_items?: unknown; source?: unknown; batch_id?: unknown; replace?: unknown; platform?: unknown }; + try { + body = await req.json(); + } catch { + return NextResponse.json({ error: "invalid JSON" }, { status: 400 }); + } + + if (body.replace !== true) return NextResponse.json({ error: "replace: true is required — this endpoint only replaces" }, { status: 400 }); + if (body.source !== "intake" && body.source !== "pantry") return NextResponse.json({ error: "source must be 'intake' or 'pantry'" }, { status: 400 }); + const batchId = typeof body.batch_id === "string" && body.batch_id ? body.batch_id : null; + const platform = typeof body.platform === "string" && body.platform ? body.platform : null; + + try { + const lineItems = validateLineItems(body.line_items); + const result = await replaceLineItems(transactionId, lineItems, { source: body.source, platform, batchId }); + return NextResponse.json({ + transaction_id: result.transaction_id, + metadata_id: result.metadata_id, + created_metadata: result.created_metadata, + previous_line_items: result.previous_line_items, + line_items: result.line_items, + }); + } catch (e) { + if (e instanceof LineItemsValidationError) return NextResponse.json({ error: e.message }, { status: 400 }); + if (e instanceof TransactionNotFoundError) return NextResponse.json({ error: e.message }, { status: 404 }); + const message = e instanceof Error ? e.message : String(e); + return NextResponse.json({ error: message }, { status: 500 }); + } +} diff --git a/src/lib/intake-actions.ts b/src/lib/intake-actions.ts new file mode 100644 index 0000000..b7d8970 --- /dev/null +++ b/src/lib/intake-actions.ts @@ -0,0 +1,118 @@ +import { loadFiles } from "@/lib/intake-files"; +import type { PantryForward } from "@/lib/intake-pantry"; +import type { IntakeSource } from "@/lib/intake-rules"; +import { + createFromBatch, + loadBatch, + rememberKind, + resolvedCard, + routeFood, + routeGrocery, + type BatchRow, + type RouteContext, +} from "@/lib/intake-router"; +import { replaceLineItems, TransactionNotFoundError } from "@/lib/line-items"; +import { enrichedCard, errorCard, type Block } from "@/lib/intake-blocks"; +import { queryRaw } from "@/lib/db"; + +/** + * The four intake taps (plan §11 C5), as one function the Slack route calls. + * + * Every tap acts on the persisted batch, never on the card's text. A tap on a + * batch that is already resolved returns its current card rather than acting + * twice — a pressed button that looked dead and got pressed again is how three + * splits were lost on the order nudge, and the same shape applies here. + */ + +export const INTAKE_ACTIONS = ["intake_use_candidate", "intake_create", "intake_grocery", "intake_meal"] as const; +export type IntakeActionId = (typeof INTAKE_ACTIONS)[number]; + +export function isIntakeAction(id: unknown): id is IntakeActionId { + return typeof id === "string" && (INTAKE_ACTIONS as readonly string[]).includes(id); +} + +export type IntakeActionValue = { batch_id?: string; transaction_id?: number }; + +export function parseActionValue(raw: unknown): IntakeActionValue { + if (typeof raw !== "string") return {}; + try { + const v = JSON.parse(raw) as IntakeActionValue; + return v && typeof v === "object" ? v : {}; + } catch { + return {}; + } +} + +const note = (text: string): { blocks: Block[]; text: string; replace: boolean } => ({ + text, + blocks: [{ type: "section", text: { type: "mrkdwn", text } }], + replace: false, +}); + +const ctxFor = (batch: BatchRow): RouteContext | null => + batch.header + ? { + batchId: batch.id, + source: batch.source as IntakeSource, + sharedBy: batch.shared_by, + platform: batch.header.platform, + header: batch.header, + items: batch.items ?? [], + } + : null; + +export async function handleIntakeAction( + actionId: IntakeActionId, + value: IntakeActionValue, + deps: { pantry: PantryForward } +): Promise<{ blocks: Block[]; text: string; replace: boolean }> { + if (!value.batch_id) return note("That button carried no batch id — I can't tell which upload it was."); + const batch = await loadBatch(value.batch_id); + if (!batch) return note("I can't find that upload any more."); + if (batch.status === "enriched" || batch.status === "created" || batch.status === "forwarded_to_pantry") { + return { ...resolvedCard(batch), replace: true }; + } + const ctx = ctxFor(batch); + if (!ctx) return note("That upload never got a readable header — share it again with the top of the screen."); + + try { + if (actionId === "intake_use_candidate") { + if (!Number.isInteger(value.transaction_id)) return note("That button carried no transaction id."); + const txnId = value.transaction_id as number; + await replaceLineItems(txnId, ctx.items, { source: "intake", platform: ctx.platform, batchId: batch.id }); + await queryRaw( + `UPDATE intake_batches SET status = 'enriched', kind = COALESCE(kind, 'food'), transaction_id = $2, resolved_at = now() WHERE id = $1::uuid`, + [batch.id, txnId] + ); + return { + ...enrichedCard({ batchId: batch.id, merchant: ctx.header.merchant, total: ctx.header.total, orderDate: ctx.header.order_date, platform: ctx.platform, itemCount: ctx.items.length, transactionId: txnId }), + replace: true, + }; + } + + if (actionId === "intake_create") { + const kind = batch.kind === "grocery" ? "grocery" : "food"; + const r = await createFromBatch(batch, kind); + return { blocks: r.slack_blocks ?? [], text: r.text, replace: true }; + } + + if (actionId === "intake_meal") { + await rememberKind(ctx.header.merchant, "food"); + const r = await routeFood(ctx); + return { blocks: r.slack_blocks ?? [], text: r.text, replace: true }; + } + + // intake_grocery + await rememberKind(ctx.header.merchant, "grocery"); + const files = await loadFiles(batch.file_sha256); + if (!files.length) { + return note("I remembered that as groceries, but the original files are gone — share it again and it will go straight to the pantry board."); + } + const r = await routeGrocery(ctx, files, deps.pantry); + return { blocks: r.slack_blocks ?? [], text: r.text, replace: r.status !== "error" }; + } catch (e) { + if (e instanceof TransactionNotFoundError) return note("That transaction is no longer in the ledger."); + const detail = e instanceof Error ? e.message : String(e); + return { ...errorCard(detail), replace: false }; + } +} diff --git a/src/lib/intake-blocks.ts b/src/lib/intake-blocks.ts new file mode 100644 index 0000000..ba35ec9 --- /dev/null +++ b/src/lib/intake-blocks.ts @@ -0,0 +1,143 @@ +import { platformLabel } from "@/lib/intake-rules"; + +/** + * The intake reply as Block Kit (plan §11 C1/C5). Built here, never in n8n + * expressions, so the button payloads the interactive handler round-trips are + * produced by one piece of code and tested like one. + * + * Button `value`s are JSON `{batch_id, ...}` — the batch row is what a later + * tap acts on. action_ids are the C5 contract: `intake_use_candidate`, + * `intake_create`, `intake_grocery`, `intake_meal`. + */ + +export type Block = Record; + +export type Candidate = { transaction_id: number; description: string | null; transaction_date: string; amount: string | number }; + +export function financePublicUrl(): string { + return (process.env.FINANCE_PUBLIC_URL || "https://finance.bosecamp.com").replace(/\/$/, ""); +} + +/** The transactions page searches on `q`; the merchant is the most selective thing a card knows. */ +export function transactionLink(merchant: string | null): string { + return `${financePublicUrl()}/transactions?q=${encodeURIComponent(merchant ?? "")}`; +} + +const money = (n: number | string) => `$${Number(n).toFixed(2)}`; +const section = (text: string): Block => ({ type: "section", text: { type: "mrkdwn", text } }); +const button = (action_id: string, text: string, value: unknown, style?: "primary" | "danger"): Block => ({ + type: "button", + action_id, + text: { type: "plain_text", text, emoji: true }, + value: JSON.stringify(value), + ...(style ? { style } : {}), +}); + +export type CardInput = { + batchId: string; + merchant: string | null; + total: number | null; + orderDate: string | null; + platform: string | null; + itemCount: number; +}; + +export function enrichedCard(c: CardInput & { transactionId: number }): { blocks: Block[]; text: string } { + const what = [c.total !== null ? money(c.total) : null, platformLabel(c.platform) || c.merchant, c.platform && c.merchant ? `(${c.merchant})` : null] + .filter(Boolean) + .join(" "); + const text = `Added ${c.itemCount} item${c.itemCount === 1 ? "" : "s"} to your ${what} order ✓`; + return { + text, + blocks: [ + { + type: "section", + text: { type: "mrkdwn", text: `:receipt: ${text}` }, + accessory: { type: "button", text: { type: "plain_text", text: "View" }, url: transactionLink(c.merchant), action_id: "intake_view" }, + }, + ], + }; +} + +export function createdCard(c: CardInput & { transactionId: number }): { blocks: Block[]; text: string } { + const text = `Created a ${c.total !== null ? money(c.total) + " " : ""}${c.merchant ?? "new"} transaction with ${c.itemCount} item${c.itemCount === 1 ? "" : "s"} ✓`; + return { + text, + blocks: [ + { + type: "section", + text: { type: "mrkdwn", text: `:receipt: ${text}` }, + accessory: { type: "button", text: { type: "plain_text", text: "View" }, url: transactionLink(c.merchant), action_id: "intake_view" }, + }, + ], + }; +} + +export function candidatesCard(c: CardInput, candidates: Candidate[]): { blocks: Block[]; text: string } { + const label = platformLabel(c.platform) || ""; + const head = `No matching ${label ? label + " " : ""}order for *${c.merchant ?? "this merchant"}* ${c.total !== null ? money(c.total) : ""} on ${c.orderDate ?? "?"}.`; + const text = `${head.replace(/\*/g, "")} ${candidates.length ? "Pick one below or create it." : "Create it?"}`; + const blocks: Block[] = [section(`:mag: ${head}${candidates.length ? "\nIs it one of these?" : ""}`)]; + for (const cand of candidates) { + blocks.push({ + type: "section", + text: { type: "mrkdwn", text: `${cand.transaction_date} · ${cand.description ?? "—"} · ${money(cand.amount)}` }, + accessory: button("intake_use_candidate", "Use this", { batch_id: c.batchId, transaction_id: cand.transaction_id }), + }); + } + blocks.push({ + type: "actions", + block_id: "intake_create", + elements: [button("intake_create", `Create new ${c.total !== null ? money(c.total) + " " : ""}${c.merchant ?? ""} transaction`.replace(/\s+/g, " ").trim(), { batch_id: c.batchId }, "primary")], + }); + return { blocks, text }; +} + +export function forwardedCard(c: CardInput & { captureId: number; enriching: boolean }): { blocks: Block[]; text: string } { + const text = `${c.merchant ?? "Grocery receipt"} ${c.total !== null ? money(c.total) : ""} → pantry board (${c.itemCount} item${c.itemCount === 1 ? "" : "s"}) — finance ${c.enriching ? "updates" : "books it"} on confirm`.replace(/\s+/g, " "); + return { text, blocks: [section(`:shopping_trolley: ${text}`)] }; +} + +export function needsHeaderCard(): { blocks: Block[]; text: string } { + const text = "I can see items but no order total or date — share the top of the screen too so I can match it."; + return { text, blocks: [section(`:warning: ${text}`)] }; +} + +export function heldCard(reason: string): { blocks: Block[]; text: string } { + const text = `Items don't add up to the total — is a shot missing? (${reason})`; + return { text, blocks: [section(`:warning: ${text}`)] }; +} + +export function askCard(c: CardInput): { blocks: Block[]; text: string } { + const text = `${c.merchant ?? "This upload"} ${c.total !== null ? money(c.total) : ""} — groceries or a meal?`.replace(/\s+/g, " "); + return { + text, + blocks: [ + section(`:thinking_face: ${text}`), + { + type: "actions", + block_id: "intake_ask", + elements: [ + button("intake_grocery", "Groceries", { batch_id: c.batchId }), + button("intake_meal", "A meal", { batch_id: c.batchId }), + ], + }, + ], + }; +} + +export function duplicateCard(c: { merchant: string | null; total: number | null; transactionId: number | null; captureId: number | null; batchId: string | null }): { blocks: Block[]; text: string } { + const where = c.transactionId ? `transaction ${c.transactionId}` : c.captureId ? `pantry capture ${c.captureId}` : c.batchId ? `batch ${c.batchId}` : "an earlier upload"; + const text = `Already handled — ${c.merchant ?? "this order"} ${c.total !== null ? money(c.total) : ""} is ${where}. Nothing was changed.`.replace(/\s+/g, " "); + return { text, blocks: [section(`:repeat: ${text}`)] }; +} + +export function unsupportedCard(detail?: string): { blocks: Block[]; text: string } { + const text = `I can read receipts, statements and order screens — not this.${detail ? ` (${detail})` : ""}`; + return { text, blocks: [section(`:no_entry_sign: ${text}`)] }; +} + +export function errorCard(detail: string): { blocks: Block[]; text: string } { + const text = `Couldn't process that upload: ${detail}`; + return { text, blocks: [section(`:x: ${text}`)] }; +} diff --git a/src/lib/intake-extract.ts b/src/lib/intake-extract.ts new file mode 100644 index 0000000..189b752 --- /dev/null +++ b/src/lib/intake-extract.ts @@ -0,0 +1,199 @@ +import { extractText, getDocumentProxy } from "unpdf"; +import type { ExtractedOrder, ExtractedItem } from "@/lib/intake-rules"; + +/** + * Extraction for the intake router: one joint model call per batch. + * + * The model EXTRACTS; it does not classify or route. The source tap and the + * merchant pattern decide where the result goes (plan §2 — "a classifier is + * confidently wrong exactly when it matters"). So this prompt asks for the + * header and the lines, verbatim, and nothing else. + * + * One call with every part in order, never one per image: scrolling screenshots + * overlap, and extracting each and concatenating double-counts the row both + * shots contain and strands the lower shot with no header (plan §3, Codex #7). + * Dedupe of the overlap is the model's job; the reconciliation check in + * `intake-rules.ts` is what catches it getting that wrong. + */ + +export type InputPart = + | { kind: "image"; mime: string; bytes: Buffer; name: string } + | { kind: "text"; text: string; name: string }; + +/** + * Digital receipts are text, not pixels. Reading the text beats photographing + * the screen: nothing is misread, and the model parses a page rather than an + * image. Same approach as pantry-app's `pdf-text.ts`. + */ +export async function extractPdfText(bytes: Buffer): Promise { + const document = await getDocumentProxy(new Uint8Array(bytes)); + const { text } = await extractText(document, { mergePages: true }); + return (Array.isArray(text) ? text.join("\n") : text).trim(); +} + +export const ORDER_ITEM_PROMPT = `You are reading a purchase receipt: an online food or grocery order (Uber Eats, DoorDash, Coles, Woolworths, ...) or an in-store till receipt. It arrives as one or more screenshots in scroll order, as the text of a PDF, or as pasted text. + +Return ONE JSON object with exactly these keys: +{ "merchant", "order_date", "total", "subtotal", "currency", "items": [{ "quantity", "description", "unit_price", "line_total", "options" }], "adjustments": [{ "label", "amount" }] } + +Rules: +- merchant: the store or restaurant name exactly as printed, including any branch in parentheses (e.g. "Coles (Wyndham Vale)", "Burgertory (Werribee)"). Never the delivery platform. null if not shown. +- order_date: the order or purchase date printed, as YYYY-MM-DD. null if not shown. +- total: the amount actually charged — the "Total" / amount paid — as a number. null if not shown. +- subtotal: the printed item subtotal before fees and discounts, as a number, or null. +- currency: ISO code if printed (default "AUD"). +- items: every purchased line. Keep description VERBATIM as printed — do not rename, canonicalise, translate, shorten or summarise it. A burger called "Michael" stays "Michael"; "Coles RSPCA Approved Chicken Thigh Cutlets approx. 1.1kg" stays exactly that. Put selected options and modifiers (e.g. "Milk Bun", "Chicken") and weighed or substitution notes (e.g. "Final 1.16 kg @ $11.00/kg", "Replaced Coles Beef Slow Cook Chuck Roast approx. 1.3kg") in options, as an array of strings, not in the description. +- quantity: the number ordered (a number; 1 if a line shows none). unit_price: the price per unit as printed, or null. line_total: the money for the whole line as printed, or quantity × unit_price when only a unit price is shown; for weighed lines multiply the final weight by the per-kg price. null when nothing is shown. +- Do not put fees, tips, delivery or service charges, discounts, promotions, savings, tax lines or payment lines in items. List those in adjustments instead, fees positive and discounts negative. +- Multiple screenshots are consecutive scrolls of ONE order and overlap: each physical line appears exactly once in items, never twice. +- Never invent a value. If something is not legible or not present, use null.`; + +const NUM_OR_NULL = { anyOf: [{ type: "number" }, { type: "null" }] }; +const STR_OR_NULL = { anyOf: [{ type: "string" }, { type: "null" }] }; + +/** Strict json_schema: every property listed in `required`, nullability via anyOf. */ +export const EXTRACT_RESPONSE_FORMAT = { + type: "json_schema", + json_schema: { + name: "order_items", + strict: true, + schema: { + type: "object", + additionalProperties: false, + required: ["merchant", "order_date", "total", "subtotal", "currency", "items", "adjustments"], + properties: { + merchant: STR_OR_NULL, + order_date: STR_OR_NULL, + total: NUM_OR_NULL, + subtotal: NUM_OR_NULL, + currency: STR_OR_NULL, + items: { + type: "array", + items: { + type: "object", + additionalProperties: false, + required: ["quantity", "description", "unit_price", "line_total", "options"], + properties: { + quantity: NUM_OR_NULL, + description: { type: "string" }, + unit_price: NUM_OR_NULL, + line_total: NUM_OR_NULL, + options: { type: "array", items: { type: "string" } }, + }, + }, + }, + adjustments: { + type: "array", + items: { + type: "object", + additionalProperties: false, + required: ["label", "amount"], + properties: { label: { type: "string" }, amount: { type: "number" } }, + }, + }, + }, + }, + }, +} as const; + +/** The OpenAI-format message content: prompt first, then every part in the order shared. */ +export function buildMessageContent(parts: InputPart[]): unknown[] { + const content: unknown[] = [{ type: "text", text: ORDER_ITEM_PROMPT }]; + parts.forEach((p, i) => { + if (p.kind === "text") { + content.push({ type: "text", text: `\n--- Input ${i + 1} of ${parts.length} (${p.name}) ---\n${p.text}` }); + } else { + content.push({ type: "text", text: `\n--- Input ${i + 1} of ${parts.length}: image ${p.name} ---` }); + content.push({ type: "image_url", image_url: { url: `data:${p.mime || "image/jpeg"};base64,${p.bytes.toString("base64")}` } }); + } + }); + return content; +} + +const num = (v: unknown): number | null => { + if (typeof v === "number" && Number.isFinite(v)) return v; + if (typeof v === "string" && v.trim()) { + const n = Number(v.replace(/[^0-9.-]/g, "")); + return Number.isFinite(n) ? n : null; + } + return null; +}; + +/** Tolerant of fences and a stray preamble; strict about the shape it hands on. */ +export function parseExtraction(content: string): ExtractedOrder { + const unfenced = content.trim().replace(/^```(?:json)?\s*/i, "").replace(/\s*```$/, ""); + const start = unfenced.indexOf("{"); + const end = unfenced.lastIndexOf("}"); + const candidate = start >= 0 && end > start ? unfenced.slice(start, end + 1) : unfenced; + const raw = JSON.parse(candidate) as Record; + const items: ExtractedItem[] = Array.isArray(raw.items) + ? (raw.items as Record[]) + .filter((i) => i && typeof i.description === "string") + .map((i) => ({ + quantity: num(i.quantity), + description: String(i.description), + unit_price: num(i.unit_price), + line_total: num(i.line_total), + options: Array.isArray(i.options) ? i.options.map(String) : [], + })) + : []; + return { + merchant: typeof raw.merchant === "string" ? raw.merchant : null, + order_date: typeof raw.order_date === "string" ? raw.order_date : null, + total: num(raw.total), + subtotal: num(raw.subtotal), + currency: typeof raw.currency === "string" && raw.currency ? raw.currency : "AUD", + items, + adjustments: Array.isArray(raw.adjustments) + ? (raw.adjustments as Record[]) + .filter((a) => a && typeof a.label === "string" && num(a.amount) !== null) + .map((a) => ({ label: String(a.label), amount: num(a.amount) as number })) + : [], + }; +} + +export type Extractor = (parts: InputPart[]) => Promise; + +export class ExtractionError extends Error {} + +/** + * LiteLLM, OpenAI-compatible, alias `cloud-budget` — the same alias pantry's + * capture uses, so a model swap is one config line in `litellm-config.yaml`. + * `temperature: 0`; `response_format` is the strict schema above. + */ +export function liteLLMExtractor(opts: { + baseUrl?: string; + key?: string; + model?: string; + fetchImpl?: typeof fetch; +} = {}): Extractor { + return async (parts) => { + const baseUrl = (opts.baseUrl ?? process.env.LITELLM_BASE_URL ?? "").replace(/\/$/, ""); + const key = opts.key ?? process.env.LITELLM_KEY; + if (!baseUrl || !key) throw new ExtractionError("LITELLM_BASE_URL / LITELLM_KEY are not configured"); + const doFetch = opts.fetchImpl ?? fetch; + const response = await doFetch(`${baseUrl}/v1/chat/completions`, { + method: "POST", + headers: { Authorization: `Bearer ${key}`, "Content-Type": "application/json" }, + body: JSON.stringify({ + model: opts.model ?? "cloud-budget", + temperature: 0, + max_tokens: 16384, + response_format: EXTRACT_RESPONSE_FORMAT, + messages: [{ role: "user", content: buildMessageContent(parts) }], + }), + }); + if (!response.ok) throw new ExtractionError(`LiteLLM returned ${response.status}`); + const raw = (await response.json()) as { choices?: { message?: { content?: string }; finish_reason?: string }[] }; + // A truncated reply can still parse once the JSON is closed, which quietly + // drops the tail of a long receipt. Losing lines silently is worse than failing. + if (raw.choices?.[0]?.finish_reason === "length") throw new ExtractionError("The receipt was too long to read in one pass."); + const content = raw.choices?.[0]?.message?.content; + if (!content) throw new ExtractionError("LiteLLM returned no content"); + try { + return parseExtraction(content); + } catch (e) { + throw new ExtractionError(`could not parse extraction: ${e instanceof Error ? e.message : String(e)}`); + } + }; +} diff --git a/src/lib/intake-files.ts b/src/lib/intake-files.ts new file mode 100644 index 0000000..b4d010a --- /dev/null +++ b/src/lib/intake-files.ts @@ -0,0 +1,51 @@ +import { createHash } from "node:crypto"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import path from "node:path"; + +/** + * The attachment store: the original files of a batch, keyed by sha256. + * + * Two reasons it exists. A "Groceries or a meal?" answer arrives minutes after + * the upload, in a fresh request carrying only the batch id, and "groceries" + * means forwarding the ORIGINAL files to pantry — so they have to still be + * here. And plan §8: "the source file is retrievable from the attachment store + * by its sha256". + * + * Plain files under `INTAKE_FILE_DIR` (default `/app/data/intake`); a sidecar + * `.json` keeps the name and MIME type. No database column holds bytes. + */ + +export type StoredFile = { name: string; type: string; bytes: Buffer; sha256: string }; + +export function intakeFileDir(): string { + return process.env.INTAKE_FILE_DIR || "/app/data/intake"; +} + +export function sha256Of(bytes: Buffer): string { + return createHash("sha256").update(bytes).digest("hex"); +} + +export async function storeFiles(files: StoredFile[]): Promise { + const dir = intakeFileDir(); + await mkdir(dir, { recursive: true }); + for (const f of files) { + await writeFile(path.join(dir, `${f.sha256}.bin`), f.bytes); + await writeFile(path.join(dir, `${f.sha256}.json`), JSON.stringify({ name: f.name, type: f.type })); + } +} + +/** The files that are still on disk, in the order asked for; missing ones are skipped. */ +export async function loadFiles(shas: string[]): Promise { + const dir = intakeFileDir(); + const out: StoredFile[] = []; + for (const sha of shas) { + try { + const bytes = await readFile(path.join(dir, `${sha}.bin`)); + const meta = JSON.parse(await readFile(path.join(dir, `${sha}.json`), "utf-8")) as { name?: string; type?: string }; + out.push({ name: meta.name ?? `${sha}.bin`, type: meta.type ?? "application/octet-stream", bytes, sha256: sha }); + } catch { + /* gone — the caller decides what that means */ + } + } + return out; +} diff --git a/src/lib/intake-pantry.ts b/src/lib/intake-pantry.ts new file mode 100644 index 0000000..89da07c --- /dev/null +++ b/src/lib/intake-pantry.ts @@ -0,0 +1,76 @@ +import type { StoredFile } from "@/lib/intake-files"; + +/** + * finance → pantry hand-off for a grocery upload (plan §11 C3). + * + * The ORIGINAL files go across, not the extracted items: pantry owns grocery + * extraction (canonical names, categories, staples) and its capture board is + * the review surface. What finance adds is the routing decision — which + * transaction the confirm should enrich, or that it should create one — carried + * in the `intake` field so pantry's confirm seam (C4) can suppress the ING-10 + * create that would otherwise double-book an order the email already booked. + */ + +export type PantryForwardArgs = { + files: StoredFile[]; + batchId: string; + platform: string | null; + sharedBy: string | null; + financeTargetTransactionId: number | null; + financeAction: "enrich" | "create"; +}; + +export type PantryForwardResult = { captureId: number; itemCount: number | null }; + +export class PantryDuplicateError extends Error { + constructor( + public readonly captureId: number | null, + public readonly reason: string + ) { + super(reason); + } +} + +export class PantryUnavailableError extends Error {} + +export type PantryForward = (args: PantryForwardArgs) => Promise; + +export function pantryForwarder(opts: { baseUrl?: string; token?: string; fetchImpl?: typeof fetch } = {}): PantryForward { + return async (args) => { + const baseUrl = (opts.baseUrl ?? process.env.PANTRY_BASE_URL ?? "").replace(/\/$/, ""); + const token = opts.token ?? process.env.RECEIPT_INGEST_TOKEN; + if (!baseUrl || !token) throw new PantryUnavailableError("PANTRY_BASE_URL / RECEIPT_INGEST_TOKEN are not configured"); + const doFetch = opts.fetchImpl ?? fetch; + + const form = new FormData(); + for (const f of args.files) { + form.append("image", new Blob([new Uint8Array(f.bytes)], { type: f.type }), f.name); + } + form.append("source", "receipt"); + form.append( + "intake", + JSON.stringify({ + batch_id: args.batchId, + platform: args.platform, + origin: "intake", + shared_by: args.sharedBy, + finance_target_transaction_id: args.financeTargetTransactionId, + finance_action: args.financeAction, + }) + ); + + let response: Response; + try { + response = await doFetch(`${baseUrl}/api/capture`, { method: "POST", headers: { "x-ingest-token": token }, body: form }); + } catch (e) { + throw new PantryUnavailableError(`pantry unreachable: ${e instanceof Error ? e.message : String(e)}`); + } + const body = (await response.json().catch(() => ({}))) as { captureId?: number; error?: string; items?: unknown[]; reason?: string }; + if (response.status === 409) { + throw new PantryDuplicateError(typeof body.captureId === "number" ? body.captureId : null, body.reason ?? body.error ?? "already captured"); + } + if (!response.ok) throw new PantryUnavailableError(`pantry returned ${response.status}: ${body.error ?? ""}`.trim()); + if (typeof body.captureId !== "number") throw new PantryUnavailableError("pantry returned no captureId"); + return { captureId: body.captureId, itemCount: Array.isArray(body.items) ? body.items.length : null }; + }; +} diff --git a/src/lib/intake-router.ts b/src/lib/intake-router.ts new file mode 100644 index 0000000..90a3164 --- /dev/null +++ b/src/lib/intake-router.ts @@ -0,0 +1,505 @@ +import { randomUUID } from "node:crypto"; +import { queryRaw, queryRow } from "@/lib/db"; +import { DEFAULT_OWNER_ID } from "@/lib/order-ingestion"; +import { + classifyInput, + merchantKey, + merchantMatches, + platformForSource, + reconcileItems, + resolveMerchantKind, + toLineItems, + validateHeader, + type ExtractedOrder, + type IntakeSource, + type MerchantKind, + type OrderHeader, + type StoredLineItem, +} from "@/lib/intake-rules"; +import { extractPdfText, ExtractionError, type Extractor, type InputPart } from "@/lib/intake-extract"; +import { storeFiles, type StoredFile } from "@/lib/intake-files"; +import { PantryDuplicateError, PantryUnavailableError, type PantryForward } from "@/lib/intake-pantry"; +import { replaceLineItems } from "@/lib/line-items"; +import { + askCard, + candidatesCard, + createdCard, + duplicateCard, + enrichedCard, + errorCard, + forwardedCard, + heldCard, + needsHeaderCard, + unsupportedCard, + type Block, + type Candidate, + type CardInput, +} from "@/lib/intake-blocks"; + +/** + * The intake router (plan §11 C1): one upload, every applicable step. + * + * Order of operations matters and is the plan's: + * persist the batch → classify each file → ONE extraction call → header → + * merchant kind → duplicate check → reconciliation → match → act. + * + * The batch row is written before anything else so that a Slack tap minutes + * later has something to act on, and so a re-share is recognisable. Nothing + * else is written until the header is whole: a half-read upload produces a + * reply, not a row (plan §2). + */ + +export const MAX_FILES = 6; + +export type IntakeKind = "grocery" | "food" | "ask" | "held" | "unsupported"; +export type IntakeStatus = + | "received" + | "enriched" + | "created" + | "candidates" + | "forwarded_to_pantry" + | "needs_header" + | "duplicate" + | "unsupported" + | "ask" + | "held" + | "error"; + +export type IntakeResult = { + kind: IntakeKind; + status: IntakeStatus; + batch_id: string; + transaction_id?: number; + capture_id?: number; + duplicate_of?: { batch_id: string; transaction_id: number | null; capture_id: number | null }; + candidates?: Candidate[]; + header?: OrderHeader | null; + items?: StoredLineItem[]; + slack_blocks: Block[] | null; + text: string; + http_status: number; +}; + +export type IntakeDeps = { extractor: Extractor; pantry: PantryForward }; + +export type BatchRow = { + id: string; + source: IntakeSource; + shared_by: string | null; + file_sha256: string[]; + header: (OrderHeader & { platform: string | null; subtotal?: number | null; currency?: string | null }) | null; + items: StoredLineItem[] | null; + kind: string | null; + status: IntakeStatus; + transaction_id: number | null; + capture_id: number | null; +}; + + +// ─── batch persistence ────────────────────────────────────────────────────── + +export async function loadBatch(id: string): Promise { + return queryRow( + `SELECT id::text, source, shared_by, file_sha256, header, items, kind, status, transaction_id, capture_id + FROM intake_batches WHERE id = $1::uuid`, + [id] + ); +} + +async function insertBatch(b: { id: string; source: IntakeSource; sharedBy: string | null; shas: string[] }) { + await queryRaw( + `INSERT INTO intake_batches (id, source, shared_by, file_sha256, status) + VALUES ($1::uuid, $2, $3, $4::text[], 'received')`, + [b.id, b.source, b.sharedBy, b.shas] + ); +} + +async function updateBatch( + id: string, + patch: { + header?: unknown; + items?: unknown; + kind?: string | null; + status: IntakeStatus; + transaction_id?: number | null; + capture_id?: number | null; + } +) { + await queryRaw( + `UPDATE intake_batches + SET header = COALESCE($2::jsonb, header), + items = COALESCE($3::jsonb, items), + kind = COALESCE($4, kind), + status = $5, + transaction_id = COALESCE($6, transaction_id), + capture_id = COALESCE($7, capture_id), + resolved_at = CASE WHEN $5 IN ('enriched','created','forwarded_to_pantry','duplicate') THEN now() ELSE resolved_at END + WHERE id = $1::uuid`, + [ + id, + patch.header === undefined ? null : JSON.stringify(patch.header), + patch.items === undefined ? null : JSON.stringify(patch.items), + patch.kind ?? null, + patch.status, + patch.transaction_id ?? null, + patch.capture_id ?? null, + ] + ); +} + +// ─── lookups ──────────────────────────────────────────────────────────────── + +export async function rememberedKind(merchant: string | null): Promise<"grocery" | "food" | null> { + const key = merchantKey(merchant); + if (!key) return null; + const row = await queryRow<{ kind: "grocery" | "food" }>(`SELECT kind FROM intake_merchant_kinds WHERE merchant_key = $1`, [key]); + return row?.kind ?? null; +} + +export async function rememberKind(merchant: string | null, kind: "grocery" | "food"): Promise { + const key = merchantKey(merchant); + if (!key) return; + await queryRaw( + `INSERT INTO intake_merchant_kinds (merchant_key, kind) VALUES ($1, $2) + ON CONFLICT (merchant_key) DO UPDATE SET kind = EXCLUDED.kind`, + [key, kind] + ); +} + +/** The owner a batch's rows belong to: the sharer's participant, else the default. */ +export async function ownerForEmail(email: string | null): Promise { + if (!email) return DEFAULT_OWNER_ID; + const row = await queryRow<{ id: number }>(`SELECT id FROM participants WHERE lower(email) = lower($1) LIMIT 1`, [email]); + return row?.id ?? DEFAULT_OWNER_ID; +} + +/** + * A batch already resolved from the same files, or from the same + * (platform, merchant, date, total) header (plan §5, Codex #5). Only RESOLVED + * batches count: a `needs_header` share followed by a fuller one must go + * through, and a pending candidates card re-shared is harmless. + */ +export async function findDuplicateBatch( + exceptId: string, + shas: string[], + header: (OrderHeader & { platform: string | null }) | null +): Promise { + const bySha = await queryRow( + `SELECT id::text, source, shared_by, file_sha256, header, items, kind, status, transaction_id, capture_id + FROM intake_batches + WHERE id <> $1::uuid AND status IN ('enriched','created','forwarded_to_pantry') + AND file_sha256 && $2::text[] + ORDER BY created_at DESC LIMIT 1`, + [exceptId, shas] + ); + if (bySha) return bySha; + if (!header || !header.merchant) return null; + const rows = await queryRaw( + `SELECT id::text, source, shared_by, file_sha256, header, items, kind, status, transaction_id, capture_id + FROM intake_batches + WHERE id <> $1::uuid AND status IN ('enriched','created','forwarded_to_pantry') + AND header->>'order_date' = $2 + AND (header->>'total')::numeric = $3::numeric + AND COALESCE(header->>'platform', '') = COALESCE($4, '') + ORDER BY created_at DESC`, + [exceptId, header.order_date, header.total, header.platform] + ); + return rows.find((r) => merchantMatches(r.header?.merchant, header.merchant)) ?? null; +} + +type MatchRow = { transaction_id: number; description: string | null; transaction_date: string; amount: string; merchant: string | null; platform: string | null }; + +/** + * The match rule (plan §5, D2): `(platform, merchant, date, total)`, amount and + * date EXACT, never widened silently. Platform is required when the source + * implies one; an in-store receipt has none and matches on the other three. + * The merchant compares normalised, with containment ("Coles" vs "Coles + * (Wyndham Vale)") — that is spelling, not widening. A unique hit is a match; + * two hits are an ambiguity and go to the candidates card like a miss. + */ +export async function matchTransaction(header: OrderHeader, platform: string | null): Promise<{ match: MatchRow | null; ambiguous: boolean }> { + if (!header.merchant) return { match: null, ambiguous: false }; + const rows = await queryRaw( + `SELECT DISTINCT ON (t.id) + t.id AS transaction_id, t.description, t.transaction_date::text, t.amount::text, + COALESCE(em.merchant_normalized, t.merchant_normalized, t.merchant_name, t.description) AS merchant, + em.platform + FROM transactions t + LEFT JOIN expense_metadata em ON em.transaction_id = t.id OR em.matched_transaction_id = t.id + WHERE t.superseded_by_id IS NULL + AND t.transaction_date = $1::date + AND t.amount = $2::numeric + AND ($3::text IS NULL OR em.platform = $3) + ORDER BY t.id, (em.transaction_id = t.id) DESC NULLS LAST`, + [header.order_date, header.total, platform] + ); + const hits = rows.filter((r) => merchantMatches(r.merchant, header.merchant) || merchantMatches(r.description, header.merchant)); + if (hits.length === 1) return { match: hits[0], ambiguous: false }; + return { match: null, ambiguous: hits.length > 1 }; +} + +/** Nearest candidates for the card: same platform and merchant, ±3 days, any amount. */ +export async function findCandidates(header: OrderHeader, platform: string | null): Promise { + if (!header.merchant) return []; + const rows = await queryRaw( + `SELECT DISTINCT ON (t.id) + t.id AS transaction_id, t.description, t.transaction_date::text, t.amount::text, + COALESCE(em.merchant_normalized, t.merchant_normalized, t.merchant_name, t.description) AS merchant, + em.platform + FROM transactions t + LEFT JOIN expense_metadata em ON em.transaction_id = t.id OR em.matched_transaction_id = t.id + WHERE t.superseded_by_id IS NULL + AND t.transaction_date BETWEEN $1::date - 3 AND $1::date + 3 + AND ($2::text IS NULL OR em.platform = $2) + ORDER BY t.id, (em.transaction_id = t.id) DESC NULLS LAST`, + [header.order_date, platform] + ); + return rows + .filter((r) => merchantMatches(r.merchant, header.merchant) || merchantMatches(r.description, header.merchant)) + .sort((a, b) => { + const da = Math.abs(Date.parse(a.transaction_date) - Date.parse(header.order_date)); + const db = Math.abs(Date.parse(b.transaction_date) - Date.parse(header.order_date)); + if (da !== db) return da - db; + return Math.abs(Number(a.amount) - header.total) - Math.abs(Number(b.amount) - header.total); + }) + .slice(0, 5) + .map((r) => ({ transaction_id: r.transaction_id, description: r.description, transaction_date: r.transaction_date, amount: r.amount })); +} + +// ─── the two destinations ─────────────────────────────────────────────────── + +export type RouteContext = { + batchId: string; + source: IntakeSource; + sharedBy: string | null; + platform: string | null; + header: OrderHeader; + items: StoredLineItem[]; + subtotal?: number | null; +}; + +const cardInput = (ctx: RouteContext): CardInput => ({ + batchId: ctx.batchId, + merchant: ctx.header.merchant, + total: ctx.header.total, + orderDate: ctx.header.order_date, + platform: ctx.platform, + itemCount: ctx.items.length, +}); + +const finish = (ctx: RouteContext, kind: IntakeKind, status: IntakeStatus, card: { blocks: Block[]; text: string }, extra: Partial = {}): IntakeResult => ({ + kind, + status, + batch_id: ctx.batchId, + header: ctx.header, + items: ctx.items, + slack_blocks: card.blocks, + text: card.text, + http_status: 200, + ...extra, +}); + +/** Food: enrich the matched order's line_items, or show candidates. NEVER auto-creates. */ +export async function routeFood(ctx: RouteContext): Promise { + const { match } = await matchTransaction(ctx.header, ctx.platform); + if (match) { + await replaceLineItems(match.transaction_id, ctx.items, { source: "intake", platform: ctx.platform, batchId: ctx.batchId }); + await updateBatch(ctx.batchId, { kind: "food", status: "enriched", transaction_id: match.transaction_id, items: ctx.items }); + return finish(ctx, "food", "enriched", enrichedCard({ ...cardInput(ctx), transactionId: match.transaction_id }), { transaction_id: match.transaction_id }); + } + const candidates = await findCandidates(ctx.header, ctx.platform); + await updateBatch(ctx.batchId, { kind: "food", status: "candidates", items: ctx.items }); + return finish(ctx, "food", "candidates", candidatesCard(cardInput(ctx), candidates), { candidates }); +} + +/** Grocery: the original files go to pantry's board; finance rides along on its confirm (C3/C4). */ +export async function routeGrocery(ctx: RouteContext, files: StoredFile[], pantry: PantryForward): Promise { + const { match } = await matchTransaction(ctx.header, ctx.platform); + try { + const r = await pantry({ + files, + batchId: ctx.batchId, + platform: ctx.platform, + sharedBy: ctx.sharedBy, + financeTargetTransactionId: match?.transaction_id ?? null, + financeAction: match ? "enrich" : "create", + }); + await updateBatch(ctx.batchId, { kind: "grocery", status: "forwarded_to_pantry", transaction_id: match?.transaction_id ?? null, capture_id: r.captureId, items: ctx.items }); + const card = forwardedCard({ ...cardInput(ctx), itemCount: r.itemCount ?? ctx.items.length, captureId: r.captureId, enriching: !!match }); + return finish(ctx, "grocery", "forwarded_to_pantry", card, { capture_id: r.captureId, transaction_id: match?.transaction_id }); + } catch (e) { + if (e instanceof PantryDuplicateError) { + await updateBatch(ctx.batchId, { kind: "grocery", status: "duplicate", capture_id: e.captureId, items: ctx.items }); + return finish(ctx, "grocery", "duplicate", duplicateCard({ merchant: ctx.header.merchant, total: ctx.header.total, transactionId: match?.transaction_id ?? null, captureId: e.captureId, batchId: null }), { + capture_id: e.captureId ?? undefined, + }); + } + const detail = e instanceof PantryUnavailableError ? e.message : e instanceof Error ? e.message : String(e); + await updateBatch(ctx.batchId, { kind: "grocery", status: "error", items: ctx.items }); + return { ...finish(ctx, "grocery", "error", errorCard(detail)), http_status: 502 }; + } +} + +/** Create the transaction ING-10-style: manual, `statement_id NULL`, owner set — only ever on an explicit tap. */ +export async function createFromBatch(batch: BatchRow, kind: "grocery" | "food"): Promise { + if (!batch.header) throw new Error("batch has no header"); + const header = batch.header; + const items = batch.items ?? []; + const ownerId = await ownerForEmail(batch.shared_by); + const merchant = header.merchant ?? "Unknown merchant"; + const description = header.platform ? `Order - ${merchant}` : merchant; + const category = kind === "grocery" ? "groceries" : "dining"; + // source/source_ref make a second press of the same card a no-op rather than a second row. + const existing = await queryRow<{ id: number }>(`SELECT id FROM transactions WHERE source = 'intake' AND source_ref = $1`, [batch.id]); + const txnId = + existing?.id ?? + ( + await queryRow<{ id: number }>( + `INSERT INTO transactions ( + transaction_date, description, amount, amount_aud, category, payment_method, + merchant_name, merchant_normalized, transaction_type, owner_id, source, source_ref + ) VALUES ($1::date, $2, $3::numeric, $3::numeric, $4, NULL, $5, $5, 'debit', $6, 'intake', $7) + RETURNING id`, + [header.order_date, description, header.total, category, merchant, ownerId, batch.id] + ) + )!.id; + await replaceLineItems(txnId, items, { source: "intake", platform: header.platform, batchId: batch.id }); + await updateBatch(batch.id, { kind, status: "created", transaction_id: txnId }); + const ctx: RouteContext = { batchId: batch.id, source: batch.source, sharedBy: batch.shared_by, platform: header.platform, header, items }; + return finish(ctx, kind, "created", createdCard({ ...cardInput(ctx), transactionId: txnId }), { transaction_id: txnId }); +} + +/** The card for a batch as it stands — what a second press of a resolved card gets back. */ +export function resolvedCard(batch: BatchRow): { blocks: Block[]; text: string } { + const h = batch.header; + const base: CardInput = { batchId: batch.id, merchant: h?.merchant ?? null, total: h?.total ?? null, orderDate: h?.order_date ?? null, platform: h?.platform ?? null, itemCount: batch.items?.length ?? 0 }; + if (batch.status === "enriched" && batch.transaction_id) return enrichedCard({ ...base, transactionId: batch.transaction_id }); + if (batch.status === "created" && batch.transaction_id) return createdCard({ ...base, transactionId: batch.transaction_id }); + if (batch.status === "forwarded_to_pantry" && batch.capture_id) return forwardedCard({ ...base, captureId: batch.capture_id, enriching: !!batch.transaction_id }); + return duplicateCard({ merchant: base.merchant, total: base.total, transactionId: batch.transaction_id, captureId: batch.capture_id, batchId: batch.id }); +} + +// ─── the router ───────────────────────────────────────────────────────────── + +export type IntakeInput = { batchId?: string | null; source: IntakeSource; sharedBy: string | null; files: StoredFile[] }; + +export async function runIntake(input: IntakeInput, deps: IntakeDeps): Promise { + const batchId = input.batchId || randomUUID(); + const shas = input.files.map((f) => f.sha256); + const platform = platformForSource(input.source); + + if (await loadBatch(batchId)) { + return { kind: "unsupported", status: "duplicate", batch_id: batchId, slack_blocks: null, text: `batch ${batchId} was already received`, http_status: 409 }; + } + await insertBatch({ id: batchId, source: input.source, sharedBy: input.sharedBy, shas }); + + const fail = (kind: IntakeKind, status: IntakeStatus, card: { blocks: Block[]; text: string }, http = 200, extra: Partial = {}): IntakeResult => ({ + kind, + status, + batch_id: batchId, + slack_blocks: card.blocks, + text: card.text, + http_status: http, + ...extra, + }); + + // 1. classify each file and build the parts, in the order shared. + const parts: InputPart[] = []; + const usable: StoredFile[] = []; + for (const f of input.files) { + const kind = classifyInput(f); + if (kind === "pdf") { + let text = ""; + try { + text = await extractPdfText(f.bytes); + } catch { + /* unreadable PDF — reported below */ + } + if (!text) { + await updateBatch(batchId, { kind: "unsupported", status: "unsupported" }); + return fail("unsupported", "unsupported", unsupportedCard(`${f.name} has no readable text — try a screenshot of it`)); + } + parts.push({ kind: "text", text, name: f.name }); + } else if (kind === "image") { + parts.push({ kind: "image", mime: f.type || "image/jpeg", bytes: f.bytes, name: f.name }); + } else if (kind === "text") { + parts.push({ kind: "text", text: f.bytes.toString("utf-8"), name: f.name }); + } else { + continue; + } + usable.push(f); + } + if (!parts.length) { + await updateBatch(batchId, { kind: "unsupported", status: "unsupported" }); + return fail("unsupported", "unsupported", unsupportedCard()); + } + + // A re-share of the very same files, before spending a model call on it. + const dupBySha = await findDuplicateBatch(batchId, shas, null); + if (dupBySha) { + await updateBatch(batchId, { status: "duplicate", transaction_id: dupBySha.transaction_id, capture_id: dupBySha.capture_id }); + return fail("unsupported", "duplicate", duplicateCard({ merchant: dupBySha.header?.merchant ?? null, total: dupBySha.header?.total ?? null, transactionId: dupBySha.transaction_id, captureId: dupBySha.capture_id, batchId: dupBySha.id }), 200, { + kind: (dupBySha.kind as IntakeKind) ?? "food", + duplicate_of: { batch_id: dupBySha.id, transaction_id: dupBySha.transaction_id, capture_id: dupBySha.capture_id }, + }); + } + + // Keep the originals: a later "Groceries" tap forwards them to pantry. + try { + await storeFiles(usable); + } catch (e) { + console.warn("intake: could not store files", e instanceof Error ? e.message : e); + } + + // 2. ONE extraction call for the whole batch. + let extracted: ExtractedOrder; + try { + extracted = await deps.extractor(parts); + } catch (e) { + const detail = e instanceof ExtractionError ? e.message : e instanceof Error ? e.message : String(e); + await updateBatch(batchId, { status: "error" }); + return fail("held", "error", errorCard(detail), 502); + } + const items = toLineItems(extracted.items); + + // 3. header. + const headerCheck = validateHeader(extracted); + if (!headerCheck.ok) { + await updateBatch(batchId, { header: { merchant: extracted.merchant, order_date: extracted.order_date, total: extracted.total, platform }, items, status: "needs_header" }); + return fail("held", "needs_header", needsHeaderCard(), 200, { items }); + } + const header = headerCheck.header; + const storedHeader = { ...header, platform, subtotal: extracted.subtotal ?? null, currency: extracted.currency ?? "AUD" }; + + // 4. merchant → grocery | food | ask. + const kind: MerchantKind = resolveMerchantKind(input.source, header.merchant, await rememberedKind(header.merchant)); + + // 5. the same order, already handled under other files. + const dup = await findDuplicateBatch(batchId, shas, storedHeader); + if (dup) { + await updateBatch(batchId, { header: storedHeader, items, kind, status: "duplicate", transaction_id: dup.transaction_id, capture_id: dup.capture_id }); + return fail(kind, "duplicate", duplicateCard({ merchant: header.merchant, total: header.total, transactionId: dup.transaction_id, captureId: dup.capture_id, batchId: dup.id }), 200, { + header, + items, + duplicate_of: { batch_id: dup.id, transaction_id: dup.transaction_id, capture_id: dup.capture_id }, + }); + } + + // 6. do the lines add up? (missing-screenshot check) + const recon = reconcileItems(extracted); + if (!recon.ok) { + await updateBatch(batchId, { header: storedHeader, items, kind, status: "held" }); + return fail("held", "held", heldCard(recon.reason ?? "mismatch"), 200, { header, items }); + } + + await updateBatch(batchId, { header: storedHeader, items, kind, status: "received" }); + const ctx: RouteContext = { batchId, source: input.source, sharedBy: input.sharedBy, platform, header, items, subtotal: extracted.subtotal }; + + // 7. act. + if (kind === "ask") { + await updateBatch(batchId, { kind: "ask", status: "ask" }); + return finish(ctx, "ask", "ask", askCard(cardInput(ctx))); + } + if (kind === "grocery") return routeGrocery(ctx, usable, deps.pantry); + return routeFood(ctx); +} diff --git a/src/lib/intake-rules.ts b/src/lib/intake-rules.ts new file mode 100644 index 0000000..69c9bc3 --- /dev/null +++ b/src/lib/intake-rules.ts @@ -0,0 +1,228 @@ +/** + * Intake router — the pure decisions. + * + * Everything here is a function of its arguments: what kind of input a file is, + * which platform a source tap implies, whether a merchant is a grocer, whether a + * header is complete, whether the items add up. Kept free of the database and the + * model so each rule can be unit-tested against the exact strings real receipts + * print. The orchestration that calls these lives in `intake-router.ts`. + * + * Plan: smarthome `plans/intake-router-slice-plan.md` §2, §3, §5, §11 C1. + */ + +export const INTAKE_SOURCES = ["uber_eats", "doordash", "coles", "woolworths", "other"] as const; +export type IntakeSource = (typeof INTAKE_SOURCES)[number]; + +export function isIntakeSource(value: unknown): value is IntakeSource { + return typeof value === "string" && (INTAKE_SOURCES as readonly string[]).includes(value); +} + +export type InputKind = "image" | "pdf" | "text" | "unsupported"; + +/** Input type picks the extractor; it is independent of source and merchant. */ +export function classifyInput(file: { type?: string | null; name?: string | null }): InputKind { + const type = (file.type ?? "").toLowerCase(); + const name = file.name ?? ""; + if (type === "application/pdf" || /\.pdf$/i.test(name)) return "pdf"; + if (type.startsWith("image/") || /\.(png|jpe?g|webp|heic|gif)$/i.test(name)) return "image"; + if (type.startsWith("text/") || /\.(txt|md)$/i.test(name)) return "text"; + return "unsupported"; +} + +/** + * The platform the source tap implies, in `expense_metadata.platform` vocabulary. + * Null for an in-store receipt or `other`: those rows carry no platform, so the + * match falls back to (merchant, date, total) with no platform constraint. + */ +export function platformForSource(source: IntakeSource): string | null { + if (source === "uber_eats") return "ubereats"; + if (source === "doordash") return "doordash"; + return null; +} + +export const PLATFORM_LABEL: Record = { + ubereats: "Uber Eats", + doordash: "DoorDash", +}; + +export function platformLabel(platform: string | null): string { + return platform ? PLATFORM_LABEL[platform] ?? platform : ""; +} + +/** + * A small pattern list, not an exact lookup (plan §2, Codex #6). Uber Eats prints + * `Coles (Wyndham Vale)`, a PDF may print `Woolworths Group Limited`; substring on + * the normalised name covers both without a merchant master-data subsystem. + */ +export const GROCERY_PATTERN = /coles|woolworths|woolies|aldi|iga|foodworks|costco/i; + +/** + * The normalised merchant used for matching and for the remembered-kind table. + * Lower-case, punctuation dropped, whitespace collapsed — so `Coles (Wyndham Vale)` + * and `coles wyndham vale` are one key, while `Coles` stays distinct from both. + */ +export function merchantKey(name: string | null | undefined): string { + return (name ?? "") + .toLowerCase() + .replace(/[^\p{L}\p{N}]+/gu, " ") + .trim() + .replace(/\s+/g, " "); +} + +export type MerchantKind = "grocery" | "food" | "ask"; + +const UNKNOWN_MERCHANT = /^(unknown|n ?a|none|null|undefined|merchant)$/; + +/** + * Grocery-vs-food is decided by the merchant, not the platform — "Uber Eats" + * covers both a meal and a grocery run. The source tap is ground truth where it + * says so (Coles/Woolworths taps are grocers by definition); the pattern list + * decides the obvious delivery cases; the remembered answer covers the + * `Romeo's Food Hall`s; and `other` with nothing else to go on asks rather than + * silently falling through to "meal" (plan §2, D3 holding state). + */ +export function resolveMerchantKind( + source: IntakeSource, + merchant: string | null | undefined, + remembered: "grocery" | "food" | null = null +): MerchantKind { + if (source === "coles" || source === "woolworths") return "grocery"; + const key = merchantKey(merchant); + if (!key || UNKNOWN_MERCHANT.test(key)) return "ask"; + if (GROCERY_PATTERN.test(key)) return "grocery"; + if (remembered) return remembered; + if (source === "other") return "ask"; + return "food"; +} + +/** What the extractor returns — names verbatim, options kept (plan §3). */ +export type ExtractedItem = { + quantity: number | null; + description: string; + unit_price?: number | null; + line_total?: number | null; + options?: string[] | null; +}; + +export type ExtractedOrder = { + merchant: string | null; + order_date: string | null; + total: number | null; + subtotal?: number | null; + currency?: string | null; + items: ExtractedItem[]; + adjustments?: { label: string; amount: number }[] | null; +}; + +export type OrderHeader = { merchant: string | null; order_date: string; total: number }; + +/** + * "Missing the header/total shot → nothing is written." A header is the date and + * the total; the merchant may be absent and still route (it becomes an ask). + */ +export function validateHeader( + x: ExtractedOrder +): { ok: true; header: OrderHeader } | { ok: false; missing: string[] } { + const missing: string[] = []; + if (typeof x.order_date !== "string" || !/^\d{4}-\d{2}-\d{2}$/.test(x.order_date)) missing.push("date"); + if (typeof x.total !== "number" || !Number.isFinite(x.total) || x.total <= 0) missing.push("total"); + if (missing.length) return { ok: false, missing }; + return { + ok: true, + header: { + merchant: typeof x.merchant === "string" && x.merchant.trim() ? x.merchant.trim() : null, + order_date: x.order_date as string, + total: round2(x.total as number), + }, + }; +} + +export const round2 = (n: number) => Number(n.toFixed(2)); + +/** A line's money, as printed or derived; null when the receipt gave neither. */ +export function itemAmount(item: ExtractedItem): number | null { + if (typeof item.line_total === "number" && Number.isFinite(item.line_total)) return round2(item.line_total); + if (typeof item.unit_price === "number" && Number.isFinite(item.unit_price)) { + return round2(item.unit_price * (item.quantity ?? 1)); + } + return null; +} + +export type Reconciliation = { + ok: boolean; + basis: "subtotal" | "total" | "none"; + items_sum: number; + reference: number | null; + diff: number; + reason?: string; +}; + +/** + * The missing-screenshot check (plan §3, Codex #7): the lines must reconcile + * within $1 or 2%, else the batch is held. + * + * Reconciled to the printed ITEM SUBTOTAL when there is one, not the total — a + * delivery order's total is items plus fees minus promotions ($118.94 of items + * became $85.74 on the real Coles export), so against the total every delivery + * order would be held. The total stays the match key; the subtotal is what the + * items can be checked against. Without a printed subtotal the total is the only + * reference there is. A receipt whose lines carry no prices at all (pasted text) + * cannot be checked and is not held for it. + */ +export function reconcileItems(x: ExtractedOrder): Reconciliation { + const amounts = (x.items ?? []).map(itemAmount); + if (!x.items?.length) { + return { ok: false, basis: "none", items_sum: 0, reference: null, diff: 0, reason: "no items" }; + } + if (amounts.every((a) => a === null)) { + return { ok: true, basis: "none", items_sum: 0, reference: null, diff: 0 }; + } + const itemsSum = round2(amounts.reduce((s, a) => s + (a ?? 0), 0)); + const subtotal = typeof x.subtotal === "number" && Number.isFinite(x.subtotal) ? x.subtotal : null; + const basis: Reconciliation["basis"] = subtotal !== null ? "subtotal" : "total"; + const reference = subtotal ?? (typeof x.total === "number" ? x.total : null); + if (reference === null) return { ok: false, basis, items_sum: itemsSum, reference: null, diff: 0, reason: "no total" }; + const diff = round2(itemsSum - reference); + const tolerance = Math.max(1, 0.02 * reference); + const ok = Math.abs(diff) <= tolerance; + return ok + ? { ok, basis, items_sum: itemsSum, reference, diff } + : { ok, basis, items_sum: itemsSum, reference, diff, reason: `items sum to ${itemsSum.toFixed(2)} but the ${basis} is ${reference.toFixed(2)}` }; +} + +/** + * The stored shape — the same keys the order lane writes and + * `components/order-details.tsx` renders (`qty`, `description`, `amount`, + * `options`). A burger called "Michael" is stored as "Michael". + */ +export type StoredLineItem = { + qty: number; + description: string; + amount: number | null; + unit_price?: number | null; + options?: string[]; +}; + +export function toLineItems(items: ExtractedItem[]): StoredLineItem[] { + return (items ?? []) + .filter((i) => typeof i.description === "string" && i.description.trim()) + .map((i) => { + const out: StoredLineItem = { + qty: typeof i.quantity === "number" && i.quantity > 0 ? i.quantity : 1, + description: i.description.trim(), + amount: itemAmount(i), + }; + if (typeof i.unit_price === "number" && Number.isFinite(i.unit_price)) out.unit_price = round2(i.unit_price); + const options = (i.options ?? []).map((o) => String(o).trim()).filter(Boolean); + if (options.length) out.options = options; + return out; + }); +} + +/** Does the stored merchant name denote the extracted one? Exact key, or one contains the other. */ +export function merchantMatches(stored: string | null | undefined, extracted: string | null | undefined): boolean { + const a = merchantKey(stored); + const b = merchantKey(extracted); + if (!a || !b) return false; + return a === b || a.includes(b) || b.includes(a); +} diff --git a/src/lib/line-items.ts b/src/lib/line-items.ts new file mode 100644 index 0000000..b94c846 --- /dev/null +++ b/src/lib/line-items.ts @@ -0,0 +1,124 @@ +import { queryRow, queryRaw } from "@/lib/db"; + +/** + * Replace the itemised receipt behind a transaction (plan §11 C2). + * + * An idempotent UPDATE, never an INSERT of a second row: `expense_metadata. + * transaction_id` is UNIQUE and the order the email pipeline booked already has + * its row — with `line_items = []`, because the items sat behind an auth-gated + * link. Idempotence is also what lets pantry re-run it after every confirm, + * writing the board's current state (plan §5, Codex #3), and what lets undo + * restore `previous_line_items` through the same call. + * + * The row is found on `transaction_id = :id OR matched_transaction_id = :id` — + * a card-settled order has only the latter (CLAUDE.md trap). When the + * transaction has no row at all (a manual transaction, a statement line), one + * is created so the items have somewhere to live; a transaction is NEVER + * created here. + */ + +export type LineItemsSource = "intake" | "pantry"; + +export class TransactionNotFoundError extends Error {} +export class LineItemsValidationError extends Error {} + +export type LineItemRecord = Record & { description: string }; + +/** + * Permissive on purpose: the order lane writes `{qty, description, amount, + * options}`, the receipt lane `{description, qty, amount, unit, category}`, and + * the detail panel reads the overlap. What every producer must supply is a + * description — an item with no name renders as nothing. + */ +export function validateLineItems(input: unknown): LineItemRecord[] { + if (!Array.isArray(input)) throw new LineItemsValidationError("line_items must be an array"); + return input.map((raw, i) => { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) { + throw new LineItemsValidationError(`line_items[${i}] must be an object`); + } + const item = { ...(raw as Record) }; + if (typeof item.description !== "string" || !item.description.trim()) { + throw new LineItemsValidationError(`line_items[${i}].description is required`); + } + item.description = item.description.trim(); + // `quantity` is the extractor's word; `qty` is what the renderer reads. + if (item.qty === undefined && typeof item.quantity === "number") item.qty = item.quantity; + return item as LineItemRecord; + }); +} + +export type ReplaceResult = { + transaction_id: number; + metadata_id: number; + created_metadata: boolean; + previous_line_items: unknown[]; + line_items: LineItemRecord[]; +}; + +export async function replaceLineItems( + transactionId: number, + lineItems: LineItemRecord[], + opts: { source: LineItemsSource; platform?: string | null; batchId?: string | null } +): Promise { + const txn = await queryRow<{ + id: number; + merchant_normalized: string | null; + merchant_name: string | null; + amount: string; + transaction_date: string; + }>( + `SELECT id, merchant_normalized, merchant_name, amount::text, transaction_date::text + FROM transactions WHERE id = $1`, + [transactionId] + ); + if (!txn) throw new TransactionNotFoundError(`transaction ${transactionId} does not exist`); + + // Prefer the row that IS this transaction over one that merely points at it — + // a statement line can be the `matched_transaction_id` of a receipt row and + // carry its own row too, and the items belong on the receipt. + const existing = await queryRow<{ id: number; line_items: unknown }>( + `SELECT id, line_items FROM expense_metadata + WHERE transaction_id = $1 OR matched_transaction_id = $1 + ORDER BY (transaction_id = $1) DESC, id + LIMIT 1`, + [transactionId] + ); + + const json = JSON.stringify(lineItems); + if (existing) { + await queryRaw(`UPDATE expense_metadata SET line_items = $2::jsonb WHERE id = $1`, [existing.id, json]); + return { + transaction_id: transactionId, + metadata_id: existing.id, + created_metadata: false, + previous_line_items: Array.isArray(existing.line_items) ? existing.line_items : [], + line_items: lineItems, + }; + } + + const reference = opts.batchId ? `intake:${opts.batchId}` : `intake:txn:${transactionId}`; + const created = await queryRow<{ id: number }>( + `INSERT INTO expense_metadata ( + transaction_id, source, platform, order_reference, line_items, + merchant_normalized, amount, transaction_date, currency, flags, extraction_model + ) VALUES ($1, $2, $3, $4, $5::jsonb, $6, $7::numeric, $8::date, 'AUD', '[]'::jsonb, 'cloud-budget') + RETURNING id`, + [ + transactionId, + opts.source, + opts.platform ?? null, + reference, + json, + txn.merchant_normalized ?? txn.merchant_name ?? null, + txn.amount, + txn.transaction_date, + ] + ); + return { + transaction_id: transactionId, + metadata_id: created!.id, + created_metadata: true, + previous_line_items: [], + line_items: lineItems, + }; +}