diff --git a/docs/PROGRESS.md b/docs/PROGRESS.md index a3eff28..60b39c6 100644 --- a/docs/PROGRESS.md +++ b/docs/PROGRESS.md @@ -5,9 +5,9 @@ commit history + [`PIVOT_PLAN.md`](./PIVOT_PLAN.md) + project memory. Update thi file at the **end of every step** so we never jump ahead on a deferred item or skip a main phase. -- **Last updated:** 2026-07-23 - see [PROGRESS §7 Changelog](#7-changelog). This field carries the date only; the narrative lives in §7. -- **Active branch:** `docs/b5b-scoping-record` - this docs-only PR RECORDS the **B5b upload-content-hardening scoping** as register **#50** and creates the decision record [`b5b-scoping.md`](./b5b-scoping.md): the established+verified constraints (the real ceiling is Vercel's hard ~4.5 MB edge cap, not the app's broken 50 MB promise; duration is the unmeasured true limit), the corrections (the cap does not stop ZIP bombs; `PIVOT_PLAN.md:269`'s 50 MB claim is false), the B5b-1/B5b-2 split plus enablers (b1)/(b2), and **seven OPEN founder decisions** with an OUTSTANDING duration measurement. Touches **only** two files: `docs/b5b-scoping.md` (new) and this file (one new register row #50, one new §7 block, and these three header fields). **No code, no route/validation/size-constant change, no other register row's substance, no §1-§6 phase content; B5b §1 row stays ⬜ (scoping is not progress), `PIVOT_PLAN.md` untouched.** Register **#17** stays OPEN. -- **Main tip:** `1e37c03` (PR **#68** merge - the docs-only close of register #45 and logging of design items #46-#49), on top of `ecb1ed5` (PR **#67** - the duplicate Vercel ingestion service removed from `vercel.json` + `INGESTION_TOKEN` rotated) and `2d0856d` (PR **#66** - the §7 changelog migration off the line-8 header field, all 17 entry bodies byte-identical). Both required checks green on `1e37c03` (`tsc`, `db-types`). +- **Last updated:** 2026-07-25 - see [PROGRESS §7 Changelog](#7-changelog). This field carries the date only; the narrative lives in §7. +- **Active branch:** `feat/b1-railway-owns-chunk-persistence` - implements **enabler (b1)** from register #50 and logs it as register **#51**: chunk + embedding persistence moves **into the Railway ingestion service with FULL ownership** (`/convert` renamed **`/ingest`**; it now persists chunks and writes the terminal `documents` status itself, returning only a small ack), killing the **9-13× return blob** and the **stuck-at-`processing` orphan class**. Railway authenticates as the **uploading user via a forwarded JWT (Option B)** on the **public anon key** — **no `SUPABASE_SERVICE_ROLE_KEY` anywhere in the service, its Dockerfile, its env, or the route** (register **#45** blast radius: that service was publicly duplicable, so an RLS-bypassing credential there would have been a full-database breach). Touches **four** files: `services/ingestion/main.py`, `services/ingestion/requirements.txt`, `src/app/api/ingest/route.ts`, and this file (one new register row #51, one new §7 block, these three header fields). **This is the B6 ENABLER, not B6** (ingestion stays synchronous — Next still awaits) **and not B5b progress: the B5b §1 row stays ⬜.** No migration, no DB change; `backfill.py` deploy posture untouched; `PIVOT_PLAN.md` untouched. Register **#17** stays OPEN. +- **Main tip:** `140bc31` (PR **#69** merge - the docs-only record of the B5b upload-hardening scoping, register #50 + `docs/b5b-scoping.md`), on top of `1e37c03` (PR **#68** - the close of register #45 and logging of design items #46-#49) and `ecb1ed5` (PR **#67** - the duplicate Vercel ingestion service removed from `vercel.json` + `INGESTION_TOKEN` rotated). Both required checks green on `140bc31` (`tsc`, `db-types`). > ✅ **RESOLVED 2026-07-04 — Supabase project back online (register #21).** The > project **auto-paused** (free-tier inactivity) and then hit Supabase's free @@ -147,6 +147,7 @@ project memory, and per-step review debates. "Where" is the target phase/trigger | 48 🔎 **OPEN — product-design item, raised 2026-07-23 (Abo Jad design review). DESIGN-QUALITY, NOT A BUG.** | **Pre-login landing page has no navigation menu.** | Not a defect — a missing-nav/UX item for the mobile app; no phase assigned. | Wanted: a **hamburger (three-line) icon** opposite the KnowFlow wordmark in the header, opening a **main dropdown menu** for the mobile app. | | 49 🔎 **OPEN — product-design item, raised 2026-07-23 (Abo Jad design review). Carries a correctness rider, not pure polish.** | **The landing page's terminal-style demo window shows static text.** **Cross-ref #20:** that window currently displays **"Ready in 0.4s"**, a timing claim **#20 already flags**, and this session proved **ingestion is synchronous and takes real seconds** — so the number MUST be corrected or removed as part of this work rather than re-shipping a false claim. | Not pure polish — the embedded "0.4s" IS a false claim (see **#20**), so this row carries a correctness rider; no phase assigned. | Wanted: **animate the terminal** so the text **types out live from start to finish**; as part of the same work, **correct or remove "Ready in 0.4s"** (per **#20**) rather than re-ship it. | | 50 🔎 **SCOPING RECORDED 2026-07-23 — B5b upload content hardening. Decisions 1-7 OPEN pending founder rulings; the duration measurement OUTSTANDING. Scoping is NOT progress — B5b (§1) stays ⬜.** | **B5b (deep upload content hardening: magic-byte, decompression-bomb / nested-archive limits, content scanning) is now scoped in [`b5b-scoping.md`](./b5b-scoping.md).** Established + verified this session: the real upload ceiling is **Vercel's hard ~4.5 MB edge cap** (platform constant, unraisable), **not** the 50 MB the app promises in **7 places** (`route.ts:51/:52`, `DropZone.tsx:26`, four i18n strings — the `:52` message is hardcoded English, register-#17 class); ingestion runs on **Railway** (`INGESTION_SERVICE_URL`), so the `route.ts`→`/convert` hop faces no second Vercel cap but a **9-13× return-path blob** (each chunk's 1024-float embedding inline + full markdown, materialized by `await pyResponse.json()`); and **no `maxDuration` is set anywhere** (grep-empty), so the true user-facing ceiling is an **UNMEASURED duration limit**, not size. Corrections recorded so they are not re-inherited: the 4.5 MB cap does **NOT** mitigate decompression bombs (compressed bytes are bounded, expansion is not); the pre-auth body-buffering concern is **bounded** by the edge cap, so the `getUser`-above-`formData` reorder is a low-priority ride-along (`checkDocumentLimit` **cannot** move — needs `kbId` from the body; `enforceLimit` **should not** — it would burn an upload credit on a rejected request); and **`PIVOT_PLAN.md:269`'s "Size cap (50 MB) already exists" is FALSE** (the effective cap is the platform's ~4.5 MB). | Not a bug and not deferred work — a **decision record** written before implementation so the founder's rulings are captured, not inferred. The split is recorded (B5b-1 request-path magic-byte + declared-size ZIP-ratio pre-check + the auth reorder, doable now; B5b-2 enforced expansion bounds + nested depth + AV/scanning, genuinely gated on **B6**; plus enabler moves (b1) persist chunks from Railway and (b2) signed-URL direct-to-storage). **No code, no route/validation/size-constant change; `PIVOT_PLAN.md` untouched.** | **Seven OPEN founder decisions** (each stated with consequences in the doc, none picked here): (1) max file size — accept ~4.5 MB vs. deliver real 50 MB via (b2); (2) format list — keep all six vs. drop the docx/pptx/xlsx ZIP trio that carries all bomb risk; (3) scanning scope — AV (nothing in the image today), PII, and **prompt-injection in extracted text** (flows unfiltered into Haiku via `` interpolation — currently unmitigated, needs at minimum a documented accepted-risk decision); (4) failure posture — reject-at-upload (415) vs. accept-then-quarantine; (5) enforcement point — Next, Railway, or both (Railway trusts its caller on a single bearer token, so Next-only is bypassed if `INGESTION_TOKEN` leaks); (6) the three ZIP numbers (expansion ratio, absolute expanded cap, nesting depth) — each measured from the spoofable declared central directory (cheap) or actual expansion (accurate, needs B6); (7) the signature table (`%PDF-`; `PK\x03\x04` + inner part-path for the OOXML trio; the txt/md non-signature UTF-8/NUL rule). **OUTSTANDING measurement:** at what file size does the upload time out — plain-`.txt`-on-Preview procedure + failure-signature table recorded in the doc §5 (each upload burns a rate-limit credit and the #22 free tier — throwaway subject, delete rows after). Related: **#45** (the duplicate ingestion service closed while scoping this), **#22** (bounds (b2) and the §5 test budget), **#24** (why `enforceLimit` cannot move), **#17** (the `:52` hardcoded-English class). | +| 51 🔎 **LANDED (pending Abo Jad's pre-merge verification) 2026-07-25 — enabler (b1) from register #50. This is the B6 ENABLER, not B6 and not B5b progress: ingestion is still synchronous and B5b (§1) stays ⬜.** | **Chunk + embedding persistence moved into the Railway ingestion service, with FULL ownership.** The service now runs convert → chunk → embed → **persist chunks** → **write the terminal `documents` status**, and returns only a small ack `{document_id, chunk_count, status}`. `/convert` is **renamed `/ingest`** (both request and response shapes are incompatible with the old contract, so a version-skewed caller gets a loud **404** instead of silently persisting a null `markdown_content` over a document the service already finished correctly). The Next route still creates the `documents` row and forwards, still awaits, and keeps both the browser reply and the `recordStudyEvent('material_uploaded')` emit gated on the ack — but it **no longer receives or inserts chunks** and no longer materializes the **9-13× return blob** that register #50 measured (`await pyResponse.json()` of every chunk's 1024-float embedding + full markdown). | Not a bug fix — the enabler that makes **B6** (async ingestion) possible: the slow work and its writes now live entirely in the only component without a duration ceiling, so making it fire-and-forget later requires no further ownership change. **Deliberately NOT done here: B6 itself** (Next still awaits Railway — fire-and-forget is a separate step), and **`backfill.py`'s deploy posture** (untouched, still a local-only script). | **AUTH DECISION — Option B (forwarded user JWT) chosen over a service-role key, and this is the load-bearing part.** Railway authenticates as the **uploading user**: Next lifts the verified session's `access_token` and forwards it in an `X-Supabase-Token` header (`Authorization` already carries `INGESTION_TOKEN`, so the two credentials stay separate); the service builds a **per-request** Supabase client with the **public anon key** + that bearer, so `auth.uid()` resolves to the uploader and **every existing RLS policy applies unchanged** (`001_initial_schema.sql:53`, `20260501_rag_pgvector.sql:31-41`). **Blast-radius reasoning (register #45):** this exact service was proven **publicly duplicable with a shared bearer token** — a `SUPABASE_SERVICE_ROLE_KEY` in that process would have upgraded #45 from "a duplicate converter" to a **full-database read/wipe breach**, which is precisely what `backfill.py`'s own docstring forbids. The client is built **per request, never at module level**, because a shared client holds the bearer header and would leak one user's token into another user's writes. **Hardening that came with it:** the service reads the target row **through RLS before any expensive work** (invisible row → 404, so a foreign `document_id` never reaches Voyage) and takes `kb_id` **from the database**, rejecting a caller-supplied mismatch; the ready-update checks it **matched a row**, since an RLS-filtered `UPDATE` returns 200/0-rows rather than an error; persistence reuses `backfill.py`'s **idempotent** shape (`delete().eq(document_id)` then batch-insert at 50). **Error ownership moved with the work:** Railway writes its own `status='error'` + `error_message`; Next keeps **unconditional** error writes only **pre-forward**, and every **post-forward** failure (including a thrown/timed-out `fetch`, which previously fell to the outer catch and stranded the row) now writes through a **`.eq('status','processing')` guard** — that closes the original orphan class **without** opening its mirror image, where Next stomps `error` over a `ready` row the service had already written before the ack was lost. | **Founder-owned open items (PIVOT_PLAN §10) — not engineering-blocked, tracked for launch:** Next.js hosting decision (Vercel Pro vs self-host, ~Phase 7/10) · **Supabase backend @@ -208,6 +209,10 @@ bodies total 37,095 bytes. **This retires the Option C frozen-tail invariant by existed only to police a boundary inside an unreviewable single line, and the append-only rule above supersedes it. No bespoke hash is needed for future updates: the diff is the proof. +### 2026-07-25 - (b1) chunk+embedding persistence moved into Railway with full ownership (register #51) + +(**Enabler (b1) from register #50 is IMPLEMENTED — the Railway ingestion service now owns chunk persistence AND the terminal `documents` status. This is the B6 ENABLER, not B6: ingestion is still synchronous, Next still awaits Railway, and B5b (§1) stays ⬜ — (b1) is not B5b progress.** **The move:** `/convert` is **renamed `/ingest`** and now runs convert → chunk → embed → **persist chunks** → **write the terminal status**, returning only `{document_id, chunk_count, status}`. The rename is deliberate rather than an in-place change: both the request shape (it now needs `document_id`, `kb_id`, and the user's token) and the response shape are incompatible with the old contract, so a **version-skewed deploy gets a loud 404** instead of an old Next silently overwriting `markdown_content` with `null` on a document the service had just finished correctly — which means **deploy order matters** (either order is loud and recoverable; uploads in the window fail to `status='error'`). `src/app/api/ingest/route.ts` still creates the `documents` row, still forwards, still awaits, and still gates both the browser reply and `recordStudyEvent('material_uploaded')` on the ack — but it **no longer receives or inserts chunks**, and the **9-13× return blob register #50 measured** (`await pyResponse.json()` over every chunk's 1024-float embedding plus the full markdown) is gone from the Next function's memory. **AUTH — Option B (forwarded user JWT), chosen over a service-role key, blocking constraint honored:** Railway writes **as the uploading user**. Next lifts the already-verified session's `access_token` (`getUser()` above it is what verified it) and forwards it in **`X-Supabase-Token`**, kept separate from `Authorization` which still carries the service-to-service `INGESTION_TOKEN`; the service builds a **per-request** Supabase client from the **public anon key** + that bearer, so `auth.uid()` is the uploader and the **existing RLS policies apply unchanged** (`001_initial_schema.sql:53` for `documents`, `20260501_rag_pgvector.sql:31-41` for `chunks` select/insert/delete). **Why not a service-role key — the register #45 blast radius:** #45 proved *this exact service* was **publicly duplicable and its bearer token shared**; an RLS-bypassing credential living in that process would have turned #45 from "a duplicated converter" into a **full-database read/wipe breach**, the precise failure `backfill.py`'s docstring forbids. **No service-role key was added to the service, its Dockerfile, its env, or the route — verified by grep across all three.** The client is built **per request, never module-level**, because the client object holds the bearer header and a shared one would leak one user's token into another user's writes. **Mechanism verified offline before shipping:** on the pinned `supabase==2.29.0`, `AsyncClientOptions(headers={'Authorization': ...})` reaches PostgREST as `apiKey: ` + `Authorization: Bearer ` (printed and checked), and supabase-py's `create()` skips its own session lookup when that header is supplied, so the forwarded token is the client's only identity. **Hardening that rode along:** the service reads the target row **through RLS before any expensive work** — an invisible row is a 404, so a foreign `document_id` never costs a Voyage batch — and takes **`kb_id` from the database**, rejecting a caller-supplied mismatch, so a holder of `INGESTION_TOKEN` cannot file chunks under one KB while pointing them at a document in another; the ready-update **checks it matched a row**, because an RLS-filtered `UPDATE` returns **200 with zero rows**, not an error, and would otherwise ack `ready` for a document whose status never moved; persistence reuses **`backfill.py`'s idempotent shape** (`delete().eq(document_id)` then batch-insert at **50**), so a retry of a half-written document converges instead of duplicating; and `/health` now reports `supabase_configured` as a **boolean only** — never the URL or the key — so the deploy can be checked without an upload. **ERROR OWNERSHIP moved with the work, in both directions.** Railway writes its own `status='error'` + `error_message` on its own failures (best-effort, never masking the real error). Next keeps **unconditional** error writes only **pre-forward** (misconfig, storage, insert — nothing else can have touched the row yet); every **post-forward** failure now goes through a **`.eq('status','processing')` guard**, including a **thrown or timed-out `fetch`**, which previously fell through to the outer catch and left the row at `processing` **forever**. That is the orphan class this change exists to kill, and the guard closes it **without opening its mirror image**: if Railway succeeded and the ack was merely lost, the row is already `ready` and the guarded write is a **no-op** rather than stomping a correct document into `error`. (Honest residual: in that lost-ack case the browser is told the upload failed while the document is in fact `ready` — data is correct, the message is pessimistic; `router.refresh()` shows the real row. Fixing the *message* is B6's territory, not this step's.) **Deliberately NOT in this PR:** B6 itself (fire-and-forget — Next still awaits), and any change to `backfill.py`'s deploy posture (**untouched**, still local-only). **Scope:** `services/ingestion/main.py`, `services/ingestion/requirements.txt` (adds `supabase==2.29.0`, the version already proven in the local venv `backfill.py` runs against), `src/app/api/ingest/route.ts`, and this file. **No migration, no DB change** — so `db-types` cannot drift by construction; `tsc --noEmit` clean; `main.py` compiles. `PIVOT_PLAN.md` untouched; §1 B5b row untouched; register **#17** stays OPEN.) + ### 2026-07-23 - B5b upload-hardening scoping recorded (register #50, docs/b5b-scoping.md) (**B5b (deep upload content hardening) is now SCOPED, not started — register #50 opened, `docs/b5b-scoping.md` created; §1 B5b stays ⬜ because scoping is not progress on the gate.** The doc records this session's established + verified facts, its corrections, the split, and seven OPEN founder decisions. **Established (each verified against the code this session):** the real upload ceiling is Vercel's **hard ~4.5 MB edge cap** (platform constant, unraisable — `413 FUNCTION_PAYLOAD_TOO_LARGE` before the handler), **not** the 50 MB the app promises in **7 places** (`route.ts:51` check / `:52` hardcoded-English 413 message — register-#17 class, `DropZone.tsx:26`, and four i18n strings in `en.ts`/`ar.ts`); ingestion runs on **Railway** (`INGESTION_SERVICE_URL = knowflow-production.up.railway.app`), so the `route.ts`→`/convert` hop faces no second Vercel cap but a **9-13× return-path blob** — `/convert` returns each chunk's **1024-float embedding inline plus full markdown** (`main.py`: `EMBED_DIM=1024`, `CHUNK_TOKENS=512`, `CHUNK_OVERLAP=64`), and `await pyResponse.json()` materializes it in the Next function's memory; and **no `maxDuration` is configured anywhere** (grep-empty across `vercel.json`/`next.config.ts`/`package.json`/`src`), with Voyage embeds batched 128-at-60s and Supabase inserts batched 50, all sequential in one invocation — so the true user-facing ceiling is an **UNMEASURED duration limit**, not size. **Corrections recorded so they are not re-inherited:** the 4.5 MB cap does **NOT** mitigate decompression bombs (it bounds compressed bytes, never the expansion — 4.5 MB of crafted ZIP still exhausts the service); the pre-auth body-buffering concern is **bounded** by the edge cap, making the `getUser`-above-`formData` reorder a low-priority ride-along (`checkDocumentLimit` **cannot** move — it needs `kbId` from the body; `enforceLimit` **should not** — it would burn an upload credit on a rejected request, register #24); and **`PIVOT_PLAN.md:269`'s "Size cap (50 MB) already exists" is a documented FALSE claim** (the effective cap is the platform's ~4.5 MB), corrected here, `PIVOT_PLAN.md` deliberately untouched. **The split:** B5b-1 (request-path, doable now — magic-byte verification, a declared-size ZIP-ratio pre-check from the central directory, the auth reorder); B5b-2 (gated on **B6** — enforced expansion bounds, nested-archive depth, AV/content scanning, all needing actual expansion or a worker); plus enabler moves **(b1)** move chunk+embedding persistence into the Railway service (kills the 9-13× blob, relocates slow work to the only component without a duration ceiling — the enabler of B6, not B6 itself) and **(b2)** signed-URL direct-to-Supabase-storage upload (the only way past the 4.5 MB inbound cap; gated on the file-size ruling; #22's 5 GB/mo egress limits its payoff to ~100 uploads/mo at 50 MB). **Seven OPEN decisions (stated with consequences, none picked): (1)** max file size — ~4.5 MB vs. real 50 MB via (b2), gating the enforcement point and sync/async posture; **(2)** format list — keep six vs. drop the docx/pptx/xlsx ZIP trio that carries all bomb risk; **(3)** scanning scope — AV (nothing in the image today), PII, and **prompt-injection in extracted text** (flows unfiltered into Haiku via `` interpolation across summarize/quiz/agent — unmitigated, needs at minimum a documented accepted-risk decision); **(4)** failure posture — reject-at-upload (415) vs. accept-then-quarantine; **(5)** enforcement point — Next, Railway, or both (Railway trusts its caller on a single bearer token, so Next-only is bypassed if `INGESTION_TOKEN` leaks); **(6)** the three ZIP numbers (expansion ratio, absolute expanded cap, nesting depth) — each measured from the spoofable declared central directory vs. actual expansion; **(7)** the signature table (`%PDF-`; `PK\x03\x04` + inner part-path for the OOXML trio; the txt/md non-signature UTF-8/NUL rule — txt/md have no magic bytes at all). **OUTSTANDING measurement:** the duration ceiling — plain-`.txt`-on-Preview procedure + a failure-signature table (`200`+chunk_count=pass; `413`+Vercel HTML=edge cap; `413`+app JSON=app check; `500`+"Ingestion failed"; `504`/none=duration), each upload burning a rate-limit credit and the #22 free tier. Related registers: **#45** (the duplicate ingestion service closed while scoping this), **#22**, **#24**, **#17**. **Docs-only: no code, no migration, no DB, no route/validation/size-constant change; only `docs/b5b-scoping.md` (new) and this file changed; `Phase 7` stays ⬜, §1 B5b row untouched, `PIVOT_PLAN.md` untouched, register #17 stays OPEN.**) diff --git a/services/ingestion/main.py b/services/ingestion/main.py index f4c1208..65b032b 100644 --- a/services/ingestion/main.py +++ b/services/ingestion/main.py @@ -6,10 +6,11 @@ import httpx import tiktoken -from fastapi import FastAPI, UploadFile, File, Header, HTTPException, Body +from fastapi import FastAPI, UploadFile, File, Form, Header, HTTPException, Body from fastapi.responses import JSONResponse from markitdown import MarkItDown from pydantic import BaseModel +from supabase import acreate_client, AsyncClient, AsyncClientOptions app = FastAPI() @@ -19,6 +20,23 @@ VOYAGE_URL = "https://api.voyageai.com/v1/embeddings" EMBED_DIM = 1024 +# (b1) Persistence config. This service now writes chunks and owns the terminal +# `documents` status, so it needs Supabase — but it authenticates as the UPLOADING +# USER, using the access token forwarded per request, on top of the PUBLIC anon +# key. There is deliberately NO service-role key here and there must never be one: +# register #45 proved this exact service was publicly duplicable and its bearer +# token shared, so an RLS-bypassing credential in this process would have turned +# that incident into a full-database breach. backfill.py's docstring forbids the +# same thing for the same reason; it stays a local-only script. Every write below +# therefore lands under the existing RLS policies (001_initial_schema.sql:53, +# 20260501_rag_pgvector.sql:31-41) and can only touch rows the user already owns. +SUPABASE_URL = os.environ.get("SUPABASE_URL") +SUPABASE_ANON_KEY = os.environ.get("SUPABASE_ANON_KEY") + +# Matches backfill.py's CHUNK_INSERT_BATCH: keeps each insert under Supabase's +# request-size limit (a 1024-float vector per row is not small). +CHUNK_INSERT_BATCH = 50 + # Embedding provider seam (PIVOT_PLAN.md §4): swap providers with one env var, no # schema change. Default 'voyage' (the only provisioned path); 'bge_m3' is a # documented stub. Normalized so trailing space / casing can't cause a silent miss. @@ -133,6 +151,132 @@ async def _embed_voyage(texts: list[str], input_type: Literal["query", "document return out +async def _user_client(user_token: str) -> AsyncClient: + """Build a Supabase client that acts AS THE UPLOADING USER (b1, Option B). + + The anon key is the apiKey; the user's forwarded access token is the + Authorization bearer, so PostgREST evaluates `auth.uid()` as that user and + every RLS policy applies unchanged. supabase-py's `create()` skips its own + session lookup when an Authorization header is supplied in options, so this + token is the only identity the client ever has. + + Built PER REQUEST on purpose. A module-level client would be shared across + concurrent uploads and would leak one user's bearer token into another + user's writes — the client holds the header, not the call. + """ + if not SUPABASE_URL or not SUPABASE_ANON_KEY: + raise HTTPException( + status_code=503, + detail="SUPABASE_URL / SUPABASE_ANON_KEY are not set", + ) + return await acreate_client( + SUPABASE_URL, + SUPABASE_ANON_KEY, + AsyncClientOptions(headers={"Authorization": f"Bearer {user_token}"}), + ) + + +async def _load_owned_document(sb: AsyncClient, document_id: str, kb_id: str) -> dict: + """Read the target row THROUGH RLS before doing any expensive work. + + Two jobs. (1) Ownership pre-check: the row is invisible to this token unless + the user owns its knowledge base, so a missing row means "not yours" and we + stop before spending a Voyage batch on it. (2) We take `kb_id` from the + DATABASE, not from the caller — the request's kb_id is cross-checked against + it and a mismatch is rejected, so a caller holding INGESTION_TOKEN cannot + file chunks under one KB while pointing them at a document in another. + """ + res = await ( + sb.table("documents").select("id, kb_id").eq("id", document_id).execute() + ) + rows = res.data or [] + if not rows: + raise HTTPException( + status_code=404, + detail="Document not found or not owned by this user", + ) + row = rows[0] + if str(row["kb_id"]) != str(kb_id): + raise HTTPException(status_code=400, detail="kb_id does not match document") + return row + + +async def _persist( + sb: AsyncClient, + document_id: str, + kb_id: str, + markdown_text: str, + chunks: list[dict], +) -> None: + """Write the chunk set and flip the document to its terminal ready state. + + Same idempotent shape backfill.py uses: clear any partial chunk set left by a + prior failed attempt for this document, then batch-insert. A retry of a + half-written document therefore converges instead of accumulating duplicates. + """ + await sb.table("chunks").delete().eq("document_id", document_id).execute() + + rows = [ + { + "document_id": document_id, + "kb_id": kb_id, + "chunk_index": c["chunk_index"], + "content": c["content"], + "token_count": c["token_count"], + "embedding": c["embedding"], + } + for c in chunks + ] + for i in range(0, len(rows), CHUNK_INSERT_BATCH): + await sb.table("chunks").insert(rows[i : i + CHUNK_INSERT_BATCH]).execute() + + # The terminal status write. This service owns it now: once the ack leaves + # here the row is already `ready`, so a caller that times out waiting for the + # ack cannot strand a document at `processing`. + res = await ( + sb.table("documents") + .update( + { + "markdown_content": markdown_text, + "status": "ready", + "embedding_status": "ready", + "chunk_count": len(rows), + "error_message": None, + } + ) + .eq("id", document_id) + .execute() + ) + # An UPDATE filtered out by RLS is not an error in PostgREST — it matches + # zero rows and returns 200. Without this check we would ack "ready" for a + # document whose status never moved. + if not (res.data or []): + raise RuntimeError("Ready update matched no document row") + + +async def _mark_error(sb: AsyncClient, document_id: str, message: str) -> None: + """Best-effort terminal failure write. Never raises. + + If this cannot land (expired token, Supabase down), the caller still sees a + non-2xx ack and applies its own fallback, so the row does not stay stuck. + """ + try: + await ( + sb.table("documents") + .update( + { + "status": "error", + "embedding_status": "error", + "error_message": message[:500], + } + ) + .eq("id", document_id) + .execute() + ) + except Exception as e: # noqa: BLE001 - diagnostics only; must not mask the real error + print(f"_mark_error failed for {document_id}: {e}", flush=True) + + @app.get("/health") async def health(): return { @@ -140,48 +284,91 @@ async def health(): "embed_provider": EMBEDDING_PROVIDER, "embed_model": VOYAGE_MODEL, "embed_dim": EMBED_DIM, + # (b1) Booleans only — never the URL or the key. Lets the deploy check + # "is persistence configured" without an upload and without leaking + # anything to an unauthenticated caller. + "supabase_configured": bool(SUPABASE_URL and SUPABASE_ANON_KEY), } -@app.post("/convert") -async def convert( - file: UploadFile = File(...), - authorization: str | None = Header(default=None), -): - """Convert a document to markdown, chunk it, and return chunks with embeddings.""" - _check_auth(authorization) - +def _convert_to_markdown(file: UploadFile) -> str: fd, tmp_path = tempfile.mkstemp(suffix=f"_{_safe_basename(file.filename)}") os.close(fd) try: with open(tmp_path, "wb") as buffer: shutil.copyfileobj(file.file, buffer) - md = MarkItDown() result = md.convert(tmp_path) - markdown_text = result.text_content or "" + return result.text_content or "" + finally: + if os.path.exists(tmp_path): + try: + os.remove(tmp_path) + except OSError: + pass + + +@app.post("/ingest") +async def ingest( + file: UploadFile = File(...), + document_id: str = Form(...), + kb_id: str = Form(...), + authorization: str | None = Header(default=None), + x_supabase_token: str | None = Header(default=None), +): + """Convert -> chunk -> embed -> PERSIST, and own the terminal document status. + + (b1) Replaces `/convert`, which returned every chunk's 1024-float embedding + plus the full markdown inline for the caller to insert. Two things are fixed + by moving the writes here: that 9-13x return blob no longer crosses the + network or gets materialized in the caller's memory, and the document can no + longer be stranded at `processing` by a caller that dies after we succeed — + because the `ready` transition happens here, before the ack is sent. + + RENAMED rather than changed in place on purpose. The request and response + shapes are both incompatible with the old contract, so a version-skewed + caller gets a loud 404 instead of silently persisting a null markdown_content + over a document this service already finished correctly. + + Two credentials, two different jobs. `Authorization` is INGESTION_TOKEN and + proves the caller is our own backend. `X-Supabase-Token` is the END USER's + Supabase access token and is the only thing that authorizes a database write + — it is used as a bearer on the anon key, so RLS decides what may be touched. + """ + _check_auth(authorization) + if not x_supabase_token: + raise HTTPException(status_code=401, detail="Missing X-Supabase-Token") + + # Built before the expensive work so a bad/expired user token fails fast, and + # so the failure writer below is always available once conversion starts. + sb = await _user_client(x_supabase_token) + await _load_owned_document(sb, document_id, kb_id) + try: + markdown_text = _convert_to_markdown(file) chunks = _chunk_text(markdown_text) if chunks: embeddings = await _embed([c["content"] for c in chunks], input_type="document") + if len(embeddings) != len(chunks): + raise RuntimeError("chunk/embedding count mismatch") for c, emb in zip(chunks, embeddings): c["embedding"] = emb - return { - "markdown": markdown_text, - "filename": file.filename, - "chunks": chunks, - } - except HTTPException: + await _persist(sb, document_id, kb_id, markdown_text, chunks) + except HTTPException as e: + await _mark_error(sb, document_id, str(e.detail)) raise except Exception as e: + await _mark_error(sb, document_id, str(e)) return JSONResponse(status_code=500, content={"error": str(e)}) - finally: - if os.path.exists(tmp_path): - try: - os.remove(tmp_path) - except OSError: - pass + + # Small ack by design: no embeddings, no markdown. The caller already knows + # the filename and does not need the content back — it is in the database. + return { + "document_id": document_id, + "chunk_count": len(chunks), + "status": "ready", + } class EmbedRequest(BaseModel): diff --git a/services/ingestion/requirements.txt b/services/ingestion/requirements.txt index 1e92d39..481a37b 100644 --- a/services/ingestion/requirements.txt +++ b/services/ingestion/requirements.txt @@ -4,3 +4,4 @@ uvicorn==0.30.0 python-multipart==0.0.9 httpx==0.27.2 tiktoken==0.7.0 +supabase==2.29.0 diff --git a/src/app/api/ingest/route.ts b/src/app/api/ingest/route.ts index 903d518..f8d3355 100644 --- a/src/app/api/ingest/route.ts +++ b/src/app/api/ingest/route.ts @@ -5,11 +5,15 @@ import { enforceLimit } from '@/lib/rate-limit'; import { recordStudyEvent } from '@/lib/study-events'; import { ALLOWED_FILE_TYPES, type FileType } from '@/types'; -interface IngestionChunk { - chunk_index: number; - content: string; - token_count: number; - embedding: number[]; +// (b1) The ingestion service's ack. Deliberately tiny: it persists the chunks and +// writes the terminal document status itself, so nothing comes back here except +// confirmation. The old contract returned every chunk with its 1024-float +// embedding plus the full markdown, and `await pyResponse.json()` materialized +// all of it in this function's memory. +interface IngestionAck { + document_id?: string; + chunk_count?: number; + status?: string; } // B5a: upload allowlist — only the formats MarkItDown handles well. Maps each @@ -140,6 +144,24 @@ export async function POST(request: Request) { return NextResponse.json({ success: false, error: docError?.message }, { status: 500 }); } + // (b1) Once the request leaves for the ingestion service, THAT service owns + // the terminal status — it writes `ready` before it acks us. So a failure on + // OUR side after the forward must never blind-write `error`: the service may + // well have succeeded and had its ack lost to a timeout, and stomping a + // correctly-finished `ready` row would destroy a document that is actually + // fine. The `.eq('status', 'processing')` guard makes this write a no-op in + // exactly that case, and still rescues the row when the service never got + // far enough to set anything. Pre-forward failures below keep their + // unconditional writes — nothing else can have touched the row yet. + const failIfStillProcessing = async (message: string) => { + const { error: guardErr } = await supabase + .from('documents') + .update({ status: 'error', embedding_status: 'error', error_message: message }) + .eq('id', docRecord.id) + .eq('status', 'processing'); + if (guardErr) console.error('Post-forward error write failed:', guardErr); + }; + const pythonServiceUrl = process.env.INGESTION_SERVICE_URL || 'http://localhost:8000'; const ingestionToken = process.env.INGESTION_TOKEN; if (!ingestionToken) { @@ -148,97 +170,79 @@ export async function POST(request: Request) { return NextResponse.json({ success: false, error: 'Server misconfigured' }, { status: 500 }); } + // (b1) The ingestion service writes to Supabase AS THIS USER, so it needs the + // user's own access token. It is NOT given a service-role key — see the note + // in services/ingestion/main.py and register #45: that service was publicly + // duplicable, so an RLS-bypassing credential there is a full-database breach. + // `getUser()` above already verified the session against the auth server; + // this only lifts the token that verification was performed on. + const { data: { session } } = await supabase.auth.getSession(); + const accessToken = session?.access_token; + if (!accessToken) { + console.error('No access token on an authenticated session'); + await supabase.from('documents').update({ status: 'error', embedding_status: 'error' }).eq('id', docRecord.id); + return NextResponse.json({ success: false, error: 'Server misconfigured' }, { status: 500 }); + } + + // document_id + kb_id travel with the file so the service knows which row to + // write; the user token travels in its own header (Authorization is already + // carrying the service-to-service token) and is never put in the form body. const pyFormData = new FormData(); pyFormData.append('file', file); - - const pyResponse = await fetch(`${pythonServiceUrl}/convert`, { - method: 'POST', - headers: { Authorization: `Bearer ${ingestionToken}` }, - body: pyFormData, - }); - - if (!pyResponse.ok) { - await supabase.from('documents').update({ status: 'error', embedding_status: 'error' }).eq('id', docRecord.id); + pyFormData.append('document_id', docRecord.id); + pyFormData.append('kb_id', kbId); + + let pyResponse: Response; + try { + pyResponse = await fetch(`${pythonServiceUrl}/ingest`, { + method: 'POST', + headers: { + Authorization: `Bearer ${ingestionToken}`, + 'X-Supabase-Token': accessToken, + }, + body: pyFormData, + }); + } catch (fetchErr) { + // The orphan class this whole change exists to kill: the request died in + // flight (timeout, socket reset) and we have no idea whether the service + // finished. Previously this fell to the outer catch and left the row at + // `processing` forever. Now the service has already written `ready` if it + // succeeded, and the guard below only touches the row if it did not. + console.error('Ingestion service unreachable:', fetchErr); + await failIfStillProcessing('Ingestion service unreachable'); return NextResponse.json({ success: false, error: 'Ingestion failed' }, { status: 500 }); } - const result: { markdown?: string; chunks?: IngestionChunk[] } = await pyResponse.json(); - const chunks = result.chunks ?? []; - - if (chunks.length > 0) { - const rows = chunks.map((c) => ({ - document_id: docRecord.id, - kb_id: kbId, - chunk_index: c.chunk_index, - content: c.content, - token_count: c.token_count, - // chunks.embedding is pgvector `vector`, generated as `string`. postgREST - // accepts a JSON number[] and coerces it to the vector on insert, which is - // how ingestion has always worked. Type-only assertion; the value written - // is still the number[] from the chunker, so ingestion behaviour is - // unchanged. - embedding: c.embedding as unknown as string, - })); - - // Insert in batches to avoid Supabase request-size limits. - const BATCH = 50; - for (let i = 0; i < rows.length; i += BATCH) { - const { error: chunkErr } = await supabase.from('chunks').insert(rows.slice(i, i + BATCH)); - if (chunkErr) { - console.error('Chunk insert error:', chunkErr); - await supabase - .from('documents') - .update({ status: 'error', embedding_status: 'error', error_message: chunkErr.message }) - .eq('id', docRecord.id); - return NextResponse.json({ success: false, error: 'Failed to store chunks' }, { status: 500 }); - } - } + if (!pyResponse.ok) { + // The service sets its own `error` status on its own failures; this is the + // fallback for the cases where it could not (misconfig, 401, crash). + await failIfStillProcessing('Ingestion failed'); + return NextResponse.json({ success: false, error: 'Ingestion failed' }, { status: 500 }); } - // We still keep the raw markdown for debugging / re-chunking, but it's no - // longer used at query time. - const { error: readyErr } = await supabase - .from('documents') - .update({ - markdown_content: result.markdown ?? null, - status: 'ready', - embedding_status: 'ready', - chunk_count: chunks.length, - }) - .eq('id', docRecord.id); - if (readyErr) { - // PostgREST reports DB failures in `error`, not by throwing, so an unchecked - // ready write would fall through to the emit below with the row still at - // `status: 'processing'`. Guard it like every other status write in this - // route: log, flip the row to `error`, and exit non-2xx. The update is a - // single atomic UPDATE, so a returned error means none of the four fields - // landed — flipping to `error` is not fighting a half-applied ready write. - console.error('Ready status update error:', readyErr); - await supabase - .from('documents') - .update({ status: 'error', embedding_status: 'error', error_message: readyErr.message }) - .eq('id', docRecord.id); - return NextResponse.json({ success: false, error: 'Failed to finalize document' }, { status: 500 }); + // Small ack only: {document_id, chunk_count, status}. No chunks, no + // embeddings, no markdown — none of it is materialized here any more. + const ack: IngestionAck = await pyResponse.json(); + if (ack.status !== 'ready' || typeof ack.chunk_count !== 'number') { + console.error('Unexpected ingestion ack:', ack); + await failIfStillProcessing('Ingestion returned an unexpected result'); + return NextResponse.json({ success: false, error: 'Ingestion failed' }, { status: 500 }); } - // P5.2 study event. You asked whether this route can emit while the document is - // "still processing" — it now cannot. `status: 'processing'` is written at the - // insert above, but the route then BLOCKS on the converter fetch, the chunk - // inserts, and the final `status: 'ready'` update; every one of those has an - // error return that flips the row to `status: 'error'` and exits non-2xx — - // INCLUDING the ready write just above, whose `error` is now captured rather - // than dropped. So the guarantee is: the row is `ready` unless we verified - // otherwise and bailed. By the time control reaches this line the ready - // transition was confirmed successful, making this an unambiguous success - // point. (A future move to background processing would break that and the emit - // would have to follow the work, not the request.) + // P5.2 study event. The emit stays gated on a CONFIRMED success, which is now + // the service's ack rather than a ready-write made here: reaching this line + // means the service reported `status: 'ready'`, and it only reports that + // after its chunk inserts AND its `documents` update both landed (it verifies + // the update matched a row, since an RLS-filtered UPDATE returns 200/0 rows). + // Every failure path above returns non-2xx before this point. // // Past the 400, the 413, both 415s, the 401, the per-KB 403, the rate-limit - // denial, the storage 500, the insert 500, the misconfig 500, the converter 500, - // the chunk-insert 500, and the ready-write 500. Fails open; never throws. + // denial, the storage 500, the insert 500, the misconfig 500, the missing-token + // 500, the unreachable-service 500, the non-2xx ack 500, and the bad-ack 500. + // Fails open; never throws. await recordStudyEvent(supabase, 'material_uploaded'); - return NextResponse.json({ success: true, document_id: docRecord.id, chunk_count: chunks.length }); + return NextResponse.json({ success: true, document_id: docRecord.id, chunk_count: ack.chunk_count }); } catch (error) { console.error('Ingest API error:', error); return NextResponse.json({ success: false, error: 'Internal Server Error' }, { status: 500 });