Skip to content

2026-08-16 — Persist source PDFs to Supabase Storage (stop re-downloading the source)

Date: 2026-08-16 Branch: backend-rag Status: Plan — not yet implemented (design only) Related: app/services/pdf_fetcher.py, app/services/ingestion.py, scripts/backfill_assets.py

Goal

Today every ingest downloads-and-discards the source PDF (fetch_pdf(ref["pdf_source"]) at ingestion.py:820). The founder wants two things:

  1. Save each source PDF once into the self-hosted Supabase Storage on the VPS (private bucket curriculum-pdfs).
  2. Stop relying on the external source — after a PDF is archived, read the stored copy instead of re-downloading from koutoubi.mr / bsimr.com.

This is a write-through cache of the PDF bytes keyed on the reference. The retrieval and chunking pipeline is unchanged; only where the bytes come from changes.

Non-goals

  • No change to parsing, OCR, chunking, embedding, or retrieval.
  • No change to the documents/chunks schema. New state lives on references only.
  • No new always-on service, worker, or broker (respect the shared-VPS resource envelope).
  • Not switching the app off the Supabase client API (no direct S3/Postgres driver).

Ground truth this builds on (verified)

  • fetch_pdf(url, timeout, transport) is SSRF-guarded. Its guard (validate_pdf_source_url) already special-cases Supabase Storage presigned reads: _is_supabase_presigned accepts a URL whose scheme+host+port match SUPABASE_URL and whose path starts with /storage/v1/object/sign/. So a signed read-back URL passes the existing guard with no allowlist change — PDF_SOURCE_ALLOWED_HOSTS does not need the bucket/host added. ⚠️ The guard rejects any path containing % or .. (file_api line-normalization defense). The signed URL path is /storage/v1/object/sign/<bucket>/<storage_path> — so storage_path must contain no characters that get percent-encoded (see key sanitization).
  • Supabase SDK is supabase==2.31.0 / storage3==2.31.0. Verified signatures:
  • client.storage.from_(bucket).upload(path, file_bytes, {"content-type": ..., "upsert": "true"}) → POST; upsert: "true" maps to the x-upsert header (overwrite-in-place).
  • client.storage.from_(bucket).create_signed_url(path, expires_in) → returns a dict with keys signedURL and signedUrl (both present; use ["signedURL"]).
  • Service-role client (create_client(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY), exposed as the supabase_service lazy proxy and IngestionService.self.supabase) bypasses Storage RLS — service-role is the storage owner, uploads/signs need no bucket policy.
  • references already carries nullable storage_path TEXT, content_sha256 TEXT, file_size BIGINT, archived_at TIMESTAMPTZ + a partial index WHERE content_sha256 IS NULL (owned by the migration/ Flyway project — see Migration prerequisite). Intended key convention from the migration comment: <source>/<record_key>.pdf.

Design

1. app/services/storage_service.py (new)

A thin, stateless module. Functions take the Supabase client explicitly (mockable; mirrors how backfill_assets.py uses the client). No new class/service is registered in dependencies.py.

import hashlib, re
from app.core.config import settings
from app.core.logging import our_logs           # match existing import site
from app.services.pdf_fetcher import fetch_pdf

_SAFE_KEY = re.compile(r"[^A-Za-z0-9._-]+")

def compute_sha256(pdf_bytes: bytes) -> str:
    """Hex SHA-256 of the raw PDF bytes (deterministic content id)."""
    return hashlib.sha256(pdf_bytes).hexdigest()

def _sanitize(part: str) -> str:
    # Keep the object key percent-encoding-free so the signed-read URL passes the
    # SSRF guard's `%`-in-path rejection. Collapse everything else to '-'.
    return _SAFE_KEY.sub("-", (part or "").strip()).strip("-") or "unknown"

def derive_storage_path(reference: dict) -> str:
    """`<source>/<record_key>.pdf`; falls back to the reference id when record_key is null."""
    source = _sanitize(reference.get("source") or "unknown")
    key = reference.get("record_key") or reference.get("id")
    return f"{source}/{_sanitize(str(key))}.pdf"

def is_archived(reference: dict) -> bool:
    """True once a PDF has been stored for this reference."""
    return bool(reference.get("content_sha256") and reference.get("storage_path"))

def archive_pdf(supabase, reference: dict, pdf_bytes: bytes) -> dict:
    """Upload bytes to the curriculum-pdfs bucket (idempotent upsert). Returns metadata.
    Raises on a hard storage failure — callers decide best-effort vs fatal."""
    storage_path = derive_storage_path(reference)
    meta = {
        "storage_path": storage_path,
        "content_sha256": compute_sha256(pdf_bytes),
        "file_size": len(pdf_bytes),
    }
    supabase.storage.from_(settings.SUPABASE_PDF_BUCKET).upload(
        storage_path, pdf_bytes,
        {"content-type": "application/pdf", "upsert": "true"},   # overwrite-in-place = idempotent
    )
    return meta

def signed_url_for(supabase, storage_path: str, expires_in: int = 600) -> str:
    """Presigned GET URL for read-back. Accepted by fetch_pdf's SSRF guard."""
    res = supabase.storage.from_(settings.SUPABASE_PDF_BUCKET).create_signed_url(
        storage_path, expires_in)
    return res["signedURL"]

Idempotency / "already exists": upsert: "true" makes re-upload of the same key a harmless overwrite (same bytes → byte-identical object). We therefore never need to special-case a 409/"Duplicate" — but the resolver still avoids re-uploading when the reference is already archived (see below), so the steady state is zero redundant uploads. Keep the upload inside archive_pdf narrow; do not swallow errors here — the caller owns the best-effort policy so the same helper is reusable by a stricter backfill.

2. Resolver — the core behavior change

resolve_pdf_bytes replaces the raw fetch_pdf(ref["pdf_source"]) call. Lives in storage_service.py (keeps ingestion.py thin; matches "fat services" rule).

def resolve_pdf_bytes(supabase, reference: dict, *, timeout: float = 60.0) -> bytes:
    """Return the reference's PDF bytes, preferring the archived copy.

    - Archive feature off  -> always fetch from source (exact legacy behavior).
    - Already archived      -> read the stored copy via a signed URL (through fetch_pdf).
                               On stored-read failure: log + fall back to source (availability
                               wins), and DO NOT clear storage_path.
    - Not archived yet      -> fetch from source, then best-effort archive + persist metadata
                               (write-through). Archive failure never blocks ingest.
    """
    if not settings.PDF_ARCHIVE_ENABLED:
        return fetch_pdf(reference["pdf_source"], timeout=timeout)

    if is_archived(reference):
        try:
            url = signed_url_for(supabase, reference["storage_path"])
            return fetch_pdf(url, timeout=timeout)              # reads the VPS-local copy
        except Exception as e:                                  # noqa: BLE001
            our_logs.error(f"PDF stored-read failed for {reference.get('id')}: {e}; "
                           f"falling back to source")
            return fetch_pdf(reference["pdf_source"], timeout=timeout)

    # Not archived: fetch from source once, seed storage, record metadata.
    pdf_bytes = fetch_pdf(reference["pdf_source"], timeout=timeout)
    if settings.PDF_ARCHIVE_ENABLED:
        try:
            meta = archive_pdf(supabase, reference, pdf_bytes)
            _persist_archive_metadata(supabase, reference["id"], meta)
            reference.update(meta)          # so downstream is_archived() sees it in-process
        except Exception as e:              # noqa: BLE001 — archive is best-effort
            our_logs.error(f"PDF archive failed for {reference.get('id')}: {e}; "
                           f"ingest continues from source bytes")
    return pdf_bytes

def _persist_archive_metadata(supabase, reference_id: str, meta: dict) -> None:
    supabase.table("references").update(
        {**meta, "archived_at": datetime.utcnow().isoformat()}
    ).eq("id", reference_id).execute()

Policy rationale (recommended, safe): - Stored-read failure → fall back to source, keep storage_path. Availability beats a hard "never touch source" rule; a transient Storage hiccup must not fail an ingest. Ops sees the logged PDF stored-read failed line and can investigate. We do not auto-clear storage_path (avoids flapping a good archive to null on a blip). - Archive is best-effort. A failed upload leaves storage_path null → the next ingest re-fetches source and retries the archive. This matches the existing best-effort pattern for chapter tagging and asset extraction (ingestion.py 971–1009). It also means the feature degrades safely if the migration columns are not yet live (the metadata UPDATE fails, is swallowed and logged, and ingest proceeds exactly as today). - Idempotent + re-runnable. Metadata write keyed on reference_id; upload keyed on the deterministic storage_path with upsert. Re-ingesting an already-archived reference reads storage and uploads nothing. - Source-updated-at-origin case: once archived, we deliberately read the frozen copy and do not notice a changed source (that is the whole point — "stop relying on the source"). Re-archiving an updated source is an explicit action (backfill re-run or a future admin "re-archive" button), not an automatic ingest side effect. Flagged as an open question.

3. Wiring into ingestion.py

Single-line swap in execute_job, inside the parse_and_chunk stage (currently line ~820):

# before
file_bytes = fetch_pdf(ref["pdf_source"], timeout=60.0)
# after
file_bytes = resolve_pdf_bytes(self.supabase, ref, timeout=60.0)
  • Keep the surrounding try/except (PDFSourceNotAllowedError, httpx.HTTPError) exactly as is: resolve_pdf_bytes only raises those when it genuinely can't get bytes from either storage or source, so the existing _fail_or_requeue retry/terminal logic (Issue #311) still applies unchanged. (Archive failures never propagate — they're swallowed inside the resolver.)
  • ref is the in-memory dict from select("*"); reference.update(meta) mutates it in place so any later is_archived(ref) check within the same job is consistent.
  • The metadata columns are written inside the resolver, right after a successful archive — not folded into the finalize UPDATE at line 1055. Rationale: if a later stage fails and the job re-queues, the archive is already recorded, so the retry reads from storage and never re-hits the source. This is strictly better than waiting for finalize.
  • Admin dry-run / preview call sites (admin.py ~62/121/149/220 wrapped in def work():) are lower priority. They can adopt resolve_pdf_bytes(supabase_service, ref_row) in PR2 or a follow-up so previews also read the local copy; behavior is otherwise identical. Not on the critical path.

4. Config additions (app/core/config.py)

# PDF persistence (2026-08-16). Archive source PDFs to self-hosted Supabase Storage so
# ingestion stops re-downloading from the external curriculum source.
SUPABASE_PDF_BUCKET: str = "curriculum-pdfs"
PDF_ARCHIVE_ENABLED: bool = True   # kill switch: False => exact legacy download-and-discard
  • No PDF_SOURCE_ALLOWED_HOSTS change — presigned Storage reads are already whitelisted by _is_supabase_presigned (confirmed in pdf_fetcher.py).
  • PDF_ARCHIVE_ENABLED=False is a full revert: the resolver ignores storage_path and always fetches source, so ops can disable the feature without a redeploy of code.

5. Backfill script — scripts/backfill_pdf_archive.py

Mirror backfill_assets.py (uses setup_dependencies(), imports supabase_service, resumable, runs via docker cp + docker exec). One-time seed of storage from source.

  • Selection (uses the partial index): references where content_sha256 IS NULL and pdf_source is not null. Resumable = the IS NULL filter itself; a re-run skips rows a prior run already archived.
  • Per row: fetch_pdf(ref["pdf_source"]) (the deliberate one last source download), archive_pdf(...), _persist_archive_metadata(...). Reuse the exact storage_service functions so the key convention and hashing can't drift from the ingest path.
  • Flags: --dry-run (log the derived storage_path/size, no upload/no DB write), --limit N (cap rows per run). Progress logging [i/total] <path> -> archived size=… sha=…, a per-row try/except that counts failures without aborting the run, and a summary line.
  • Consider a per-row signal.alarm timeout like backfill_assets.py for wedged downloads.

6. Tests (tests/services/test_storage_service.py, mock all storage/DB calls)

  • compute_sha256 determinism + matches hashlib.sha256(b).hexdigest().
  • derive_storage_path: normal <source>/<record_key>.pdf; null record_key → uses id; unsafe chars (spaces, /, Arabic, ..) are sanitized to a percent-encoding-free key.
  • is_archived: true only when both content_sha256 and storage_path present.
  • archive_pdf: calls storage.from_(bucket).upload(path, bytes, {...upsert:"true"}) with the right path + content-type; returns correct content_sha256/file_size.
  • resolve_pdf_bytes branch selection (mock fetch_pdf, signed_url_for):
  • archive disabled → fetches source, no archive.
  • already archived → signs + reads stored copy, no source fetch.
  • not archived → fetches source, archives, persists metadata, mutates reference in place.
  • stored-read raises → falls back to source, storage_path untouched, logged.
  • archive raises → returns source bytes, ingest not affected (no raise), logged.
  • Signed-url passes the SSRF guard: feed a representative signed URL ({SUPABASE_URL}/storage/v1/object/sign/curriculum-pdfs/koutoubi/<key>.pdf?token=…) to validate_pdf_source_url and assert it does not raise (guards the %/.. constraint).
  • Idempotent re-archive: calling resolve_pdf_bytes on an already-archived reference issues zero uploads.

PR breakdown (reviewer-sized)

  • PR1 — storage primitives + config + unit tests. storage_service.py (compute_sha256, derive_storage_path, is_archived, archive_pdf, signed_url_for), the two config settings, and their unit tests. No behavior change to ingestion yet (nothing calls the new code). Also add the mkdocs nav entry for this doc so mkdocs build --strict passes, and a docs/96_implementation/ record. Depends on: nothing (columns can be absent — code isn't wired in yet).
  • PR2 — resolver + ingest wiring + tests. Add resolve_pdf_bytes + _persist_archive_metadata, swap the ingestion.py:820 call, add branch tests. Optionally switch the admin.py preview call sites. This PR is the behavior change and should land only after the migration columns are confirmed live in prod (otherwise it runs, but every archive best-effort-fails and logs until the columns exist). Depends on: PR1 + Migration prerequisite.
  • PR3 — backfill script. scripts/backfill_pdf_archive.py, run manually post-merge to seed storage for the existing corpus. Depends on: PR1 (reuses its functions) + columns live.

Migration prerequisite

The storage_path / content_sha256 / file_size / archived_at columns + the partial index on references are owned by the migration/ project (Flyway, branch db-migrations), not by rag/ (this checkout has no db/migrations/). Confirm they are live in prod before merging PR2 (a separate agent is verifying). If they are not yet live, ship the Flyway migration sql/V<timestamp>__references_pdf_archive_columns.sql there first. Safety net: because the metadata write is best-effort, PR2 does not crash without the columns — it just logs archive failures and keeps ingesting from source until they exist.

Risks & mitigations

  • RAM on the shared VPS. No new peak: full PDF bytes are already held in memory today for parse/OCR. The extra buffer is one hex digest + one upload of the same bytes. Backend mem_limit is 1g; the one-at-a-time ingestion drainer already bounds concurrency to one PDF.
  • Disk growth. Every archived PDF now persists on the VPS (host Postgres-adjacent Storage volume). Curriculum corpus is bounded (hundreds of PDFs, tens of MB each), but this is monotonic. Mitigation: private bucket, one object per reference (upsert, not versioned), and a note to add a Storage volume disk-usage check to the ops runbook.
  • Bucket RLS. Service-role bypasses RLS (it is the Storage owner), so uploads/signs need no policy. The bucket stays private; reads happen only via short-lived signed URLs generated server-side. Confirmed no anon exposure.
  • Partial write / corruption. Upload is a single PUT/POST of the full byte buffer — no streaming partial state. We write content_sha256 at archive time; a future integrity check can re-hash a stored read and compare. Metadata is written only after a successful upload, so a null storage_path always means "not archived", never "half archived".
  • Signed-URL / SSRF constraint. The read path depends on storage_path being percent-encoding-free (guard rejects % in path). derive_storage_path sanitization enforces this; the SSRF-guard test locks it in.
  • "Source now dead but not yet archived." A reference whose source 404s before its first successful archive can never be ingested — same failure mode as today. The backfill seeds storage for the current corpus to shrink this window; going forward, the first successful ingest archives the bytes, so subsequent source death is survivable. There is no way to archive a source that was never reachable — flagged as an accepted limitation.

Open questions for the founder

  1. Re-archiving a changed source. Once archived we freeze the copy and ignore source updates (intended). Do you want a manual admin "re-archive from source" action (and/or a backfill --force that re-downloads and overwrites) for when a curriculum PDF is corrected upstream? Default plan: no automatic refresh.
  2. Retention. Keep every archived PDF forever (simplest, monotonic disk), or add a cleanup for references that are deleted/superseded? Default plan: keep forever.
  3. Rollout order. Confirm the migration columns are live in prod before PR2 merges — OK to gate PR2 on that, or prefer shipping PR2 behind PDF_ARCHIVE_ENABLED=False first and flipping the flag once columns are confirmed?