From fb720cddac168c3482c602fb7d5ac6a8f75e2239 Mon Sep 17 00:00:00 2001 From: oskar Date: Thu, 6 Aug 2026 14:26:51 +0200 Subject: [PATCH] feat(kb-mail): adapter IMAP + model stanu synca + migracja 005 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Krok 7 fazy mailowej, warstwa wspoldzielona. Realizuje decyzje (a), (b), (f) reconu kb/audits/mail-sync-2026-08-06.md (zatwierdzone przez operatora 2026-08-06): jeden adapter IMAP na oba konta, stan synca jako tabela w bazie. Nowe moduly w packages/kb-mail: - imap.py — ImapAccount/ImapClient nad stdlib imaplib (zero nowych zaleznosci). Foldery otwierane READ-ONLY (EXAMINE) i pobierane przez BODY.PEEK[], zeby job nie ustawial \Seen na skrzynce operatora. Wybor folderu po atrybucie SPECIAL-USE, nigdy po nazwie — Gmail lokalizuje "[Gmail]/All Mail". search_from_uid filtruje zakres po stronie klienta, bo n:* zwraca ostatnia wiadomosc takze gdy przedzial pusty. - sync_state.py — tabela mail_sync_state + czyste funkcje: plan_folder_sync (pierwszy tick / przyrost / uniewaznienie UIDVALIDITY) i contiguous_last_uid (kursor przesuwa sie tylko po nieprzerwanym ciagu sukcesow — bledna wiadomosc jest ponawiana, nie przeskakiwana). - headers.py / message.py — parse_headers(+fallback) z gmail-header-backfill oraz message_id/parse_date/parse_attachments/eml_ref z gmail-bulk-import, przeniesione zamiast skopiowane. Klucz dedup musi pochodzic z jednej implementacji: kazdy insert przyrostowki trafia na 225 030 istniejacych id. Stare joby re-eksportuja te nazwy — ich CLI i testy bez zmian. kb_mail.db.insert_envelope zwraca teraz command tag (+ rows_affected, envelope_source) — bez tego nie da sie odroznic zwyklego duplikatu od kolizji Message-ID miedzy kontami (recon §2.4). Migracja 005_mail_sync_state.sql: addytywna, klucz (account, folder). Testy: 285 passed (111 kb-mail w tym 37 adaptera IMAP na fake serwerze i 26 planera kursora; 174 istniejace suity jobow bez zmian po ekstrakcji). Co-Authored-By: Claude Opus 5 (1M context) --- .../src/gmail_bulk_import/importer.py | 98 +---- .../src/gmail_header_backfill/backfill.py | 136 +----- packages/kb-mail/src/kb_mail/db.py | 31 +- packages/kb-mail/src/kb_mail/headers.py | 168 +++++++ packages/kb-mail/src/kb_mail/imap.py | 413 ++++++++++++++++++ packages/kb-mail/src/kb_mail/message.py | 116 +++++ packages/kb-mail/src/kb_mail/sync_state.py | 167 +++++++ packages/kb-mail/tests/test_imap.py | 330 ++++++++++++++ packages/kb-mail/tests/test_message.py | 94 ++++ packages/kb-mail/tests/test_migration.py | 37 ++ packages/kb-mail/tests/test_sync_state.py | 182 ++++++++ .../kb-postgres/init/005_mail_sync_state.sql | 27 ++ 12 files changed, 1580 insertions(+), 219 deletions(-) create mode 100644 packages/kb-mail/src/kb_mail/headers.py create mode 100644 packages/kb-mail/src/kb_mail/imap.py create mode 100644 packages/kb-mail/src/kb_mail/message.py create mode 100644 packages/kb-mail/src/kb_mail/sync_state.py create mode 100644 packages/kb-mail/tests/test_imap.py create mode 100644 packages/kb-mail/tests/test_message.py create mode 100644 packages/kb-mail/tests/test_sync_state.py create mode 100644 services/kb-postgres/init/005_mail_sync_state.sql diff --git a/jobs/gmail-bulk-import/src/gmail_bulk_import/importer.py b/jobs/gmail-bulk-import/src/gmail_bulk_import/importer.py index e936b21..9a0c22f 100644 --- a/jobs/gmail-bulk-import/src/gmail_bulk_import/importer.py +++ b/jobs/gmail-bulk-import/src/gmail_bulk_import/importer.py @@ -29,12 +29,10 @@ from __future__ import annotations import argparse import asyncio import email.policy -import hashlib import json import mailbox import sys -from datetime import datetime, timezone -from email.utils import parsedate_to_datetime +from datetime import datetime from pathlib import Path from typing import Optional @@ -43,94 +41,26 @@ import structlog from kb_mail.archive import save_eml from kb_mail.envelope import Envelope -from kb_mail.text import sanitize_surrogates as _sanitize +# id / ts / attachment-manifest derivation moved to packages/kb-mail when jobs/mail-imap-sync +# needed the identical four functions (recon kb/audits/mail-sync-2026-08-06.md §2.1). The +# dedup key MUST come from ONE implementation: every insert the poller makes is matched +# against the 225 030 ids this job wrote. Aliased to the old private names so the rest of this +# module and its tests are untouched. +from kb_mail.message import EPOCH as _EPOCH +from kb_mail.message import eml_ref as _archive_ref +from kb_mail.message import message_id as _message_id # noqa: F401 +from kb_mail.message import parse_attachments as _parse_attachments # noqa: F401 +from kb_mail.message import parse_date as _parse_date # noqa: F401 _log = structlog.get_logger(__name__) BATCH_SIZE = 500 -_EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc) - -# Mirrors archive._UNSAFE — used to reconstruct raw_ref for already-archived messages. -_ARCHIVE_UNSAFE = str.maketrans({"/": "_", "\\": "_", ":": "_", "<": "", ">": ""}) def _eml_ref(envelope_id: str, ts: datetime) -> str: - """Compute the archive-relative .eml path without writing (mirrors save_eml logic).""" - safe_id = envelope_id.translate(_ARCHIVE_UNSAFE) - return f"gmail/{ts.year:04d}/{ts.month:02d}/{safe_id}.eml" - - -def _message_id(msg: mailbox.mboxMessage) -> str: - """Return a stable envelope id from Message-ID header or SHA-256 content hash. - - compat32 .get() returns an email.header.Header (not str) when the raw value - holds 8-bit bytes — str() + _sanitize keeps the id jsonb-safe and gives - .strip() a real string to work on. An unguarded .strip() on a Header raised - AttributeError and, because _message_id runs before the per-message try, - killed the whole run (the archive has proven 8-bit header bytes). - """ - mid = _sanitize(str(msg.get("Message-ID", ""))).strip().strip("<>") - if mid: - return mid - return "sha256-" + hashlib.sha256(msg.as_bytes()).hexdigest()[:32] - - -def _parse_date(msg: mailbox.mboxMessage) -> datetime: - """Parse Date header to UTC-aware datetime. Returns epoch as last resort. - - str() first: a compat32 8-bit Date surfaces as an email.header.Header, and - parsedate_to_datetime() raises on a Header (→ needless epoch_fallback) even - when str(header) is perfectly parseable. - """ - date_str = str(msg.get("Date", "")) - if date_str: - try: - dt = parsedate_to_datetime(date_str) - if dt.tzinfo is None: - dt = dt.replace(tzinfo=timezone.utc) - return dt.astimezone(timezone.utc) - except Exception: - pass - return _EPOCH - - -def _parse_attachments(msg: mailbox.mboxMessage) -> list[dict]: - """Return attachment descriptors from MIME structure. - - A part is an attachment if it has a filename, Content-Disposition: attachment, - or is not inline text/plain or text/html. - Errors on individual parts are logged and skipped — never propagated. - """ - attachments = [] - for part in msg.walk(): - if part.get_content_maintype() == "multipart": - continue - content_type = part.get_content_type() - disposition = part.get_content_disposition() or "" - filename = part.get_filename() - - if ( - content_type in ("text/plain", "text/html") - and disposition != "attachment" - and not filename - ): - continue - - try: - payload = part.get_payload(decode=True) - if payload is None: - continue - attachments.append({ - "type": "attachment", - "filename": filename, - "content_type": content_type, - "size": len(payload), - "sha256": hashlib.sha256(payload).hexdigest(), - }) - except Exception: - _log.warning("attachment_parse_failed", - filename=filename, content_type=content_type) - return attachments + """Archive-relative .eml path for an already-archived gmail message (source is fixed here; + `kb_mail.message.eml_ref` takes it as a parameter for the multi-account poller).""" + return _archive_ref(envelope_id, "gmail", ts) async def _insert_batch(conn: asyncpg.Connection, envs: list[Envelope]) -> None: diff --git a/jobs/gmail-header-backfill/src/gmail_header_backfill/backfill.py b/jobs/gmail-header-backfill/src/gmail_header_backfill/backfill.py index 7cb46f9..d1a877a 100644 --- a/jobs/gmail-header-backfill/src/gmail_header_backfill/backfill.py +++ b/jobs/gmail-header-backfill/src/gmail_header_backfill/backfill.py @@ -34,19 +34,18 @@ 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 -from kb_mail.text import sanitize_surrogates as _sanitize +# Both parsers moved to packages/kb-mail when jobs/mail-imap-sync needed them too (recon +# kb/audits/mail-sync-2026-08-06.md §6 poz. 3). Re-exported here so this job's CLI, its tests, +# and every existing `from gmail_header_backfill.backfill import parse_headers` keep working. +from kb_mail.headers import parse_headers, parse_headers_fallback # noqa: F401 _log = structlog.get_logger(__name__) @@ -64,133 +63,6 @@ WHERE id = $1 """ -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 - - # compat32 .get() returns an email.header.Header (not str) when the raw - # value contains 8-bit bytes — str() + _sanitize keeps it JSON/jsonb-safe. - # An unguarded Header here crashed the original full run mid-slice - # (TypeError at json.dumps), silently losing the rest of the slice. - date_header = msg_compat.get("Date") - date_raw = _sanitize(str(date_header)) if date_header is not None else None - - return { - "type": "headers", - "from": from_obj, - "to": to_list, - "cc": cc_list, - "delivered_to": delivered_to, - "subject": subject, - "date_raw": date_raw, - } - - -def parse_headers_fallback(raw: bytes) -> dict: - """Degraded parse for messages the typed (policy.default) path rejects. - - Real-world triggers found in the archive (diagnosis 2026-07-14, 9 of - 225 030): RFC 2047 encoded-words that decode to text with CR/LF in a - display name (ValueError in headerregistry), group syntax in To: - ("unlisted-recipients:;"), and a _header_value_parser bug on malformed - display names (fixed in newer CPython, present on PIHA's 3.11). - - Everything comes from a compat32 parse: address fields are split into - {name, address} by email.utils.getaddresses over the RAW header text — no - RFC 2047 decoding, so names may keep literal =?...?= encoded-words. - Subject and delivered_to stay raw strings too. The returned shape is - exactly parse_headers()'s §4.1 shape — only parse QUALITY degrades, never - the schema. Callers count these separately (stats["parsed_fallback"]). - """ - msg = email.message_from_bytes(raw, policy=email.policy.compat32) - - from_headers = [str(h) for h in (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([from_headers[0]]) - if from_addrs and from_addrs[0][1]: - name, address = from_addrs[0] - from_obj = {"name": _sanitize(name) or None, "address": _sanitize(address)} - - to_list = [ - {"name": _sanitize(name) or None, "address": _sanitize(address)} - for name, address in email.utils.getaddresses( - [str(h) for h in (msg.get_all("To") or [])] - ) - if address - ] - - cc_list = [ - {"name": _sanitize(name) or None, "address": _sanitize(address)} - for name, address in email.utils.getaddresses( - [str(h) for h in (msg.get_all("Cc") or [])] - ) - if address - ] - - delivered_to = [_sanitize(str(h)) for h in (msg.get_all("Delivered-To") or [])] - - subject_header = msg.get("Subject") - subject = _sanitize(str(subject_header)) if subject_header is not None else None - - date_raw = _sanitize(msg.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: diff --git a/packages/kb-mail/src/kb_mail/db.py b/packages/kb-mail/src/kb_mail/db.py index f3acf42..18ab49e 100644 --- a/packages/kb-mail/src/kb_mail/db.py +++ b/packages/kb-mail/src/kb_mail/db.py @@ -12,10 +12,15 @@ from .envelope import Envelope _log = structlog.get_logger(__name__) -async def insert_envelope(conn: asyncpg.Connection, env: Envelope) -> None: - """Insert envelope into kb-postgres. Silently ignores duplicate ids (ON CONFLICT DO NOTHING).""" +async def insert_envelope(conn: asyncpg.Connection, env: Envelope) -> str: + """Insert envelope into kb-postgres. Silently ignores duplicate ids (ON CONFLICT DO NOTHING). + + Returns the postgres command tag (`INSERT 0 1` on a real insert, `INSERT 0 0` when the id + was already present) so a caller that needs to tell "new" from "already had it" can, via + `rows_affected()`. Callers that don't care simply ignore it — this used to return None. + """ geo_param = json.dumps(env.geo) if env.geo is not None else None - await conn.execute( + tag = await conn.execute( """ INSERT INTO envelope (id, source, ts, geo, raw_ref, entities) VALUES ($1, $2, $3, $4::jsonb, $5, $6::jsonb) @@ -29,6 +34,26 @@ async def insert_envelope(conn: asyncpg.Connection, env: Envelope) -> None: json.dumps(env.entities), ) _log.info("envelope.inserted", id=env.id, source=env.source) + return tag + + +def rows_affected(command_tag: str) -> int: + """Row count out of a postgres command tag (`INSERT 0 1` -> 1).""" + return int(command_tag.rsplit(" ", 1)[-1]) + + +async def envelope_source(conn: asyncpg.Connection, envelope_id: str) -> Optional[str]: + """The `source` of an existing envelope, or None if there is no such row. + + Used to classify a suppressed insert. `envelope.id` is a bare Message-ID and therefore + GLOBALLY unique across accounts, so a mail present in both mailboxes (list traffic, a CC + to both addresses, a forward) is stored once under whichever source inserted it first — + correct for the index, but it means "new fastmail mails" is systematically undercounted by + the shared part. The recon's answer (§2.4) is to keep the behaviour and make the + phenomenon visible: count it as `envelopes_conflict_other_source` instead of letting it + hide inside an ordinary duplicate count. + """ + return await conn.fetchval("SELECT source FROM envelope WHERE id = $1", envelope_id) async def get_envelope(conn: asyncpg.Connection, envelope_id: str) -> Optional[Envelope]: diff --git a/packages/kb-mail/src/kb_mail/headers.py b/packages/kb-mail/src/kb_mail/headers.py new file mode 100644 index 0000000..95ce01c --- /dev/null +++ b/packages/kb-mail/src/kb_mail/headers.py @@ -0,0 +1,168 @@ +"""Header parsing shared by every job that turns `.eml` bytes into an +`entities[type=headers]` object (kb/phases/kb-m5-faza2.md §4.1). + +Extracted from `gmail_header_backfill.backfill` when `jobs/mail-imap-sync` needed the same +two functions (recon `kb/audits/mail-sync-2026-08-06.md` §6 poz. 3): a poller that inserts +envelopes without headers would give every new mail the chunk prefix +`Temat: (brak tematu) | Od: ?` — silently, with no error, just worse retrieval (§2.5 i). +Same reason `kb_mail.chunking` was extracted in Krok 0: one tested implementation, not a copy. + +`gmail_header_backfill.backfill` re-exports both names, so its own tests and CLI are unchanged. +""" +from __future__ import annotations + +import email +import email.policy +import email.utils +from typing import Optional + +import structlog + +from .text import sanitize_surrogates as _sanitize + +_log = structlog.get_logger(__name__) + + +def parse_headers(raw: bytes) -> dict: + """Parse `.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 + + # compat32 .get() returns an email.header.Header (not str) when the raw + # value contains 8-bit bytes — str() + _sanitize keeps it JSON/jsonb-safe. + # An unguarded Header here crashed the original full backfill run mid-slice + # (TypeError at json.dumps), silently losing the rest of the slice. + date_header = msg_compat.get("Date") + date_raw = _sanitize(str(date_header)) if date_header is not None else None + + return { + "type": "headers", + "from": from_obj, + "to": to_list, + "cc": cc_list, + "delivered_to": delivered_to, + "subject": subject, + "date_raw": date_raw, + } + + +def parse_headers_fallback(raw: bytes) -> dict: + """Degraded parse for messages the typed (policy.default) path rejects. + + Real-world triggers found in the archive (diagnosis 2026-07-14, 9 of + 225 030): RFC 2047 encoded-words that decode to text with CR/LF in a + display name (ValueError in headerregistry), group syntax in To: + ("unlisted-recipients:;"), and a _header_value_parser bug on malformed + display names (fixed in newer CPython, present on PIHA's 3.11). + + Everything comes from a compat32 parse: address fields are split into + {name, address} by email.utils.getaddresses over the RAW header text — no + RFC 2047 decoding, so names may keep literal =?...?= encoded-words. + Subject and delivered_to stay raw strings too. The returned shape is + exactly parse_headers()'s §4.1 shape — only parse QUALITY degrades, never + the schema. Callers count these separately (stats["parsed_fallback"] / + stats["headers_fallback"]). + """ + msg = email.message_from_bytes(raw, policy=email.policy.compat32) + + from_headers = [str(h) for h in (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([from_headers[0]]) + if from_addrs and from_addrs[0][1]: + name, address = from_addrs[0] + from_obj = {"name": _sanitize(name) or None, "address": _sanitize(address)} + + to_list = [ + {"name": _sanitize(name) or None, "address": _sanitize(address)} + for name, address in email.utils.getaddresses( + [str(h) for h in (msg.get_all("To") or [])] + ) + if address + ] + + cc_list = [ + {"name": _sanitize(name) or None, "address": _sanitize(address)} + for name, address in email.utils.getaddresses( + [str(h) for h in (msg.get_all("Cc") or [])] + ) + if address + ] + + delivered_to = [_sanitize(str(h)) for h in (msg.get_all("Delivered-To") or [])] + + subject_header = msg.get("Subject") + subject = _sanitize(str(subject_header)) if subject_header is not None else None + + date_raw = _sanitize(msg.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 parse_headers_resilient(raw: bytes) -> tuple[dict, bool]: + """`parse_headers` with the compat32 fallback already wired in. + + Returns `(headers_entity, used_fallback)`. The poller inserts envelopes one at a time and + must never lose a mail to a header quirk, so it always takes this path; the backfill job + keeps its own inline try/except because it counts the two outcomes into different slice + stats. A message neither parser can handle raises — the caller decides whether that is a + skipped message or a failed run. + """ + try: + return parse_headers(raw), False + except Exception as typed_exc: + entity = parse_headers_fallback(raw) + _log.info("headers.parsed_fallback", typed_error=repr(typed_exc)) + return entity, True diff --git a/packages/kb-mail/src/kb_mail/imap.py b/packages/kb-mail/src/kb_mail/imap.py new file mode 100644 index 0000000..e7b9ca6 --- /dev/null +++ b/packages/kb-mail/src/kb_mail/imap.py @@ -0,0 +1,413 @@ +"""IMAP transport for the mail pillar — one adapter, two accounts (gmail + fastmail). + +Realizes decisions (a) and (b) of `kb/audits/mail-sync-2026-08-06.md`, approved by the +operator 2026-08-06: both live mailboxes are read over plain IMAP with an app password, not +Gmail API and not JMAP. The reason is written down in `kb/subsystems/kb-overview.md` §61 — +*"protokół, nie provider"* — and it has a concrete payoff here: one class, one test suite, one +8-bit-header hardening path, instead of two transports producing identical rows. + +Deliberately synchronous. `imaplib` is stdlib (PIHA runs 3.11.2), so the poller adds **zero +dependencies**; at ~37 messages a day an async client would buy nothing but a third-party +package to keep patched. The job runs these calls under `asyncio.to_thread` so its asyncpg +side stays async. + +Two properties this module is responsible for, both easy to get silently wrong: + +* **Never mutate the mailbox.** Folders are opened with `EXAMINE` (`readonly=True`) and + bodies fetched with `BODY.PEEK[]`. A plain `SELECT` + `FETCH RFC822` would set `\\Seen` on + every message the poller touches — turning a read-only KB job into something that visibly + edits the operator's inbox. +* **Never hardcode a folder name.** Gmail localizes `[Gmail]/All Mail` (Polish UI: + `[Gmail]/Wszystkie`), so the folder is resolved by its SPECIAL-USE attribute (`\\All`) from + the `LIST` response. Literal names stay available for Fastmail's classic `INBOX`/`Archive`/ + `Sent` layout (recon Decyzja (e)). + +Folder names are carried as the exact string the server returned. Non-ASCII names are IMAP +modified-UTF-7 on the wire and are neither decoded nor re-encoded here — they only ever need +to round-trip back to the server and into `mail_sync_state.folder`, which they do byte-for-byte. +""" +from __future__ import annotations + +import imaplib +import re +import ssl +from dataclasses import dataclass, field +from datetime import date +from typing import Callable, Optional + +import structlog + +_log = structlog.get_logger(__name__) + +DEFAULT_PORT = 993 +DEFAULT_TIMEOUT_S = 60.0 + +INITIAL_MODES = ("new-only", "since", "full") + +# IMAP SEARCH dates are `dd-Mon-yyyy` with English month abbreviations regardless of locale. +_IMAP_MONTHS = ("Jan", "Feb", "Mar", "Apr", "May", "Jun", + "Jul", "Aug", "Sep", "Oct", "Nov", "Dec") + +_LIST_RE = re.compile( + rb'^\((?P[^)]*)\)\s+(?P"(?:[^"\\]|\\.)*"|NIL)\s+(?P.+)$' +) +_UID_IN_FETCH_RE = re.compile(rb"\bUID\s+(\d+)") + + +class ImapError(RuntimeError): + """An IMAP command returned a non-OK status, or a response could not be parsed. + + Raised instead of returning a sentinel so a folder whose protocol exchange went wrong + fails loudly and leaves `mail_sync_state` untouched — the next tick then retries from the + same cursor rather than skipping a range nobody ever fetched. + """ + + +@dataclass(frozen=True) +class ImapAccount: + """One mailbox's connection details and sync scope. + + `name` doubles as `envelope.source` and `mail_sync_state.account` — 'gmail' | 'fastmail'. + + `password` is `repr=False` on purpose: this dataclass is passed around the job and dumped + into structlog context in places, and the repo has two documented secret leaks to session + transcripts already (recon Decyzja (c) rule 2). + """ + + name: str + host: str + user: str + password: str = field(repr=False) + port: int = DEFAULT_PORT + folders: tuple[str, ...] = () # literal mailbox names (fastmail) + special_use: tuple[str, ...] = () # SPECIAL-USE attributes to resolve (gmail: '\\All') + initial_mode: str = "new-only" # what the FIRST tick on a folder fetches + initial_since: Optional[date] = None # required when initial_mode == 'since' + + def __post_init__(self) -> None: + if not self.folders and not self.special_use: + raise ValueError( + f"account {self.name}: no scope — set folders and/or special_use " + f"(recon Decyzja (e): gmail = \\All, fastmail = INBOX,Archive,Sent)" + ) + if self.initial_mode not in INITIAL_MODES: + raise ValueError( + f"account {self.name}: initial_mode={self.initial_mode!r} " + f"must be one of {INITIAL_MODES}" + ) + if self.initial_mode == "since" and self.initial_since is None: + raise ValueError( + f"account {self.name}: initial_mode='since' requires initial_since (YYYY-MM-DD)" + ) + + +@dataclass(frozen=True) +class FolderStatus: + """What the server says about a folder right now. + + `uidnext` is the UID the *next* delivered message will get, so `uidnext - 1` is the + high-water mark a 'start from now' first tick records without fetching anything. + `messages` is only populated by `status()` (the sizing measurement); `select()` leaves it + None because EXAMINE's EXISTS count is not what the runbook asks for. + """ + + name: str + uidvalidity: int + uidnext: int + messages: Optional[int] = None + + +def imap_date(value: date) -> str: + """`dd-Mon-yyyy` as IMAP SEARCH wants it (`15-Jun-2026`), month names always English.""" + return f"{value.day:02d}-{_IMAP_MONTHS[value.month - 1]}-{value.year:04d}" + + +def quote_mailbox(name: str) -> str: + """Quote a mailbox name for the wire. imaplib passes mailbox arguments through verbatim, + so `[Gmail]/All Mail` without quotes is parsed by the server as two arguments.""" + escaped = name.replace("\\", "\\\\").replace('"', '\\"') + return f'"{escaped}"' + + +def _unquote(raw: bytes) -> str: + text = raw.decode("utf-8", errors="replace").strip() + if len(text) >= 2 and text.startswith('"') and text.endswith('"'): + text = text[1:-1].replace('\\"', '"').replace("\\\\", "\\") + return text + + +def parse_list_line(line) -> Optional[tuple[frozenset[str], str]]: + """One `LIST` response item -> (lowercased attribute set, mailbox name). + + imaplib hands back either a bytes line or, when the server sent the name as a literal, + a `(prefix, literal)` tuple — both shapes occur in the wild, so both are handled. + Returns None for anything that does not parse as a LIST line (servers occasionally + interleave other untagged data). + """ + if isinstance(line, tuple): + prefix, literal = line[0], line[1] + match = _LIST_RE.match(prefix.rstrip(b"{0123456789}").strip()) + if match is None: + head = _LIST_RE.match(prefix.strip()) + if head is None: + return None + attrs = head.group("attrs") + else: + attrs = match.group("attrs") + name = literal.decode("utf-8", errors="replace") + else: + match = _LIST_RE.match(line.strip()) + if match is None: + return None + attrs = match.group("attrs") + name = _unquote(match.group("name")) + + attr_set = frozenset( + a.decode("ascii", errors="replace").lower() + for a in attrs.split() + ) + return attr_set, name + + +class ImapClient: + """Thin, read-only wrapper over `imaplib.IMAP4_SSL` for one account. + + Use as a context manager — `__exit__` always attempts LOGOUT, so a folder that raises + mid-sync still closes its connection instead of leaning on the server's idle timeout. + + `connection_factory` exists so the whole protocol layer is unit-testable against a fake + IMAP server: the tests exercise this class's real parsing, not a mock of it. + """ + + def __init__( + self, + account: ImapAccount, + *, + timeout_s: float = DEFAULT_TIMEOUT_S, + connection_factory: Optional[Callable[[ImapAccount, float], object]] = None, + ) -> None: + self.account = account + self.timeout_s = timeout_s + self._factory = connection_factory or _default_connection + self._imap = None + self._selected: Optional[str] = None + + # -- lifecycle --------------------------------------------------------------- + + def __enter__(self) -> "ImapClient": + self.connect() + return self + + def __exit__(self, *exc_info) -> None: + self.logout() + + def connect(self) -> None: + self._imap = self._factory(self.account, self.timeout_s) + # LOGIN failures raise imaplib.IMAP4.error; the message carries the server's + # rejection text, never the credential. + self._imap.login(self.account.user, self.account.password) + _log.info("imap.connected", account=self.account.name, + host=self.account.host, port=self.account.port) + + def logout(self) -> None: + if self._imap is None: + return + try: + self._imap.logout() + except Exception: + _log.warning("imap.logout_failed", account=self.account.name) + finally: + self._imap = None + self._selected = None + + def _conn(self): + if self._imap is None: + raise ImapError(f"account {self.account.name}: not connected") + return self._imap + + def _check(self, command: str, result) -> list: + typ, data = result + if typ != "OK": + raise ImapError( + f"account {self.account.name}: {command} returned {typ}: {data!r}" + ) + return data + + # -- folder discovery -------------------------------------------------------- + + def list_folders(self) -> list[tuple[frozenset[str], str]]: + data = self._check("LIST", self._conn().list()) + parsed = [parse_list_line(line) for line in data if line] + return [p for p in parsed if p is not None] + + def resolve_folders(self) -> list[str]: + """The account's configured scope as concrete mailbox names, in a stable order. + + SPECIAL-USE attributes are resolved against `LIST`; a configured attribute the server + does not advertise is an error, not a silent skip — a gmail account whose `\\All` + vanished would otherwise sync nothing while reporting a clean run every hour. + Literal names are trusted as configured (a typo surfaces on SELECT, loudly). + Duplicates are collapsed, so listing a folder that also matches an attribute costs + nothing. + """ + resolved: list[str] = [] + + if self.account.special_use: + listing = self.list_folders() + for attr in self.account.special_use: + wanted = attr.lower() + matches = [name for attrs, name in listing if wanted in attrs] + if not matches: + advertised = sorted({a for attrs, _ in listing for a in attrs if a.startswith("\\")}) + raise ImapError( + f"account {self.account.name}: no folder advertises SPECIAL-USE " + f"{attr!r}; server advertises {advertised}" + ) + if len(matches) > 1: + _log.warning("imap.special_use_ambiguous", account=self.account.name, + attribute=attr, matches=matches) + resolved.append(matches[0]) + + resolved.extend(self.account.folders) + + seen: set[str] = set() + unique = [] + for name in resolved: + if name not in seen: + seen.add(name) + unique.append(name) + return unique + + # -- folder state ------------------------------------------------------------ + + def status(self, folder: str) -> FolderStatus: + """`STATUS (MESSAGES UIDNEXT UIDVALIDITY)` — read-only and cheap. + + This is the measurement the runbook takes on Fastmail before deciding whether to pull + its history (recon Decyzja (e): decide on the number, do not guess it). + """ + data = self._check( + "STATUS", + self._conn().status(quote_mailbox(folder), "(MESSAGES UIDNEXT UIDVALIDITY)"), + ) + blob = b" ".join(part for part in data if isinstance(part, bytes)) + return FolderStatus( + name=folder, + uidvalidity=_status_field(blob, b"UIDVALIDITY", folder), + uidnext=_status_field(blob, b"UIDNEXT", folder), + messages=_status_field(blob, b"MESSAGES", folder), + ) + + def examine(self, folder: str) -> FolderStatus: + """Open a folder READ-ONLY (EXAMINE) and return its UIDVALIDITY/UIDNEXT. + + Both values normally arrive as untagged responses to EXAMINE; a server that omits + UIDNEXT there is covered by falling back to STATUS rather than by guessing, because + `uidnext - 1` is exactly the cursor a first tick persists. + """ + conn = self._conn() + self._check("EXAMINE", conn.select(quote_mailbox(folder), readonly=True)) + self._selected = folder + + uidvalidity = _untagged_int(conn, "UIDVALIDITY") + uidnext = _untagged_int(conn, "UIDNEXT") + if uidvalidity is None or uidnext is None: + fallback = self.status(folder) + uidvalidity = uidvalidity if uidvalidity is not None else fallback.uidvalidity + uidnext = uidnext if uidnext is not None else fallback.uidnext + # STATUS on some servers implicitly deselects; re-EXAMINE so the UID commands + # below still have a mailbox open. + self._check("EXAMINE", conn.select(quote_mailbox(folder), readonly=True)) + + return FolderStatus(name=folder, uidvalidity=int(uidvalidity), uidnext=int(uidnext)) + + # -- searching --------------------------------------------------------------- + + def _uid_search(self, *criteria: str) -> list[int]: + data = self._check("UID SEARCH", self._conn().uid("SEARCH", None, *criteria)) + uids: list[int] = [] + for blob in data: + if not blob: + continue + uids.extend(int(token) for token in blob.split()) + return sorted(uids) + + def search_all(self) -> list[int]: + return self._uid_search("ALL") + + def search_since(self, since: date) -> list[int]: + """`UID SEARCH SINCE dd-Mon-yyyy` — filters on the server's INTERNALDATE (when the + message arrived), NOT the sender's `Date:` header. That is the right side of the + distinction the recon draws in §2.2 wariant C: a mail delivered today with a + month-old header date is still caught.""" + return self._uid_search("SINCE", imap_date(since)) + + def search_from_uid(self, start_uid: int) -> list[int]: + """UIDs at or above `start_uid`. + + The client-side filter is not redundant: `n:*` is a RANGE, and when `n` exceeds the + highest existing UID the server resolves it as `highest:n` and returns the last + message anyway. Without the filter, the 'new mails this tick' counter would never + read zero on an idle mailbox and every observability claim built on it would be a + small, permanent lie (recon §2.2). + """ + uids = self._uid_search("UID", f"{start_uid}:*") + return [uid for uid in uids if uid >= start_uid] + + # -- fetching ---------------------------------------------------------------- + + def fetch_message(self, uid: int) -> Optional[bytes]: + """Raw RFC822 bytes for one UID via `BODY.PEEK[]` — never sets `\\Seen`. + + Returns None when the UID no longer exists (deleted or expunged between SEARCH and + FETCH — an ordinary race on a live mailbox, not an error). + """ + data = self._check("UID FETCH", self._conn().uid("FETCH", str(uid), "(BODY.PEEK[])")) + for item in data: + if not isinstance(item, tuple) or len(item) < 2: + continue + prefix, payload = item[0], item[1] + match = _UID_IN_FETCH_RE.search(prefix or b"") + # A server that echoes a different UID than requested means the response stream + # is out of step with our request — refuse the payload rather than archive a + # message under the wrong cursor. + if match is not None and int(match.group(1)) != uid: + raise ImapError( + f"account {self.account.name}: FETCH for UID {uid} returned UID " + f"{int(match.group(1))}" + ) + if payload: + return bytes(payload) + return None + + +def _default_connection(account: ImapAccount, timeout_s: float): + return imaplib.IMAP4_SSL( + host=account.host, + port=account.port, + ssl_context=ssl.create_default_context(), + timeout=timeout_s, + ) + + +def _status_field(blob: bytes, key: bytes, folder: str) -> int: + match = re.search(key + rb"\s+(\d+)", blob) + if match is None: + raise ImapError(f"STATUS for {folder!r} has no {key.decode()}: {blob!r}") + return int(match.group(1)) + + +def _untagged_int(conn, key: str) -> Optional[int]: + try: + typ, data = conn.response(key) + except Exception: + return None + if typ != key and typ != "OK": + return None + for blob in data or []: + if not blob: + continue + match = re.search(rb"(\d+)", blob if isinstance(blob, bytes) else bytes(blob)) + if match: + return int(match.group(1)) + return None diff --git a/packages/kb-mail/src/kb_mail/message.py b/packages/kb-mail/src/kb_mail/message.py new file mode 100644 index 0000000..68d92bc --- /dev/null +++ b/packages/kb-mail/src/kb_mail/message.py @@ -0,0 +1,116 @@ +"""Message-level derivations shared by every job that turns `.eml` bytes into an `Envelope`: +the envelope id, the timestamp, the attachment manifest, and the archive path they imply. + +Extracted from `gmail_bulk_import.importer` when `jobs/mail-imap-sync` needed the identical +four derivations (recon `kb/audits/mail-sync-2026-08-06.md` §2.1) — the poller's dedup rests +entirely on producing byte-identical ids to the 225 030 rows the bulk import already wrote, so +a second implementation of `message_id()` is the one thing that must not exist. +`gmail_bulk_import.importer` re-exports these under its old private names, unchanged. +""" +from __future__ import annotations + +import email.message +import hashlib +from datetime import datetime, timezone +from email.utils import parsedate_to_datetime + +import structlog + +from .text import sanitize_surrogates as _sanitize + +_log = structlog.get_logger(__name__) + +EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc) + +# Mirrors archive._UNSAFE — used to reconstruct raw_ref for already-archived messages +# without writing them again. +_ARCHIVE_UNSAFE = str.maketrans({"/": "_", "\\": "_", ":": "_", "<": "", ">": ""}) + + +def eml_ref(envelope_id: str, source: str, ts: datetime) -> str: + """The archive-relative `.eml` path `save_eml` would write, computed without writing. + + Kept in lockstep with `kb_mail.archive.save_eml` — callers use it on `FileExistsError` + ("already archived") to fill `Envelope.raw_ref` for a message they did not just write. + """ + safe_id = envelope_id.translate(_ARCHIVE_UNSAFE) + return f"{source}/{ts.year:04d}/{ts.month:02d}/{safe_id}.eml" + + +def message_id(msg: email.message.Message) -> str: + """Return a stable envelope id from the Message-ID header, or a SHA-256 content hash. + + This IS the dedup key (`envelope.id`, `ON CONFLICT (id) DO NOTHING`), so its output must + stay byte-identical across jobs and years — 9 of the 225 030 existing rows carry the + `sha256-` form, the rest a bare Message-ID. + + compat32 `.get()` returns an `email.header.Header` (not str) when the raw value holds + 8-bit bytes — `str()` + `_sanitize` keeps the id jsonb-safe and gives `.strip()` a real + string to work on. An unguarded `.strip()` on a Header raised AttributeError and killed a + whole bulk-import run (the archive has proven 8-bit header bytes). + """ + mid = _sanitize(str(msg.get("Message-ID", ""))).strip().strip("<>") + if mid: + return mid + return "sha256-" + hashlib.sha256(msg.as_bytes()).hexdigest()[:32] + + +def parse_date(msg: email.message.Message) -> datetime: + """Parse the Date header to a UTC-aware datetime. Returns `EPOCH` as last resort. + + `str()` first: a compat32 8-bit Date surfaces as an `email.header.Header`, and + `parsedate_to_datetime()` raises on a Header (→ needless epoch fallback) even when + `str(header)` is perfectly parseable. + + Note this is the SENDER's clock, which is why it must never be used as a sync cursor — + 2 559 envelopes in the live corpus sit at epoch 1970 (recon §2.2 wariant C). + """ + date_str = str(msg.get("Date", "")) + if date_str: + try: + dt = parsedate_to_datetime(date_str) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt.astimezone(timezone.utc) + except Exception: + pass + return EPOCH + + +def parse_attachments(msg: email.message.Message) -> list[dict]: + """Return `entities[type=attachment]` descriptors from the MIME structure. + + A part is an attachment if it has a filename, `Content-Disposition: attachment`, or is not + inline `text/plain`/`text/html`. Errors on individual parts are logged and skipped — never + propagated, since one unwalkable part must not cost the whole message. + """ + attachments = [] + for part in msg.walk(): + if part.get_content_maintype() == "multipart": + continue + content_type = part.get_content_type() + disposition = part.get_content_disposition() or "" + filename = part.get_filename() + + if ( + content_type in ("text/plain", "text/html") + and disposition != "attachment" + and not filename + ): + continue + + try: + payload = part.get_payload(decode=True) + if payload is None: + continue + attachments.append({ + "type": "attachment", + "filename": filename, + "content_type": content_type, + "size": len(payload), + "sha256": hashlib.sha256(payload).hexdigest(), + }) + except Exception: + _log.warning("attachment_parse_failed", + filename=filename, content_type=content_type) + return attachments diff --git a/packages/kb-mail/src/kb_mail/sync_state.py b/packages/kb-mail/src/kb_mail/sync_state.py new file mode 100644 index 0000000..a16c7bb --- /dev/null +++ b/packages/kb-mail/src/kb_mail/sync_state.py @@ -0,0 +1,167 @@ +"""Per-folder IMAP sync cursor: the `mail_sync_state` table (migration 005) plus the pure +logic that decides what a tick fetches and how far the cursor is allowed to move. + +The rules encoded here are the whole correctness story of the poller, so they live as pure +functions with their own tests rather than inline in the job's loop: + +* **UIDVALIDITY governs everything.** When the server changes it, every stored UID becomes + meaningless. The only safe response is to sweep the folder and lean on Message-ID dedup — + a poller that trusts `last_uid` across a UIDVALIDITY change loses mail with no error + anywhere (recon `kb/audits/mail-sync-2026-08-06.md` §2.2). +* **The cursor moves only over messages that are fully durable**, archive file and envelope + row both. It moves over a CONTIGUOUS prefix of successes, so a message that failed is + re-fetched next tick instead of being stepped over silently. +* **A first tick is a policy decision, not a default.** `new-only` starts the corpus from + now; `since` closes a known gap (gmail's corpus stops at 2026-06-19); `full` pulls a + mailbox's whole history. Which one is right depends on a measurement the runbook takes. +""" +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date, datetime +from typing import Iterable, Optional + +import asyncpg +import structlog + +from .imap import FolderStatus, ImapAccount + +_log = structlog.get_logger(__name__) + +# What a plan tells the caller to run against the server. +SEARCH_NONE = "none" # fetch nothing this tick; just record the high-water mark +SEARCH_ALL = "all" # UID SEARCH ALL — full sweep, dedup does the rest +SEARCH_SINCE = "since" # UID SEARCH SINCE +SEARCH_FROM_UID = "from-uid" # UID SEARCH UID :* + + +@dataclass(frozen=True) +class FolderSyncState: + """One row of `mail_sync_state`.""" + + account: str + folder: str + uidvalidity: int + last_uid: int + last_sync_ts: Optional[datetime] = None + + +@dataclass(frozen=True) +class SyncPlan: + """What this tick will do to one folder. + + `baseline_uid` is the cursor to persist when the plan turns up NO messages — never a + floor to combine with partial results, because on a full sweep it equals `uidnext - 1` + and combining it with a partially-failed batch would step over everything that failed. + """ + + mode: str + search: str + baseline_uid: int + start_uid: Optional[int] = None + since: Optional[date] = None + uidvalidity_reset: bool = False + + +def plan_folder_sync( + account: ImapAccount, + status: FolderStatus, + state: Optional[FolderSyncState], +) -> SyncPlan: + """Decide this tick's work for one folder from the stored cursor and the server's state.""" + high_water = max(status.uidnext - 1, 0) + + if state is None: + if account.initial_mode == "new-only": + # Start the corpus here: record the high-water mark, fetch nothing. Everything + # already in the mailbox stays out of KB unless the operator re-runs with a + # different initial_mode after clearing the row. + return SyncPlan(mode="initial-new-only", search=SEARCH_NONE, baseline_uid=high_water) + if account.initial_mode == "since": + return SyncPlan(mode="initial-since", search=SEARCH_SINCE, + baseline_uid=high_water, since=account.initial_since) + return SyncPlan(mode="initial-full", search=SEARCH_ALL, baseline_uid=high_water) + + if state.uidvalidity != status.uidvalidity: + _log.warning("imap.uidvalidity_reset", account=account.name, folder=status.name, + stored=state.uidvalidity, server=status.uidvalidity) + return SyncPlan(mode="uidvalidity-reset", search=SEARCH_ALL, + baseline_uid=high_water, uidvalidity_reset=True) + + return SyncPlan(mode="incremental", search=SEARCH_FROM_UID, + baseline_uid=state.last_uid, start_uid=state.last_uid + 1) + + +def contiguous_last_uid(floor: int, attempted: Iterable[int], succeeded: set[int]) -> int: + """How far the cursor may move: the last UID of the unbroken run of successes. + + A failure stops the cursor at the message before it. The consequences are deliberate and + worth naming: nothing is ever lost, the re-fetch costs nothing (dedup), and a message + that fails *permanently* stalls its folder — visibly, as a non-zero error counter every + tick and a cursor that stops moving, with the offending UID in the log. That is strictly + better than the alternative of stepping over it, which drops mail while reporting success. + The runbook documents the manual `UPDATE mail_sync_state` escape hatch for that case. + """ + last = floor + for uid in sorted(attempted): + if uid not in succeeded: + break + last = max(last, uid) + return last + + +async def get_folder_state( + conn: asyncpg.Connection, account: str, folder: str +) -> Optional[FolderSyncState]: + row = await conn.fetchrow( + "SELECT account, folder, uidvalidity, last_uid, last_sync_ts " + "FROM mail_sync_state WHERE account = $1 AND folder = $2", + account, folder, + ) + if row is None: + return None + return FolderSyncState( + account=row["account"], + folder=row["folder"], + uidvalidity=int(row["uidvalidity"]), + last_uid=int(row["last_uid"]), + last_sync_ts=row["last_sync_ts"], + ) + + +async def upsert_folder_state( + conn: asyncpg.Connection, account: str, folder: str, uidvalidity: int, last_uid: int +) -> None: + """Persist the cursor and stamp `last_sync_ts`. + + Called on every tick including empty ones: `last_sync_ts` is how one answers "is this + folder being polled at all", which stays true when nothing has arrived for a week. + """ + await conn.execute( + """ + INSERT INTO mail_sync_state (account, folder, uidvalidity, last_uid, last_sync_ts) + VALUES ($1, $2, $3, $4, now()) + ON CONFLICT (account, folder) DO UPDATE + SET uidvalidity = EXCLUDED.uidvalidity, + last_uid = EXCLUDED.last_uid, + last_sync_ts = EXCLUDED.last_sync_ts + """, + account, folder, uidvalidity, last_uid, + ) + _log.info("sync_state.saved", account=account, folder=folder, + uidvalidity=uidvalidity, last_uid=last_uid) + + +async def fetch_all_state(conn: asyncpg.Connection) -> list[FolderSyncState]: + """Every cursor, for the runbook's inspection queries and the job's summary line.""" + rows = await conn.fetch( + "SELECT account, folder, uidvalidity, last_uid, last_sync_ts " + "FROM mail_sync_state ORDER BY account, folder" + ) + return [ + FolderSyncState( + account=r["account"], folder=r["folder"], uidvalidity=int(r["uidvalidity"]), + last_uid=int(r["last_uid"]), last_sync_ts=r["last_sync_ts"], + ) + for r in rows + ] diff --git a/packages/kb-mail/tests/test_imap.py b/packages/kb-mail/tests/test_imap.py new file mode 100644 index 0000000..d391453 --- /dev/null +++ b/packages/kb-mail/tests/test_imap.py @@ -0,0 +1,330 @@ +"""Unit tests for the IMAP adapter — no network, no real imaplib connection. + +`ImapClient` is exercised against `_FakeIMAP`, a stand-in for `imaplib.IMAP4_SSL` that speaks +real IMAP response shapes (bytes lines, `(prefix, literal)` FETCH tuples, untagged responses). +The point is to test this module's OWN parsing rather than a mock of it — every response shape +here is one imaplib actually hands back. +""" +from __future__ import annotations + +from datetime import date + +import pytest + +from kb_mail.imap import ( + DEFAULT_PORT, + ImapAccount, + ImapClient, + ImapError, + imap_date, + parse_list_line, + quote_mailbox, +) + + +class _FakeIMAP: + """Minimal IMAP4_SSL stand-in. Records commands so tests can assert on the wire traffic — + notably that folders are opened READ-ONLY and bodies fetched with BODY.PEEK[].""" + + def __init__( + self, + *, + list_data=None, + messages=None, + uidvalidity=42, + uidnext=101, + status_line=None, + search_ok=True, + select_ok=True, + untagged=True, + ): + self.list_data = list_data if list_data is not None else [ + b'(\\HasNoChildren) "/" "INBOX"', + b'(\\HasNoChildren \\All) "/" "[Gmail]/Wszystkie"', + b'(\\HasNoChildren \\Sent) "/" "[Gmail]/Wys&AWI-ane"', + ] + self.messages = messages or {} + self.uidvalidity = uidvalidity + self.uidnext = uidnext + self.status_line = status_line + self.search_ok = search_ok + self.select_ok = select_ok + self.untagged = untagged + self.commands: list[tuple] = [] + self.logged_in = False + self.logged_out = False + + def login(self, user, password): + self.commands.append(("LOGIN", user)) + self.logged_in = True + return ("OK", [b"LOGIN completed"]) + + def logout(self): + self.logged_out = True + return ("BYE", [b"logging out"]) + + def list(self, directory='""', pattern="*"): + self.commands.append(("LIST",)) + return ("OK", self.list_data) + + def select(self, mailbox, readonly=False): + self.commands.append(("SELECT", mailbox, readonly)) + if not self.select_ok: + return ("NO", [b"no such mailbox"]) + return ("OK", [str(len(self.messages)).encode()]) + + def status(self, mailbox, names): + self.commands.append(("STATUS", mailbox, names)) + line = self.status_line or ( + f'"{mailbox}" (MESSAGES {len(self.messages)} ' + f"UIDNEXT {self.uidnext} UIDVALIDITY {self.uidvalidity})" + ).encode() + return ("OK", [line]) + + def response(self, key): + if not self.untagged: + return (key, [None]) + if key == "UIDVALIDITY": + return (key, [str(self.uidvalidity).encode()]) + if key == "UIDNEXT": + return (key, [str(self.uidnext).encode()]) + return (key, [None]) + + def uid(self, command, *args): + self.commands.append(("UID", command) + args) + if command == "SEARCH": + if not self.search_ok: + return ("NO", [b"search failed"]) + criteria = args[1:] + return ("OK", [self._search(criteria)]) + if command == "FETCH": + uid = int(args[0]) + raw = self.messages.get(uid) + if raw is None: + return ("OK", [None]) + prefix = f"1 (UID {uid} BODY[] {{{len(raw)}}}".encode() + return ("OK", [(prefix, raw), b")"]) + raise AssertionError(f"unexpected UID command {command}") + + def _search(self, criteria): + uids = sorted(self.messages) + if criteria and criteria[0] == "UID": + start = int(criteria[1].split(":")[0]) + hits = [u for u in uids if u >= start] + # Real servers resolve `n:*` as a range and return the highest message even when + # n is beyond it. Reproduced deliberately — the client-side filter exists for this. + if not hits and uids: + hits = [uids[-1]] + return " ".join(str(u) for u in hits).encode() + return " ".join(str(u) for u in uids).encode() + + +def _account(**kwargs) -> ImapAccount: + base = dict(name="gmail", host="imap.example.com", user="u@example.com", + password="secret", special_use=("\\All",)) + base.update(kwargs) + return ImapAccount(**base) + + +def _client(fake: _FakeIMAP, account=None) -> ImapClient: + account = account or _account() + client = ImapClient(account, connection_factory=lambda acc, timeout: fake) + client.connect() + return client + + +class TestImapAccount: + def test_defaults_to_implicit_tls_port(self): + assert _account().port == DEFAULT_PORT + + def test_password_is_not_in_repr(self): + # The repo has two documented secret leaks to session transcripts (recon Decyzja (c)). + assert "secret" not in repr(_account()) + + def test_rejects_account_with_no_scope(self): + with pytest.raises(ValueError, match="no scope"): + ImapAccount(name="x", host="h", user="u", password="p") + + def test_rejects_unknown_initial_mode(self): + with pytest.raises(ValueError, match="initial_mode"): + _account(initial_mode="yesterday") + + def test_since_mode_requires_a_date(self): + with pytest.raises(ValueError, match="initial_since"): + _account(initial_mode="since") + + def test_since_mode_with_date_is_valid(self): + assert _account(initial_mode="since", initial_since=date(2026, 6, 15)).initial_since + + +class TestHelpers: + def test_imap_date_uses_english_month_abbreviations(self): + assert imap_date(date(2026, 6, 15)) == "15-Jun-2026" + + def test_imap_date_zero_pads_the_day(self): + assert imap_date(date(2026, 12, 5)) == "05-Dec-2026" + + def test_quote_mailbox_wraps_names_with_spaces(self): + assert quote_mailbox("[Gmail]/All Mail") == '"[Gmail]/All Mail"' + + def test_quote_mailbox_escapes_quotes_and_backslashes(self): + assert quote_mailbox('we"ird\\name') == '"we\\"ird\\\\name"' + + +class TestParseListLine: + def test_quoted_name_with_attributes(self): + attrs, name = parse_list_line(b'(\\HasNoChildren \\All) "/" "[Gmail]/All Mail"') + assert name == "[Gmail]/All Mail" + assert "\\all" in attrs + + def test_unquoted_atom_name(self): + attrs, name = parse_list_line(b'(\\HasNoChildren) "/" INBOX') + assert name == "INBOX" + + def test_nil_delimiter(self): + attrs, name = parse_list_line(b'(\\Noselect) NIL "Archive"') + assert name == "Archive" + + def test_literal_name_arrives_as_a_tuple(self): + attrs, name = parse_list_line((b'(\\HasNoChildren \\Archive) "/" {7}', b"Archiwa")) + assert name == "Archiwa" + assert "\\archive" in attrs + + def test_unparseable_line_returns_none(self): + assert parse_list_line(b"* SOMETHING ELSE") is None + + +class TestResolveFolders: + def test_special_use_is_matched_by_attribute_not_name(self): + # The Polish-UI name is what makes hardcoding '[Gmail]/All Mail' a silent zero-mail sync. + fake = _FakeIMAP() + assert _client(fake).resolve_folders() == ["[Gmail]/Wszystkie"] + + def test_literal_folders_are_used_as_configured(self): + fake = _FakeIMAP() + account = _account(name="fastmail", special_use=(), + folders=("INBOX", "Archive", "Sent")) + assert _client(fake, account).resolve_folders() == ["INBOX", "Archive", "Sent"] + + def test_missing_special_use_raises_instead_of_syncing_nothing(self): + fake = _FakeIMAP(list_data=[b'(\\HasNoChildren) "/" "INBOX"']) + with pytest.raises(ImapError, match="SPECIAL-USE"): + _client(fake).resolve_folders() + + def test_duplicates_between_attribute_and_literal_are_collapsed(self): + fake = _FakeIMAP() + account = _account(special_use=("\\All",), folders=("[Gmail]/Wszystkie", "INBOX")) + assert _client(fake, account).resolve_folders() == ["[Gmail]/Wszystkie", "INBOX"] + + def test_no_list_call_when_only_literal_folders_are_configured(self): + fake = _FakeIMAP() + account = _account(special_use=(), folders=("INBOX",)) + _client(fake, account).resolve_folders() + assert not any(c[0] == "LIST" for c in fake.commands) + + +class TestExamine: + def test_opens_folder_read_only(self): + # A plain SELECT + FETCH RFC822 would set \Seen on the operator's mail. + fake = _FakeIMAP(messages={1: b"raw"}) + _client(fake).examine("INBOX") + select_cmds = [c for c in fake.commands if c[0] == "SELECT"] + assert select_cmds and select_cmds[0][2] is True + + def test_returns_uidvalidity_and_uidnext_from_untagged_responses(self): + fake = _FakeIMAP(uidvalidity=7, uidnext=55) + status = _client(fake).examine("INBOX") + assert (status.uidvalidity, status.uidnext) == (7, 55) + assert not any(c[0] == "STATUS" for c in fake.commands) + + def test_falls_back_to_status_when_server_omits_untagged_values(self): + fake = _FakeIMAP(uidvalidity=9, uidnext=30, untagged=False) + status = _client(fake).examine("INBOX") + assert (status.uidvalidity, status.uidnext) == (9, 30) + assert any(c[0] == "STATUS" for c in fake.commands) + # ... and re-opens the mailbox, since STATUS can deselect it. + assert len([c for c in fake.commands if c[0] == "SELECT"]) == 2 + + def test_non_ok_select_raises(self): + fake = _FakeIMAP(select_ok=False) + with pytest.raises(ImapError, match="EXAMINE"): + _client(fake).examine("Nope") + + +class TestStatus: + def test_parses_message_count_for_the_sizing_measurement(self): + fake = _FakeIMAP(messages={1: b"a", 2: b"b"}, uidnext=3, uidvalidity=11) + status = _client(fake).status("INBOX") + assert (status.messages, status.uidnext, status.uidvalidity) == (2, 3, 11) + + def test_missing_field_raises_rather_than_defaulting(self): + fake = _FakeIMAP(status_line=b'"INBOX" (MESSAGES 2)') + with pytest.raises(ImapError, match="has no UIDVALIDITY"): + _client(fake).status("INBOX") + + +class TestSearch: + def test_search_all_returns_sorted_uids(self): + fake = _FakeIMAP(messages={3: b"c", 1: b"a", 2: b"b"}) + assert _client(fake).search_all() == [1, 2, 3] + + def test_search_since_sends_an_imap_date(self): + fake = _FakeIMAP(messages={1: b"a"}) + _client(fake).search_since(date(2026, 6, 15)) + search = [c for c in fake.commands if c[:2] == ("UID", "SEARCH")][0] + assert search[-2:] == ("SINCE", "15-Jun-2026") + + def test_search_from_uid_filters_the_range_trailer(self): + # `n:*` past the end returns the LAST message on real servers; without the + # client-side filter every idle tick would report one "new" mail forever. + fake = _FakeIMAP(messages={1: b"a", 2: b"b"}) + assert _client(fake).search_from_uid(3) == [] + + def test_search_from_uid_returns_only_new_uids(self): + fake = _FakeIMAP(messages={1: b"a", 2: b"b", 3: b"c"}) + assert _client(fake).search_from_uid(2) == [2, 3] + + def test_non_ok_search_raises(self): + fake = _FakeIMAP(messages={1: b"a"}, search_ok=False) + with pytest.raises(ImapError, match="UID SEARCH"): + _client(fake).search_all() + + +class TestFetch: + def test_returns_raw_bytes_and_uses_body_peek(self): + fake = _FakeIMAP(messages={5: b"From: a@b\r\n\r\nhello"}) + assert _client(fake).fetch_message(5) == b"From: a@b\r\n\r\nhello" + fetch = [c for c in fake.commands if c[:2] == ("UID", "FETCH")][0] + assert fetch[-1] == "(BODY.PEEK[])" + + def test_missing_uid_returns_none_not_an_error(self): + # Expunged between SEARCH and FETCH — an ordinary race on a live mailbox. + fake = _FakeIMAP(messages={5: b"x"}) + assert _client(fake).fetch_message(6) is None + + def test_mismatched_uid_in_response_raises(self): + fake = _FakeIMAP(messages={5: b"x"}) + client = _client(fake) + fake.uid = lambda command, *args: ("OK", [(b"1 (UID 9 BODY[] {1}", b"x"), b")"]) + with pytest.raises(ImapError, match="returned UID 9"): + client.fetch_message(5) + + +class TestLifecycle: + def test_context_manager_logs_in_and_out(self): + fake = _FakeIMAP() + with ImapClient(_account(), connection_factory=lambda acc, t: fake): + assert fake.logged_in + assert fake.logged_out + + def test_logout_runs_even_when_the_body_raises(self): + fake = _FakeIMAP() + with pytest.raises(RuntimeError): + with ImapClient(_account(), connection_factory=lambda acc, t: fake): + raise RuntimeError("boom") + assert fake.logged_out + + def test_commands_before_connect_raise(self): + client = ImapClient(_account(), connection_factory=lambda acc, t: _FakeIMAP()) + with pytest.raises(ImapError, match="not connected"): + client.search_all() diff --git a/packages/kb-mail/tests/test_message.py b/packages/kb-mail/tests/test_message.py new file mode 100644 index 0000000..4479a63 --- /dev/null +++ b/packages/kb-mail/tests/test_message.py @@ -0,0 +1,94 @@ +"""Tests for the message-level derivations extracted out of gmail-bulk-import. + +The derivations themselves have been exercised by that job's suite (and by 225 030 real +messages) since June; what is new here is that they are now shared, and that `eml_ref` takes +`source` as a parameter instead of hardcoding `gmail/`. These tests pin the properties the +poller depends on: ids identical to the ones already in the table, and archive paths that +agree with `save_eml` for BOTH sources. +""" +from __future__ import annotations + +import email +import email.policy +from datetime import datetime, timezone +from pathlib import Path + +from kb_mail.archive import save_eml +from kb_mail.message import EPOCH, eml_ref, message_id, parse_attachments, parse_date + + +def _msg(raw: bytes): + return email.message_from_bytes(raw, policy=email.policy.compat32) + + +class TestMessageId: + def test_strips_angle_brackets(self): + assert message_id(_msg(b"Message-ID: \r\n\r\nbody")) == "abc@example.com" + + def test_missing_header_falls_back_to_a_content_hash(self): + mid = message_id(_msg(b"Subject: no id\r\n\r\nbody")) + assert mid.startswith("sha256-") + assert len(mid) == len("sha256-") + 32 + + def test_content_hash_is_stable_for_identical_bytes(self): + raw = b"Subject: no id\r\n\r\nbody" + assert message_id(_msg(raw)) == message_id(_msg(raw)) + + def test_eight_bit_header_bytes_do_not_raise(self): + # compat32 hands back an email.header.Header here, not a str — an unguarded + # .strip() on it killed a full bulk-import run once. + mid = message_id(_msg(b"Message-ID: <\xc4\x85bc@example.com>\r\n\r\nbody")) + assert mid + + +class TestParseDate: + def test_parses_to_utc(self): + ts = parse_date(_msg(b"Date: Mon, 15 Jun 2026 14:00:00 +0200\r\n\r\nbody")) + assert ts == datetime(2026, 6, 15, 12, 0, tzinfo=timezone.utc) + + def test_missing_date_is_epoch(self): + assert parse_date(_msg(b"Subject: x\r\n\r\nbody")) == EPOCH + + def test_unparseable_date_is_epoch(self): + assert parse_date(_msg(b"Date: not a date at all\r\n\r\nbody")) == EPOCH + + def test_naive_date_is_assumed_utc(self): + ts = parse_date(_msg(b"Date: Mon, 15 Jun 2026 14:00:00 -0000\r\n\r\nbody")) + assert ts.tzinfo is timezone.utc + + +class TestParseAttachments: + def test_inline_text_is_not_an_attachment(self): + assert parse_attachments(_msg(b"Content-Type: text/plain\r\n\r\nhello")) == [] + + def test_attachment_manifest_carries_sha256_and_size(self): + raw = ( + b'Content-Type: multipart/mixed; boundary="b"\r\n\r\n--b\r\n' + b"Content-Type: text/plain\r\n\r\nhello\r\n--b\r\n" + b"Content-Type: application/pdf\r\n" + b'Content-Disposition: attachment; filename="doc.pdf"\r\n' + b"Content-Transfer-Encoding: base64\r\n\r\nJVBERg==\r\n--b--\r\n" + ) + [att] = parse_attachments(_msg(raw)) + assert att["type"] == "attachment" + assert att["filename"] == "doc.pdf" + assert att["size"] == 4 # decoded payload, not the base64 text + assert len(att["sha256"]) == 64 + + +class TestEmlRef: + def test_layout_is_source_year_month(self): + ts = datetime(2026, 6, 15, tzinfo=timezone.utc) + assert eml_ref("abc@example.com", "fastmail", ts) == "fastmail/2026/06/abc@example.com.eml" + + def test_unsafe_characters_are_translated(self): + ts = datetime(2026, 6, 15, tzinfo=timezone.utc) + assert eml_ref("", "gmail", ts) == "gmail/2026/06/a_b_c.eml" + + async def test_agrees_with_save_eml_for_a_new_source(self, tmp_path: Path): + # The two must not drift: on FileExistsError the caller fills raw_ref from eml_ref + # for a file save_eml wrote on an earlier run. + ts = datetime(2026, 6, 15, tzinfo=timezone.utc) + written = await save_eml(tmp_path, "id/with:chars", "fastmail", ts, b"raw") + assert written == eml_ref("id/with:chars", "fastmail", ts) + assert (tmp_path / written).read_bytes() == b"raw" diff --git a/packages/kb-mail/tests/test_migration.py b/packages/kb-mail/tests/test_migration.py index d4bac31..7798c08 100644 --- a/packages/kb-mail/tests/test_migration.py +++ b/packages/kb-mail/tests/test_migration.py @@ -87,3 +87,40 @@ def test_migration_002_uses_idempotent_ddl(): if line.strip().upper().startswith("CREATE INDEX"): assert "IF NOT EXISTS" in line.upper(), \ f"CREATE INDEX must be idempotent: {line.strip()}" + + +def test_migration_005_exists(): + assert (INIT_DIR / "005_mail_sync_state.sql").exists(), \ + f"Expected {INIT_DIR / '005_mail_sync_state.sql'} to exist" + + +def test_migration_005_creates_mail_sync_state_idempotently(): + sql = (INIT_DIR / "005_mail_sync_state.sql").read_text() + assert "CREATE TABLE IF NOT EXISTS mail_sync_state" in sql + + +def test_migration_005_has_all_cursor_columns(): + sql = (INIT_DIR / "005_mail_sync_state.sql").read_text() + for col in ("account", "folder", "uidvalidity", "last_uid", "last_sync_ts"): + assert col in sql, f"Column '{col}' missing from 005_mail_sync_state.sql" + + +def test_migration_005_is_keyed_per_account_and_folder(): + # Folder scope is per account (recon Decyzja (e)): gmail syncs one \All folder, + # fastmail syncs three. A single-column key could not express that. + sql = (INIT_DIR / "005_mail_sync_state.sql").read_text() + assert "PRIMARY KEY (account, folder)" in sql + + +def test_migration_005_uidvalidity_and_last_uid_are_not_nullable(): + sql = (INIT_DIR / "005_mail_sync_state.sql").read_text() + for col in ("uidvalidity", "last_uid"): + line = next(l for l in sql.splitlines() if l.strip().startswith(col)) + assert "NOT NULL" in line.upper(), f"{col} must be NOT NULL but found: {line.strip()}" + + +def test_migration_005_is_additive(): + sql = (INIT_DIR / "005_mail_sync_state.sql").read_text().upper() + for table in ("ENVELOPE", "DOCUMENT_CHUNK", "DOCUMENT_SUMMARY"): + assert f"ALTER TABLE {table}" not in sql + assert f"DROP TABLE {table}" not in sql diff --git a/packages/kb-mail/tests/test_sync_state.py b/packages/kb-mail/tests/test_sync_state.py new file mode 100644 index 0000000..5bb242d --- /dev/null +++ b/packages/kb-mail/tests/test_sync_state.py @@ -0,0 +1,182 @@ +"""Unit tests for the sync cursor: the plan a tick makes, and how far the cursor may move. + +These two pure functions carry the whole correctness argument of the poller — a wrong branch +here loses mail with no error anywhere — so they are tested directly, without a DB. The +asyncpg helpers are tested against an in-memory fake connection, the same style as +`jobs/gmail-header-backfill/tests`. +""" +from __future__ import annotations + +from datetime import date, datetime, timezone + +import pytest + +from kb_mail.imap import FolderStatus, ImapAccount +from kb_mail.sync_state import ( + SEARCH_ALL, + SEARCH_FROM_UID, + SEARCH_NONE, + SEARCH_SINCE, + FolderSyncState, + contiguous_last_uid, + fetch_all_state, + get_folder_state, + plan_folder_sync, + upsert_folder_state, +) + + +def _account(**kwargs) -> ImapAccount: + base = dict(name="gmail", host="h", user="u", password="p", special_use=("\\All",)) + base.update(kwargs) + return ImapAccount(**base) + + +def _status(uidvalidity=42, uidnext=101) -> FolderStatus: + return FolderStatus(name="[Gmail]/All Mail", uidvalidity=uidvalidity, uidnext=uidnext) + + +def _state(uidvalidity=42, last_uid=90) -> FolderSyncState: + return FolderSyncState(account="gmail", folder="[Gmail]/All Mail", + uidvalidity=uidvalidity, last_uid=last_uid) + + +class TestPlanFirstTick: + def test_new_only_fetches_nothing_and_records_the_high_water_mark(self): + plan = plan_folder_sync(_account(initial_mode="new-only"), _status(uidnext=101), None) + assert plan.search == SEARCH_NONE + assert plan.baseline_uid == 100 + assert plan.mode == "initial-new-only" + + def test_new_only_on_an_empty_mailbox_records_zero_not_minus_one(self): + plan = plan_folder_sync(_account(initial_mode="new-only"), _status(uidnext=1), None) + assert plan.baseline_uid == 0 + + def test_since_mode_carries_the_configured_date(self): + account = _account(initial_mode="since", initial_since=date(2026, 6, 15)) + plan = plan_folder_sync(account, _status(), None) + assert plan.search == SEARCH_SINCE + assert plan.since == date(2026, 6, 15) + + def test_full_mode_sweeps_everything(self): + plan = plan_folder_sync(_account(initial_mode="full"), _status(), None) + assert plan.search == SEARCH_ALL + assert plan.mode == "initial-full" + + def test_first_tick_is_never_flagged_as_a_uidvalidity_reset(self): + plan = plan_folder_sync(_account(initial_mode="full"), _status(), None) + assert plan.uidvalidity_reset is False + + +class TestPlanIncremental: + def test_resumes_from_the_uid_after_the_cursor(self): + plan = plan_folder_sync(_account(), _status(uidvalidity=42), _state(last_uid=90)) + assert plan.search == SEARCH_FROM_UID + assert plan.start_uid == 91 + assert plan.mode == "incremental" + + def test_baseline_is_the_stored_cursor_so_an_empty_tick_changes_nothing(self): + plan = plan_folder_sync(_account(), _status(), _state(last_uid=90)) + assert plan.baseline_uid == 90 + + def test_initial_mode_is_ignored_once_a_cursor_exists(self): + # Otherwise flipping initial_mode to 'full' in .env would re-sweep every folder + # on the next tick instead of only affecting folders with no state. + account = _account(initial_mode="since", initial_since=date(2026, 1, 1)) + plan = plan_folder_sync(account, _status(), _state()) + assert plan.search == SEARCH_FROM_UID + + +class TestPlanUidvalidityReset: + def test_changed_uidvalidity_forces_a_full_sweep(self): + plan = plan_folder_sync(_account(), _status(uidvalidity=99), _state(uidvalidity=42)) + assert plan.search == SEARCH_ALL + assert plan.uidvalidity_reset is True + assert plan.mode == "uidvalidity-reset" + + def test_stored_cursor_is_discarded_not_reused_as_a_floor(self): + # The stored last_uid means nothing under a new UIDVALIDITY; keeping it as a floor + # would skip every message whose new UID happens to fall below it. + plan = plan_folder_sync(_account(), _status(uidvalidity=99, uidnext=51), + _state(uidvalidity=42, last_uid=9000)) + assert plan.start_uid is None + assert plan.baseline_uid == 50 + + +class TestContiguousLastUid: + def test_all_succeeded_advances_to_the_last_uid(self): + assert contiguous_last_uid(90, [91, 92, 93], {91, 92, 93}) == 93 + + def test_stops_before_the_first_failure(self): + assert contiguous_last_uid(90, [91, 92, 93], {91, 93}) == 91 + + def test_failure_on_the_first_uid_leaves_the_cursor_where_it_was(self): + assert contiguous_last_uid(90, [91, 92], {92}) == 90 + + def test_nothing_attempted_returns_the_floor(self): + assert contiguous_last_uid(90, [], set()) == 90 + + def test_sparse_uid_sets_advance_over_gaps(self): + # A SINCE search returns non-contiguous UIDs; skipped-by-design UIDs are not failures. + assert contiguous_last_uid(0, [95, 97, 100], {95, 97, 100}) == 100 + + def test_unsorted_input_is_ordered_before_walking(self): + assert contiguous_last_uid(0, [93, 91, 92], {91, 92}) == 92 + + def test_full_sweep_floor_of_zero_with_an_early_failure_keeps_the_cursor_at_zero(self): + # This is why baseline_uid (uidnext-1) must NOT be used as a floor for partial + # results: it would step straight over everything that failed. + assert contiguous_last_uid(0, [1, 2, 3], {2, 3}) == 0 + + +class _FakeConn: + def __init__(self, row=None, rows=None): + self.row = row + self.rows = rows or [] + self.executed: list[tuple] = [] + self.queries: list[tuple] = [] + + async def fetchrow(self, query, *params): + self.queries.append((query, params)) + return self.row + + async def fetch(self, query, *params): + self.queries.append((query, params)) + return self.rows + + async def execute(self, query, *params): + self.executed.append((query, params)) + return "INSERT 0 1" + + +class TestStateHelpers: + async def test_get_folder_state_returns_none_when_absent(self): + assert await get_folder_state(_FakeConn(), "gmail", "INBOX") is None + + async def test_get_folder_state_decodes_a_row(self): + ts = datetime(2026, 8, 6, tzinfo=timezone.utc) + conn = _FakeConn(row={"account": "gmail", "folder": "INBOX", "uidvalidity": 42, + "last_uid": 90, "last_sync_ts": ts}) + state = await get_folder_state(conn, "gmail", "INBOX") + assert (state.uidvalidity, state.last_uid, state.last_sync_ts) == (42, 90, ts) + + async def test_upsert_is_an_on_conflict_update(self): + conn = _FakeConn() + await upsert_folder_state(conn, "gmail", "INBOX", 42, 91) + query, params = conn.executed[-1] + assert "ON CONFLICT (account, folder) DO UPDATE" in query + assert params == ("gmail", "INBOX", 42, 91) + + async def test_upsert_stamps_last_sync_ts_server_side(self): + # last_sync_ts answers "is this folder being polled at all" — it must advance on + # empty ticks too, so it is set by the statement rather than passed in. + conn = _FakeConn() + await upsert_folder_state(conn, "gmail", "INBOX", 42, 91) + assert "now()" in conn.executed[-1][0] + + async def test_fetch_all_state_is_ordered(self): + conn = _FakeConn(rows=[{"account": "fastmail", "folder": "INBOX", "uidvalidity": 1, + "last_uid": 2, "last_sync_ts": None}]) + states = await fetch_all_state(conn) + assert "ORDER BY account, folder" in conn.queries[-1][0] + assert states[0].account == "fastmail" diff --git a/services/kb-postgres/init/005_mail_sync_state.sql b/services/kb-postgres/init/005_mail_sync_state.sql new file mode 100644 index 0000000..ddf2a47 --- /dev/null +++ b/services/kb-postgres/init/005_mail_sync_state.sql @@ -0,0 +1,27 @@ +-- KB spine: mail_sync_state — per-folder IMAP sync cursor for jobs/mail-imap-sync +-- Additive: does not modify 001_envelope.sql .. 004_summaries.sql. +-- Version: 005 — mail_sync_state (faza mailowa Krok 7, recon kb/audits/mail-sync-2026-08-06.md +-- §2.2 wariant A + Decyzja (f), zatwierdzona przez operatora 2026-08-06) +-- +-- Why a table and not a file in /opt/homelab/state/: the sync cursor and the envelopes it +-- describes must restore together or not at all. A state file surviving a DB restore makes the +-- poller silently skip everything between the restored envelope set and the file's last_uid — +-- a failure with no symptom, noticed months later as missing mail. Backing it into the same +-- database buys that invariant for the price of this one migration. +-- +-- Keyed (account, folder) because the folder scope is per account (Decyzja (e)): gmail syncs a +-- single SPECIAL-USE \All folder, fastmail syncs INBOX + Archive + Sent. Adding a folder is +-- then a config change, not a migration — Message-ID dedup absorbs any overlap. +-- +-- account matches envelope.source ('gmail' | 'fastmail'); folder is the IMAP mailbox name +-- exactly as the server returned it in LIST (never a hardcoded literal — Gmail localizes +-- '[Gmail]/All Mail', so the name is resolved by SPECIAL-USE attribute at runtime). + +CREATE TABLE IF NOT EXISTS mail_sync_state ( + account TEXT NOT NULL, -- == envelope.source: gmail | fastmail + folder TEXT NOT NULL, -- IMAP mailbox name as returned by LIST + uidvalidity BIGINT NOT NULL, -- server's UIDVALIDITY; a change invalidates last_uid + last_uid BIGINT NOT NULL, -- highest UID processed to completion (archive + envelope durable) + last_sync_ts TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (account, folder) +);