From 028364ac4e3b49cf546f8aff5a399136ab1425b9 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 13:31:53 +0000 Subject: [PATCH 1/9] feat(config): add CRON_SECRET, JOB_DRAIN_INLINE and DEMO_MODE with serverless drain budget (#59) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DJ5vaKvTYiMvdngT4d3xo1 --- src/config/env.test.ts | 55 +++++++++++++++++++++++++++++++++++++++++- src/config/env.ts | 34 ++++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 1 deletion(-) diff --git a/src/config/env.test.ts b/src/config/env.test.ts index 046fa89..4a544bd 100644 --- a/src/config/env.test.ts +++ b/src/config/env.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { loadConfig } from "./env"; +import { loadConfig, SERVERLESS_DRAIN } from "./env"; const valid = { DATABASE_URL: "postgres://app_rw:rw-secret@localhost:5432/requestflow", @@ -98,4 +98,57 @@ describe("loadConfig", () => { expect(loadConfig({ ...valid, ERP_TOKEN: placeholder }).erp.token).toBe(placeholder); expect(() => loadConfig({ ...valid, APP_ENV: "showcase", BETTER_AUTH_SECRET: "s".repeat(40), ERP_TOKEN: placeholder })).toThrow(/ERP_TOKEN/); }); + + describe("serverless job drain and demo mode (#59)", () => { + // Values that fit one processing job into the drain function (60 s × 3 files). + const fits = { AI_SERVICE_TIMEOUT_MS: "60000", UPLOAD_MAX_FILES: "3" }; + const secret = "c".repeat(32); + + it("keeps the drain route, inline drain and demo banner off by default – empty values count as unset", () => { + const defaults = loadConfig(valid); + const empty = loadConfig({ ...valid, CRON_SECRET: "", JOB_DRAIN_INLINE: "", DEMO_MODE: "" }); + + expect(defaults.jobs).toEqual({ cronSecret: undefined, drainInline: false }); + expect(defaults.demoMode).toBe(false); + expect(empty.jobs).toEqual({ cronSecret: undefined, drainInline: false }); + expect(empty.demoMode).toBe(false); + }); + + it("reads CRON_SECRET, JOB_DRAIN_INLINE=true and DEMO_MODE=true", () => { + const config = loadConfig({ ...valid, ...fits, CRON_SECRET: secret, JOB_DRAIN_INLINE: "true", DEMO_MODE: "true" }); + + expect(config.jobs).toEqual({ cronSecret: secret, drainInline: true }); + expect(config.demoMode).toBe(true); + }); + + it("refuses a CRON_SECRET shorter than 24 characters and non-boolean switches, naming only the variable", () => { + let message = ""; + try { + loadConfig({ ...valid, ...fits, CRON_SECRET: "too-short-secret" }); + } catch (error) { + message = String(error); + } + + expect(message).toMatch(/CRON_SECRET/); + expect(message).not.toMatch(/too-short-secret/); + expect(() => loadConfig({ ...valid, JOB_DRAIN_INLINE: "yes" })).toThrow(/JOB_DRAIN_INLINE/); + expect(() => loadConfig({ ...valid, DEMO_MODE: "1" })).toThrow(/DEMO_MODE/); + }); + + it("refuses a serverless drain whose worst-case processing job cannot finish within the function limit", () => { + // Defaults: 120 s AI timeout × 10 files – far beyond one 300 s function run. + expect(() => loadConfig({ ...valid, CRON_SECRET: secret })).toThrow(/AI_SERVICE_TIMEOUT_MS.*UPLOAD_MAX_FILES/); + expect(() => loadConfig({ ...valid, JOB_DRAIN_INLINE: "true" })).toThrow(/AI_SERVICE_TIMEOUT_MS.*UPLOAD_MAX_FILES/); + expect(() => loadConfig({ ...valid, ...fits, CRON_SECRET: secret, JOB_DRAIN_INLINE: "true" })).not.toThrow(); + expect(() => loadConfig({ ...valid, AI_SERVICE_TIMEOUT_MS: "60000", UPLOAD_MAX_FILES: "4", CRON_SECRET: secret })).toThrow(/UPLOAD_MAX_FILES/); + }); + + it("keeps the worst case of one drain run within the function limit", () => { + const { processMs, exportMs, marginMs, maxDurationSeconds } = SERVERLESS_DRAIN; + const worstCaseMs = processMs + 60_000 * 3 + exportMs + 20_000 + marginMs; + + expect(maxDurationSeconds).toBeLessThanOrEqual(300); + expect(worstCaseMs).toBeLessThanOrEqual(maxDurationSeconds * 1000); + }); + }); }); diff --git a/src/config/env.ts b/src/config/env.ts index 6045695..143fac6 100644 --- a/src/config/env.ts +++ b/src/config/env.ts @@ -1,5 +1,16 @@ import { z } from "zod"; +// Empty values count as unset (a `.env` line `CRON_SECRET=` must not fail the length check). +const optional = (inner: T) => z.preprocess((value) => (value === "" ? undefined : value), inner.optional()); +const flag = z.preprocess((value) => (value === "" ? undefined : value), z.enum(["true", "false"]).default("false")); + +/** + * One serverless drain run (#59, ADR-0001 D2/D11): the route and `after()` stop starting new jobs after + * `processMs` resp. `exportMs`, but the job in hand always finishes. `maxDurationSeconds` must equal the + * function limit in `vercel.json` and the routes' `maxDuration` (≤ 300 s on Vercel Hobby). + */ +export const SERVERLESS_DRAIN = { maxDurationSeconds: 300, processMs: 50_000, exportMs: 20_000, marginMs: 30_000 } as const; + const schema = z.object({ DATABASE_URL: z.url({ protocol: /^postgres(ql)?$/ }), S3_ENDPOINT: z.url(), @@ -33,6 +44,11 @@ const schema = z.object({ .string() .default("") .refine((value) => value.split(",").map((entry) => entry.trim()).filter(Boolean).every((entry) => ["503", "lost", "timeout"].includes(entry))), + // Serverless showcase (#59): shared secret of the drain route (unset → route answers 404), inline + // drain after upload/approval/reprocess, and the demo banner. + CRON_SECRET: optional(z.string().min(24)), + JOB_DRAIN_INLINE: flag, + DEMO_MODE: flag, }); const LOCAL_PLACEHOLDER_SECRETS = new Set(["local-dev-only-secret-change-me-0123456789"]); @@ -71,6 +87,13 @@ export interface AppConfig { timeoutMs: number; mock: { enabled: boolean; faults: string }; }; + jobs: { + /** Bearer secret of `/api/jobs/drain`; undefined → the route is off (404). */ + cronSecret: string | undefined; + /** Drain once via `after()` right after upload, approval and reprocess (serverless runtimes). */ + drainInline: boolean; + }; + demoMode: boolean; } // Errors list variable names only – values may be secrets and end up in logs. @@ -89,6 +112,15 @@ export function loadConfig(source: Record = process. if (env.APP_ENV !== "local" && env.ERP_TOKEN && LOCAL_PLACEHOLDER_ERP_TOKENS.has(env.ERP_TOKEN)) { throw new Error("Invalid or missing configuration: ERP_TOKEN"); } + // A serverless drain cannot outlive its function: the worst-case processing job (every document hits + // the AI timeout) plus the export and loop windows must fit, or the platform kills the job mid-run. + if (env.CRON_SECRET !== undefined || env.JOB_DRAIN_INLINE === "true") { + const { processMs, exportMs, marginMs, maxDurationSeconds } = SERVERLESS_DRAIN; + const worstCaseMs = processMs + env.AI_SERVICE_TIMEOUT_MS * env.UPLOAD_MAX_FILES + exportMs + env.ERP_TIMEOUT_MS + marginMs; + if (worstCaseMs > maxDurationSeconds * 1000) { + throw new Error("Invalid or missing configuration: AI_SERVICE_TIMEOUT_MS, UPLOAD_MAX_FILES"); + } + } return { databaseUrl: env.DATABASE_URL, storage: { @@ -113,5 +145,7 @@ export function loadConfig(source: Record = process. timeoutMs: env.ERP_TIMEOUT_MS, mock: { enabled: env.ERP_MOCK_ENABLED === "true", faults: env.ERP_MOCK_FAULTS }, }, + jobs: { cronSecret: env.CRON_SECRET, drainInline: env.JOB_DRAIN_INLINE === "true" }, + demoMode: env.DEMO_MODE === "true", }; } From d2b9f0ab29a8c57479df4fdbae062d4a685d4499 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 13:31:53 +0000 Subject: [PATCH 2/9] feat(jobs): bounded drain route and after() drain for runtimes without a worker (#59) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DJ5vaKvTYiMvdngT4d3xo1 --- src/app/_server/drain-request.test.ts | 129 ++++++++++++++++++ src/app/_server/drain-request.ts | 52 ++++++++ src/app/_server/drain.ts | 40 ++++++ src/app/api/jobs/drain/route.ts | 27 ++++ src/app/api/requests/route.ts | 4 + src/app/requests/[id]/actions.ts | 10 +- src/app/requests/[id]/page.tsx | 2 + src/app/requests/actions.ts | 2 + src/app/requests/page.tsx | 2 + src/job-drain.ts | 58 ++++++++ src/worker.ts | 20 +-- tests/integration/drain-route.test.ts | 81 ++++++++++++ tests/integration/drain.test.ts | 184 ++++++++++++++++++++++++++ 13 files changed, 596 insertions(+), 15 deletions(-) create mode 100644 src/app/_server/drain-request.test.ts create mode 100644 src/app/_server/drain-request.ts create mode 100644 src/app/_server/drain.ts create mode 100644 src/app/api/jobs/drain/route.ts create mode 100644 src/job-drain.ts create mode 100644 tests/integration/drain-route.test.ts create mode 100644 tests/integration/drain.test.ts diff --git a/src/app/_server/drain-request.test.ts b/src/app/_server/drain-request.test.ts new file mode 100644 index 0000000..460a1e3 --- /dev/null +++ b/src/app/_server/drain-request.test.ts @@ -0,0 +1,129 @@ +import { after } from "next/server"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { captureLogs } from "@/features/observability"; +import { handleDrainRequest, scheduleAfterResponse } from "./drain-request"; + +// `after()` is the Next.js runtime boundary (it needs a request scope): replaced by a recorder. +vi.mock("next/server", async (original) => ({ ...(await original()), after: vi.fn() })); + +// The drain route's gate (#59): shared secret, constant-time comparison, feature off without a secret. +// The drain itself is injected here; tests/integration/drain.test.ts drains real queues. +const SECRET = "cron-secret-synthetic-0123456789"; +const summary = { processing: { processed: 1, failed: 0, deadLettered: 0 }, exports: { exported: 1, failed: 0, deadLettered: 0 } }; + +describe("handleDrainRequest", () => { + let lines: string[]; + let restore: () => void; + const run = vi.fn(async () => summary); + const call = (authorization: string | null, { cronSecret = SECRET as string | undefined, method = "GET" } = {}) => + handleDrainRequest( + new Request("http://localhost:3000/api/jobs/drain", { method, headers: authorization === null ? {} : { authorization } }), + { cronSecret, run }, + ); + + beforeEach(() => { + lines = []; + restore = captureLogs(lines); + run.mockClear(); + }); + afterEach(() => restore()); + + it("answers 404 and drains nothing when CRON_SECRET is not configured (feature off)", async () => { + const response = await call(`Bearer ${SECRET}`, { cronSecret: "" }); + + expect(response.status).toBe(404); + expect(run).not.toHaveBeenCalled(); + }); + + it.each([ + ["no Authorization header", null], + ["a wrong secret", "Bearer cron-secret-synthetic-WRONG-000000"], + ["a secret of a different length", "Bearer short"], + ["the right secret without the Bearer scheme", SECRET], + ["the right secret with another scheme", `Basic ${SECRET}`], + ["an empty bearer", "Bearer "], + ])("answers 401 and drains nothing for %s", async (_case, authorization) => { + const response = await call(authorization); + + expect(response.status).toBe(401); + expect(response.headers.get("www-authenticate")).toBe("Bearer"); + expect(run).not.toHaveBeenCalled(); + }); + + it("drains once with the right secret (GET for Vercel Cron and POST) and returns counts only", async () => { + const get = await call(`Bearer ${SECRET}`); + const post = await call(`Bearer ${SECRET}`, { method: "POST" }); + + expect(get.status).toBe(200); + expect(post.status).toBe(200); + expect(await get.json()).toEqual(summary); + expect(get.headers.get("cache-control")).toBe("no-store"); + expect(run).toHaveBeenCalledTimes(2); + }); + + it("never writes the secret or the presented credential into the log", async () => { + await call("Bearer presented-wrong-credential-0000"); + await call(`Bearer ${SECRET}`); + + const log = lines.join("\n"); + expect(log).not.toContain(SECRET); + expect(log).not.toContain("presented-wrong-credential"); + expect(lines.some((line) => JSON.parse(line).event === "jobs.drain_unauthorized")).toBe(true); + }); + + it("answers 503 with a generic body when the drain fails and logs the error class only", async () => { + run.mockRejectedValueOnce(Object.assign(new Error("connect ECONNREFUSED postgres://app_rw:pw@db"), { name: "ConnectionError" })); + + const response = await call(`Bearer ${SECRET}`); + + expect(response.status).toBe(503); + expect(JSON.stringify(await response.json())).not.toMatch(/ECONNREFUSED|postgres/); + expect(lines.join("\n")).not.toMatch(/ECONNREFUSED|app_rw:pw/); + expect(lines.some((line) => JSON.parse(line).code === "ConnectionError")).toBe(true); + }); +}); + +describe("scheduleAfterResponse (JOB_DRAIN_INLINE)", () => { + const afterMock = vi.mocked(after); + let lines: string[]; + let restore: () => void; + + beforeEach(() => { + afterMock.mockReset(); + lines = []; + restore = captureLogs(lines); + }); + afterEach(() => restore()); + + it("schedules nothing when the inline drain is off", () => { + const run = vi.fn(async () => summary); + + scheduleAfterResponse(false, run); + + expect(afterMock).not.toHaveBeenCalled(); + expect(run).not.toHaveBeenCalled(); + }); + + it("schedules exactly one drain after the response when it is on", async () => { + const run = vi.fn(async () => summary); + + scheduleAfterResponse(true, run); + + expect(afterMock).toHaveBeenCalledTimes(1); + expect(run).not.toHaveBeenCalled(); // not before the response is sent + await (afterMock.mock.calls[0]![0] as () => Promise)(); + expect(run).toHaveBeenCalledTimes(1); + }); + + it("swallows a failing drain and logs its class only", async () => { + const run = vi.fn(async () => { + throw Object.assign(new Error("password authentication failed for user app_rw"), { name: "DatabaseError" }); + }); + + scheduleAfterResponse(true, run); + await expect((afterMock.mock.calls[0]![0] as () => Promise)()).resolves.toBeUndefined(); + + expect(lines.join("\n")).not.toMatch(/password|app_rw/); + expect(lines.map((line) => JSON.parse(line))).toContainEqual(expect.objectContaining({ event: "jobs.drain_failed", code: "DatabaseError" })); + }); +}); diff --git a/src/app/_server/drain-request.ts b/src/app/_server/drain-request.ts new file mode 100644 index 0000000..c04859f --- /dev/null +++ b/src/app/_server/drain-request.ts @@ -0,0 +1,52 @@ +import { createHash, timingSafeEqual } from "node:crypto"; +import { after } from "next/server"; +import { logEvent } from "@/features/observability"; + +// Gate of the drain route (#59): `Authorization: Bearer ` – what Vercel Cron sends when +// CRON_SECRET is set. No request body is read and neither the secret nor the presented credential is +// ever logged. Without a configured secret the route does not exist (404). +export interface DrainRequestOptions { + cronSecret: string | undefined; + run: () => Promise; +} + +const json = (status: number, body: unknown, headers: Record = {}) => + Response.json(body, { status, headers: { "cache-control": "no-store", ...headers } }); + +// Hash both sides first: timingSafeEqual needs equal lengths, and the digest hides the secret's length. +const digest = (value: string) => createHash("sha256").update(value, "utf8").digest(); + +export function isAuthorized(authorization: string | null, secret: string): boolean { + const presented = authorization?.startsWith("Bearer ") ? authorization.slice("Bearer ".length) : ""; + return timingSafeEqual(digest(presented), digest(secret)) && presented.length > 0; +} + +/** + * Runs `run` via `after()` once the response is sent – only when `enabled` (JOB_DRAIN_INLINE=true). + * Failures never reach the user's response; they are logged by error class only. + */ +export function scheduleAfterResponse(enabled: boolean, run: () => Promise): void { + if (!enabled) return; + after(async () => { + try { + await run(); + } catch (error) { + logEvent("error", "jobs.drain_failed", {}, { code: error instanceof Error ? error.name : "unknown" }); + } + }); +} + +export async function handleDrainRequest(request: Request, options: DrainRequestOptions): Promise { + if (!options.cronSecret) return json(404, { error: { title: "Nicht gefunden." } }); + if (!isAuthorized(request.headers.get("authorization"), options.cronSecret)) { + logEvent("warn", "jobs.drain_unauthorized", {}, { status: 401 }); + return json(401, { error: { title: "Nicht autorisiert." } }, { "www-authenticate": "Bearer" }); + } + try { + return json(200, await options.run()); + } catch (error) { + // Error messages can quote SQL or connection strings: the class only (IDs/codes rule, #28). + logEvent("error", "jobs.drain_failed", {}, { code: error instanceof Error ? error.name : "unknown" }); + return json(503, { error: { title: "Verarbeitung derzeit nicht möglich." } }); + } +} diff --git a/src/app/_server/drain.ts b/src/app/_server/drain.ts new file mode 100644 index 0000000..224c32d --- /dev/null +++ b/src/app/_server/drain.ts @@ -0,0 +1,40 @@ +import { SERVERLESS_DRAIN } from "@/config/env"; +import { buildJobDeps, drainRound, handledJobs, type DrainRoundResult, type JobDeps } from "@/job-drain"; +import { logEvent } from "@/features/observability"; +import { scheduleAfterResponse } from "./drain-request"; +import { getJobClient, getRuntime } from "./runtime"; + +// Serverless drain of the web process (#59, ADR-0001 D2): the showcase has no worker, so the route +// `/api/jobs/drain` (Vercel Cron, manual trigger) and `after()` following upload, approval and reprocess +// run one bounded round of the SAME handlers as src/worker.ts. Budgets: SERVERLESS_DRAIN (config). +let deps: Promise | undefined; + +function jobDeps(): Promise { + deps ??= getJobClient() + .then((boss) => { + const { config, tenancy, storage } = getRuntime(); + return buildJobDeps(config, { tenancy, storage, boss }); + }) + .catch((error: unknown) => { + deps = undefined; + throw error; + }); + return deps; +} + +/** One bounded round incl. pg-boss maintenance (no supervising worker here). */ +export async function drainNow(): Promise { + const started = Date.now(); + const result = await drainRound(await jobDeps(), { + processMs: SERVERLESS_DRAIN.processMs, + exportMs: SERVERLESS_DRAIN.exportMs, + maintenance: true, + }); + logEvent("info", "jobs.drain_run", {}, { count: handledJobs(result), durationMs: Date.now() - started }); + return result; +} + +/** After upload, approval or reprocess: drain once after the response – only with JOB_DRAIN_INLINE=true. */ +export function drainAfterResponse(): void { + scheduleAfterResponse(getRuntime().config.jobs.drainInline, drainNow); +} diff --git a/src/app/api/jobs/drain/route.ts b/src/app/api/jobs/drain/route.ts new file mode 100644 index 0000000..5286d3f --- /dev/null +++ b/src/app/api/jobs/drain/route.ts @@ -0,0 +1,27 @@ +import { drainNow } from "@/app/_server/drain"; +import { handleDrainRequest } from "@/app/_server/drain-request"; +import { getRuntime } from "@/app/_server/runtime"; +import { logEvent } from "@/features/observability"; + +export const dynamic = "force-dynamic"; +// Must equal SERVERLESS_DRAIN.maxDurationSeconds (config) and vercel.json – a segment-config literal. +export const maxDuration = 300; + +// GET (Vercel Cron) and POST (manual/operator trigger) /api/jobs/drain – one bounded drain round of +// processing and export jobs for runtimes without a worker (#59). `Authorization: Bearer `; +// 404 while CRON_SECRET is unset. No request body is read. +async function handle(request: Request): Promise { + let cronSecret: string | undefined; + try { + cronSecret = getRuntime().config.jobs.cronSecret; + } catch (error) { + // Invalid configuration: generic answer; the log names the variables (never values, see loadConfig). + const names = error instanceof Error ? /configuration: (.+)$/.exec(error.message)?.[1]?.split(", ") : undefined; + logEvent("error", "jobs.drain_config_invalid", {}, { code: "config", names }); + return Response.json({ error: { title: "Verarbeitung derzeit nicht möglich." } }, { status: 503, headers: { "cache-control": "no-store" } }); + } + return handleDrainRequest(request, { cronSecret, run: drainNow }); +} + +export const GET = handle; +export const POST = handle; diff --git a/src/app/api/requests/route.ts b/src/app/api/requests/route.ts index 02c8935..1383066 100644 --- a/src/app/api/requests/route.ts +++ b/src/app/api/requests/route.ts @@ -1,9 +1,12 @@ +import { drainAfterResponse } from "@/app/_server/drain"; import { currentActor, getJobClient, getRuntime } from "@/app/_server/runtime"; import { AuthorizationError } from "@/features/identity"; import { submitUpload, UploadRejected } from "@/features/intake"; import { logEvent } from "@/features/observability"; export const dynamic = "force-dynamic"; +// The inline drain (`after()`, JOB_DRAIN_INLINE) runs inside this function's limit (SERVERLESS_DRAIN). +export const maxDuration = 300; const problem = (status: number, title: string) => Response.json({ error: { title } }, { status }); @@ -36,6 +39,7 @@ export async function POST(request: Request): Promise { try { const boss = await getJobClient(); const result = await submitUpload({ tenancy, storage, boss, limits: config.upload }, actor, files); + drainAfterResponse(); // serverless runtimes only (JOB_DRAIN_INLINE=true): process right away return Response.json(result, { status: 201 }); } catch (error) { if (error instanceof UploadRejected) return problem(422, error.message); diff --git a/src/app/requests/[id]/actions.ts b/src/app/requests/[id]/actions.ts index aed9178..b91dcac 100644 --- a/src/app/requests/[id]/actions.ts +++ b/src/app/requests/[id]/actions.ts @@ -3,6 +3,7 @@ import { headers } from "next/headers"; import { notFound, redirect } from "next/navigation"; import { z } from "zod"; +import { drainAfterResponse } from "@/app/_server/drain"; import { currentActor, getJobClient, getRuntime } from "@/app/_server/runtime"; import { AuthorizationError } from "@/features/identity"; import { approveRequest, confirmNotDuplicate, correctField, rejectAsDuplicate, rejectRequest, ReviewRefused } from "@/features/review"; @@ -54,7 +55,14 @@ export async function approveAction(formData: FormData): Promise { const actor = await actorOrLogin(); const requestId = requestIdOf(formData); const boss = await getJobClient(); - await guarded(requestId, () => approveRequest({ tenancy: getRuntime().tenancy, boss }, actor, requestId), "approved"); + await guarded( + requestId, + async () => { + await approveRequest({ tenancy: getRuntime().tenancy, boss }, actor, requestId); + drainAfterResponse(); // serverless runtimes only (JOB_DRAIN_INLINE=true): export right away + }, + "approved", + ); } export async function rejectAction(formData: FormData): Promise { diff --git a/src/app/requests/[id]/page.tsx b/src/app/requests/[id]/page.tsx index 1548659..e10064b 100644 --- a/src/app/requests/[id]/page.tsx +++ b/src/app/requests/[id]/page.tsx @@ -8,6 +8,8 @@ import { approveAction, confirmNotDuplicateAction, correctFieldAction, rejectAct import { DocumentList, needsAttention, Source, STATUS_LABEL, StatusBadge } from "./review-parts"; export const dynamic = "force-dynamic"; +// Server actions of this page may drain inline via `after()` (JOB_DRAIN_INLINE) – SERVERLESS_DRAIN limit. +export const maxDuration = 300; const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; const dateFormat = new Intl.DateTimeFormat("de-DE", { day: "2-digit", month: "2-digit", year: "numeric", hour: "2-digit", minute: "2-digit", timeZone: "Europe/Berlin" }); diff --git a/src/app/requests/actions.ts b/src/app/requests/actions.ts index f78b07d..ea71c5f 100644 --- a/src/app/requests/actions.ts +++ b/src/app/requests/actions.ts @@ -3,6 +3,7 @@ import { headers } from "next/headers"; import { redirect } from "next/navigation"; import { z } from "zod"; +import { drainAfterResponse } from "@/app/_server/drain"; import { currentActor, getJobClient, getRuntime } from "@/app/_server/runtime"; import { AuthorizationError } from "@/features/identity"; import { reprocessRequest, ReprocessRefused } from "@/features/jobs"; @@ -16,6 +17,7 @@ export async function reprocessAction(formData: FormData): Promise { if (!requestId.success) redirect("/requests?error=refused"); try { await reprocessRequest({ tenancy: getRuntime().tenancy, boss: await getJobClient() }, actor, requestId.data); + drainAfterResponse(); // serverless runtimes only (JOB_DRAIN_INLINE=true): retry right away } catch (error) { if (error instanceof ReprocessRefused || error instanceof AuthorizationError) redirect("/requests?error=refused"); throw error; diff --git a/src/app/requests/page.tsx b/src/app/requests/page.tsx index b1c540a..5711c10 100644 --- a/src/app/requests/page.tsx +++ b/src/app/requests/page.tsx @@ -10,6 +10,8 @@ import { StatusPill } from "./status-pill"; import { UploadForm } from "./upload-form"; export const dynamic = "force-dynamic"; +// Server actions of this page may drain inline via `after()` (JOB_DRAIN_INLINE) – SERVERLESS_DRAIN limit. +export const maxDuration = 300; const dateFormat = new Intl.DateTimeFormat("de-DE", { day: "2-digit", month: "2-digit", year: "numeric", hour: "2-digit", minute: "2-digit", timeZone: "Europe/Berlin" }); const STAGE_LABEL: Record = { processing: "Verarbeitung", export: "Export" }; diff --git a/src/job-drain.ts b/src/job-drain.ts new file mode 100644 index 0000000..564d797 --- /dev/null +++ b/src/job-drain.ts @@ -0,0 +1,58 @@ +// Shared job wiring (module `jobs`, ADR-0001 D2): builds the processing and export dependencies from the +// configuration and runs one bounded drain round. Used by the long-running worker (src/worker.ts) and by +// the serverless drain (src/app/_server/drain.ts: route `/api/jobs/drain` and `after()`), so both run +// the same handlers against the same queues. Composition only – no connections are opened here. +import type { AppConfig } from "@/config/env"; +import { createErpClient, drainExports, type ExportDrainDeps, type ExportDrainResult } from "@/features/export"; +import { createAiServiceClient } from "@/features/extraction"; +import { drain, QUEUES, type DrainDeps, type DrainResult, type JobRunner } from "@/features/jobs"; +import { currentFieldValues } from "@/features/review"; +import type { S3BlobStore } from "@/features/storage"; +import type { Tenancy } from "@/features/tenancy"; + +export interface JobDeps { + processing: DrainDeps; + exports: ExportDrainDeps; +} + +export interface DrainRoundOptions { + /** Stop starting processing jobs after this many ms (the job in hand always finishes). */ + processMs: number; + /** Stop starting export jobs after this many ms. */ + exportMs: number; + /** Run pg-boss maintenance (expiry → retry, dead letter, retention) first – runtimes without a supervising worker. */ + maintenance?: boolean; + /** Queue names – tests use dedicated queues. */ + queues?: { process: { process: string; dead: string }; export: { export: string; dead: string } }; +} + +export interface DrainRoundResult { + processing: DrainResult; + exports: ExportDrainResult; +} + +/** Fail-closed: a missing AI_SERVICE_TOKEN or ERP_TOKEN throws (the error names the variable only). */ +export function buildJobDeps(config: AppConfig, parts: { tenancy: Tenancy; storage: Pick; boss: JobRunner }): JobDeps { + const { tenancy, storage, boss } = parts; + return { + processing: { tenancy, storage, boss, ai: createAiServiceClient(config.aiService) }, + // The export reads the reviewed values through the review module (injected – no module cycle). + exports: { tenancy, boss, erp: createErpClient(config.erp), fieldValues: currentFieldValues }, + }; +} + +export async function drainRound(deps: JobDeps, options: DrainRoundOptions): Promise { + const queues = options.queues ?? { + process: { process: QUEUES.processRequest, dead: QUEUES.processRequestDead }, + export: { export: QUEUES.exportRequest, dead: QUEUES.exportRequestDead }, + }; + if (options.maintenance) { + for (const name of [queues.process.process, queues.process.dead, queues.export.export, queues.export.dead]) await deps.processing.boss.supervise(name); + } + const processing = await drain(deps.processing, { maxMs: options.processMs, queues: queues.process }); + const exports = await drainExports(deps.exports, { maxMs: options.exportMs, queues: queues.export }); + return { processing, exports }; +} + +export const handledJobs = ({ processing, exports }: DrainRoundResult): number => + processing.processed + processing.failed + processing.deadLettered + exports.exported + exports.failed + exports.deadLettered; diff --git a/src/worker.ts b/src/worker.ts index d574298..e2c734b 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -5,13 +5,11 @@ import { setTimeout as sleep } from "node:timers/promises"; import { loadConfig } from "@/config/env"; import { createDatabase } from "@/db"; import { createJobQueue } from "@/db/job-queue-client"; -import { createErpClient, drainExports } from "@/features/export"; -import { createAiServiceClient } from "@/features/extraction"; -import { assertProcessingBudget, drain } from "@/features/jobs"; +import { assertProcessingBudget } from "@/features/jobs"; import { logEvent } from "@/features/observability"; -import { currentFieldValues } from "@/features/review"; import { S3BlobStore } from "@/features/storage"; import { createTenancy } from "@/features/tenancy"; +import { buildJobDeps, drainRound, handledJobs } from "@/job-drain"; const DRAIN_BUDGET_MS = 30_000; const IDLE_MS = 2_000; @@ -19,15 +17,11 @@ const IDLE_MS = 2_000; async function main(): Promise { const config = loadConfig(); assertProcessingBudget({ aiTimeoutMs: config.aiService.timeoutMs, maxFiles: config.upload.maxFiles }); - const ai = createAiServiceClient(config.aiService); - const erp = createErpClient(config.erp); const database = createDatabase(config.databaseUrl, { max: 4 }); const storage = new S3BlobStore(config.storage); const boss = await createJobQueue(config.databaseUrl, { supervise: true, onError: (error) => logEvent("error", "jobs.error", {}, { code: error.name }) }); - const tenancy = createTenancy(database.db); - const deps = { tenancy, storage, ai, boss }; - // The export reads the reviewed values through the review module (injected – no module cycle). - const exportDeps = { tenancy, erp, boss, fieldValues: currentFieldValues }; + // Same handlers as the serverless drain (src/job-drain.ts); pg-boss supervises itself here. + const deps = buildJobDeps(config, { tenancy: createTenancy(database.db), storage, boss }); let running = true; const stop = (signal: string) => { @@ -40,10 +34,8 @@ async function main(): Promise { logEvent("info", "worker.started"); while (running) { try { - const processing = await drain(deps, { maxMs: DRAIN_BUDGET_MS }); - const exports = await drainExports(exportDeps, { maxMs: DRAIN_BUDGET_MS }); - const handled = processing.processed + processing.failed + processing.deadLettered + exports.exported + exports.failed + exports.deadLettered; - if (handled === 0) await sleep(IDLE_MS); + const round = await drainRound(deps, { processMs: DRAIN_BUDGET_MS, exportMs: DRAIN_BUDGET_MS }); + if (handledJobs(round) === 0) await sleep(IDLE_MS); } catch (error) { // Infrastructure hiccup (database/storage): log the class only, back off, keep running. logEvent("error", "worker.drain_failed", {}, { code: error instanceof Error ? error.name : "unknown" }); diff --git a/tests/integration/drain-route.test.ts b/tests/integration/drain-route.test.ts new file mode 100644 index 0000000..6a23cd5 --- /dev/null +++ b/tests/integration/drain-route.test.ts @@ -0,0 +1,81 @@ +import { after } from "next/server"; +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { GET as drainGet, POST as drainPost } from "@/app/api/jobs/drain/route"; +import { POST as upload } from "@/app/api/requests/route"; +import { getJobClient, getRuntime } from "@/app/_server/runtime"; +import { SERVERLESS_DRAIN } from "@/config/env"; +import { drainRound } from "@/job-drain"; +import { companyWithAdmin, createStack, type Stack } from "./helpers/stack"; + +// Wiring of the serverless triggers (#59) in the web process's own runtime (same env as `next start`): +// the route's secret gate with the real configuration, and the upload's `after()` hook with +// JOB_DRAIN_INLINE=true. The drain round itself is replaced here – the shared queues of other runs must +// stay untouched; tests/integration/drain.test.ts drains real jobs with the real round. +const SECRET = "cron-secret-synthetic-0123456789"; +Object.assign(process.env, { + CRON_SECRET: SECRET, + JOB_DRAIN_INLINE: "true", + // A worst-case processing job must fit into one function run, or loadConfig refuses the drain. + AI_SERVICE_TIMEOUT_MS: "60000", + UPLOAD_MAX_FILES: "3", + AI_SERVICE_TOKEN: "t".repeat(24), + ERP_TOKEN: "local-dev-only-erp-token-0123456789", +}); + +const summary = { processing: { processed: 0, failed: 0, deadLettered: 0 }, exports: { exported: 0, failed: 0, deadLettered: 0 } }; +vi.mock("@/job-drain", async (original) => ({ ...(await original()), drainRound: vi.fn(async () => summary) })); +vi.mock("next/server", async (original) => ({ ...(await original()), after: vi.fn() })); + +describe("drain route and inline drain wiring", () => { + let stack: Stack; + let cookie: string; + const request = (authorization?: string) => new Request("http://localhost:3000/api/jobs/drain", { headers: authorization ? { authorization } : {} }); + + beforeAll(async () => { + stack = createStack(); + cookie = (await companyWithAdmin(stack)).cookie; + }); + beforeEach(() => { + vi.mocked(drainRound).mockClear(); + vi.mocked(after).mockClear(); + }); + afterAll(async () => { + await (await getJobClient()).stop({ graceful: false }); + await getRuntime().database.pool.end(); + await stack.close(); + }); + + it("refuses requests without or with a wrong secret (401) and drains nothing", async () => { + expect((await drainGet(request())).status).toBe(401); + expect((await drainGet(request("Bearer cron-secret-synthetic-WRONG-000000"))).status).toBe(401); + expect((await drainPost(request(`Basic ${SECRET}`))).status).toBe(401); + expect(drainRound).not.toHaveBeenCalled(); + }); + + it("with the configured secret runs one round with maintenance and the serverless budgets", async () => { + const response = await drainGet(request(`Bearer ${SECRET}`)); + + expect(response.status).toBe(200); + expect(await response.json()).toEqual(summary); + expect(drainRound).toHaveBeenCalledTimes(1); + expect(vi.mocked(drainRound).mock.calls[0]![1]).toEqual({ processMs: SERVERLESS_DRAIN.processMs, exportMs: SERVERLESS_DRAIN.exportMs, maintenance: true }); + }); + + it("an accepted upload schedules one drain after the response (JOB_DRAIN_INLINE=true)", async () => { + const form = new FormData(); + form.append("files", new File(["%PDF-1.7\n% synthetic inline drain\n"], "anfrage.pdf", { type: "application/pdf" })); + const encoded = new Response(form); + const body = new Uint8Array(await encoded.arrayBuffer()); + const headers = { "content-type": encoded.headers.get("content-type")!, "content-length": String(body.byteLength), cookie }; + + const created = await upload(new Request("http://localhost:3000/api/requests", { method: "POST", body, headers })); + const refused = await upload(new Request("http://localhost:3000/api/requests", { method: "POST", body, headers: { ...headers, cookie: "" } })); + + expect(created.status).toBe(201); + expect(refused.status).toBe(401); + expect(after).toHaveBeenCalledTimes(1); + expect(drainRound).not.toHaveBeenCalled(); // only once the response is sent + await (vi.mocked(after).mock.calls[0]![0] as () => Promise)(); + expect(drainRound).toHaveBeenCalledTimes(1); + }); +}); diff --git a/tests/integration/drain.test.ts b/tests/integration/drain.test.ts new file mode 100644 index 0000000..63dd723 --- /dev/null +++ b/tests/integration/drain.test.ts @@ -0,0 +1,184 @@ +import { randomUUID } from "node:crypto"; +import { createServer, type IncomingMessage, type Server, type ServerResponse } from "node:http"; +import type { AddressInfo } from "node:net"; +import type { PgBoss } from "pg-boss"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { loadConfig, type AppConfig } from "@/config/env"; +import { sendInTransaction } from "@/db/job-queue"; +import { createJobQueue, installJobQueues } from "@/db/job-queue-client"; +import { listAuditEvents } from "@/features/audit"; +import { insertDocuments } from "@/features/documents"; +import { createErpMock, MemoryMockStore, type ErpMock } from "@/features/erp-mock"; +import { drainExports, getExportRecord } from "@/features/export"; +import { persistExtractionRun } from "@/features/extraction"; +import { syntheticExtractResponse } from "@/features/extraction/fixtures"; +import { getActor, type Actor } from "@/features/identity"; +import { QUEUES } from "@/features/jobs"; +import { createRequest, getRequest, lockRequest, transitionRequest } from "@/features/requests"; +import { approveRequest, correctField } from "@/features/review"; +import { S3BlobStore } from "@/features/storage"; +import { createTenancy, type Tenancy } from "@/features/tenancy"; +import { buildJobDeps, drainRound, type DrainRoundOptions, type JobDeps } from "@/job-drain"; +import { companyWithAdmin, createStack, type Stack } from "./helpers/stack"; + +// The serverless drain round (#59) – what `/api/jobs/drain` and `after()` run – against real Postgres, +// pg-boss and S3 storage. AI service and ERP are one local HTTP stub (external I/O boundary; the ERP side +// is the real mock module). Dedicated queues keep the shared queues of other runs untouched. +const TOKEN = "local-dev-only-erp-token-0123456789"; +const suffix = randomUUID().slice(0, 8); +const QUEUES_UNDER_TEST = { + process: { process: `test-drain-process-${suffix}`, dead: `test-drain-process-dead-${suffix}` }, + export: { export: `test-drain-export-${suffix}`, dead: `test-drain-export-dead-${suffix}` }, +}; +const ROUND: DrainRoundOptions = { processMs: 5_000, exportMs: 5_000, maintenance: true, queues: QUEUES_UNDER_TEST }; + +describe("serverless drain round: processing and export jobs, exactly once alongside the worker", () => { + let stack: Stack; + let server: Server; + let storage: S3BlobStore; + let tenancy: Tenancy; + let serverlessBoss: PgBoss; + let workerBoss: PgBoss; + let serverless: JobDeps; + let worker: JobDeps; + let admin: Actor; + let mock: ErpMock; + const erpCalls: string[] = []; + + const requestOf = (id: string) => tenancy.withTenant(admin.companyId, (tx) => getRequest(tx, id)); + + /** A NEW request with a stored mail and a processing job on the test queue – like intake. */ + async function uploaded() { + const requestId = randomUUID(); + const documentId = randomUUID(); + const storageKey = S3BlobStore.documentKey(admin.companyId, requestId, documentId); + await storage.put(storageKey, new TextEncoder().encode("From: einkauf@example.com\r\nSubject: Anfrage\r\n\r\nMusterbau Beispiel GmbH\r\n"), "message/rfc822"); + await tenancy.withTenant(admin.companyId, async (tx) => { + await createRequest(tx, { id: requestId, createdBy: admin.userId }); + await insertDocuments(tx, [{ id: documentId, requestId, filename: "anfrage.eml", contentType: "message/rfc822", kind: "eml", sizeBytes: 10, sha256: "x".repeat(64), storageKey }]); + await sendInTransaction(serverlessBoss, tx, QUEUES_UNDER_TEST.process.process, { requestId, companyId: admin.companyId }, { singletonKey: requestId }); + }); + return requestId; + } + + /** An approved request (review module) whose export job sits on the test queue. */ + async function approved() { + const requestId = randomUUID(); + const documentId = randomUUID(); + await tenancy.withTenant(admin.companyId, async (tx) => { + await createRequest(tx, { id: requestId, createdBy: admin.userId, subject: "Anfrage Flansche DN 100" }); + await insertDocuments(tx, [{ id: documentId, requestId, filename: "anfrage.eml", contentType: "message/rfc822", kind: "eml", sizeBytes: 10, sha256: "x".repeat(64), storageKey: `${admin.companyId}/${requestId}/${documentId}` }]); + const row = await transitionRequest(tx, (await lockRequest(tx, requestId))!, "processing.started", { attempts: 1 }); + await persistExtractionRun(tx, { requestId, jobId: randomUUID(), outcomes: [{ documentId, response: syntheticExtractResponse(documentId) }] }); + await transitionRequest(tx, row, "processing.succeeded"); + }); + await correctField(tenancy, admin, requestId, "company", "Musterbau Beispiel GmbH & Co. KG"); + await approveRequest({ tenancy, boss: serverlessBoss }, admin, requestId); + await stack.database.pool.query("delete from pgboss.job where name = $1 and singleton_key = $2", [QUEUES.exportRequest, requestId]); + await enqueueExport(requestId); + return requestId; + } + const enqueueExport = (requestId: string) => + tenancy.withTenant(admin.companyId, (tx) => sendInTransaction(serverlessBoss, tx, QUEUES_UNDER_TEST.export.export, { requestId, companyId: admin.companyId }, { singletonKey: requestId })); + const exportedEvents = async (id: string) => + (await tenancy.withTenant(admin.companyId, (tx) => listAuditEvents(tx, "request", id))).filter((event) => event.action === "request.exported"); + + beforeAll(async () => { + mock = createErpMock({ token: TOKEN, store: new MemoryMockStore() }); + // One stub for both outbound services: /v1/extract answers synthetic fields, /v1/quote-requests is the ERP mock. + server = createServer((request: IncomingMessage, response: ServerResponse) => { + const chunks: Buffer[] = []; + request.on("data", (chunk: Buffer) => chunks.push(chunk)); + request.on("end", async () => { + const body = Buffer.concat(chunks); + if (request.url === "/v1/extract") { + const documentId = /name="documentId"\r\n\r\n([^\r]+)/.exec(body.toString("latin1"))?.[1] ?? ""; + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify(syntheticExtractResponse(documentId))); + return; + } + erpCalls.push(String(request.headers["idempotency-key"] ?? "")); + const headers = new Headers(Object.entries(request.headers).flatMap(([name, value]) => (typeof value === "string" ? [[name, value] as [string, string]] : []))); + const answer = await mock.handle(new Request(`http://stub${request.url}`, { method: request.method, headers, body })); + response.writeHead(answer.status, Object.fromEntries(answer.headers)); + response.end(Buffer.from(await answer.arrayBuffer())); + }); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const stubUrl = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; + + await installJobQueues(process.env.MIGRATION_DATABASE_URL!, [ + { name: QUEUES_UNDER_TEST.process.dead, policy: "standard" }, + { name: QUEUES_UNDER_TEST.process.process, policy: "exclusive", retryLimit: 1, retryDelay: 0, retryBackoff: false, deadLetter: QUEUES_UNDER_TEST.process.dead }, + { name: QUEUES_UNDER_TEST.export.dead, policy: "standard" }, + { name: QUEUES_UNDER_TEST.export.export, policy: "exclusive", retryLimit: 1, retryDelay: 0, retryBackoff: false, deadLetter: QUEUES_UNDER_TEST.export.dead }, + ]); + const base = loadConfig(); + const config: AppConfig = { + ...base, + aiService: { baseUrl: stubUrl, token: "t".repeat(24), timeoutMs: 2_000 }, + erp: { ...base.erp, baseUrl: stubUrl, token: TOKEN, timeoutMs: 2_000 }, + }; + stack = createStack(); + tenancy = createTenancy(stack.database.db); + storage = new S3BlobStore(base.storage); + // Two pg-boss clients: the serverless function and a long-running worker are separate processes. + serverlessBoss = await createJobQueue(base.databaseUrl); + workerBoss = await createJobQueue(base.databaseUrl); + serverless = buildJobDeps(config, { tenancy, storage, boss: serverlessBoss }); + worker = buildJobDeps(config, { tenancy, storage, boss: workerBoss }); + admin = (await getActor(stack.auth, stack.database.db, new Headers({ cookie: (await companyWithAdmin(stack)).cookie })))!; + }); + + afterAll(async () => { + await serverlessBoss.stop({ graceful: false }); + await workerBoss.stop({ graceful: false }); + storage.destroy(); + server.closeAllConnections(); + server.close(); + await stack.close(); + }); + + it("one round drains a queued processing job (→ REVIEW) and a queued export job (→ EXPORTED) within its budget", async () => { + const processed = await uploaded(); + const exported = await approved(); + const started = Date.now(); + + const round = await drainRound(serverless, ROUND); + + expect(Date.now() - started).toBeLessThan(ROUND.processMs + ROUND.exportMs); + expect(round.processing).toMatchObject({ processed: 1, failed: 0 }); + expect(round.exports).toMatchObject({ exported: 1, failed: 0 }); + expect((await requestOf(processed))?.status).toBe("REVIEW"); + expect((await requestOf(exported))?.status).toBe("EXPORTED"); + }); + + it("an empty queue ends the round at once", async () => { + const started = Date.now(); + + const round = await drainRound(serverless, ROUND); + + expect(round).toEqual({ processing: { processed: 0, failed: 0, deadLettered: 0 }, exports: { exported: 0, failed: 0, deadLettered: 0 } }); + expect(Date.now() - started).toBeLessThan(2_000); + }); + + it("serverless drain and worker race for the same export, then a redelivery: one ERP record, one export row, one audit event", async () => { + const requestId = await approved(); + + await Promise.all([ + drainRound(serverless, ROUND), + drainExports(worker.exports, { maxMs: 5_000, queues: QUEUES_UNDER_TEST.export }), + drainRound(serverless, ROUND), + ]); + // At-least-once delivery: the same export arrives again after it succeeded. + await enqueueExport(requestId); + const redelivered = await drainRound(serverless, ROUND); + + expect((await requestOf(requestId))?.status).toBe("EXPORTED"); + expect(erpCalls.filter((key) => key === requestId)).toHaveLength(1); + expect(mock.created()).toBe(2); // this request + the one from the first test + expect(await tenancy.withTenant(admin.companyId, (tx) => getExportRecord(tx, requestId))).toMatchObject({ status: "succeeded", idempotencyKey: requestId, attempts: 1 }); + expect(await exportedEvents(requestId)).toHaveLength(1); + expect(redelivered.exports).toMatchObject({ exported: 1, failed: 0 }); // job completed as "skipped", no ERP call + }); +}); From ad8cd51e8bb544aaa7a5442b261bfd470f090ec9 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 13:31:53 +0000 Subject: [PATCH 3/9] feat(app): demo banner when DEMO_MODE=true (#59) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DJ5vaKvTYiMvdngT4d3xo1 --- src/app/_components/demo-banner.test.ts | 19 +++++++++++++++++++ src/app/_components/demo-banner.tsx | 9 +++++++++ src/app/globals.css | 10 ++++++++++ src/app/layout.tsx | 13 +++++++++++++ 4 files changed, 51 insertions(+) create mode 100644 src/app/_components/demo-banner.test.ts create mode 100644 src/app/_components/demo-banner.tsx diff --git a/src/app/_components/demo-banner.test.ts b/src/app/_components/demo-banner.test.ts new file mode 100644 index 0000000..3618764 --- /dev/null +++ b/src/app/_components/demo-banner.test.ts @@ -0,0 +1,19 @@ +import { createElement } from "react"; +import { renderToStaticMarkup } from "react-dom/server"; +import { describe, expect, it } from "vitest"; +import { DemoBanner } from "./demo-banner"; + +// DEMO_MODE (#59): the showcase says on every page that it holds synthetic data only (root layout). +describe("DemoBanner", () => { + it("shows a labelled note „Demo – nur synthetische Daten“ when demo mode is on", () => { + const html = renderToStaticMarkup(createElement(DemoBanner, { enabled: true })); + + expect(html).toContain('role="note"'); + expect(html).toContain('class="demo-banner"'); + expect(html).toContain("Demo – nur synthetische Daten"); + }); + + it("renders nothing when demo mode is off", () => { + expect(renderToStaticMarkup(createElement(DemoBanner, { enabled: false }))).toBe(""); + }); +}); diff --git a/src/app/_components/demo-banner.tsx b/src/app/_components/demo-banner.tsx new file mode 100644 index 0000000..474de7a --- /dev/null +++ b/src/app/_components/demo-banner.tsx @@ -0,0 +1,9 @@ +// Showcase notice (DEMO_MODE=true, ADR-0001 D11): visible on every page, above the app header. +export function DemoBanner({ enabled }: { enabled: boolean }) { + if (!enabled) return null; + return ( +
+ Demo – nur synthetische Daten +
+ ); +} diff --git a/src/app/globals.css b/src/app/globals.css index 917b0ed..f04496a 100644 --- a/src/app/globals.css +++ b/src/app/globals.css @@ -108,6 +108,16 @@ code { /* ---------- App shell ---------- */ +/* Showcase notice (DEMO_MODE) – above the sticky header, text 8.8:1 on --warn-bg */ +.demo-banner { + background: var(--warn-bg); + color: var(--warn-text); + border-bottom: 1px solid var(--warn-line); + padding: 6px 28px; + text-align: center; + font-size: 13px; + font-weight: 600; +} .app-header { background: var(--header); border-bottom: 1px solid var(--border); diff --git a/src/app/layout.tsx b/src/app/layout.tsx index 57193eb..49f12fa 100644 --- a/src/app/layout.tsx +++ b/src/app/layout.tsx @@ -1,6 +1,8 @@ import type { Metadata } from "next"; import type { ReactNode } from "react"; +import { getRuntime } from "./_server/runtime"; import { AppHeader } from "./_components/app-header"; +import { DemoBanner } from "./_components/demo-banner"; // Self-hosted IBM Plex (Latin subset, only the weights the design uses) – no third-party requests. import "@fontsource/ibm-plex-sans/latin-400.css"; import "@fontsource/ibm-plex-sans/latin-500.css"; @@ -17,10 +19,21 @@ export const metadata: Metadata = { description: "Quote requests, reviewed beside their source", }; +// Without a valid configuration (e.g. while `next build` prerenders) there is no banner; the pages +// themselves report the configuration error. +function demoMode(): boolean { + try { + return getRuntime().config.demoMode; + } catch { + return false; + } +} + export default function RootLayout({ children }: { children: ReactNode }) { return ( + {children} From 5e08d81b240398b48de506eaf696b802dd7b4a80 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 13:31:53 +0000 Subject: [PATCH 4/9] chore(deploy): vercel.json with drain function limit and daily cron; document new variables (#59) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DJ5vaKvTYiMvdngT4d3xo1 --- .env.example | 12 ++++++++++++ src/config/deployment.test.ts | 32 ++++++++++++++++++++++++++++++++ vercel.json | 11 +++++++++++ 3 files changed, 55 insertions(+) create mode 100644 src/config/deployment.test.ts create mode 100644 vercel.json diff --git a/.env.example b/.env.example index 83aabd3..95e2cbb 100644 --- a/.env.example +++ b/.env.example @@ -76,6 +76,18 @@ ERP_TIMEOUT_MS=10000 ERP_MOCK_ENABLED=true ERP_MOCK_FAULTS= +# --- Serverless showcase (Vercel, #59 – docs/technical/deployment-vercel.md) ---------------------- +# Without a worker process, `/api/jobs/drain` runs one bounded round of processing + export jobs. +# Bearer secret of that route (at least 24 chars; Vercel Cron sends it). Unset → the route answers 404. +# Generate: `openssl rand -base64 32`. Docker Compose runs the worker and leaves it unset. +# CRON_SECRET= +# "true": also drain once via after() right after upload, approval and reprocess (serverless only). +# With CRON_SECRET or this switch set, AI_SERVICE_TIMEOUT_MS × UPLOAD_MAX_FILES must fit into one +# 300 s function run (e.g. 60000 × 3) – otherwise the configuration is refused. +JOB_DRAIN_INLINE=false +# "true": banner „Demo – nur synthetische Daten" on every page (showcase). +DEMO_MODE=false + # --- Web --------------------------------------------------------------------------------------- # Host port of the web container. WEB_PORT=3000 diff --git a/src/config/deployment.test.ts b/src/config/deployment.test.ts new file mode 100644 index 0000000..e9b5114 --- /dev/null +++ b/src/config/deployment.test.ts @@ -0,0 +1,32 @@ +import { readFileSync } from "node:fs"; +import { describe, expect, it } from "vitest"; +import { SERVERLESS_DRAIN } from "./env"; + +// vercel.json and the routes' segment config (a literal Next.js reads statically) must agree with the +// budget loadConfig checks – otherwise a drain could be killed mid-job (#59). +const read = (path: string) => readFileSync(new URL(`../../${path}`, import.meta.url), "utf8"); + +describe("serverless deployment settings", () => { + const vercel = JSON.parse(read("vercel.json")) as { + framework: string; + functions: Record; + crons: Array<{ path: string; schedule: string }>; + }; + + it("gives the drain route the function limit SERVERLESS_DRAIN assumes and a daily cron (Hobby)", () => { + expect(vercel.framework).toBe("nextjs"); + expect(vercel.functions["src/app/api/jobs/drain/route.ts"]?.maxDuration).toBe(SERVERLESS_DRAIN.maxDurationSeconds); + expect(vercel.crons).toEqual([{ path: "/api/jobs/drain", schedule: expect.stringMatching(/^\d+ \d+ \* \* \*$/) }]); + }); + + it.each(["src/app/api/jobs/drain/route.ts", "src/app/api/requests/route.ts", "src/app/requests/page.tsx", "src/app/requests/[id]/page.tsx"])( + "%s exports the same maxDuration (it runs a drain directly or via after())", + (path) => { + expect(read(path)).toContain(`export const maxDuration = ${SERVERLESS_DRAIN.maxDurationSeconds};`); + }, + ); + + it("carries no secrets", () => { + expect(read("vercel.json")).not.toMatch(/secret|token|password|key/i); + }); +}); diff --git a/vercel.json b/vercel.json new file mode 100644 index 0000000..510a2ea --- /dev/null +++ b/vercel.json @@ -0,0 +1,11 @@ +{ + "$schema": "https://openapi.vercel.sh/vercel.json", + "framework": "nextjs", + "installCommand": "pnpm install --frozen-lockfile", + "buildCommand": "pnpm build", + "regions": ["fra1"], + "functions": { + "src/app/api/jobs/drain/route.ts": { "maxDuration": 300 } + }, + "crons": [{ "path": "/api/jobs/drain", "schedule": "0 5 * * *" }] +} From b2a6ebabbb0bd083cfad85eccb286d648b5f1a7d Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 13:34:27 +0000 Subject: [PATCH 5/9] chore(db): Supabase role bootstrap script mirroring the local roles init (#59) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DJ5vaKvTYiMvdngT4d3xo1 --- scripts/supabase-bootstrap.sql | 41 ++++++++++++++++++++++++++++++++++ 1 file changed, 41 insertions(+) create mode 100644 scripts/supabase-bootstrap.sql diff --git a/scripts/supabase-bootstrap.sql b/scripts/supabase-bootstrap.sql new file mode 100644 index 0000000..fd01fc8 --- /dev/null +++ b/scripts/supabase-bootstrap.sql @@ -0,0 +1,41 @@ +-- One-time role bootstrap for the Supabase showcase database (ADR-0001 D7, D11 amendment 2026-09-24). +-- Mirrors docker/postgres/init/01-roles.sh, which only runs in the local container: +-- app_owner – owns schema `app` and `pgboss`, runs migrations (`pnpm setup:deploy`) +-- app_rw – runtime role: no superuser, no CREATEROLE/CREATEDB, NOBYPASSRLS → row-level security +-- always applies +-- Run ONCE as the Supabase `postgres` user over the session pooler (port 5432) or the direct connection, +-- never over the transaction pooler. Passwords are supplied by the operator as psql variables – they are +-- never written into this file, the repository or a migration: +-- +-- read -rs APP_OWNER_PASSWORD; read -rs APP_RW_PASSWORD +-- psql "" -v ON_ERROR_STOP=1 \ +-- -v owner_pw="$APP_OWNER_PASSWORD" -v rw_pw="$APP_RW_PASSWORD" -f scripts/supabase-bootstrap.sql +-- +-- Afterwards the first migration checks the roles again and refuses to run if either could bypass RLS. +\set ON_ERROR_STOP on + +-- Fail (exit code 3) before anything is created when a password variable is missing. +\if :{?owner_pw} +\else + DO $$ BEGIN RAISE EXCEPTION 'psql variable owner_pw is required (-v owner_pw=...)'; END $$; +\endif +\if :{?rw_pw} +\else + DO $$ BEGIN RAISE EXCEPTION 'psql variable rw_pw is required (-v rw_pw=...)'; END $$; +\endif + +SELECT current_database() AS db \gset + +-- All or nothing: a failure leaves no half-created role behind. +BEGIN; + +-- psql quotes :'var' as a literal and :"var" as an identifier – no string building. +CREATE ROLE app_owner LOGIN PASSWORD :'owner_pw' NOSUPERUSER NOCREATEROLE NOBYPASSRLS; +CREATE ROLE app_rw LOGIN PASSWORD :'rw_pw' NOSUPERUSER NOCREATEROLE NOCREATEDB NOBYPASSRLS; +GRANT CONNECT ON DATABASE :"db" TO app_owner, app_rw; +GRANT CREATE ON DATABASE :"db" TO app_owner; +REVOKE CREATE ON SCHEMA public FROM PUBLIC; +COMMIT; + +-- Show the result without secrets: both roles must read f / f / f. +SELECT rolname, rolsuper, rolbypassrls, rolcreaterole FROM pg_roles WHERE rolname IN ('app_owner', 'app_rw') ORDER BY rolname; From 7697c2b06fbb84d8ced07cb7532dd3633db9258f Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 13:34:27 +0000 Subject: [PATCH 6/9] docs: ADR-0001 D11 amendment (Supabase), Vercel runbook, drain route contract (#59) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DJ5vaKvTYiMvdngT4d3xo1 --- CHANGELOG.md | 3 + docs/decisions/ADR-0001-pilot-architecture.md | 42 +++++- docs/decisions/INDEX.md | 3 +- docs/technical/api.md | 17 +++ docs/technical/architecture.md | 16 ++- docs/technical/deployment-vercel.md | 129 ++++++++++++++++++ docs/technical/operations.md | 4 +- 7 files changed, 204 insertions(+), 10 deletions(-) create mode 100644 docs/technical/deployment-vercel.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 8837bcc..0a122a7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -66,3 +66,6 @@ This file records what changes **in the product** – process and session state - Database roles `app_owner` (migrations) and `app_rw` (runtime, no RLS bypass); schema `app`. - Verify commands `pnpm verify:changed`, `pnpm verify`, `pnpm verify:full`; CI runs integration tests against real PostgreSQL + SeaweedFS. +- Showcase preparation (Vercel + Supabase): without a worker, jobs run right after upload, approval and + reprocess and via a protected drain route (daily cron); `DEMO_MODE` shows „Demo – nur synthetische + Daten" on every page; runbook `docs/technical/deployment-vercel.md` and a role bootstrap script. diff --git a/docs/decisions/ADR-0001-pilot-architecture.md b/docs/decisions/ADR-0001-pilot-architecture.md index 958b2a8..2839973 100644 --- a/docs/decisions/ADR-0001-pilot-architecture.md +++ b/docs/decisions/ADR-0001-pilot-architecture.md @@ -3,6 +3,8 @@ - **Status:** Accepted – 2026-09-22 by Fluory (orchestrator); every decision D1–D11 was confirmed individually - **Deviation from the draft:** D8 – the orchestrator chose a Python AI service from day 1 instead of the drafted recommendation (full TypeScript); the draft recommendation is kept as alternative 1 in D8 +- **Amendment 2026-09-24 (D11, also touches D3/D5):** the showcase uses Supabase Postgres + Supabase + Storage instead of Neon + R2 – see "Amendment 2026-09-24" at the end of D11 - **Deciders:** Fluory (orchestrator) · drafted by a Claude session - **Inputs:** `docs/input/2026-09-22-kundenanfrage.md` (customer request), `PROJECT-START.md` (discovery) - **Facts verified:** 2026-09-22 against official docs, registries and provider terms (sources at the end). @@ -610,6 +612,44 @@ exceptions register (`docs/technical/architecture.md`). acceptance, including the AI-service hosting spike. *Heuristic.* **Revisit when** the customer environment is known. +### Amendment 2026-09-24 – Supabase instead of Neon + R2 (decided by Fluory, orchestrator; #59) + +**Decision.** The showcase keeps Vercel (Hobby) for the TS app but uses **Supabase** in +`eu-central-1` (Frankfurt) for both stateful parts: **Supabase Postgres** replaces Neon (D3) and +**Supabase Storage through its S3-compatible API** replaces Cloudflare R2 (D5). One vendor less, +one dashboard, the same region for database and files. Local, CI and the production path are +unchanged (PostgreSQL 17 container, SeaweedFS, the customer's choice). Nothing Supabase-specific +enters the code: the app still talks plain PostgreSQL (Drizzle, pg-boss) and the S3 API. + +**Consequences.** +- **Connections.** At runtime `app_rw` connects through the **Supavisor transaction pooler (port + 6543)** – serverless functions open many short connections. This is pooler-safe: `withTenant()` + sets the tenant with `set_config('app.company_id', …, true)`, i.e. transaction-local, and pg-boss + polls (no LISTEN/NOTIFY, D4). Migrations (`pnpm setup:deploy`, `app_owner`) use the **session + pooler** (port 5432) or the direct connection – DDL and the pg-boss installer need a session. + *To verify at the first deploy:* the pool's `statement_timeout` startup parameter through Supavisor. +- **Roles.** `app_owner`/`app_rw` are created once by the operator with `scripts/supabase-bootstrap.sql` + (mirrors `docker/postgres/init/01-roles.sh`; `app_rw` NOBYPASSRLS); the first migration still + refuses unsafe roles. The Supabase `postgres`/`service_role` credentials are never given to the app. +- **Data API.** The schemas `app`, `pgboss` and our auth schema must **not** be added to the exposed + schemas of the Supabase Data API (PostgREST/GraphQL); the app never uses the Data API, and `anon`/ + `authenticated` get no grants on our schemas. +- **Schema `auth`.** Supabase reserves the schema `auth` for its own Auth service – our Better Auth + schema of the same name cannot be deployed there. It is renamed to `identity` in a separate issue + (#60); the showcase deploy waits for it. +- **Storage.** A **private** bucket; S3 access keys from the Storage settings (server-side only), + endpoint `https://.storage.supabase.co/storage/v1/s3`, path-style URLs. The R2 + "EU jurisdiction on the free plan" open point is obsolete. +- **Jobs.** Unchanged from D2: no worker on Vercel. A protected route `/api/jobs/drain` + (`CRON_SECRET`) runs one bounded round of `drain()` + `drainExports()` incl. pg-boss maintenance; + `after()` triggers it after upload, approval and reprocess (`JOB_DRAIN_INLINE=true`). The Hobby cron + runs **at most once per day** – the exceptions-register entry "no unattended retries" still applies. + Worst case of one run (AI timeout × files + windows) must fit into the 300 s function limit; + `loadConfig` enforces it. +- **AI service host** is still open (spike). Recommendation: **Google Cloud Run in the EU** (same GCP + project as Vertex `eu`, container image already exists, no 5 GB package or 300 s limit for docling/OCR). +- **Runbook:** `docs/technical/deployment-vercel.md`. + --- ## Summary of the challenged decisions @@ -688,7 +728,7 @@ explicit requirements. - pg-boss maintenance API for serverless `drain()`. - docling EML/MSG coverage and its provenance granularity for XLSX cells. - Google Gen AI SDK configuration for the Vertex `eu` endpoint. -- R2 EU jurisdiction on the free plan. +- ~~R2 EU jurisdiction on the free plan.~~ Obsolete – Supabase Storage instead (D11 amendment 2026-09-24). - The AI-service showcase host (spike). ## Sources (verified 2026-09-22) diff --git a/docs/decisions/INDEX.md b/docs/decisions/INDEX.md index 17ef4db..0f225b6 100644 --- a/docs/decisions/INDEX.md +++ b/docs/decisions/INDEX.md @@ -1,7 +1,8 @@ # Decision index ADRs are immutable once accepted; a new ADR supersedes an old one with a note ("supersedes ADR-x"). +A dated amendment decided by the orchestrator is appended to the decision it changes and listed here. | ADR | Date | Title | Status | |---|---|---|---| -| [ADR-0001](ADR-0001-pilot-architecture.md) | 2026-09-22 | Pilot architecture baseline (D1–D11) | Accepted | +| [ADR-0001](ADR-0001-pilot-architecture.md) | 2026-09-22 | Pilot architecture baseline (D1–D11) | Accepted; amended 2026-09-24 (D11: Supabase instead of Neon + R2) | diff --git a/docs/technical/api.md b/docs/technical/api.md index 56e9c15..5b2b26c 100644 --- a/docs/technical/api.md +++ b/docs/technical/api.md @@ -41,6 +41,23 @@ Request (NEW), documents, audit event and the `request-process` job commit in on Duplicates: same `Message-ID` (read from `.eml` only – `.msg` Message-ID extraction is a follow-up) or the same set of file hashes within the company; checks are serialised per company. +## `GET` / `POST /api/jobs/drain` (shared secret, only with `CRON_SECRET` set) + +One bounded drain round for runtimes without a worker (#59): pg-boss maintenance, then processing jobs +(new ones for up to 50 s), then export jobs (up to 20 s) – the job in hand always finishes. Vercel Cron +calls it with GET; operators may use POST. Header `Authorization: Bearer `, compared in +constant time; no body is read, neither secret nor header is logged. Function limit 300 s. + +| Status | Body | +|---|---| +| 200 | `{"processing":{"processed":n,"failed":n,"deadLettered":n},"exports":{"exported":n,"failed":n,"deadLettered":n}}` – counts only | +| 401 | `{"error":{"title":"Nicht autorisiert."}}` + `WWW-Authenticate: Bearer` – missing or wrong secret | +| 404 | `CRON_SECRET` not configured (feature off, e.g. Docker Compose with its worker) | +| 503 | `{"error":{"title":"Verarbeitung derzeit nicht möglich."}}` – invalid configuration or infrastructure error (log: variable names / error class) | + +Responses carry `cache-control: no-store`. Concurrent calls, `after()` drains and a worker are safe: +pg-boss hands each job to one caller, and the export stays exactly-once (D9). + ## `POST /api/erp-mock/v1/quote-requests` (ERP mock, only with `ERP_MOCK_ENABLED=true`) The simulated ERP of the pilot (ADR-0001 D9) – contract `contracts/erp-export.openapi.yaml`. Without the diff --git a/docs/technical/architecture.md b/docs/technical/architecture.md index b021e64..fe75129 100644 --- a/docs/technical/architecture.md +++ b/docs/technical/architecture.md @@ -33,12 +33,12 @@ Every new file belongs to one of these modules – otherwise add the module here | `identity` | `src/features/identity/` | Better Auth, users, companies, roles | public login route | personal (staff) | rate limit, invite-only | built: Better Auth (invite-only, organization + admin plugins), `authorize()`, audited invite, user management (`/users`: roles, deactivate/reactivate, last-admin rule), seed | | `tenancy` | `src/features/tenancy/` | `withTenant()`, RLS policies | internal | – | forced RLS, `app_rw` without BYPASSRLS | built: `withTenant()`, forced RLS on `app.*`, guard test (every `app` table: `company_id`, forced RLS, only company policies; allow-list empty) | | `audit` | `src/features/audit/` | append-only audit events | internal | personal (staff) | INSERT/SELECT only | partial: `recordAudit()` (append-only enforced by grants) | -| `jobs` | `src/features/jobs/`, entrypoint `src/worker.ts` | pg-boss, job handlers, `drain()`, worker entrypoint | internal | IDs only | transactional enqueue | built: queues, transactional enqueue, handler, `drain()`, dead letter → ERROR, reprocess, worker loop | +| `jobs` | `src/features/jobs/`, entrypoint `src/worker.ts`, shared wiring `src/job-drain.ts` | pg-boss, job handlers, `drain()`, worker entrypoint, one drain round (processing + export) for worker and serverless | internal | IDs only | transactional enqueue | built: queues, transactional enqueue, handler, `drain()`, dead letter → ERROR, reprocess, worker loop, serverless drain (#59) | | `storage` | `src/features/storage/` | `BlobStore` port + S3 adapter | internal | confidential | private bucket, access via app routes | built: S3 adapter (put/get/delete, bucket setup, ping) | | `observability` | `src/features/observability/` | logger, health, request-list ops data | `/api/health` | IDs only | no PII in logs | built: pino JSON logs with a fixed key set (IDs + codes only, tested on a full run), health (database, storage; informational AI-service reachability and queue backlog) | | `db` | `src/db/`, deploy step `src/setup.ts` | Drizzle schema, migrations, DB roles | internal | – | migrations as owner role | built: roles check, schema `app`, default grants for `app_rw` | -| `config` | `src/config/` | typed runtime configuration, validated at start (zod) | internal | secrets (in memory only) | errors name variables, never values | built | -| `app` | `src/app/` | Next.js routes and pages; composition root `src/app/_server/` (pool, storage client) | `/`, `/login`, `/signup`, `/invite`, `/requests`, `/requests/:id`, `/api/requests`, `/api/documents/:id`, `/api/auth/*`, `/api/health`, `/api/erp-mock/v1/quote-requests` (flag) | – | calls module APIs only (dependency-cruiser) | partial: login, sign-up, invite, home, requests, review page, ERP mock route | +| `config` | `src/config/` | typed runtime configuration, validated at start (zod); serverless drain budget | internal | secrets (in memory only) | errors name variables, never values | built | +| `app` | `src/app/` | Next.js routes and pages; composition root `src/app/_server/` (pool, storage client, serverless drain) | `/`, `/login`, `/signup`, `/invite`, `/requests`, `/requests/:id`, `/api/requests`, `/api/documents/:id`, `/api/auth/*`, `/api/health`, `/api/erp-mock/v1/quote-requests` (flag), `/api/jobs/drain` (`CRON_SECRET`) | – | calls module APIs only (dependency-cruiser); drain route: bearer secret, constant-time | partial: login, sign-up, invite, home, requests, review page, ERP mock route | | AI service | `services/ai/` | docling parsing, extraction, grounding, evals | internal HTTP | confidential + personal (transient) | bearer token, stateless, no DB/storage access | built: eval runner + 15 weighted synthetic cases + replay gate in CI (#24); `POST /v1/extract` – EML, MSG (recursive attachments), PDF (text layer; scans via OCR with `AI_PDF_OCR=auto`), XLSX (rows), DOCX (paragraphs, table cells) → segments with stable locators, bounded OOXML/MSG parsing, 6 header fields + line items (schema v2, prompt `extract_v2`), normalisers (German numbers, units, dates, calendar weeks → at most `uncertain`), grounding verifier, bearer auth; Vertex adapter with recorded responses (live call unverified) | | Contracts | `contracts/` | OpenAPI: AI service, ERP export | – | – | contract tests | built: `ai-service.openapi.yaml` (generated from the service), `erp-export.openapi.yaml` (hand-written); TS types + drift tests | @@ -49,7 +49,7 @@ Deliberately accepted risks – without an entry here a deviation counts as a de | Exception | Why accepted | Owner | Expires | |---|---|---|---| | No RLS on the `auth` and `pgboss` schemas | Not company-owned business data; reachable only by server code (ADR-0001 D7) | Fluory | 2026-12-31 (review at M3) | -| Showcase without unattended retries (Vercel Hobby cron once/day) | Showcase only; production runs a worker (D2) | Fluory | when a production-like demo is needed | +| Showcase without unattended retries (Vercel Hobby cron once/day) | Showcase only; production runs a worker (D2). Jobs run via `after()` on upload/approval/reprocess and `/api/jobs/drain` (#59) | Fluory | when a production-like demo is needed | | Better Auth admin plugin mounted without any holder of its admin role | ADR-0001 D6 names the plugin; decided in #30: kept – its `banned` field implements deactivation (sign-in blocked by the plugin). Nobody holds `platform-admin`, so `/api/auth/admin/*` rejects every caller (tested); user management runs through `identity` | Fluory | 2026-12-31 (review at M3) | | Upload endpoint without a per-user rate limit | Authenticated staff only; body bounded by `Content-Length` + `UPLOAD_MAX_REQUEST_BYTES` before reading | Fluory | before any public deployment (#19) | | `.msg` uploads checked by OLE signature only | Structure check of Outlook messages needs a CFB parser; files are served only as attachments with `nosniff` and parsed later by the stateless AI service | Fluory | with #23 (MSG parsing) | @@ -63,6 +63,7 @@ worker ─► jobs.drain ─► extraction ─► AI service (bytes in, segments ─► requests(REVIEW) + fields + audit ── one transaction review ─► corrections + approve ─► requests(APPROVED) + export job + audit worker ─► export ─► ERP (Idempotency-Key) ─► requests(EXPORTED) +showcase (no worker): after() / cron ─► /api/jobs/drain ─► the same drain round (src/job-drain.ts) failure at any step ─► retry with backoff ─► dead letter ─► requests(ERROR, visible cause) ``` @@ -70,8 +71,9 @@ failure at any step ─► retry with backoff ─► dead letter ─► requests | Service | Purpose | Environments | |---|---|---| -| PostgreSQL 17 | all state incl. queue and auth | local container · showcase Neon (aws-eu-central-1) | -| S3-compatible storage | original mails and attachments | local SeaweedFS · showcase Cloudflare R2 (EU jurisdiction) | +| PostgreSQL 17 | all state incl. queue and auth | local container · showcase Supabase Postgres (eu-central-1, Supavisor transaction pooler) | +| S3-compatible storage | original mails and attachments | local SeaweedFS · showcase Supabase Storage (S3 API, private bucket) | +| Vercel (Hobby) | showcase runtime of the TS app, daily cron | showcase only – runbook [deployment-vercel.md](deployment-vercel.md) (ADR-0001 D11 amendment 2026-09-24) | | Vertex AI (`eu` endpoint, gemini-3.5-flash) | extraction | showcase + customer; local dev may use the Gemini free tier with synthetic data | | ERP | export target | pilot: `erp-mock`; contract `contracts/erp-export.openapi.yaml` | @@ -81,7 +83,7 @@ No secrets in this document; every variable is documented in `.env.example`. | Role | Created by | Used by | Properties | |---|---|---|---| -| `app_owner` | `docker/postgres/init/01-roles.sh` (password from env) | migrations (`src/setup.ts`, `MIGRATION_DATABASE_URL`) | owns schema `app`; no superuser, NOBYPASSRLS | +| `app_owner` | `docker/postgres/init/01-roles.sh` (password from env); Supabase: `scripts/supabase-bootstrap.sql` | migrations (`src/setup.ts`, `MIGRATION_DATABASE_URL`) | owns schema `app`; no superuser, NOBYPASSRLS | | `app_rw` | same | web + worker (`DATABASE_URL`) | USAGE on `app`, no CREATE; DML via default privileges; no superuser, NOBYPASSRLS – RLS always applies | The first migration refuses to run if either role is missing or could bypass RLS. diff --git a/docs/technical/deployment-vercel.md b/docs/technical/deployment-vercel.md new file mode 100644 index 0000000..adce36e --- /dev/null +++ b/docs/technical/deployment-vercel.md @@ -0,0 +1,129 @@ +# Showcase deployment – Vercel + Supabase + +> Runbook for the operator (orchestrator). Decision: ADR-0001 D11, amendment 2026-09-24. Everything +> here uses **synthetic data only**. No secret belongs in this file, the repository, an issue or a chat – +> keep them in a password manager and the Vercel / Supabase / GCP settings. + +## 0. Before you start + +| Blocker | Why | +|---|---| +| #60 merged (Better Auth schema `auth` → `identity`) | Supabase reserves the schema `auth`; the migrations fail on Supabase until then | +| AI service host chosen and running | The drain calls it; recommendation: **Google Cloud Run in the EU** (same GCP project as Vertex `eu`, existing image `services/ai`, no 300 s / package limits) | + +What runs where: Vercel (Hobby) runs the Next.js app, the drain route and the ERP mock +(`/api/erp-mock`). Supabase (`eu-central-1`) holds Postgres (incl. the pg-boss queue) and the files. +There is **no worker process** – jobs run through `/api/jobs/drain` (see step 7). + +## 1. Supabase project + +1. Create a project in region **eu-central-1 (Frankfurt)**. Store the generated `postgres` password in + the password manager – the app never uses it. +2. **Data API:** Project Settings → Data API: keep the exposed schemas at `public` (or turn the Data API + off). Never add `app`, `pgboss` or `identity` – the app does not use the Data API. +3. **Storage:** create a **private** bucket `requestflow-documents` (no public access, no RLS policies + for `anon`/`authenticated`). Storage → S3 connection: enable the S3 protocol and create an **S3 access + key** (access key id + secret). Note the endpoint `https://.storage.supabase.co/storage/v1/s3`. +4. **Connection strings** (Connect dialog), host `aws-0-eu-central-1.pooler.supabase.com` (check yours): + - transaction pooler, port **6543** – runtime (`DATABASE_URL`, user `app_rw.`) + - session pooler, port **5432** – bootstrap and migrations (users `postgres.`, `app_owner.`) + +## 2. Database roles (once) + +Generate two strong passwords, then as the `postgres` user over the **session pooler**: + +```bash +read -rs APP_OWNER_PASSWORD; read -rs APP_RW_PASSWORD +psql "postgresql://postgres.@:5432/postgres" -v ON_ERROR_STOP=1 \ + -v owner_pw="$APP_OWNER_PASSWORD" -v rw_pw="$APP_RW_PASSWORD" -f scripts/supabase-bootstrap.sql +``` + +The last output lists `app_owner` and `app_rw` with `f | f | f` (no superuser, no RLS bypass, no role +creation). The script is all-or-nothing; running it twice fails on "role already exists" – that is fine. + +## 3. AI service + +Deploy `services/ai` (Cloud Run EU recommended) with Vertex `eu` credentials (a service account of the +GCP project, never an API key in the repo) and a random `AI_SERVICE_TOKEN` (≥ 24 chars). The Gemini +free tier is **not** allowed for the showcase (D8). Note its HTTPS URL. + +## 4. Vercel project + +Import the GitHub repository (framework Next.js is set by `vercel.json`, function region `fra1`, +drain route limit 300 s, one cron). Set the variables below for **Production**. Do not give Preview +deployments the showcase database – leave Preview variables empty (previews then fail closed) or use a +separate Supabase project. + +| Variable | Value | +|---|---| +| `APP_ENV` | `showcase` | +| `DATABASE_URL` | `postgresql://app_rw.:@:6543/postgres` (transaction pooler) | +| `S3_ENDPOINT` / `S3_REGION` / `S3_BUCKET` | `https://.storage.supabase.co/storage/v1/s3` / `eu-central-1` / `requestflow-documents` | +| `S3_ACCESS_KEY_ID` / `S3_SECRET_ACCESS_KEY` | the Supabase S3 access key | +| `S3_FORCE_PATH_STYLE` | `true` | +| `BETTER_AUTH_SECRET` | `openssl rand -base64 32` | +| `BETTER_AUTH_URL` | `https://.vercel.app` | +| `AUTH_IP_HEADERS` | `x-real-ip` (set by Vercel; confirm in the smoke check) | +| `AI_SERVICE_URL` / `AI_SERVICE_TOKEN` | URL from step 3 / the same token as the AI service | +| `AI_SERVICE_TIMEOUT_MS` / `UPLOAD_MAX_FILES` | `60000` / `3` – one job must fit into one 300 s run (enforced at start) | +| `ERP_BASE_URL` | `https://.vercel.app/api/erp-mock` | +| `ERP_TOKEN` | random, ≥ 24 chars (not the local default) | +| `ERP_MOCK_ENABLED` | `true` | +| `CRON_SECRET` | `openssl rand -base64 32` – Vercel Cron sends it as `Authorization: Bearer …` | +| `JOB_DRAIN_INLINE` | `true` | +| `DEMO_MODE` | `true` | + +`.env.example` documents every variable. `MIGRATION_DATABASE_URL` and `SEED_PASSWORD` are **not** set on +Vercel – they are only used from the operator's shell (steps 5 and 6). + +## 5. First migration + +From a checkout on the operator's machine (`pnpm install`), export the Production variables of step 4 in +the shell plus `MIGRATION_DATABASE_URL=postgresql://app_owner.:@:5432/postgres` +(**session** pooler), then run `pnpm setup:deploy`. It applies the migrations, installs the pg-boss +queues and checks the bucket. It is idempotent – repeat it after every release with new migrations, +**before** promoting the deployment. + +## 6. Demo accounts + +With the same shell variables and a strong `SEED_PASSWORD` (≥ 12 chars, password manager), `pnpm seed:demo` +creates the two synthetic companies with their admins (invite-only; see `src/seed.ts`). Hand the +accounts out only to people who may see the demo. + +## 7. Jobs without a worker + +- `after()`: with `JOB_DRAIN_INLINE=true`, upload, approval and "Erneut verarbeiten" drain once after the + response – the normal path needs nothing else. +- Cron: `vercel.json` calls `GET /api/jobs/drain` **once a day** (05:00 UTC) – the Hobby maximum. It picks + up retries with backoff that no user action triggered. Faster unattended retries need a paid plan or a + worker (exceptions register). +- Manual: `curl -fsS -H "Authorization: Bearer $CRON_SECRET" https:///api/jobs/drain` (GET or + POST) returns counts only. Without the right secret → 401; without `CRON_SECRET` → 404. +- One run stops taking new jobs after 50 s (processing) + 20 s (export); a job in hand always finishes. + If the platform still kills a run, the job becomes visible again after its pg-boss expiry (1 h). + +## 8. Smoke check + +1. `GET /api/health` → 200, `database` and `storage` `ok`, `aiService` `ok`. +2. Drain route: without header → 401; with the secret → 200 JSON. +3. The banner „Demo – nur synthetische Daten" is visible on the login page. +4. Sign in as the seeded admin, upload a synthetic `.eml` (like the one in `tests/e2e/review-smoke.spec.ts`) + → the request reaches *Zur Prüfung* within about a minute; approve it → *Exportiert* with an ERP reference. +5. Vercel function logs show `jobs.drain_run` lines (IDs and counts only). Confirm the client IP header + (rate limit) and that the pool's `statement_timeout` is accepted by Supavisor. + +## 9. Switch off and roll back + +- Pause the showcase: remove `CRON_SECRET` and set `JOB_DRAIN_INLINE=false` (route → 404, no drains), or + pause the Supabase project. +- Code: Vercel "Instant Rollback" to the previous deployment. Migrations are forward-only + (`docs/technical/operations.md` → Rollback); roll back code only to a version that knows the schema. +- Rotate a leaked secret in its settings page, then redeploy (Vercel reads variables at deploy time). + +## Known limits + +- The ERP mock keeps its idempotency store in memory **per function instance**, so its replay answer + holds only within one instance. The app side (unique export row + row lock, D9) still sends each + approved request once; keep `ERP_MOCK_FAULTS` empty on the showcase (a simulated lost answer could be + replayed on another instance). +- Unattended retries only once a day (see 7). diff --git a/docs/technical/operations.md b/docs/technical/operations.md index 89a7267..d9cf131 100644 --- a/docs/technical/operations.md +++ b/docs/technical/operations.md @@ -101,7 +101,9 @@ migration aborts if `app_owner`/`app_rw` are missing or could bypass RLS – fix `docker/postgres/init/01-roles.sh` runs only in the local container. On any other PostgreSQL (showcase, customer) an operator creates `app_owner` and `app_rw` once with the same statements (passwords from -the secret manager), before the first `setup` run; the first migration refuses to run otherwise. +the secret manager), before the first `setup` run; the first migration refuses to run otherwise. For +Supabase use `scripts/supabase-bootstrap.sql`; the whole showcase setup (Vercel + Supabase, jobs without +a worker, switch-off and rollback) is in [deployment-vercel.md](deployment-vercel.md). ## Login rate limit and client IP From 308ac9b13d1bdfad1a6579f0fa78edef30594dd6 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 13:40:25 +0000 Subject: [PATCH 7/9] fix(config): name ERP_TIMEOUT_MS in the drain budget error; runbook: Fluid compute (#59) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DJ5vaKvTYiMvdngT4d3xo1 --- docs/technical/deployment-vercel.md | 3 ++- src/config/env.test.ts | 4 ++-- src/config/env.ts | 2 +- 3 files changed, 5 insertions(+), 4 deletions(-) diff --git a/docs/technical/deployment-vercel.md b/docs/technical/deployment-vercel.md index adce36e..9ef3124 100644 --- a/docs/technical/deployment-vercel.md +++ b/docs/technical/deployment-vercel.md @@ -50,7 +50,8 @@ free tier is **not** allowed for the showcase (D8). Note its HTTPS URL. ## 4. Vercel project Import the GitHub repository (framework Next.js is set by `vercel.json`, function region `fra1`, -drain route limit 300 s, one cron). Set the variables below for **Production**. Do not give Preview +drain route limit 300 s, one cron). Keep **Fluid compute** enabled (the Hobby default) – without it the +300 s function limit is not available; verify it in Project Settings → Functions. Set the variables below for **Production**. Do not give Preview deployments the showcase database – leave Preview variables empty (previews then fail closed) or use a separate Supabase project. diff --git a/src/config/env.test.ts b/src/config/env.test.ts index 4a544bd..8cd527d 100644 --- a/src/config/env.test.ts +++ b/src/config/env.test.ts @@ -137,8 +137,8 @@ describe("loadConfig", () => { it("refuses a serverless drain whose worst-case processing job cannot finish within the function limit", () => { // Defaults: 120 s AI timeout × 10 files – far beyond one 300 s function run. - expect(() => loadConfig({ ...valid, CRON_SECRET: secret })).toThrow(/AI_SERVICE_TIMEOUT_MS.*UPLOAD_MAX_FILES/); - expect(() => loadConfig({ ...valid, JOB_DRAIN_INLINE: "true" })).toThrow(/AI_SERVICE_TIMEOUT_MS.*UPLOAD_MAX_FILES/); + expect(() => loadConfig({ ...valid, CRON_SECRET: secret })).toThrow(/AI_SERVICE_TIMEOUT_MS, ERP_TIMEOUT_MS, UPLOAD_MAX_FILES/); + expect(() => loadConfig({ ...valid, JOB_DRAIN_INLINE: "true" })).toThrow(/AI_SERVICE_TIMEOUT_MS, ERP_TIMEOUT_MS, UPLOAD_MAX_FILES/); expect(() => loadConfig({ ...valid, ...fits, CRON_SECRET: secret, JOB_DRAIN_INLINE: "true" })).not.toThrow(); expect(() => loadConfig({ ...valid, AI_SERVICE_TIMEOUT_MS: "60000", UPLOAD_MAX_FILES: "4", CRON_SECRET: secret })).toThrow(/UPLOAD_MAX_FILES/); }); diff --git a/src/config/env.ts b/src/config/env.ts index 143fac6..d26cb3d 100644 --- a/src/config/env.ts +++ b/src/config/env.ts @@ -118,7 +118,7 @@ export function loadConfig(source: Record = process. const { processMs, exportMs, marginMs, maxDurationSeconds } = SERVERLESS_DRAIN; const worstCaseMs = processMs + env.AI_SERVICE_TIMEOUT_MS * env.UPLOAD_MAX_FILES + exportMs + env.ERP_TIMEOUT_MS + marginMs; if (worstCaseMs > maxDurationSeconds * 1000) { - throw new Error("Invalid or missing configuration: AI_SERVICE_TIMEOUT_MS, UPLOAD_MAX_FILES"); + throw new Error("Invalid or missing configuration: AI_SERVICE_TIMEOUT_MS, ERP_TIMEOUT_MS, UPLOAD_MAX_FILES"); } } return { From 438551a8588a83adc2b604be1a1a797f3a7b6788 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 16:46:23 +0000 Subject: [PATCH 8/9] fix(deploy): upload cap per person, drain budget counts the request's own time, role-level statement timeout (#59 review) - UPLOAD_MAX_PER_HOUR (required with APP_ENV=showcase): 429 before anything is stored - after() drains subtract the time the invocation already spent - bootstrap: passwords via \prompt (not on the command line), ALTER ROLE app_rw SET statement_timeout - Bearer scheme case-insensitive; runbook and ADR details Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DJ5vaKvTYiMvdngT4d3xo1 --- .env.example | 3 +++ CHANGELOG.md | 2 ++ docs/decisions/ADR-0001-pilot-architecture.md | 8 ++++--- docs/technical/architecture.md | 2 +- docs/technical/deployment-vercel.md | 20 ++++++++++------ scripts/supabase-bootstrap.sql | 22 ++++++++++------- src/app/_server/drain-request.test.ts | 20 +++++++++++++++- src/app/_server/drain-request.ts | 11 ++++++++- src/app/_server/drain.ts | 20 ++++++++++------ src/app/api/requests/route.ts | 6 +++-- src/app/requests/[id]/actions.ts | 3 ++- src/app/requests/actions.ts | 3 ++- src/config/env.test.ts | 9 +++++++ src/config/env.ts | 11 ++++++++- src/features/intake/files.ts | 8 +++++++ src/features/intake/index.ts | 2 +- src/features/intake/submit.ts | 16 ++++++++++--- src/features/requests/index.ts | 1 + src/features/requests/repository.ts | 12 +++++++++- tests/integration/intake.test.ts | 24 ++++++++++++++++++- 20 files changed, 164 insertions(+), 39 deletions(-) diff --git a/.env.example b/.env.example index 95e2cbb..20bd28c 100644 --- a/.env.example +++ b/.env.example @@ -53,6 +53,9 @@ UPLOAD_MAX_FILE_BYTES=20971520 UPLOAD_MAX_FILES=10 # Cap of one whole upload request (all files + form overhead); requests without Content-Length are refused. UPLOAD_MAX_REQUEST_BYTES=41943040 +# Uploads per person and hour – a cost cap (every upload starts AI calls). Empty = no cap (local, CI); +# required with APP_ENV=showcase. +UPLOAD_MAX_PER_HOUR= # --- AI service (services/ai) -------------------------------------------------------------------- # Called by the worker only. The token must equal the service's AI_SERVICE_TOKEN (at least 24 chars). diff --git a/CHANGELOG.md b/CHANGELOG.md index 0a122a7..275a67a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -69,3 +69,5 @@ This file records what changes **in the product** – process and session state - Showcase preparation (Vercel + Supabase): without a worker, jobs run right after upload, approval and reprocess and via a protected drain route (daily cron); `DEMO_MODE` shows „Demo – nur synthetische Daten" on every page; runbook `docs/technical/deployment-vercel.md` and a role bootstrap script. +- Upload cap per person and hour (`UPLOAD_MAX_PER_HOUR`, required on the showcase): the upload answers + 429 before anything is stored. diff --git a/docs/decisions/ADR-0001-pilot-architecture.md b/docs/decisions/ADR-0001-pilot-architecture.md index 2839973..ad86e8a 100644 --- a/docs/decisions/ADR-0001-pilot-architecture.md +++ b/docs/decisions/ADR-0001-pilot-architecture.md @@ -624,10 +624,12 @@ enters the code: the app still talks plain PostgreSQL (Drizzle, pg-boss) and the **Consequences.** - **Connections.** At runtime `app_rw` connects through the **Supavisor transaction pooler (port 6543)** – serverless functions open many short connections. This is pooler-safe: `withTenant()` - sets the tenant with `set_config('app.company_id', …, true)`, i.e. transaction-local, and pg-boss - polls (no LISTEN/NOTIFY, D4). Migrations (`pnpm setup:deploy`, `app_owner`) use the **session + sets the tenant with `set_config('app.company_id', …, true)`, i.e. transaction-local; node-postgres + and Drizzle use unnamed prepared statements; pg-boss polls (no LISTEN/NOTIFY, D4) and its maintenance + takes transaction-scoped advisory locks. `statement_timeout` is also set on the role `app_rw` + (bootstrap script), so it holds even if the pooler drops the client's startup parameter. Migrations (`pnpm setup:deploy`, `app_owner`) use the **session pooler** (port 5432) or the direct connection – DDL and the pg-boss installer need a session. - *To verify at the first deploy:* the pool's `statement_timeout` startup parameter through Supavisor. + The smoke check (runbook step 8) reads `SHOW statement_timeout` as `app_rw`. - **Roles.** `app_owner`/`app_rw` are created once by the operator with `scripts/supabase-bootstrap.sql` (mirrors `docker/postgres/init/01-roles.sh`; `app_rw` NOBYPASSRLS); the first migration still refuses unsafe roles. The Supabase `postgres`/`service_role` credentials are never given to the app. diff --git a/docs/technical/architecture.md b/docs/technical/architecture.md index fe75129..96041a7 100644 --- a/docs/technical/architecture.md +++ b/docs/technical/architecture.md @@ -51,7 +51,7 @@ Deliberately accepted risks – without an entry here a deviation counts as a de | No RLS on the `auth` and `pgboss` schemas | Not company-owned business data; reachable only by server code (ADR-0001 D7) | Fluory | 2026-12-31 (review at M3) | | Showcase without unattended retries (Vercel Hobby cron once/day) | Showcase only; production runs a worker (D2). Jobs run via `after()` on upload/approval/reprocess and `/api/jobs/drain` (#59) | Fluory | when a production-like demo is needed | | Better Auth admin plugin mounted without any holder of its admin role | ADR-0001 D6 names the plugin; decided in #30: kept – its `banned` field implements deactivation (sign-in blocked by the plugin). Nobody holds `platform-admin`, so `/api/auth/admin/*` rejects every caller (tested); user management runs through `identity` | Fluory | 2026-12-31 (review at M3) | -| Upload endpoint without a per-user rate limit | Authenticated staff only; body bounded by `Content-Length` + `UPLOAD_MAX_REQUEST_BYTES` before reading | Fluory | before any public deployment (#19) | +| Upload cap per person only where configured | Local and CI run without `UPLOAD_MAX_PER_HOUR`; the showcase refuses to start without it (#59); concurrent uploads may pass the check together – a cost cap, not an exact quota | Fluory | 2026-12-31 (review at M3) | | `.msg` uploads checked by OLE signature only | Structure check of Outlook messages needs a CFB parser; files are served only as attachments with `nosniff` and parsed later by the stateless AI service | Fluory | with #23 (MSG parsing) | | Gemini API free tier for local development | Synthetic data only; never showcase or customer data (D8) | Fluory | when a Vertex development budget exists | diff --git a/docs/technical/deployment-vercel.md b/docs/technical/deployment-vercel.md index 9ef3124..9f250a8 100644 --- a/docs/technical/deployment-vercel.md +++ b/docs/technical/deployment-vercel.md @@ -8,7 +8,7 @@ | Blocker | Why | |---|---| -| #60 merged (Better Auth schema `auth` → `identity`) | Supabase reserves the schema `auth`; the migrations fail on Supabase until then | +| #60 merged (PR #63: Better Auth schema `auth` → `identity`) | Supabase reserves the schema `auth`; the migrations fail on Supabase until then. The bootstrap script creates no schema, so it is not affected | | AI service host chosen and running | The drain calls it; recommendation: **Google Cloud Run in the EU** (same GCP project as Vertex `eu`, existing image `services/ai`, no 300 s / package limits) | What runs where: Vercel (Hobby) runs the Next.js app, the drain route and the ERP mock @@ -19,8 +19,8 @@ There is **no worker process** – jobs run through `/api/jobs/drain` (see step 1. Create a project in region **eu-central-1 (Frankfurt)**. Store the generated `postgres` password in the password manager – the app never uses it. -2. **Data API:** Project Settings → Data API: keep the exposed schemas at `public` (or turn the Data API - off). Never add `app`, `pgboss` or `identity` – the app does not use the Data API. +2. **Data API:** Project Settings → Data API: keep the exposed schemas at the defaults `public` and + `graphql_public` (or turn the Data API off). Never add `app`, `pgboss` or `identity` – the app does not use the Data API. 3. **Storage:** create a **private** bucket `requestflow-documents` (no public access, no RLS policies for `anon`/`authenticated`). Storage → S3 connection: enable the S3 protocol and create an **S3 access key** (access key id + secret). Note the endpoint `https://.storage.supabase.co/storage/v1/s3`. @@ -33,11 +33,13 @@ There is **no worker process** – jobs run through `/api/jobs/drain` (see step Generate two strong passwords, then as the `postgres` user over the **session pooler**: ```bash -read -rs APP_OWNER_PASSWORD; read -rs APP_RW_PASSWORD psql "postgresql://postgres.@:5432/postgres" -v ON_ERROR_STOP=1 \ - -v owner_pw="$APP_OWNER_PASSWORD" -v rw_pw="$APP_RW_PASSWORD" -f scripts/supabase-bootstrap.sql + -f scripts/supabase-bootstrap.sql ``` +psql asks for both passwords without echoing them; they never appear on the command line or in the +process list. + The last output lists `app_owner` and `app_rw` with `f | f | f` (no superuser, no RLS bypass, no role creation). The script is all-or-nothing; running it twice fails on "role already exists" – that is fine. @@ -73,6 +75,7 @@ separate Supabase project. | `CRON_SECRET` | `openssl rand -base64 32` – Vercel Cron sends it as `Authorization: Bearer …` | | `JOB_DRAIN_INLINE` | `true` | | `DEMO_MODE` | `true` | +| `UPLOAD_MAX_PER_HOUR` | e.g. `20` – uploads per person and hour (every upload starts paid AI calls); **required** with `APP_ENV=showcase` | `.env.example` documents every variable. `MIGRATION_DATABASE_URL` and `SEED_PASSWORD` are **not** set on Vercel – they are only used from the operator's shell (steps 5 and 6). @@ -95,12 +98,14 @@ accounts out only to people who may see the demo. - `after()`: with `JOB_DRAIN_INLINE=true`, upload, approval and "Erneut verarbeiten" drain once after the response – the normal path needs nothing else. -- Cron: `vercel.json` calls `GET /api/jobs/drain` **once a day** (05:00 UTC) – the Hobby maximum. It picks +- Cron: `vercel.json` calls `GET /api/jobs/drain` **once a day** (some time within 05:00–05:59 UTC – Hobby + crons are not minute-exact) – the Hobby maximum. It picks up retries with backoff that no user action triggered. Faster unattended retries need a paid plan or a worker (exceptions register). - Manual: `curl -fsS -H "Authorization: Bearer $CRON_SECRET" https:///api/jobs/drain` (GET or POST) returns counts only. Without the right secret → 401; without `CRON_SECRET` → 404. - One run stops taking new jobs after 50 s (processing) + 20 s (export); a job in hand always finishes. + An `after()` drain subtracts the time its request already used (e.g. the upload itself) from the 50 s. If the platform still kills a run, the job becomes visible again after its pg-boss expiry (1 h). ## 8. Smoke check @@ -111,7 +116,8 @@ accounts out only to people who may see the demo. 4. Sign in as the seeded admin, upload a synthetic `.eml` (like the one in `tests/e2e/review-smoke.spec.ts`) → the request reaches *Zur Prüfung* within about a minute; approve it → *Exportiert* with an ERP reference. 5. Vercel function logs show `jobs.drain_run` lines (IDs and counts only). Confirm the client IP header - (rate limit) and that the pool's `statement_timeout` is accepted by Supavisor. + (rate limit). `SHOW statement_timeout` as `app_rw` reads `30s` (set on the role by the bootstrap + script, so it holds even if Supavisor drops the pool's startup parameter). ## 9. Switch off and roll back diff --git a/scripts/supabase-bootstrap.sql b/scripts/supabase-bootstrap.sql index fd01fc8..bc27dca 100644 --- a/scripts/supabase-bootstrap.sql +++ b/scripts/supabase-bootstrap.sql @@ -4,24 +4,27 @@ -- app_rw – runtime role: no superuser, no CREATEROLE/CREATEDB, NOBYPASSRLS → row-level security -- always applies -- Run ONCE as the Supabase `postgres` user over the session pooler (port 5432) or the direct connection, --- never over the transaction pooler. Passwords are supplied by the operator as psql variables – they are --- never written into this file, the repository or a migration: +-- never over the transaction pooler. psql prompts for both passwords without echo – they never reach the +-- command line, the process list, this file, the repository or a migration: -- --- read -rs APP_OWNER_PASSWORD; read -rs APP_RW_PASSWORD --- psql "" -v ON_ERROR_STOP=1 \ --- -v owner_pw="$APP_OWNER_PASSWORD" -v rw_pw="$APP_RW_PASSWORD" -f scripts/supabase-bootstrap.sql +-- psql "" -v ON_ERROR_STOP=1 -f scripts/supabase-bootstrap.sql -- -- Afterwards the first migration checks the roles again and refuses to run if either could bypass RLS. \set ON_ERROR_STOP on --- Fail (exit code 3) before anything is created when a password variable is missing. +-- Prompt (no echo) unless the operator already set the variables, e.g. from a secrets manager. \if :{?owner_pw} \else - DO $$ BEGIN RAISE EXCEPTION 'psql variable owner_pw is required (-v owner_pw=...)'; END $$; + \prompt 'Password for app_owner: ' owner_pw \endif \if :{?rw_pw} \else - DO $$ BEGIN RAISE EXCEPTION 'psql variable rw_pw is required (-v rw_pw=...)'; END $$; + \prompt 'Password for app_rw: ' rw_pw +\endif +SELECT length(:'owner_pw') >= 16 AND length(:'rw_pw') >= 16 AS passwords_ok \gset +\if :passwords_ok +\else + DO $$ BEGIN RAISE EXCEPTION 'both passwords must have at least 16 characters'; END $$; \endif SELECT current_database() AS db \gset @@ -35,6 +38,9 @@ CREATE ROLE app_rw LOGIN PASSWORD :'rw_pw' NOSUPERUSER NOCREATEROLE NOCREA GRANT CONNECT ON DATABASE :"db" TO app_owner, app_rw; GRANT CREATE ON DATABASE :"db" TO app_owner; REVOKE CREATE ON SCHEMA public FROM PUBLIC; +-- Safety net of the export row lock (ADR-0001 D9, src/db/client.ts): set on the role, so it also holds +-- when a pooler does not pass the client's startup parameter through. +ALTER ROLE app_rw SET statement_timeout = '30s'; COMMIT; -- Show the result without secrets: both roles must read f / f / f. diff --git a/src/app/_server/drain-request.test.ts b/src/app/_server/drain-request.test.ts index 460a1e3..05ec164 100644 --- a/src/app/_server/drain-request.test.ts +++ b/src/app/_server/drain-request.test.ts @@ -1,7 +1,7 @@ import { after } from "next/server"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { captureLogs } from "@/features/observability"; -import { handleDrainRequest, scheduleAfterResponse } from "./drain-request"; +import { handleDrainRequest, isAuthorized, processBudgetMs, scheduleAfterResponse } from "./drain-request"; // `after()` is the Next.js runtime boundary (it needs a request scope): replaced by a recorder. vi.mock("next/server", async (original) => ({ ...(await original()), after: vi.fn() })); @@ -127,3 +127,21 @@ describe("scheduleAfterResponse (JOB_DRAIN_INLINE)", () => { expect(lines.map((line) => JSON.parse(line))).toContainEqual(expect.objectContaining({ event: "jobs.drain_failed", code: "DatabaseError" })); }); }); + +describe("isAuthorized (#59 review)", () => { + it("accepts the Bearer scheme case-insensitively (RFC 7235) and nothing else", () => { + expect(isAuthorized(`Bearer ${SECRET}`, SECRET)).toBe(true); + expect(isAuthorized(`bearer ${SECRET}`, SECRET)).toBe(true); + expect(isAuthorized(`Basic ${SECRET}`, SECRET)).toBe(false); + expect(isAuthorized("Bearer ", SECRET)).toBe(false); + expect(isAuthorized(null, SECRET)).toBe(false); + }); +}); + +describe("processBudgetMs (#59 review)", () => { + it("subtracts the time the invocation already spent (e.g. the upload itself) and never goes below zero", () => { + expect(processBudgetMs(50_000, 1_000, 1_000)).toBe(50_000); + expect(processBudgetMs(50_000, 1_000, 21_000)).toBe(30_000); + expect(processBudgetMs(50_000, 1_000, 90_000)).toBe(0); + }); +}); diff --git a/src/app/_server/drain-request.ts b/src/app/_server/drain-request.ts index c04859f..c9a8a52 100644 --- a/src/app/_server/drain-request.ts +++ b/src/app/_server/drain-request.ts @@ -17,10 +17,19 @@ const json = (status: number, body: unknown, headers: Record = { const digest = (value: string) => createHash("sha256").update(value, "utf8").digest(); export function isAuthorized(authorization: string | null, secret: string): boolean { - const presented = authorization?.startsWith("Bearer ") ? authorization.slice("Bearer ".length) : ""; + // The auth scheme is case-insensitive (RFC 7235); the credential is compared exactly. + const presented = authorization && /^bearer /i.test(authorization) ? authorization.slice("Bearer ".length) : ""; return timingSafeEqual(digest(presented), digest(secret)) && presented.length > 0; } +/** + * Processing budget left for a drain that shares its function invocation with other work (the upload + * itself before an `after()` drain): the whole run must still fit into the function limit (#59 review). + */ +export function processBudgetMs(budgetMs: number, invokedAt: number, now: number = Date.now()): number { + return Math.max(0, budgetMs - (now - invokedAt)); +} + /** * Runs `run` via `after()` once the response is sent – only when `enabled` (JOB_DRAIN_INLINE=true). * Failures never reach the user's response; they are logged by error class only. diff --git a/src/app/_server/drain.ts b/src/app/_server/drain.ts index 224c32d..e586fff 100644 --- a/src/app/_server/drain.ts +++ b/src/app/_server/drain.ts @@ -1,7 +1,7 @@ import { SERVERLESS_DRAIN } from "@/config/env"; import { buildJobDeps, drainRound, handledJobs, type DrainRoundResult, type JobDeps } from "@/job-drain"; import { logEvent } from "@/features/observability"; -import { scheduleAfterResponse } from "./drain-request"; +import { processBudgetMs, scheduleAfterResponse } from "./drain-request"; import { getJobClient, getRuntime } from "./runtime"; // Serverless drain of the web process (#59, ADR-0001 D2): the showcase has no worker, so the route @@ -22,11 +22,14 @@ function jobDeps(): Promise { return deps; } -/** One bounded round incl. pg-boss maintenance (no supervising worker here). */ -export async function drainNow(): Promise { +/** + * One bounded round incl. pg-boss maintenance (no supervising worker here). `invokedAt`: start of the + * function invocation – time already spent there (e.g. the upload) shortens the processing window. + */ +export async function drainNow(invokedAt: number = Date.now()): Promise { const started = Date.now(); const result = await drainRound(await jobDeps(), { - processMs: SERVERLESS_DRAIN.processMs, + processMs: processBudgetMs(SERVERLESS_DRAIN.processMs, invokedAt), exportMs: SERVERLESS_DRAIN.exportMs, maintenance: true, }); @@ -34,7 +37,10 @@ export async function drainNow(): Promise { return result; } -/** After upload, approval or reprocess: drain once after the response – only with JOB_DRAIN_INLINE=true. */ -export function drainAfterResponse(): void { - scheduleAfterResponse(getRuntime().config.jobs.drainInline, drainNow); +/** + * After upload, approval or reprocess: drain once after the response – only with JOB_DRAIN_INLINE=true. + * `invokedAt` is taken at the start of the handler, so the drain accounts for the handler's own time. + */ +export function drainAfterResponse(invokedAt: number): void { + scheduleAfterResponse(getRuntime().config.jobs.drainInline, () => drainNow(invokedAt)); } diff --git a/src/app/api/requests/route.ts b/src/app/api/requests/route.ts index 1383066..51c7325 100644 --- a/src/app/api/requests/route.ts +++ b/src/app/api/requests/route.ts @@ -1,7 +1,7 @@ import { drainAfterResponse } from "@/app/_server/drain"; import { currentActor, getJobClient, getRuntime } from "@/app/_server/runtime"; import { AuthorizationError } from "@/features/identity"; -import { submitUpload, UploadRejected } from "@/features/intake"; +import { submitUpload, UploadRateLimited, UploadRejected } from "@/features/intake"; import { logEvent } from "@/features/observability"; export const dynamic = "force-dynamic"; @@ -13,6 +13,7 @@ const problem = (status: number, title: string) => Response.json({ error: { titl // POST /api/requests – multipart upload of one request (field `files`, 1..n files). // Company and user come from the session, never from the form (ADR-0001 D7). export async function POST(request: Request): Promise { + const invokedAt = Date.now(); const actor = await currentActor(request.headers); if (!actor) return problem(401, "Nicht angemeldet."); const { config, tenancy, storage } = getRuntime(); @@ -39,9 +40,10 @@ export async function POST(request: Request): Promise { try { const boss = await getJobClient(); const result = await submitUpload({ tenancy, storage, boss, limits: config.upload }, actor, files); - drainAfterResponse(); // serverless runtimes only (JOB_DRAIN_INLINE=true): process right away + drainAfterResponse(invokedAt); // serverless runtimes only (JOB_DRAIN_INLINE=true): process right away return Response.json(result, { status: 201 }); } catch (error) { + if (error instanceof UploadRateLimited) return problem(429, error.message); if (error instanceof UploadRejected) return problem(422, error.message); if (error instanceof AuthorizationError) return problem(403, "Keine Berechtigung."); const kind = error instanceof Error ? error.name : "unknown"; diff --git a/src/app/requests/[id]/actions.ts b/src/app/requests/[id]/actions.ts index b91dcac..4071627 100644 --- a/src/app/requests/[id]/actions.ts +++ b/src/app/requests/[id]/actions.ts @@ -52,6 +52,7 @@ export async function correctFieldAction(formData: FormData): Promise { } export async function approveAction(formData: FormData): Promise { + const invokedAt = Date.now(); const actor = await actorOrLogin(); const requestId = requestIdOf(formData); const boss = await getJobClient(); @@ -59,7 +60,7 @@ export async function approveAction(formData: FormData): Promise { requestId, async () => { await approveRequest({ tenancy: getRuntime().tenancy, boss }, actor, requestId); - drainAfterResponse(); // serverless runtimes only (JOB_DRAIN_INLINE=true): export right away + drainAfterResponse(invokedAt); // serverless runtimes only (JOB_DRAIN_INLINE=true): export right away }, "approved", ); diff --git a/src/app/requests/actions.ts b/src/app/requests/actions.ts index ea71c5f..5a2d375 100644 --- a/src/app/requests/actions.ts +++ b/src/app/requests/actions.ts @@ -11,13 +11,14 @@ import { reprocessRequest, ReprocessRefused } from "@/features/jobs"; // Reprocess from the request list (#26): ERROR(processing) → NEW + processing job, ERROR(export) → // APPROVED + export job – status change, job and audit event in ONE transaction (jobs module). export async function reprocessAction(formData: FormData): Promise { + const invokedAt = Date.now(); const actor = await currentActor(await headers()); if (!actor) redirect("/login"); const requestId = z.uuid().safeParse(formData.get("requestId")); if (!requestId.success) redirect("/requests?error=refused"); try { await reprocessRequest({ tenancy: getRuntime().tenancy, boss: await getJobClient() }, actor, requestId.data); - drainAfterResponse(); // serverless runtimes only (JOB_DRAIN_INLINE=true): retry right away + drainAfterResponse(invokedAt); // serverless runtimes only (JOB_DRAIN_INLINE=true): retry right away } catch (error) { if (error instanceof ReprocessRefused || error instanceof AuthorizationError) redirect("/requests?error=refused"); throw error; diff --git a/src/config/env.test.ts b/src/config/env.test.ts index 8cd527d..c94c669 100644 --- a/src/config/env.test.ts +++ b/src/config/env.test.ts @@ -99,6 +99,15 @@ describe("loadConfig", () => { expect(() => loadConfig({ ...valid, APP_ENV: "showcase", BETTER_AUTH_SECRET: "s".repeat(40), ERP_TOKEN: placeholder })).toThrow(/ERP_TOKEN/); }); + it("caps uploads per hour only when set, and requires the cap on the showcase (#59 review)", () => { + expect(loadConfig(valid).upload.maxPerHour).toBeUndefined(); + expect(loadConfig({ ...valid, UPLOAD_MAX_PER_HOUR: "20" }).upload.maxPerHour).toBe(20); + expect(() => loadConfig({ ...valid, UPLOAD_MAX_PER_HOUR: "0" })).toThrow(/UPLOAD_MAX_PER_HOUR/); + const showcase = { ...valid, APP_ENV: "showcase", BETTER_AUTH_SECRET: "s".repeat(40), ERP_TOKEN: "e".repeat(32) }; + expect(() => loadConfig(showcase)).toThrow(/UPLOAD_MAX_PER_HOUR/); + expect(loadConfig({ ...showcase, UPLOAD_MAX_PER_HOUR: "20" }).upload.maxPerHour).toBe(20); + }); + describe("serverless job drain and demo mode (#59)", () => { // Values that fit one processing job into the drain function (60 s × 3 files). const fits = { AI_SERVICE_TIMEOUT_MS: "60000", UPLOAD_MAX_FILES: "3" }; diff --git a/src/config/env.ts b/src/config/env.ts index d26cb3d..c51ca36 100644 --- a/src/config/env.ts +++ b/src/config/env.ts @@ -28,6 +28,9 @@ const schema = z.object({ APP_ENV: z.enum(["local", "showcase", "production"]), UPLOAD_MAX_FILE_BYTES: z.coerce.number().int().positive().default(20 * 1024 * 1024), UPLOAD_MAX_FILES: z.coerce.number().int().positive().max(50).default(10), + // Uploads per person and hour (#59): unset = no cap; required on the showcase, where every upload + // starts paid AI calls. + UPLOAD_MAX_PER_HOUR: optional(z.coerce.number().int().positive()), AI_SERVICE_URL: z.url().default("http://127.0.0.1:8000"), AI_SERVICE_TOKEN: z.string().min(24).optional(), AI_SERVICE_TIMEOUT_MS: z.coerce.number().int().positive().default(120_000), @@ -80,6 +83,8 @@ export interface AppConfig { maxFileBytes: number; maxFiles: number; maxRequestBytes: number; + /** Uploads per person and hour; undefined = no cap. */ + maxPerHour: number | undefined; }; erp: { baseUrl: string; @@ -112,6 +117,10 @@ export function loadConfig(source: Record = process. if (env.APP_ENV !== "local" && env.ERP_TOKEN && LOCAL_PLACEHOLDER_ERP_TOKENS.has(env.ERP_TOKEN)) { throw new Error("Invalid or missing configuration: ERP_TOKEN"); } + // The showcase is public: without an upload cap an invited account could run up AI cost (#59 review). + if (env.APP_ENV === "showcase" && env.UPLOAD_MAX_PER_HOUR === undefined) { + throw new Error("Invalid or missing configuration: UPLOAD_MAX_PER_HOUR"); + } // A serverless drain cannot outlive its function: the worst-case processing job (every document hits // the AI timeout) plus the export and loop windows must fit, or the platform kills the job mid-run. if (env.CRON_SECRET !== undefined || env.JOB_DRAIN_INLINE === "true") { @@ -138,7 +147,7 @@ export function loadConfig(source: Record = process. trustedProxies: list(env.AUTH_TRUSTED_PROXIES), }, aiService: { baseUrl: env.AI_SERVICE_URL, token: env.AI_SERVICE_TOKEN, timeoutMs: env.AI_SERVICE_TIMEOUT_MS }, - upload: { maxFileBytes: env.UPLOAD_MAX_FILE_BYTES, maxFiles: env.UPLOAD_MAX_FILES, maxRequestBytes: env.UPLOAD_MAX_REQUEST_BYTES }, + upload: { maxFileBytes: env.UPLOAD_MAX_FILE_BYTES, maxFiles: env.UPLOAD_MAX_FILES, maxRequestBytes: env.UPLOAD_MAX_REQUEST_BYTES, maxPerHour: env.UPLOAD_MAX_PER_HOUR }, erp: { baseUrl: env.ERP_BASE_URL, token: env.ERP_TOKEN, diff --git a/src/features/intake/files.ts b/src/features/intake/files.ts index a6213ea..9445beb 100644 --- a/src/features/intake/files.ts +++ b/src/features/intake/files.ts @@ -11,6 +11,14 @@ export class UploadRejected extends Error { } } +/** Too many uploads by one person within the hour (showcase cost cap, #59) – answered with 429. */ +export class UploadRateLimited extends UploadRejected { + constructor(readonly maxPerHour: number) { + super(`Höchstens ${maxPerHour} Uploads pro Stunde – bitte später erneut versuchen.`); + this.name = "UploadRateLimited"; + } +} + export interface UploadLimits { maxFileBytes: number; } diff --git a/src/features/intake/index.ts b/src/features/intake/index.ts index ffada73..2c442fc 100644 --- a/src/features/intake/index.ts +++ b/src/features/intake/index.ts @@ -1,3 +1,3 @@ // Public API of the `intake` module: upload, validation, duplicate fingerprint. export { submitUpload, type IntakeDeps, type SubmittedRequest, type UploadedFile } from "./submit"; -export { UploadRejected, type UploadLimits } from "./files"; +export { UploadRateLimited, UploadRejected, type UploadLimits } from "./files"; diff --git a/src/features/intake/submit.ts b/src/features/intake/submit.ts index ea019ae..364bf98 100644 --- a/src/features/intake/submit.ts +++ b/src/features/intake/submit.ts @@ -3,11 +3,11 @@ import { recordAudit } from "@/features/audit"; import { insertDocuments, type NewDocument } from "@/features/documents"; import { authorize, type Actor } from "@/features/identity"; import { enqueueRequestProcessing, type JobSender } from "@/features/jobs"; -import { createRequest, findDuplicate, lockDuplicateDetection } from "@/features/requests"; +import { countRequestsCreatedBy, createRequest, findDuplicate, lockDuplicateDetection } from "@/features/requests"; import { logEvent } from "@/features/observability"; import { S3BlobStore } from "@/features/storage"; import type { Tenancy } from "@/features/tenancy"; -import { classifyUpload, UploadRejected, type UploadLimits } from "./files"; +import { classifyUpload, UploadRateLimited, UploadRejected, type UploadLimits } from "./files"; import { requestFingerprint, sha256Hex } from "./fingerprint"; import { parseMailHeaders } from "./mail-headers"; @@ -15,7 +15,8 @@ export interface IntakeDeps { tenancy: Tenancy; storage: S3BlobStore; boss: JobSender; - limits: UploadLimits & { maxFiles: number }; + /** `maxPerHour`: uploads per person and hour; undefined = no cap (local, CI). */ + limits: UploadLimits & { maxFiles: number; maxPerHour?: number }; } export interface UploadedFile { @@ -40,6 +41,15 @@ export async function submitUpload(deps: IntakeDeps, actor: Actor, files: Upload if (files.length === 0) throw new UploadRejected("Bitte mindestens eine Datei auswählen."); if (files.length > deps.limits.maxFiles) throw new UploadRejected(`Höchstens ${deps.limits.maxFiles} Dateien pro Anfrage.`); + // Cost cap of a public deployment (#59): every upload starts paid AI calls. Checked before anything is + // stored; concurrent uploads may pass the check together – a cap, not an exact quota. + const maxPerHour = deps.limits.maxPerHour; + if (maxPerHour !== undefined) { + const since = new Date(Date.now() - 60 * 60 * 1000); + const recent = await deps.tenancy.withTenant(actor.companyId, (tx) => countRequestsCreatedBy(tx, actor.userId, since)); + if (recent >= maxPerHour) throw new UploadRateLimited(maxPerHour); + } + const requestId = randomUUID(); const classified = files.map((file) => ({ file, meta: classifyUpload(file.name, file.bytes, deps.limits) })); const documents: NewDocument[] = classified.map(({ file, meta }) => { diff --git a/src/features/requests/index.ts b/src/features/requests/index.ts index 14123b9..17c81af 100644 --- a/src/features/requests/index.ts +++ b/src/features/requests/index.ts @@ -1,5 +1,6 @@ // Public API of the `requests` module: request aggregate and status machine. export { + countRequestsCreatedBy, createRequest, findDuplicate, getRequest, diff --git a/src/features/requests/repository.ts b/src/features/requests/repository.ts index 3fc6f00..d79194f 100644 --- a/src/features/requests/repository.ts +++ b/src/features/requests/repository.ts @@ -1,4 +1,4 @@ -import { and, asc, desc, eq, or, sql, type SQL } from "drizzle-orm"; +import { and, asc, desc, eq, gte, or, sql, type SQL } from "drizzle-orm"; import { requests, type RequestStatus } from "@/db/schema"; import { tenantOf, type TenantTx } from "@/features/tenancy"; import { nextStatus, type RequestEvent } from "./status"; @@ -115,3 +115,13 @@ export async function recordProcessingFailure(tx: TenantTx, id: string, failure: .set({ errorMessage: failure.message, nextRetryAt: failure.nextRetryAt }) .where(and(eq(requests.id, id), eq(requests.status, "PROCESSING"))); } + +/** Requests a user created since `since` (upload rate limit, #59 review) – within the tenant. */ +export async function countRequestsCreatedBy(tx: TenantTx, userId: string, since: Date): Promise { + tenantOf(tx); + const [row] = await tx + .select({ count: sql`count(*)::int` }) + .from(requests) + .where(and(eq(requests.createdBy, userId), gte(requests.createdAt, since))); + return row?.count ?? 0; +} diff --git a/tests/integration/intake.test.ts b/tests/integration/intake.test.ts index 0e6feab..7223b0d 100644 --- a/tests/integration/intake.test.ts +++ b/tests/integration/intake.test.ts @@ -4,7 +4,7 @@ import { loadConfig } from "@/config/env"; import { listAuditEvents } from "@/features/audit"; import { listDocuments } from "@/features/documents"; import { getActor, type Actor } from "@/features/identity"; -import { submitUpload, UploadRejected, type IntakeDeps } from "@/features/intake"; +import { submitUpload, UploadRateLimited, UploadRejected, type IntakeDeps } from "@/features/intake"; import { createJobQueue } from "@/db/job-queue-client"; import { QUEUES } from "@/features/jobs"; import { getRequest } from "@/features/requests"; @@ -205,4 +205,26 @@ describe("intake: upload a request and enqueue processing atomically", () => { client.release(); } }); + it("caps uploads per person and hour when a limit is set, before anything is stored (#59 review)", async () => { + const clerk = (await getActor(stack.auth, stack.database.db, new Headers({ cookie: (await invitedUser(stack, adminB, "clerk")).cookie })))!; + const limited = { ...deps, limits: { ...deps.limits, maxPerHour: 2 } }; + await submitUpload(limited, clerk, [{ name: "a.pdf", bytes: pdf(unique("rate-1")) }]); + await submitUpload(limited, clerk, [{ name: "b.pdf", bytes: pdf(unique("rate-2")) }]); + + const keys: string[] = []; + const spyStorage = Object.assign(Object.create(storage) as S3BlobStore, { + put: async (key: string, ...rest: unknown[]) => { + keys.push(key); + return (storage.put as (...args: unknown[]) => Promise)(key, ...rest); + }, + }); + const third = submitUpload({ ...limited, storage: spyStorage }, clerk, [{ name: "c.pdf", bytes: pdf(unique("rate-3")) }]); + await expect(third).rejects.toBeInstanceOf(UploadRateLimited); + expect(keys).toEqual([]); + + // Per person: another clerk of the same company still uploads; without a limit nobody is capped. + const colleague = (await getActor(stack.auth, stack.database.db, new Headers({ cookie: (await invitedUser(stack, adminB, "clerk")).cookie })))!; + await expect(submitUpload(limited, colleague, [{ name: "d.pdf", bytes: pdf(unique("rate-4")) }])).resolves.toMatchObject({ requestId: expect.any(String) }); + await expect(submitUpload(deps, clerk, [{ name: "e.pdf", bytes: pdf(unique("rate-5")) }])).resolves.toMatchObject({ requestId: expect.any(String) }); + }); }); From 4c6e7b746eac1b2932dd1d9a1e2afe7c7021f386 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 16:50:57 +0000 Subject: [PATCH 9/9] fix(jobs): the drain route counts its own invocation time too (#59 review) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DJ5vaKvTYiMvdngT4d3xo1 --- src/app/api/jobs/drain/route.ts | 3 ++- tests/integration/drain-route.test.ts | 6 +++++- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/src/app/api/jobs/drain/route.ts b/src/app/api/jobs/drain/route.ts index 5286d3f..6dcb007 100644 --- a/src/app/api/jobs/drain/route.ts +++ b/src/app/api/jobs/drain/route.ts @@ -11,6 +11,7 @@ export const maxDuration = 300; // processing and export jobs for runtimes without a worker (#59). `Authorization: Bearer `; // 404 while CRON_SECRET is unset. No request body is read. async function handle(request: Request): Promise { + const invokedAt = Date.now(); let cronSecret: string | undefined; try { cronSecret = getRuntime().config.jobs.cronSecret; @@ -20,7 +21,7 @@ async function handle(request: Request): Promise { logEvent("error", "jobs.drain_config_invalid", {}, { code: "config", names }); return Response.json({ error: { title: "Verarbeitung derzeit nicht möglich." } }, { status: 503, headers: { "cache-control": "no-store" } }); } - return handleDrainRequest(request, { cronSecret, run: drainNow }); + return handleDrainRequest(request, { cronSecret, run: () => drainNow(invokedAt) }); } export const GET = handle; diff --git a/tests/integration/drain-route.test.ts b/tests/integration/drain-route.test.ts index 6a23cd5..cabcee6 100644 --- a/tests/integration/drain-route.test.ts +++ b/tests/integration/drain-route.test.ts @@ -58,7 +58,11 @@ describe("drain route and inline drain wiring", () => { expect(response.status).toBe(200); expect(await response.json()).toEqual(summary); expect(drainRound).toHaveBeenCalledTimes(1); - expect(vi.mocked(drainRound).mock.calls[0]![1]).toEqual({ processMs: SERVERLESS_DRAIN.processMs, exportMs: SERVERLESS_DRAIN.exportMs, maintenance: true }); + // The processing window is the serverless budget minus the time this invocation already spent. + const options = vi.mocked(drainRound).mock.calls[0]![1]; + expect(options).toMatchObject({ exportMs: SERVERLESS_DRAIN.exportMs, maintenance: true }); + expect(options.processMs).toBeLessThanOrEqual(SERVERLESS_DRAIN.processMs); + expect(options.processMs).toBeGreaterThan(SERVERLESS_DRAIN.processMs - 5_000); }); it("an accepted upload schedules one drain after the response (JOB_DRAIN_INLINE=true)", async () => {