oskar/homelab-codex-ws
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

feat(gmail-header-backfill): one-shot job — backfill headers into envelope.entities

Browse source 
Module 5 phase 2 (docs/kb/modules/05-faza2-plan.md, §5). 225,030 existing
source='gmail' envelope rows carry only an attachment manifest — no
from/to/cc/delivered_to/subject anywhere in the DB, which blocks answering
"is this on me or my wife" / distinguishing aliases (plan §1.8). Separate
job from gmail-bulk-import per plan §2 decision 7: UPDATE semantics against
production data is a different risk profile than the historical INSERT job.

Parses headers only (no MIME-walk of attachments) from the archived .eml
files, appends {"type": "headers", ...} (plan §4.1) via the idempotent
UPDATE ... WHERE NOT EXISTS from §5.2, batched via executemany. --limit/
--offset (ORDER BY id) give stable, deterministic partitioning so the full
backfill can run and be verified in slices instead of one unattended pass.

Plain CLI (pip install -e), no Dockerfile — same convention as
gmail-bulk-import/documents-ingest, which run directly on PIHA for local
filesystem access to the .eml archive.

Verified against kb-postgres@PIHA (100-row dry-run + apply, re-run proved
idempotent no-op, 1000-row timed slice): 1096/225030 rows backfilled,
996/1000 succeeded on the timed slice (4 parse_errors — a 2014 spam message
with an RFC 2047 encoded-word decoding to an embedded newline in the From
display name, correctly caught and skipped rather than crashing the batch).
Extrapolated full-run time ~16 minutes. Full 225,030-row run is out of
scope for this change.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
oskar 2026-07-14 15:34:07 +02:00
parent cc5c7921e2
commit 80c33c487c
5 changed files with 807 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

155
jobs/gmail-header-backfill/README.md Normal file
View file

@ -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:<pw>@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.

22
jobs/gmail-header-backfill/pyproject.toml Normal file
View file

@ -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"]

0
jobs/gmail-header-backfill/src/gmail_header_backfill/__init__.py Normal file
View file

272
jobs/gmail-header-backfill/src/gmail_header_backfill/backfill.py Normal file
View file

@ -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:<pw>@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()

358
jobs/gmail-header-backfill/tests/test_backfill.py Normal file
View file

@ -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 <alice@example.com>", "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 <carol@example.com>",
"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" <jan@example.com>',
"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=?= <no-reply@warta.pl>",
"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?= <oskar@gmail.com>",
"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)