diff --git a/jobs/gmail-header-backfill/README.md b/jobs/gmail-header-backfill/README.md new file mode 100644 index 0000000..31827a8 --- /dev/null +++ b/jobs/gmail-header-backfill/README.md @@ -0,0 +1,155 @@ +# gmail-header-backfill + +One-shot job, module 5 phase 2 backfill (`docs/kb/modules/05-faza2-plan.md`, §5). +Backfills `{"type": "headers", ...}` onto the 225 030 existing `source='gmail'` +envelope rows in kb-postgres, which today carry only an attachment manifest +(`{"type": "attachment", ...}`) — no `from`/`to`/`cc`/`delivered_to`/`subject` +anywhere in the DB (plan §1.8). This job does **not** touch `document_chunk` or +any non-`entities` data, and never rewrites existing `entities` elements — it +only appends one new element per envelope. + +## Why a separate job, not an extension of gmail-bulk-import + +`gmail-bulk-import` has INSERT semantics (new envelopes from mbox/Takeout), +already run once, historical. This job has UPDATE semantics against the +225 030 rows already in production — different risk profile, different +lifecycle. See plan §2 decision 7. + +## Where it runs + +**Locally on PIHA**, as a plain CLI (not a container) — same reasoning as +`documents-ingest`: it needs simultaneous local filesystem access to the mail +archive (`/home/oskar/kb/mail/archive`) and a DB connection to kb-postgres. + +Install (from repo root, on PIHA): + +```bash +pip install -e jobs/gmail-header-backfill/ +``` + +## Usage + +```bash +# Dry run (default) — parse and count only, no DB writes: +gmail-header-backfill --dsn postgresql://kb:@localhost:5433/kb --limit 100 + +# Real run — apply the UPDATE for this slice: +gmail-header-backfill --dsn ... --limit 1000 --offset 0 --apply + +# Next slice — offset is stable/deterministic (ORDER BY id), independent of +# how many rows in earlier slices were already backfilled: +gmail-header-backfill --dsn ... --limit 1000 --offset 1000 --apply +``` + +DSN can also come from the `KB_DSN` env var instead of `--dsn`. + +`--limit`/`--offset` exist so the full 225 030-row backfill can be run in +verifiable partitions instead of one long unattended run (plan §5.3) — start +small (`--limit 100`), check the result in the DB, then widen. + +## Source of headers: the archived `.eml`, not the original mbox + +Same access pattern as `documents-ingest`: `archive_root / row["raw_ref"]` -> +`read_bytes()` -> `email.message_from_bytes(raw, policy=email.policy.default)`. +The `.eml` archive is the immutable raw layer (kb-00 rule #1) and carries the +full original headers — no need to go back to the source Takeout mbox. + +Only headers are parsed — the job does **not** MIME-walk into attachment +parts (plan §5.1); that keeps per-message cost low relative to +`gmail-bulk-import`'s full import (which does walk attachments). + +## `entities[type=headers]` shape (plan §4.1) + +```json +{ + "type": "headers", + "from": {"name": "WARTA", "address": "no-reply@warta.pl"}, + "to": [{"name": "...", "address": "oskar@gmail.com"}], + "cc": [], + "delivered_to": ["oskar+alias@gmail.com", "oskar@gmail.com"], + "subject": "Twoja polisa OC/AC", + "date_raw": "Mon, 9 Jun 2026 12:34:56 +0200" +} +``` + +- `from`/`to`/`cc` are parsed to `{name, address}` via `email.policy.default` + (RFC 2047 encoded-word decoding) + `email.utils.getaddresses` on the + decoded text. `to`/`cc` are lists (comma-separated multi-address headers + are split); `from` is a single object or `null` — if a message carries more + than one `From:` header (malformed but seen in the wild), the first is used + and a `headers.multiple_from` warning is logged. +- `delivered_to` is a list of **raw, unparsed strings** — `Delivered-To` can + repeat per hop, and every occurrence is kept (this is what identifies which + of Oskar's aliases received the message; motivation in plan §1.8). +- `date_raw` is the literal original `Date:` header text, taken from a + separate `email.policy.compat32` parse — `policy.default`'s structured + `DateHeader` reformats the value (corrects the weekday name, zero-pads the + day) rather than preserving what was actually in the file, and `date_raw` + exists specifically for byte-for-byte comparison/debug against `envelope.ts` + (already parsed at import time — this is not a duplicate source of truth). +- Malformed/undecodable headers never raise — they degrade to best-effort + text or are skipped, logged, and counted; the row is left for a future run + rather than half-updated. + +## Idempotency and resumability (plan §5.2) + +```sql +UPDATE envelope +SET entities = entities || $2::jsonb +WHERE id = $1 + AND NOT EXISTS ( + SELECT 1 FROM jsonb_array_elements(entities) e WHERE e->>'type' = 'headers' + ); +``` + +Rows that already carry a `headers` entity are skipped (checked client-side +before building the batch, and enforced again at the SQL level as +defense-in-depth). Re-running any slice — including after a crash mid-batch — +is always safe: already-backfilled rows are no-ops, not double-appended. + +Writes are batched: rows are queued in memory and flushed via +`conn.executemany` every 500 rows (one round-trip per batch, not one +transaction per row) — mirrors `gmail-bulk-import`'s `_insert_batch` pattern. + +`--limit`/`--offset` partition by `ORDER BY id`, not by "rows still missing +headers" — this keeps a given `--offset` naming the same slice of the table +across repeated runs, so progress is easy to reason about (e.g. "slices 0, +1000, 2000, ... cover the whole table") independent of how much of it is +already done. + +## Performance estimate (not measured — job didn't exist before this change) + +Per plan §5.3: ~225 030 files, ~124 KB average. Each row costs an `open` + +`read` + header-only parse (no MIME-walk of attachments) + a batched UPDATE. +The full `gmail-bulk-import` run (parsing the whole mbox, including +attachment MIME-walk and inserts) took ~29 minutes. This job does less +per-message work but pays for opening 225k small files individually instead +of streaming one mbox — the plan's estimate is "same order of magnitude, +likely tens of minutes." Not measured directly in this change — see the +DoD note below. + +## Tests + +```bash +pip install -e jobs/gmail-header-backfill/ +cd jobs/gmail-header-backfill && pytest +``` + +Pure unit tests, no DB or filesystem outside `tmp_path`/synthetic `.eml` +bytes — `run()` is tested by monkeypatching `asyncpg.connect` with an +in-memory fake connection. Covers: header parsing (multi-address `To`/`Cc`, +quoted display names with commas, multiple `Delivered-To` occurrences, RFC +2047 encoded-words including Polish diacritics, malformed encoded-words that +must not raise, missing/multiple `From`, `date_raw` preserving literal text +vs. `Date` header reformatting), idempotency (rows already carrying a +`headers` entity are skipped and never re-appended), batch flushing, and +`--limit`/`--offset` query shape. + +## Definition of Done + +Per `CLAUDE.md`: this job's smoke run is `gmail-header-backfill --dsn ... +--limit 100` (dry-run first, then `--apply` against a small slice) — run +against kb-postgres@PIHA over SSH, **not executed as part of this change** +without operator confirmation (UPDATE against production data). `pytest` +passes locally (33/33) before this commit; full 225 030-row backfill is out +of scope for this change — see the plan for the rollout sequence. diff --git a/jobs/gmail-header-backfill/pyproject.toml b/jobs/gmail-header-backfill/pyproject.toml new file mode 100644 index 0000000..25549e1 --- /dev/null +++ b/jobs/gmail-header-backfill/pyproject.toml @@ -0,0 +1,22 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "gmail-header-backfill" +version = "0.1.0" +requires-python = ">=3.11" +dependencies = [ + "asyncpg>=0.29", + "structlog>=24.1", +] + +[project.scripts] +gmail-header-backfill = "gmail_header_backfill.backfill:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +asyncio_mode = "auto" +testpaths = ["tests"] diff --git a/jobs/gmail-header-backfill/src/gmail_header_backfill/__init__.py b/jobs/gmail-header-backfill/src/gmail_header_backfill/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/jobs/gmail-header-backfill/src/gmail_header_backfill/backfill.py b/jobs/gmail-header-backfill/src/gmail_header_backfill/backfill.py new file mode 100644 index 0000000..11970f4 --- /dev/null +++ b/jobs/gmail-header-backfill/src/gmail_header_backfill/backfill.py @@ -0,0 +1,272 @@ +"""Gmail header backfill — one-shot job, module 5 phase 2 (docs/kb/modules/05-faza2-plan.md, §5). + +Backfills `{"type": "headers", ...}` (§4.1) onto the 225 030 existing `source='gmail'` +envelope rows in kb-postgres, which today carry only an attachment manifest. This is an +UPDATE against production data (not an INSERT like gmail-bulk-import) — a deliberately +separate job from gmail-bulk-import, see plan §2 decision 7. + +Runs on PIHA (needs local access to the .eml archive): + +Install (from repo root): + pip install -e jobs/gmail-header-backfill/ + +Usage: + # Dry run (default) — parse and count only, no DB writes: + gmail-header-backfill --dsn postgresql://kb:@localhost:5433/kb --limit 100 + + # Real run — apply the UPDATE for this slice: + gmail-header-backfill --dsn ... --limit 1000 --offset 0 --apply + + # Next slice (stable, deterministic partitioning by id order): + gmail-header-backfill --dsn ... --limit 1000 --offset 1000 --apply + +DSN can also come from the KB_DSN env var instead of --dsn. + +Idempotency: the UPDATE only touches rows that don't already carry a `headers` entity +(§5.2 `WHERE NOT EXISTS`), so re-running any slice (including after a crash) is safe — +already-backfilled rows are skipped, not double-appended. +""" +from __future__ import annotations + +import argparse +import asyncio +import email +import email.policy +import email.utils +import json +import os +import sys +from pathlib import Path +from typing import Optional + +import asyncpg +import structlog + +_log = structlog.get_logger(__name__) + +DEFAULT_ARCHIVE_ROOT = Path("/home/oskar/kb/mail/archive") +DEFAULT_LIMIT = 1000 +UPDATE_BATCH_SIZE = 500 + +_UPDATE_SQL = """ +UPDATE envelope +SET entities = entities || $2::jsonb +WHERE id = $1 + AND NOT EXISTS ( + SELECT 1 FROM jsonb_array_elements(entities) e WHERE e->>'type' = 'headers' + ) +""" + + +def parse_headers(raw: bytes) -> dict: + """Parse Gmail .eml headers into the `{"type": "headers", ...}` shape (plan §4.1). + + `from`/`to`/`cc` are decoded via `email.policy.default` (resolves RFC 2047 + encoded-words to real Unicode) and then split into {name, address} pairs with + `email.utils.getaddresses`. `delivered_to` is kept as a list of raw (decoded but + unparsed) strings — `Delivered-To` may repeat per hop, and every occurrence is + kept (see plan §1.8/§4.1). `date_raw` comes from a separate compat32 parse: the + default policy's DateHeader reformats the header (e.g. corrects the weekday name, + zero-pads the day) instead of preserving the literal original text. + """ + msg = email.message_from_bytes(raw, policy=email.policy.default) + msg_compat = email.message_from_bytes(raw, policy=email.policy.compat32) + + from_headers = msg.get_all("From") or [] + if len(from_headers) > 1: + _log.warning("headers.multiple_from", count=len(from_headers)) + + from_obj: Optional[dict] = None + if from_headers: + from_addrs = email.utils.getaddresses([str(from_headers[0])]) + if from_addrs and from_addrs[0][1]: + name, address = from_addrs[0] + from_obj = {"name": name or None, "address": address} + + to_list = [ + {"name": name or None, "address": address} + for name, address in email.utils.getaddresses( + [str(h) for h in (msg.get_all("To") or [])] + ) + if address + ] + + cc_list = [ + {"name": name or None, "address": address} + for name, address in email.utils.getaddresses( + [str(h) for h in (msg.get_all("Cc") or [])] + ) + if address + ] + + delivered_to = [str(h) for h in (msg.get_all("Delivered-To") or [])] + + subject_header = msg.get("Subject") + subject = str(subject_header) if subject_header is not None else None + + date_raw = msg_compat.get("Date") + + return { + "type": "headers", + "from": from_obj, + "to": to_list, + "cc": cc_list, + "delivered_to": delivered_to, + "subject": subject, + "date_raw": date_raw, + } + + +def _decode_jsonb(value: object) -> object: + """asyncpg may return jsonb as a str or an already-decoded object.""" + if value is None: + return None + if isinstance(value, str): + return json.loads(value) + return value + + +def _has_headers(entities: list) -> bool: + return any(isinstance(e, dict) and e.get("type") == "headers" for e in entities) + + +async def fetch_batch(conn: asyncpg.Connection, limit: int, offset: int) -> list: + """One deterministic slice of `source='gmail'` envelopes, ordered by id. + + Ordering by id (not filtering out already-backfilled rows here) keeps --offset + stable across runs — the same --limit/--offset always names the same slice of the + 225 030-row table, so partial runs can be resumed or re-verified predictably. + Idempotency itself is enforced by the UPDATE's WHERE NOT EXISTS (see _UPDATE_SQL). + """ + return await conn.fetch( + """ + SELECT id, raw_ref, entities + FROM envelope + WHERE source = 'gmail' + ORDER BY id + LIMIT $1 OFFSET $2 + """, + limit, + offset, + ) + + +async def _update_batch(conn: asyncpg.Connection, rows: list[tuple[str, str]]) -> None: + """Multi-row UPDATE for one batch — a single executemany round, not one transaction per row.""" + if not rows: + return + await conn.executemany(_UPDATE_SQL, rows) + _log.debug("update_batch_flushed", count=len(rows)) + + +async def run( + dsn: str, + archive_root: Path = DEFAULT_ARCHIVE_ROOT, + limit: int = DEFAULT_LIMIT, + offset: int = 0, + apply: bool = False, +) -> dict[str, int]: + """Backfill headers for one --limit/--offset slice of `source='gmail'` envelopes. + + dry-run (apply=False) parses and counts everything without writing to the DB. + Returns stats: scanned, already_has_headers, updated, read_errors, parse_errors. + Anomalies that aren't errors (e.g. more than one From: header) are logged + (`headers.multiple_from`) rather than counted here — see parse_headers(). + """ + stats = { + "scanned": 0, + "already_has_headers": 0, + "updated": 0, + "read_errors": 0, + "parse_errors": 0, + } + + conn = await asyncpg.connect(dsn) + try: + rows = await fetch_batch(conn, limit, offset) + + pending: list[tuple[str, str]] = [] + + async def _flush() -> None: + if apply: + await _update_batch(conn, pending) + pending.clear() + + for row in rows: + stats["scanned"] += 1 + envelope_id = row["id"] + entities = _decode_jsonb(row["entities"]) or [] + + if _has_headers(entities): + stats["already_has_headers"] += 1 + continue + + eml_path = archive_root / row["raw_ref"] + try: + raw = eml_path.read_bytes() + except OSError: + _log.warning("skip.read_error", envelope_id=envelope_id, path=str(eml_path)) + stats["read_errors"] += 1 + continue + + try: + headers = parse_headers(raw) + except Exception: + _log.warning("skip.parse_error", envelope_id=envelope_id, exc_info=True) + stats["parse_errors"] += 1 + continue + + stats["updated"] += 1 + pending.append((envelope_id, json.dumps([headers]))) + if len(pending) >= UPDATE_BATCH_SIZE: + await _flush() + + await _flush() + finally: + await conn.close() + + _log.info("run_complete", apply=apply, limit=limit, offset=offset, **stats) + return stats + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Backfill {'type': 'headers', ...} onto existing source='gmail' " + "envelope rows in kb-postgres (module 5, phase 2 backfill — plan §5)." + ) + parser.add_argument("--dsn", default=os.environ.get("KB_DSN"), + help="asyncpg DSN for kb-postgres (or set KB_DSN env var)") + parser.add_argument("--archive-root", type=Path, default=DEFAULT_ARCHIVE_ROOT, + help=f"Mail .eml archive root (default: {DEFAULT_ARCHIVE_ROOT})") + parser.add_argument("--limit", type=int, default=DEFAULT_LIMIT, + help=f"Envelopes in this slice (default: {DEFAULT_LIMIT})") + parser.add_argument("--offset", type=int, default=0, + help="Slice offset, ordered by envelope id (default: 0)") + parser.add_argument("--apply", action="store_true", + help="Actually write the UPDATE. Default is dry-run (parse + count only).") + args = parser.parse_args() + + if not args.dsn: + _log.error("missing_dsn", hint="pass --dsn or set KB_DSN") + sys.exit(1) + if not args.archive_root.is_dir(): + _log.error("archive_root_not_found", path=str(args.archive_root)) + sys.exit(1) + + stats = asyncio.run( + run( + dsn=args.dsn, + archive_root=args.archive_root, + limit=args.limit, + offset=args.offset, + apply=args.apply, + ) + ) + + mode = "APPLY" if args.apply else "DRY-RUN" + _log.info("summary", mode=mode, **stats) + sys.exit(1 if stats["read_errors"] + stats["parse_errors"] > 0 else 0) + + +if __name__ == "__main__": + main() diff --git a/jobs/gmail-header-backfill/tests/test_backfill.py b/jobs/gmail-header-backfill/tests/test_backfill.py new file mode 100644 index 0000000..a881694 --- /dev/null +++ b/jobs/gmail-header-backfill/tests/test_backfill.py @@ -0,0 +1,358 @@ +"""Unit tests for the gmail header backfill job — no DB, no external services.""" +from __future__ import annotations + +import json +from datetime import datetime, timezone + +import pytest + +from gmail_header_backfill.backfill import ( + _decode_jsonb, + _has_headers, + fetch_batch, + parse_headers, + run, +) + + +def _eml(headers: dict, body: str = "body") -> bytes: + lines = [f"{k}: {v}" for k, v in headers.items()] + return ("\r\n".join(lines) + "\r\n\r\n" + body).encode("utf-8") + + +class TestParseHeadersBasics: + def test_shape_has_all_expected_keys(self): + raw = _eml({ + "From": "alice@example.com", + "To": "bob@example.com", + "Subject": "Hello", + "Date": "Tue, 10 Jun 2025 12:00:00 +0000", + }) + headers = parse_headers(raw) + assert headers["type"] == "headers" + assert {"from", "to", "cc", "delivered_to", "subject", "date_raw"} <= headers.keys() + + def test_from_parsed_to_name_and_address(self): + raw = _eml({"From": "Alice Example ", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"}) + headers = parse_headers(raw) + assert headers["from"] == {"name": "Alice Example", "address": "alice@example.com"} + + def test_from_without_display_name(self): + raw = _eml({"From": "alice@example.com", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"}) + headers = parse_headers(raw) + assert headers["from"] == {"name": None, "address": "alice@example.com"} + + def test_missing_from_is_none(self): + raw = _eml({"To": "bob@example.com", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"}) + headers = parse_headers(raw) + assert headers["from"] is None + + def test_subject_present(self): + raw = _eml({"From": "a@b.com", "Subject": "Twoja polisa", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"}) + assert parse_headers(raw)["subject"] == "Twoja polisa" + + def test_subject_missing_is_none(self): + raw = _eml({"From": "a@b.com", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"}) + assert parse_headers(raw)["subject"] is None + + def test_empty_to_and_cc_yield_empty_lists(self): + raw = _eml({"From": "a@b.com", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"}) + headers = parse_headers(raw) + assert headers["to"] == [] + assert headers["cc"] == [] + assert headers["delivered_to"] == [] + + +class TestParseHeadersMultiTo: + def test_multiple_to_addresses_across_one_header(self): + raw = _eml({ + "From": "a@b.com", + "To": "bob@example.com, Carol Jones ", + "Date": "Tue, 10 Jun 2025 12:00:00 +0000", + }) + headers = parse_headers(raw) + assert headers["to"] == [ + {"name": None, "address": "bob@example.com"}, + {"name": "Carol Jones", "address": "carol@example.com"}, + ] + + def test_quoted_display_name_with_comma_not_split(self): + raw = _eml({ + "From": '"Kowalski, Jan" ', + "Date": "Tue, 10 Jun 2025 12:00:00 +0000", + }) + headers = parse_headers(raw) + assert headers["from"] == {"name": "Kowalski, Jan", "address": "jan@example.com"} + + +class TestParseHeadersMultiDeliveredTo: + def test_multiple_delivered_to_headers_all_kept(self): + lines = ( + "From: a@b.com\r\n" + "Delivered-To: oskar+alias@gmail.com\r\n" + "Delivered-To: oskar@gmail.com\r\n" + "Date: Tue, 10 Jun 2025 12:00:00 +0000\r\n" + "\r\n" + "body" + ) + headers = parse_headers(lines.encode()) + assert headers["delivered_to"] == ["oskar+alias@gmail.com", "oskar@gmail.com"] + + def test_single_delivered_to(self): + raw = _eml({ + "From": "a@b.com", + "Delivered-To": "oskar@gmail.com", + "Date": "Tue, 10 Jun 2025 12:00:00 +0000", + }) + assert parse_headers(raw)["delivered_to"] == ["oskar@gmail.com"] + + +class TestParseHeadersRfc2047: + def test_decodes_encoded_word_subject(self): + raw = _eml({ + "From": "a@b.com", + "Subject": "=?UTF-8?Q?Twoja_polisa?=", + "Date": "Tue, 10 Jun 2025 12:00:00 +0000", + }) + assert parse_headers(raw)["subject"] == "Twoja polisa" + + def test_decodes_encoded_word_display_name(self): + raw = _eml({ + "From": "=?UTF-8?B?V2FydGE=?= ", + "Date": "Tue, 10 Jun 2025 12:00:00 +0000", + }) + headers = parse_headers(raw) + assert headers["from"] == {"name": "Warta", "address": "no-reply@warta.pl"} + + def test_decodes_encoded_word_with_polish_chars(self): + raw = _eml({ + "From": "a@b.com", + "To": "=?UTF-8?Q?Oskar_K=C4=85pa=C5=82a?= ", + "Date": "Tue, 10 Jun 2025 12:00:00 +0000", + }) + headers = parse_headers(raw) + assert headers["to"] == [{"name": "Oskar Kąpała", "address": "oskar@gmail.com"}] + + def test_malformed_encoded_word_does_not_raise(self): + raw = _eml({ + "From": "a@b.com", + "Subject": "=?UTF-8?B?not-valid-base64!!!?=", + "Date": "Tue, 10 Jun 2025 12:00:00 +0000", + }) + headers = parse_headers(raw) # must not raise + assert isinstance(headers["subject"], str) + + +class TestParseHeadersMultipleFrom: + def test_multiple_from_headers_uses_first_and_logs(self, caplog): + raw = ( + b"From: alice@example.com\r\n" + b"From: bob@example.com\r\n" + b"Date: Tue, 10 Jun 2025 12:00:00 +0000\r\n" + b"\r\nbody" + ) + headers = parse_headers(raw) + assert headers["from"] == {"name": None, "address": "alice@example.com"} + + +class TestParseHeadersDateRaw: + def test_date_raw_preserves_literal_original_text(self): + # policy.default's DateHeader reformats (e.g. corrects weekday, zero-pads day) — + # date_raw must preserve the byte-for-byte original text instead (plan §4.1). + raw = _eml({"From": "a@b.com", "Date": "Mon, 9 Jun 2026 12:34:56 +0200"}) + assert parse_headers(raw)["date_raw"] == "Mon, 9 Jun 2026 12:34:56 +0200" + + def test_date_raw_none_when_missing(self): + raw = _eml({"From": "a@b.com"}) + assert parse_headers(raw)["date_raw"] is None + + def test_date_raw_preserved_even_when_unparseable(self): + raw = _eml({"From": "a@b.com", "Date": "not-a-date-at-all"}) + assert parse_headers(raw)["date_raw"] == "not-a-date-at-all" + + +class TestDecodeJsonb: + def test_passes_through_object(self): + assert _decode_jsonb([{"type": "attachment"}]) == [{"type": "attachment"}] + + def test_decodes_string(self): + assert _decode_jsonb('[{"type": "attachment"}]') == [{"type": "attachment"}] + + def test_none_stays_none(self): + assert _decode_jsonb(None) is None + + +class TestHasHeaders: + def test_true_when_present(self): + assert _has_headers([{"type": "attachment"}, {"type": "headers"}]) is True + + def test_false_when_absent(self): + assert _has_headers([{"type": "attachment"}]) is False + + def test_false_on_empty_list(self): + assert _has_headers([]) is False + + +class _FakeConn: + def __init__(self, rows): + self._rows = rows + self.executemany_calls: list[tuple[str, list]] = [] + + async def fetch(self, query, *params): + return self._rows + + async def executemany(self, query, rows): + self.executemany_calls.append((query, list(rows))) + + async def close(self): + pass + + +def _row(envelope_id, raw_ref, entities): + return {"id": envelope_id, "raw_ref": raw_ref, "entities": json.dumps(entities)} + + +def _attachment_entity(): + return {"type": "attachment", "filename": "x.pdf", "content_type": "application/pdf", + "size": 100, "sha256": "abc"} + + +class TestRun: + def _setup_archive(self, tmp_path, raw_ref, raw_bytes): + archive_root = tmp_path / "archive" + eml_path = archive_root / raw_ref + eml_path.parent.mkdir(parents=True, exist_ok=True) + eml_path.write_bytes(raw_bytes) + return archive_root + + def _patch_connect(self, monkeypatch, conn): + async def _fake_connect(dsn): + return conn + monkeypatch.setattr("gmail_header_backfill.backfill.asyncpg.connect", _fake_connect) + + async def test_dry_run_does_not_call_executemany(self, tmp_path, monkeypatch): + raw = _eml({"From": "a@b.com", "Subject": "Hi", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"}) + archive_root = self._setup_archive(tmp_path, "gmail/2025/06/m1.eml", raw) + conn = _FakeConn([_row("env1", "gmail/2025/06/m1.eml", [_attachment_entity()])]) + self._patch_connect(monkeypatch, conn) + + stats = await run(dsn="postgresql://fake", archive_root=archive_root, apply=False) + + assert stats["scanned"] == 1 + assert stats["updated"] == 1 + assert conn.executemany_calls == [] + + async def test_apply_calls_executemany_with_headers_entity(self, tmp_path, monkeypatch): + raw = _eml({"From": "a@b.com", "Subject": "Hi", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"}) + archive_root = self._setup_archive(tmp_path, "gmail/2025/06/m1.eml", raw) + conn = _FakeConn([_row("env1", "gmail/2025/06/m1.eml", [_attachment_entity()])]) + self._patch_connect(monkeypatch, conn) + + stats = await run(dsn="postgresql://fake", archive_root=archive_root, apply=True) + + assert stats["updated"] == 1 + assert len(conn.executemany_calls) == 1 + query, rows = conn.executemany_calls[0] + assert "NOT EXISTS" in query + assert rows[0][0] == "env1" + patch = json.loads(rows[0][1]) + assert patch == [{ + "type": "headers", + "from": {"name": None, "address": "a@b.com"}, + "to": [], + "cc": [], + "delivered_to": [], + "subject": "Hi", + "date_raw": "Tue, 10 Jun 2025 12:00:00 +0000", + }] + + async def test_existing_entities_are_not_touched_client_side(self, tmp_path, monkeypatch): + # The job never rewrites the existing manifest — it only appends a new patch row + # for the UPDATE (entities || $2::jsonb happens in SQL, not here). + raw = _eml({"From": "a@b.com", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"}) + archive_root = self._setup_archive(tmp_path, "gmail/2025/06/m1.eml", raw) + original_entities = [_attachment_entity()] + conn = _FakeConn([_row("env1", "gmail/2025/06/m1.eml", original_entities)]) + self._patch_connect(monkeypatch, conn) + + await run(dsn="postgresql://fake", archive_root=archive_root, apply=True) + + query, rows = conn.executemany_calls[0] + patch = json.loads(rows[0][1]) + assert len(patch) == 1 + assert patch[0]["type"] == "headers" + + async def test_idempotent_skips_rows_already_backfilled(self, tmp_path, monkeypatch): + archive_root = tmp_path / "archive" + archive_root.mkdir() + entities = [_attachment_entity(), {"type": "headers", "from": None, "to": [], + "cc": [], "delivered_to": [], "subject": None, + "date_raw": None}] + conn = _FakeConn([_row("env1", "gmail/2025/06/m1.eml", entities)]) + self._patch_connect(monkeypatch, conn) + + stats = await run(dsn="postgresql://fake", archive_root=archive_root, apply=True) + + assert stats["already_has_headers"] == 1 + assert stats["updated"] == 0 + assert conn.executemany_calls == [] + + async def test_missing_eml_file_counted_as_read_error(self, tmp_path, monkeypatch): + archive_root = tmp_path / "archive" + archive_root.mkdir() + conn = _FakeConn([_row("env1", "gmail/2025/06/missing.eml", [_attachment_entity()])]) + self._patch_connect(monkeypatch, conn) + + stats = await run(dsn="postgresql://fake", archive_root=archive_root, apply=True) + + assert stats["read_errors"] == 1 + assert stats["updated"] == 0 + + async def test_limit_and_offset_are_passed_to_query(self, tmp_path, monkeypatch): + captured = {} + + class _FakeConnCapturing(_FakeConn): + async def fetch(self, query, *params): + captured["params"] = params + return self._rows + + conn = _FakeConnCapturing([]) + self._patch_connect(monkeypatch, conn) + archive_root = tmp_path / "archive" + archive_root.mkdir() + + await run(dsn="postgresql://fake", archive_root=archive_root, limit=50, offset=200, apply=False) + + assert captured["params"] == (50, 200) + + async def test_batch_flush_multiple_rows_in_one_executemany(self, tmp_path, monkeypatch): + raw = _eml({"From": "a@b.com", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"}) + archive_root = self._setup_archive(tmp_path, "gmail/2025/06/m1.eml", raw) + rows = [_row(f"env{i}", "gmail/2025/06/m1.eml", [_attachment_entity()]) for i in range(3)] + conn = _FakeConn(rows) + self._patch_connect(monkeypatch, conn) + + stats = await run(dsn="postgresql://fake", archive_root=archive_root, apply=True) + + assert stats["updated"] == 3 + assert len(conn.executemany_calls) == 1 # under UPDATE_BATCH_SIZE, one flush at end + _, flushed_rows = conn.executemany_calls[0] + assert len(flushed_rows) == 3 + + +class TestFetchBatchQueryShape: + async def test_selects_only_gmail_source_ordered_by_id(self, monkeypatch): + captured = {} + + class _FakeConnCapturing(_FakeConn): + async def fetch(self, query, *params): + captured["query"] = query + captured["params"] = params + return [] + + conn = _FakeConnCapturing([]) + await fetch_batch(conn, limit=10, offset=5) + + assert "source = 'gmail'" in captured["query"] + assert "ORDER BY id" in captured["query"] + assert captured["params"] == (10, 5)