feat(mail-imap-sync): job przyrostowki + wpiecie w tor body-ingest
Nowy job jobs/mail-imap-sync — jedyny wlasciwy nowy kod przyrostowki (recon kb/audits/mail-sync-2026-08-06.md §6 poz. 1). Jeden tick, per konto i folder: EXAMINE -> plan -> UID SEARCH -> FETCH BODY.PEEK[] -> save_eml -> insert_envelope(entities=[headers, attachment...]) -> UPDATE mail_sync_state. Wlasciwosci, ktore latwo zgubic po cichu: - Job NIE embeduje i NIE chunkuje (recon §3.2). Pobieranie jest sieciowe i chodzi na PIHA 24/7; chunk+embed potrzebuje Ollamy na SOLARII, wylaczanej ~16 h/dobe. Spoiwem jest kolejka wynikajaca z danych: koperta bez chunkow JEST elementem kolejki, ktora drenuje mail-body-ingest --only-unchunked. - Koperta dostaje entities[type=headers] juz przy INSERCIE. Bez tego kazdy nowy mail mialby prefiks "Temat: (brak tematu) | Od: ?" — bez bledu, tylko z gorszym retrievalem (recon §2.5 i). - Kolizja Message-ID miedzy kontami jest liczona (envelopes_conflict_other_source), nie ukryta w zwyklych duplikatach (recon §2.4). - Kursor przesuwa sie tylko po nieprzerwanym ciagu w pelni trwalych wiadomosci. Bledna wiadomosc jest ponawiana (dedup czyni to darmowym), nie przeskakiwana; trwale zatrucie widac jako niezerowy licznik bledow i stojacy kursor. - Poswiadczenia wylacznie ze srodowiska — brak flagi --password/--user (Decyzja (c)). - Tryb --measure (STATUS MESSAGES/UIDNEXT/UIDVALIDITY): pomiar, na ktorym operator oprze decyzje o historii Fastmaila (Decyzja (e), celowo nieodgadywana). - Metryki .prom per konto; last_success_timestamp przenoszony przez nieudany run. Wpiecie w istniejacy tor (recon §6 poz. 4-6): - mail_body_ingest.fetch_envelopes: --sources (domyslnie gmail,fastmail) + --only-unchunked. - fetch_existing_chunk_keys zawezone do zbioru roboczego — bez tego kazdy tick czyta wszystkie 389 012 kluczy chunkow (~26 MB, ~1,0 s) na nodzie z 2,4 GB available (recon §2.5 iv). - DEFAULT_SUMMARYLESS_SOURCES += "fastmail" (Decyzja (g)) w TYM SAMYM commicie, ktory wprowadza zrodlo: bez tego koperty fastmail zaindeksowalyby sie poprawnie i byly niewidoczne w /search, bez zadnego bledu. Testy: 80 dla nowego joba (mock IMAP na poziomie imaplib, wiec testowane jest prawdziwe parsowanie protokolu) — nowe wiadomosci, uniewaznienie UIDVALIDITY, dedup, wznowienie po przerwaniu, izolacja kont, dry-run bez sieci, metryki; + 47 mail-body-ingest/kb-retrieval. Smoke: --help, blad konfiguracji -> exit 2. Zero polaczen z zywymi kontami — pierwszy zywy sync robi operator wg runbooka. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
oskar
parent
c65f0f2075
commit
f056b08574
|
|
@ -106,6 +106,9 @@ from kb_retrieval.embed import (
|
|||
_log = structlog.get_logger(__name__)
|
||||
|
||||
DEFAULT_ARCHIVE_ROOT = Path("/home/oskar/kb/mail/archive")
|
||||
# Both live mail accounts. `fastmail` joined when jobs/mail-imap-sync started writing that
|
||||
# source (recon Decyzja (b)/(g)); it is inert until then, since the source has no rows.
|
||||
DEFAULT_SOURCES = ("gmail", "fastmail")
|
||||
DEFAULT_BATCH_SIZE = 64
|
||||
DEFAULT_EMBED_RETRIES = 2
|
||||
DEFAULT_EMBED_BACKOFF_S = 1.0
|
||||
|
|
@ -366,14 +369,33 @@ def _has_threading(entities: list) -> bool:
|
|||
|
||||
|
||||
async def fetch_envelopes(
|
||||
conn: asyncpg.Connection, since: Optional[datetime], limit: Optional[int], offset: Optional[int]
|
||||
conn: asyncpg.Connection,
|
||||
since: Optional[datetime] = None,
|
||||
limit: Optional[int] = None,
|
||||
offset: Optional[int] = None,
|
||||
sources: tuple[str, ...] = DEFAULT_SOURCES,
|
||||
only_unchunked: bool = False,
|
||||
) -> list:
|
||||
"""`source='gmail'` envelopes, ordered by id for stable --limit/--offset slicing."""
|
||||
query = "SELECT id, ts, raw_ref, entities FROM envelope WHERE source = 'gmail'"
|
||||
params: list = []
|
||||
"""Mail envelopes for `sources`, ordered by id for stable --limit/--offset slicing.
|
||||
|
||||
`only_unchunked` restricts the set to envelopes with no `document_chunk` rows at all —
|
||||
the queue the incremental poller feeds (recon §3.2). It is derived from the data rather
|
||||
than kept as separate state, which makes it self-healing: an interrupted run, a mail
|
||||
skipped while Ollama was down, a message inserted with a backdated header — each stays in
|
||||
the queue until it has chunks, with nothing to reconcile. `--since` deliberately does NOT
|
||||
play that role: it filters on `envelope.ts`, the SENDER's clock, so a mail delivered today
|
||||
with a month-old header date would fall outside the window and never be chunked.
|
||||
"""
|
||||
query = "SELECT id, ts, raw_ref, entities FROM envelope e WHERE source = ANY($1)"
|
||||
params: list = [list(sources)]
|
||||
if since is not None:
|
||||
params.append(since)
|
||||
query += f" AND ts >= ${len(params)}"
|
||||
if only_unchunked:
|
||||
query += (
|
||||
" AND NOT EXISTS ("
|
||||
"SELECT 1 FROM document_chunk c WHERE c.envelope_id = e.id)"
|
||||
)
|
||||
query += " ORDER BY id"
|
||||
if limit is not None:
|
||||
params.append(limit)
|
||||
|
|
@ -384,9 +406,27 @@ async def fetch_envelopes(
|
|||
return await conn.fetch(query, *params)
|
||||
|
||||
|
||||
async def fetch_existing_chunk_keys(conn: asyncpg.Connection, model: str) -> set[tuple[str, int]]:
|
||||
async def fetch_existing_chunk_keys(
|
||||
conn: asyncpg.Connection, model: str, envelope_ids: Optional[list[str]] = None
|
||||
) -> set[tuple[str, int]]:
|
||||
"""(envelope_id, chunk_index) pairs already inserted with this model — idempotency + dry-run
|
||||
preview. The query filters `WHERE model = $1`, so the pair need not redundantly carry model."""
|
||||
preview. The query filters `WHERE model = $1`, so the pair need not redundantly carry model.
|
||||
|
||||
`envelope_ids` narrows the fetch to the current working set. Unbounded, this reads every
|
||||
chunk key in the corpus: measured 2026-08-06 at 389 012 rows, ~1.0 s server-side, ~26 MB on
|
||||
the wire and a 60-90 MB Python set — irrelevant once for a 50k backfill slice, wasteful
|
||||
every two hours on a node with 2.4 GB available, and growing with the corpus (recon
|
||||
§2.5 iv). None keeps the old unbounded behaviour for the manual full-corpus runs.
|
||||
"""
|
||||
if envelope_ids is not None:
|
||||
if not envelope_ids:
|
||||
return set()
|
||||
rows = await conn.fetch(
|
||||
"SELECT envelope_id, chunk_index FROM document_chunk "
|
||||
"WHERE model = $1 AND envelope_id = ANY($2)",
|
||||
model, envelope_ids,
|
||||
)
|
||||
else:
|
||||
rows = await conn.fetch(
|
||||
"SELECT envelope_id, chunk_index FROM document_chunk WHERE model = $1", model
|
||||
)
|
||||
|
|
@ -458,14 +498,16 @@ async def run(
|
|||
limit: Optional[int] = None,
|
||||
offset: Optional[int] = None,
|
||||
apply: bool = False,
|
||||
sources: tuple[str, ...] = DEFAULT_SOURCES,
|
||||
only_unchunked: bool = False,
|
||||
batch_size: int = DEFAULT_BATCH_SIZE,
|
||||
max_embed_failures: int = DEFAULT_MAX_EMBED_FAILURES,
|
||||
embed_retries: int = DEFAULT_EMBED_RETRIES,
|
||||
embed_backoff_s: float = DEFAULT_EMBED_BACKOFF_S,
|
||||
embed_timeout_s: float = DEFAULT_EMBED_TIMEOUT_S,
|
||||
) -> dict:
|
||||
"""Process one --limit/--offset (optionally --since-filtered) slice of `source='gmail'`
|
||||
envelopes. dry-run (apply=False): parse + quote-strip + classify + chunk + count, zero
|
||||
"""Process one --limit/--offset (optionally --since-filtered) slice of mail envelopes for
|
||||
`sources`. dry-run (apply=False): parse + quote-strip + classify + chunk + count, zero
|
||||
Ollama calls, zero writes (including the threading UPDATE). Stats must always balance:
|
||||
|
||||
mails_scanned = missing_file + read_errors + parse_errors + body_empty + mails_chunked
|
||||
|
|
@ -475,8 +517,12 @@ async def run(
|
|||
stats = _new_stats()
|
||||
conn = await asyncpg.connect(dsn)
|
||||
try:
|
||||
envelopes = await fetch_envelopes(conn, since, limit, offset)
|
||||
existing = await fetch_existing_chunk_keys(conn, model)
|
||||
envelopes = await fetch_envelopes(
|
||||
conn, since, limit, offset, sources=sources, only_unchunked=only_unchunked
|
||||
)
|
||||
existing = await fetch_existing_chunk_keys(
|
||||
conn, model, envelope_ids=[row["id"] for row in envelopes]
|
||||
)
|
||||
|
||||
session: Optional[aiohttp.ClientSession] = None
|
||||
if apply:
|
||||
|
|
@ -692,6 +738,13 @@ def _parse_since(value: str) -> datetime:
|
|||
return datetime.strptime(value, "%Y-%m-%d").replace(tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _parse_sources(value: str) -> tuple[str, ...]:
|
||||
sources = tuple(part.strip() for part in value.split(",") if part.strip())
|
||||
if not sources:
|
||||
raise SystemExit("--sources must name at least one envelope source")
|
||||
return sources
|
||||
|
||||
|
||||
def _env_num(name: str, default, cast):
|
||||
"""Env override for a numeric CLI default. A malformed value is a loud startup failure, not a
|
||||
silent fallback to the default — a typo'd MAIL_INGEST_BATCH_SIZE must not quietly produce a
|
||||
|
|
@ -707,7 +760,7 @@ def _env_num(name: str, default, cast):
|
|||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Chunk + embed source='gmail' envelope body content into document_chunk "
|
||||
description="Chunk + embed mail envelope body content into document_chunk "
|
||||
"(module 5, faza mailowa — plan §5, Krok 2)."
|
||||
)
|
||||
parser.add_argument("--dsn", default=os.environ.get("KB_DSN"),
|
||||
|
|
@ -718,8 +771,17 @@ def main() -> None:
|
|||
help=f"Ollama base URL (default: {DEFAULT_OLLAMA_URL}, or set OLLAMA_URL)")
|
||||
parser.add_argument("--model", default=os.environ.get("OLLAMA_EMBED_MODEL", DEFAULT_MODEL),
|
||||
help=f"Ollama embedding model (default: {DEFAULT_MODEL})")
|
||||
parser.add_argument("--sources", type=_parse_sources, default=DEFAULT_SOURCES,
|
||||
metavar="A,B",
|
||||
help=f"Comma-separated envelope sources to process "
|
||||
f"(default: {','.join(DEFAULT_SOURCES)})")
|
||||
parser.add_argument("--only-unchunked", action="store_true",
|
||||
help="Only envelopes that have no document_chunk rows at all — the "
|
||||
"queue mail-imap-sync feeds. This is the cyclic mode; without it "
|
||||
"the job re-scans the whole slice and relies on the chunk-key set.")
|
||||
parser.add_argument("--since", type=_parse_since, default=None, metavar="YYYY-MM-DD",
|
||||
help="Only envelopes with ts >= this date (for staged runs, plan Decyzja 9)")
|
||||
help="Only envelopes with ts >= this date (for staged runs, plan Decyzja 9). "
|
||||
"Never use it as an incremental cursor — ts is the sender's clock.")
|
||||
parser.add_argument("--limit", type=int, default=None,
|
||||
help="Max envelopes to process (default: all)")
|
||||
parser.add_argument("--offset", type=int, default=0,
|
||||
|
|
@ -772,6 +834,8 @@ def main() -> None:
|
|||
limit=args.limit,
|
||||
offset=args.offset,
|
||||
apply=args.apply,
|
||||
sources=args.sources,
|
||||
only_unchunked=args.only_unchunked,
|
||||
batch_size=args.batch_size,
|
||||
max_embed_failures=args.max_embed_failures,
|
||||
embed_retries=args.embed_retries,
|
||||
|
|
|
|||
|
|
@ -304,19 +304,59 @@ class TestFetchHelpers:
|
|||
assert "ORDER BY id" in query
|
||||
assert "LIMIT" not in query and "OFFSET" not in query
|
||||
|
||||
async def test_fetch_envelopes_defaults_to_both_live_mail_sources(self):
|
||||
# Decyzja (g): fastmail must be in the default set from the commit that creates the
|
||||
# source, or its envelopes chunk correctly and stay invisible.
|
||||
conn = _FakeConn(envelopes=[])
|
||||
await fetch_envelopes(conn)
|
||||
query, params = conn.queries[-1]
|
||||
assert "source = ANY($1)" in query
|
||||
assert params[0] == ["gmail", "fastmail"]
|
||||
|
||||
async def test_fetch_envelopes_with_since_limit_offset(self):
|
||||
conn = _FakeConn(envelopes=[])
|
||||
await fetch_envelopes(conn, since=datetime(2025, 7, 1, tzinfo=timezone.utc), limit=10, offset=5)
|
||||
query, params = conn.queries[-1]
|
||||
assert "ts >= $1" in query
|
||||
assert "LIMIT $2" in query
|
||||
assert "OFFSET $3" in query
|
||||
assert params == (datetime(2025, 7, 1, tzinfo=timezone.utc), 10, 5)
|
||||
assert "ts >= $2" in query
|
||||
assert "LIMIT $3" in query
|
||||
assert "OFFSET $4" in query
|
||||
assert params == (["gmail", "fastmail"], datetime(2025, 7, 1, tzinfo=timezone.utc), 10, 5)
|
||||
|
||||
async def test_fetch_envelopes_explicit_sources_override_the_default(self):
|
||||
conn = _FakeConn(envelopes=[])
|
||||
await fetch_envelopes(conn, sources=("gmail",))
|
||||
assert conn.queries[-1][1][0] == ["gmail"]
|
||||
|
||||
async def test_only_unchunked_adds_the_queue_predicate(self):
|
||||
conn = _FakeConn(envelopes=[])
|
||||
await fetch_envelopes(conn, only_unchunked=True)
|
||||
query, _ = conn.queries[-1]
|
||||
assert "NOT EXISTS" in query and "document_chunk c" in query
|
||||
|
||||
async def test_only_unchunked_is_off_by_default(self):
|
||||
conn = _FakeConn(envelopes=[])
|
||||
await fetch_envelopes(conn)
|
||||
assert "NOT EXISTS" not in conn.queries[-1][0]
|
||||
|
||||
async def test_fetch_existing_chunk_keys(self):
|
||||
conn = _FakeConn(existing_keys=[("m1@x", 0), ("m1@x", 1)])
|
||||
keys = await fetch_existing_chunk_keys(conn, model="bge-m3")
|
||||
assert keys == {("m1@x", 0), ("m1@x", 1)}
|
||||
assert "envelope_id = ANY" not in conn.queries[-1][0]
|
||||
|
||||
async def test_fetch_existing_chunk_keys_narrows_to_the_working_set(self):
|
||||
# Unbounded this reads all 389k chunk keys (~26 MB, ~1.0 s) every run — fine once for
|
||||
# a backfill slice, wasteful every two hours on the cyclic path (recon §2.5 iv).
|
||||
conn = _FakeConn(existing_keys=[("m1@x", 0)])
|
||||
await fetch_existing_chunk_keys(conn, model="bge-m3", envelope_ids=["m1@x"])
|
||||
query, params = conn.queries[-1]
|
||||
assert "envelope_id = ANY($2)" in query
|
||||
assert params == ("bge-m3", ["m1@x"])
|
||||
|
||||
async def test_fetch_existing_chunk_keys_short_circuits_on_an_empty_working_set(self):
|
||||
conn = _FakeConn(existing_keys=[("m1@x", 0)])
|
||||
assert await fetch_existing_chunk_keys(conn, model="bge-m3", envelope_ids=[]) == set()
|
||||
assert conn.queries == []
|
||||
|
||||
|
||||
def _env_row(envelope_id: str, subject: str, ts: datetime = datetime(2025, 8, 1, tzinfo=timezone.utc)):
|
||||
|
|
|
|||
71
jobs/mail-imap-sync/env.example
Normal file
71
jobs/mail-imap-sync/env.example
Normal file
|
|
@ -0,0 +1,71 @@
|
|||
# mail-imap-sync configuration — placeholders only. NEVER commit real values.
|
||||
#
|
||||
# On PIHA these keys are APPENDED to the existing /opt/homelab/kb/.env, which is already
|
||||
# root-owned 0600 and already holds KB_DSN, PAPERLESS_API_TOKEN and ANTHROPIC_API_KEY
|
||||
# (recon Decyzja (c)). systemd reads EnvironmentFile as root BEFORE dropping to
|
||||
# User=oskar, so the secrets reach the process without being readable by `oskar` at rest —
|
||||
# a better property than a 0600 file owned by the job's own user. Keep it that way.
|
||||
#
|
||||
# Rules, all learned the hard way in this repo:
|
||||
# 1. Never on the command line. This job has no --user/--password flag on purpose;
|
||||
# argv shows up in `ps` and in shell history.
|
||||
# 2. Never echoed to a terminal. Two secrets have already had to be rotated after
|
||||
# leaking into a session transcript (docs/sessions/2026-07-15.md,
|
||||
# docs/sessions/2026-07-21.md). Install them with an editor, not `echo >>`.
|
||||
# 3. App passwords, not account passwords — revocable one at a time, no access to
|
||||
# account settings, no bypassing 2FA.
|
||||
#
|
||||
# Full setup, including how to mint the app passwords: kb/runbooks/mail-sync-run.md
|
||||
|
||||
# --- which mailboxes to sync -------------------------------------------------------
|
||||
# Required. Comma-separated; each name is also the envelope.source value and the
|
||||
# mail_sync_state.account key. Start with one account if the other is not set up yet.
|
||||
MAIL_ACCOUNTS=gmail,fastmail
|
||||
|
||||
# --- gmail -------------------------------------------------------------------------
|
||||
# App password requires 2FA on the account. Paste it WITHOUT the spaces Google displays.
|
||||
MAIL_GMAIL_USER=you@gmail.com
|
||||
MAIL_GMAIL_APP_PASSWORD=change-me-16-char-app-password
|
||||
# Optional, defaults shown. The folder is chosen by SPECIAL-USE attribute, never by name:
|
||||
# with a Polish UI the mailbox is called "[Gmail]/Wszystkie", and a hardcoded
|
||||
# "[Gmail]/All Mail" would sync nothing while reporting success every hour.
|
||||
# MAIL_GMAIL_HOST=imap.gmail.com
|
||||
# MAIL_GMAIL_PORT=993
|
||||
# MAIL_GMAIL_SPECIAL_USE=\All
|
||||
# MAIL_GMAIL_FOLDERS=
|
||||
#
|
||||
# What the FIRST tick on a folder does — new-only | since | full. Ignored once a
|
||||
# mail_sync_state row exists for that folder.
|
||||
# new-only : start the corpus here, fetch nothing older.
|
||||
# since : fetch everything the server RECEIVED on/after MAIL_*_INITIAL_SINCE.
|
||||
# full : sweep the whole folder (dedup absorbs what is already in the DB).
|
||||
# For gmail, 'since' is the point of the exercise: the corpus stops at 2026-06-19, so a
|
||||
# 'new-only' first tick would leave that gap unfilled forever. Pick a date a few days
|
||||
# BEFORE the last envelope in the DB — dedup makes the overlap free.
|
||||
MAIL_GMAIL_INITIAL_MODE=since
|
||||
MAIL_GMAIL_INITIAL_SINCE=2026-06-15
|
||||
|
||||
# --- fastmail ----------------------------------------------------------------------
|
||||
# Fastmail app passwords can be scoped to IMAP only — use that, not a full-access one.
|
||||
MAIL_FASTMAIL_USER=you@fastmail.com
|
||||
MAIL_FASTMAIL_APP_PASSWORD=change-me-app-password
|
||||
# Optional, defaults shown. Fastmail has no \All equivalent, so the scope is a literal
|
||||
# list. Spam/Trash/Drafts are excluded on purpose (defined noise; drafts have no stable
|
||||
# Message-ID and mutate).
|
||||
# MAIL_FASTMAIL_HOST=imap.fastmail.com
|
||||
# MAIL_FASTMAIL_PORT=993
|
||||
# MAIL_FASTMAIL_FOLDERS=INBOX,Archive,Sent
|
||||
#
|
||||
# Fastmail is greenfield — zero rows, zero archive files. Whether to pull its history is
|
||||
# a decision the recon deliberately leaves open until measured:
|
||||
# mail-imap-sync --measure
|
||||
# prints MESSAGES per folder. Decide 'new-only' vs 'full' from that number, then set it
|
||||
# here BEFORE the first --apply run (after the first run the cursor exists and this key
|
||||
# no longer has any effect).
|
||||
MAIL_FASTMAIL_INITIAL_MODE=new-only
|
||||
|
||||
# --- shared ------------------------------------------------------------------------
|
||||
# Already present in /opt/homelab/kb/.env on PIHA — listed here for completeness.
|
||||
# KB_DSN=postgresql://kb:CHANGE-ME@localhost:5433/kb
|
||||
# MAIL_ARCHIVE_ROOT=/home/oskar/kb/mail/archive
|
||||
# KB_MAIL_SYNC_PROM_PATH=/opt/homelab/state/node-exporter/kb-mail-sync.prom
|
||||
31
jobs/mail-imap-sync/pyproject.toml
Normal file
31
jobs/mail-imap-sync/pyproject.toml
Normal file
|
|
@ -0,0 +1,31 @@
|
|||
[build-system]
|
||||
requires = ["setuptools>=68"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "mail-imap-sync"
|
||||
version = "0.1.0"
|
||||
requires-python = ">=3.11"
|
||||
# No IMAP dependency: imaplib is stdlib (PIHA runs 3.11.2). Recon Decyzja (a) — a synchronous
|
||||
# stdlib client is the right size for ~37 messages a day, and adds nothing to keep patched.
|
||||
dependencies = [
|
||||
"asyncpg>=0.29",
|
||||
"structlog>=24.1",
|
||||
"kb-mail",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
dev = [
|
||||
"pytest>=8.1",
|
||||
"pytest-asyncio>=0.23",
|
||||
]
|
||||
|
||||
[project.scripts]
|
||||
mail-imap-sync = "mail_imap_sync.sync:main"
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["src"]
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
asyncio_mode = "auto"
|
||||
testpaths = ["tests"]
|
||||
0
jobs/mail-imap-sync/src/mail_imap_sync/__init__.py
Normal file
0
jobs/mail-imap-sync/src/mail_imap_sync/__init__.py
Normal file
141
jobs/mail-imap-sync/src/mail_imap_sync/config.py
Normal file
141
jobs/mail-imap-sync/src/mail_imap_sync/config.py
Normal file
|
|
@ -0,0 +1,141 @@
|
|||
"""Account configuration from the environment — and only from the environment.
|
||||
|
||||
Decyzja (c) of `kb/audits/mail-sync-2026-08-06.md`: credentials come from
|
||||
`/opt/homelab/kb/.env` (root-owned `600`, read by systemd as root before dropping to
|
||||
`User=oskar`), never from argv. This job therefore has no `--password` flag and no
|
||||
`--user` flag at all — `--dsn <secret>` lands in `ps` output and in shell history, and the
|
||||
repo has already had to rotate two secrets that leaked into a session transcript.
|
||||
|
||||
Defaults encode the approved folder scope (Decyzja (e)) so a working config is short:
|
||||
|
||||
gmail -> SPECIAL-USE \\All (one fetch per message despite many labels;
|
||||
same population as the Takeout the corpus rests on)
|
||||
fastmail -> INBOX + Archive + Sent (no \\All equivalent in a classic layout)
|
||||
|
||||
Spam, Trash and Drafts are excluded from both: spam and trash are defined noise and were not
|
||||
in the Takeout either, and drafts have no stable Message-ID — they mutate.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from datetime import date, datetime
|
||||
from pathlib import Path
|
||||
from typing import Mapping, Optional
|
||||
|
||||
from kb_mail.imap import DEFAULT_PORT, INITIAL_MODES, ImapAccount
|
||||
|
||||
DEFAULT_ARCHIVE_ROOT = Path("/home/oskar/kb/mail/archive")
|
||||
|
||||
# Only a default. `MAIL_<ACCOUNT>_HOST` overrides it, and an account not listed here simply
|
||||
# has to set its host — nothing in the code special-cases a provider.
|
||||
KNOWN_HOSTS = {
|
||||
"gmail": "imap.gmail.com",
|
||||
"fastmail": "imap.fastmail.com",
|
||||
}
|
||||
|
||||
DEFAULT_SCOPE = {
|
||||
"gmail": {"special_use": ("\\All",), "folders": ()},
|
||||
"fastmail": {"special_use": (), "folders": ("INBOX", "Archive", "Sent")},
|
||||
}
|
||||
|
||||
|
||||
class ConfigError(RuntimeError):
|
||||
"""A missing or malformed setting. Always fatal at startup: an unattended hourly job
|
||||
that silently syncs a subset of what the operator configured is worse than one that
|
||||
refuses to start."""
|
||||
|
||||
|
||||
def _csv(value: Optional[str]) -> Optional[tuple[str, ...]]:
|
||||
"""Comma-separated list. An explicitly EMPTY value means "none" (a real override);
|
||||
an unset value means "no opinion" and returns None so a default can win."""
|
||||
if value is None:
|
||||
return None
|
||||
return tuple(part.strip() for part in value.split(",") if part.strip())
|
||||
|
||||
|
||||
def _parse_date(value: str, key: str) -> date:
|
||||
try:
|
||||
return datetime.strptime(value.strip(), "%Y-%m-%d").date()
|
||||
except ValueError:
|
||||
raise ConfigError(f"{key}={value!r} is not a date (expected YYYY-MM-DD)")
|
||||
|
||||
|
||||
def account_from_env(name: str, env: Mapping[str, str]) -> ImapAccount:
|
||||
"""Build one account from `MAIL_<NAME>_*` keys. `name` doubles as `envelope.source`."""
|
||||
prefix = f"MAIL_{name.upper().replace('-', '_')}_"
|
||||
|
||||
def get(key: str) -> Optional[str]:
|
||||
return env.get(prefix + key)
|
||||
|
||||
user = get("USER")
|
||||
password = get("APP_PASSWORD")
|
||||
if not user:
|
||||
raise ConfigError(f"{prefix}USER is not set")
|
||||
if not password:
|
||||
raise ConfigError(f"{prefix}APP_PASSWORD is not set")
|
||||
|
||||
host = get("HOST") or KNOWN_HOSTS.get(name)
|
||||
if not host:
|
||||
raise ConfigError(f"{prefix}HOST is not set and {name!r} has no known default")
|
||||
|
||||
port_raw = get("PORT")
|
||||
try:
|
||||
port = int(port_raw) if port_raw else DEFAULT_PORT
|
||||
except ValueError:
|
||||
raise ConfigError(f"{prefix}PORT={port_raw!r} is not an integer")
|
||||
|
||||
scope = DEFAULT_SCOPE.get(name, {"special_use": (), "folders": ()})
|
||||
special_use = _csv(get("SPECIAL_USE"))
|
||||
folders = _csv(get("FOLDERS"))
|
||||
|
||||
initial_mode = (get("INITIAL_MODE") or "new-only").strip()
|
||||
if initial_mode not in INITIAL_MODES:
|
||||
raise ConfigError(
|
||||
f"{prefix}INITIAL_MODE={initial_mode!r} must be one of {list(INITIAL_MODES)}"
|
||||
)
|
||||
since_raw = get("INITIAL_SINCE")
|
||||
initial_since = _parse_date(since_raw, prefix + "INITIAL_SINCE") if since_raw else None
|
||||
|
||||
try:
|
||||
return ImapAccount(
|
||||
name=name,
|
||||
host=host,
|
||||
port=port,
|
||||
user=user,
|
||||
password=password,
|
||||
special_use=special_use if special_use is not None else scope["special_use"],
|
||||
folders=folders if folders is not None else scope["folders"],
|
||||
initial_mode=initial_mode,
|
||||
initial_since=initial_since,
|
||||
)
|
||||
except ValueError as exc:
|
||||
raise ConfigError(str(exc))
|
||||
|
||||
|
||||
def accounts_from_env(env: Optional[Mapping[str, str]] = None) -> list[ImapAccount]:
|
||||
"""Every account named in `MAIL_ACCOUNTS`, in the order listed.
|
||||
|
||||
`MAIL_ACCOUNTS` is required rather than defaulted to "gmail,fastmail": which mailboxes
|
||||
are live is a fact about the deployment, and inferring it from which credentials happen
|
||||
to be present would turn a typo'd variable name into a silently half-synced corpus.
|
||||
"""
|
||||
env = env if env is not None else os.environ
|
||||
names = _csv(env.get("MAIL_ACCOUNTS"))
|
||||
if not names:
|
||||
raise ConfigError(
|
||||
"MAIL_ACCOUNTS is not set — list the accounts to sync, e.g. 'gmail,fastmail'"
|
||||
)
|
||||
|
||||
seen: set[str] = set()
|
||||
accounts = []
|
||||
for name in names:
|
||||
if name in seen:
|
||||
raise ConfigError(f"MAIL_ACCOUNTS lists {name!r} twice")
|
||||
seen.add(name)
|
||||
accounts.append(account_from_env(name, env))
|
||||
return accounts
|
||||
|
||||
|
||||
def archive_root_from_env(env: Optional[Mapping[str, str]] = None) -> Path:
|
||||
env = env if env is not None else os.environ
|
||||
return Path(env.get("MAIL_ARCHIVE_ROOT") or DEFAULT_ARCHIVE_ROOT)
|
||||
102
jobs/mail-imap-sync/src/mail_imap_sync/prom.py
Normal file
102
jobs/mail-imap-sync/src/mail_imap_sync/prom.py
Normal file
|
|
@ -0,0 +1,102 @@
|
|||
"""Prometheus textfile-collector output for this job.
|
||||
|
||||
Same delivery path as `kb-ingest` (recon §3.4, no new moving parts): the job writes a `.prom`
|
||||
file atomically into `/opt/homelab/state/node-exporter/`, node_exporter@PIHA picks it up
|
||||
through its existing `/:/host:ro` mount, and brain-watchdog polls Prometheus' `/api/v1/alerts`
|
||||
for Telegram delivery. No Alertmanager, deliberately.
|
||||
|
||||
`documents_ingest.cyclic_ingest` has an equivalent unlabelled renderer. It is not imported
|
||||
here: this job must not depend on the documents-ingest package (aiohttp, anthropic, a
|
||||
paperless client) to write eight numbers, and these metrics are per-account labelled, which
|
||||
that renderer cannot express. Folding both into one shared helper is worth doing the next
|
||||
time either changes — noted in kb/services/job-mail-imap-sync.md.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
DEFAULT_PROM_PATH = Path("/opt/homelab/state/node-exporter/kb-mail-sync.prom")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Sample:
|
||||
value: float
|
||||
labels: dict = field(default_factory=dict)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class MetricFamily:
|
||||
name: str
|
||||
type: str
|
||||
help: str
|
||||
samples: list = field(default_factory=list)
|
||||
|
||||
|
||||
def _escape_label(value: str) -> str:
|
||||
return value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
|
||||
|
||||
|
||||
def _format_value(value: float) -> str:
|
||||
"""Integers render without a trailing `.0` — Prometheus does not care, humans reading
|
||||
`cat kb-mail-sync.prom` during a first run do."""
|
||||
if isinstance(value, bool):
|
||||
return str(int(value))
|
||||
if isinstance(value, int) or (isinstance(value, float) and value.is_integer()):
|
||||
return str(int(value))
|
||||
return repr(float(value))
|
||||
|
||||
|
||||
def render_prom(families: list[MetricFamily]) -> str:
|
||||
lines: list[str] = []
|
||||
for family in families:
|
||||
if not family.samples:
|
||||
# A family with no samples is omitted rather than zeroed: "no accounts reported"
|
||||
# and "accounts reported zero" are different facts and must not look the same.
|
||||
continue
|
||||
lines.append(f"# HELP {family.name} {family.help}")
|
||||
lines.append(f"# TYPE {family.name} {family.type}")
|
||||
for sample in family.samples:
|
||||
if sample.labels:
|
||||
rendered = ",".join(
|
||||
f'{k}="{_escape_label(str(v))}"' for k, v in sorted(sample.labels.items())
|
||||
)
|
||||
lines.append(f"{family.name}{{{rendered}}} {_format_value(sample.value)}")
|
||||
else:
|
||||
lines.append(f"{family.name} {_format_value(sample.value)}")
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
def read_prev_value(path: Path, name: str) -> Optional[float]:
|
||||
"""Previous value of an unlabelled metric, or None.
|
||||
|
||||
Used to carry `kb_mail_sync_last_success_timestamp` forward across a failed run — the
|
||||
staleness signal must measure "time since the last good tick", not get reset to zero by
|
||||
the first transient error.
|
||||
"""
|
||||
if not path.exists():
|
||||
return None
|
||||
try:
|
||||
text = path.read_text()
|
||||
except OSError:
|
||||
return None
|
||||
pattern = re.compile(rf"^{re.escape(name)}\s+([0-9.eE+-]+)\s*$")
|
||||
for line in text.splitlines():
|
||||
match = pattern.match(line)
|
||||
if match:
|
||||
try:
|
||||
return float(match.group(1))
|
||||
except ValueError:
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def write_prom_atomic(path: Path, content: str) -> None:
|
||||
"""tmp-write + rename, so the textfile collector never reads a half-written file."""
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
tmp_path = path.with_name(f"{path.name}.tmp{os.getpid()}")
|
||||
tmp_path.write_text(content)
|
||||
tmp_path.replace(path)
|
||||
593
jobs/mail-imap-sync/src/mail_imap_sync/sync.py
Normal file
593
jobs/mail-imap-sync/src/mail_imap_sync/sync.py
Normal file
|
|
@ -0,0 +1,593 @@
|
|||
"""IMAP incremental sync — one job, two accounts (gmail + fastmail).
|
||||
|
||||
Krok 7 of the mail phase (`kb/phases/kb-m5-faza-mailowa.md` §10), built to the approved recon
|
||||
`kb/audits/mail-sync-2026-08-06.md`. Until this job existed, the mail corpus was a photograph:
|
||||
225 030 envelopes ending on 2026-06-19, with nothing in the repo that talks to a mail server.
|
||||
|
||||
What one tick does, per account and per folder:
|
||||
|
||||
EXAMINE folder (read-only) -> UIDVALIDITY, UIDNEXT
|
||||
plan the tick -> first contact / incremental / UIDVALIDITY reset
|
||||
UID SEARCH -> the UIDs this tick owns
|
||||
per UID, in ascending order:
|
||||
UID FETCH (BODY.PEEK[]) -> raw bytes, \\Seen untouched
|
||||
save_eml() -> append-only; FileExistsError just means "already have it"
|
||||
insert_envelope() -> entities = [headers, attachment...], ON CONFLICT DO NOTHING
|
||||
UPDATE mail_sync_state -> only over the contiguous prefix of fully durable messages
|
||||
|
||||
**This job does not embed and does not chunk.** That split is deliberate (recon §3.2):
|
||||
fetching is network-bound and runs hourly on PIHA, which is up 24/7; chunking and embedding
|
||||
need Ollama on SOLARIA, which is powered off ~16 h a day by design. Coupling them would mean
|
||||
mail only arrives when the desktop happens to be on. The handoff needs no queue of its own —
|
||||
an envelope with no `document_chunk` rows *is* the queue, which `mail-body-ingest
|
||||
--only-unchunked` drains whenever the GPU is reachable. That is self-healing in a way a
|
||||
timestamp cursor is not: an interrupted run, a mail inserted with a backdated header, a
|
||||
message skipped while Ollama was down — all of them simply stay in the queue.
|
||||
|
||||
Two things this job inserts that the bulk importer did not, both from recon §2.5:
|
||||
|
||||
* `entities[type=headers]` at INSERT time. Historical envelopes only have headers because a
|
||||
separate backfill job walked all 225 030 of them later; without them
|
||||
`mail_body_ingest.build_prefix` silently produces `Temat: (brak tematu) | Od: ?` — no error,
|
||||
just worse retrieval forever.
|
||||
* A count of Message-ID collisions ACROSS accounts. `envelope.id` is a bare Message-ID and so
|
||||
is globally unique: a mail present in both mailboxes is stored once, under whichever account
|
||||
inserted it first. That is right for the index, but it makes "new fastmail mails" read low,
|
||||
so the phenomenon is counted (`envelopes_conflict_other_source`) rather than left invisible.
|
||||
|
||||
Credentials come from the environment only (Decyzja (c)) — there is no `--password` flag.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import email
|
||||
import email.policy
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Callable, Optional
|
||||
|
||||
import asyncpg
|
||||
import structlog
|
||||
|
||||
from kb_mail.archive import save_eml
|
||||
from kb_mail.db import envelope_source, insert_envelope, rows_affected
|
||||
from kb_mail.envelope import Envelope
|
||||
from kb_mail.headers import parse_headers_resilient
|
||||
from kb_mail.imap import ImapAccount, ImapClient, ImapError
|
||||
from kb_mail.message import eml_ref, message_id, parse_attachments, parse_date
|
||||
from kb_mail.sync_state import (
|
||||
SEARCH_ALL,
|
||||
SEARCH_FROM_UID,
|
||||
SEARCH_NONE,
|
||||
SEARCH_SINCE,
|
||||
contiguous_last_uid,
|
||||
get_folder_state,
|
||||
plan_folder_sync,
|
||||
upsert_folder_state,
|
||||
)
|
||||
|
||||
from mail_imap_sync.config import DEFAULT_ARCHIVE_ROOT, ConfigError, accounts_from_env
|
||||
from mail_imap_sync.prom import (
|
||||
DEFAULT_PROM_PATH,
|
||||
MetricFamily,
|
||||
Sample,
|
||||
read_prev_value,
|
||||
render_prom,
|
||||
write_prom_atomic,
|
||||
)
|
||||
|
||||
_log = structlog.get_logger(__name__)
|
||||
|
||||
EXIT_CONFIG_ERROR = 2
|
||||
|
||||
_ACCOUNT_COUNTERS = (
|
||||
# Message-level outcomes — these three groups must balance (see `balance_errors`).
|
||||
"uids_seen",
|
||||
"processed",
|
||||
"vanished",
|
||||
"errors",
|
||||
"archived",
|
||||
"archive_exists",
|
||||
"envelopes_inserted",
|
||||
"envelopes_skipped_dup",
|
||||
"envelopes_conflict_other_source",
|
||||
# Labels and diagnostics — deliberately outside the balance equations.
|
||||
"headers_fallback",
|
||||
"uidvalidity_resets",
|
||||
"folder_errors",
|
||||
"account_errors",
|
||||
)
|
||||
|
||||
|
||||
def _new_counters() -> dict:
|
||||
return {name: 0 for name in _ACCOUNT_COUNTERS}
|
||||
|
||||
|
||||
def _add(into: dict, other: dict) -> None:
|
||||
for key, value in other.items():
|
||||
into[key] = into.get(key, 0) + value
|
||||
|
||||
|
||||
def total_errors(counters: dict) -> int:
|
||||
"""Message-, folder- and account-level failures together. Kept separate in the counters
|
||||
because only the message-level one participates in the balance equations — a folder that
|
||||
blew up on EXAMINE never produced a UID to account for."""
|
||||
return counters["errors"] + counters["folder_errors"] + counters["account_errors"]
|
||||
|
||||
|
||||
def balance_errors(counters: dict) -> list[str]:
|
||||
"""The three identities an apply run must satisfy. Returned rather than logged so both the
|
||||
summary line and the exit code can use them — a run whose counters do not add up has not
|
||||
told the truth about what it did, whatever else it reports."""
|
||||
problems = []
|
||||
if counters["uids_seen"] != counters["processed"] + counters["vanished"] + counters["errors"]:
|
||||
problems.append("uids_seen != processed + vanished + errors")
|
||||
if counters["processed"] != counters["archived"] + counters["archive_exists"]:
|
||||
problems.append("processed != archived + archive_exists")
|
||||
envelope_outcomes = (
|
||||
counters["envelopes_inserted"]
|
||||
+ counters["envelopes_skipped_dup"]
|
||||
+ counters["envelopes_conflict_other_source"]
|
||||
)
|
||||
if counters["processed"] != envelope_outcomes:
|
||||
problems.append("processed != inserted + skipped_dup + conflict_other_source")
|
||||
return problems
|
||||
|
||||
|
||||
def _parse_message(raw: bytes):
|
||||
"""compat32, matching `gmail-bulk-import`'s mailbox parse exactly.
|
||||
|
||||
Not a stylistic choice: `message_id()` runs over this object, and the ids it produces must
|
||||
be byte-identical to the 225 030 already in the table or dedup quietly stops working.
|
||||
"""
|
||||
return email.message_from_bytes(raw, policy=email.policy.compat32)
|
||||
|
||||
|
||||
async def _sync_folder(
|
||||
conn: asyncpg.Connection,
|
||||
client: ImapClient,
|
||||
account: ImapAccount,
|
||||
folder: str,
|
||||
archive_root: Path,
|
||||
apply: bool,
|
||||
limit: Optional[int],
|
||||
) -> dict:
|
||||
"""One folder, one tick. Returns a per-folder report; raises only on protocol failures
|
||||
that make the folder's state meaningless (those are caught one level up, per account)."""
|
||||
counters = _new_counters()
|
||||
status = await asyncio.to_thread(client.examine, folder)
|
||||
|
||||
state = await get_folder_state(conn, account.name, folder)
|
||||
plan = plan_folder_sync(account, status, state)
|
||||
if plan.uidvalidity_reset:
|
||||
counters["uidvalidity_resets"] += 1
|
||||
|
||||
if plan.search == SEARCH_NONE:
|
||||
uids: list[int] = []
|
||||
elif plan.search == SEARCH_ALL:
|
||||
uids = await asyncio.to_thread(client.search_all)
|
||||
elif plan.search == SEARCH_SINCE:
|
||||
uids = await asyncio.to_thread(client.search_since, plan.since)
|
||||
elif plan.search == SEARCH_FROM_UID:
|
||||
uids = await asyncio.to_thread(client.search_from_uid, plan.start_uid)
|
||||
else: # pragma: no cover — plan_folder_sync only emits the four above
|
||||
raise ImapError(f"unknown search plan {plan.search!r}")
|
||||
|
||||
truncated = False
|
||||
if limit is not None and len(uids) > limit:
|
||||
# Take the LOWEST UIDs: the cursor then advances from the oldest end and the rest is
|
||||
# picked up by the next tick. Taking the newest would strand the older ones behind a
|
||||
# cursor that had already moved past them.
|
||||
uids = uids[:limit]
|
||||
truncated = True
|
||||
|
||||
report = {
|
||||
"account": account.name,
|
||||
"folder": folder,
|
||||
"mode": plan.mode,
|
||||
"uidvalidity": status.uidvalidity,
|
||||
"uidnext": status.uidnext,
|
||||
"state_last_uid": state.last_uid if state else None,
|
||||
"candidates": len(uids),
|
||||
"truncated": truncated,
|
||||
}
|
||||
|
||||
if not apply:
|
||||
# Dry run stops at the plan: EXAMINE + SEARCH answer "what would this tick do" without
|
||||
# pulling a single body off the server or writing a byte. `candidates` is the whole
|
||||
# answer, so no message-level counter moves and the balance check is skipped upstream.
|
||||
report["counters"] = counters
|
||||
report["new_last_uid"] = None
|
||||
_log.info("folder.planned", **report)
|
||||
return report
|
||||
|
||||
succeeded: set[int] = set()
|
||||
for uid in uids:
|
||||
counters["uids_seen"] += 1
|
||||
try:
|
||||
raw = await asyncio.to_thread(client.fetch_message, uid)
|
||||
except Exception:
|
||||
_log.warning("skip.fetch_error", account=account.name, folder=folder,
|
||||
uid=uid, exc_info=True)
|
||||
counters["errors"] += 1
|
||||
continue
|
||||
|
||||
if raw is None:
|
||||
# Expunged between SEARCH and FETCH. Ordinary on a live mailbox, and the UID is
|
||||
# gone for good, so it counts as handled and does not block the cursor.
|
||||
_log.info("skip.vanished", account=account.name, folder=folder, uid=uid)
|
||||
counters["vanished"] += 1
|
||||
succeeded.add(uid)
|
||||
continue
|
||||
|
||||
try:
|
||||
envelope_id, inserted_new = await _store_message(
|
||||
conn, account, archive_root, raw, counters
|
||||
)
|
||||
except Exception:
|
||||
_log.warning("skip.store_error", account=account.name, folder=folder,
|
||||
uid=uid, exc_info=True)
|
||||
counters["errors"] += 1
|
||||
continue
|
||||
|
||||
succeeded.add(uid)
|
||||
_log.debug("message.stored", account=account.name, folder=folder, uid=uid,
|
||||
envelope_id=envelope_id, new=inserted_new)
|
||||
|
||||
if not uids:
|
||||
# Nothing to fetch: adopt the plan's high-water mark (or keep the cursor for an
|
||||
# incremental tick, where baseline_uid IS the stored cursor).
|
||||
new_last_uid = plan.baseline_uid
|
||||
else:
|
||||
# A floor of 0 for any sweep-shaped plan, deliberately NOT plan.baseline_uid: on a
|
||||
# full sweep that equals uidnext-1, and using it here would step the cursor straight
|
||||
# over every message that failed.
|
||||
floor = plan.start_uid - 1 if plan.search == SEARCH_FROM_UID else 0
|
||||
new_last_uid = contiguous_last_uid(floor, uids, succeeded)
|
||||
|
||||
await upsert_folder_state(conn, account.name, folder, status.uidvalidity, new_last_uid)
|
||||
|
||||
report["counters"] = counters
|
||||
report["new_last_uid"] = new_last_uid
|
||||
report["stalled"] = bool(uids) and len(succeeded) != len(uids)
|
||||
_log.info("folder.synced", **{k: v for k, v in report.items() if k != "counters"},
|
||||
**counters)
|
||||
return report
|
||||
|
||||
|
||||
async def _store_message(
|
||||
conn: asyncpg.Connection,
|
||||
account: ImapAccount,
|
||||
archive_root: Path,
|
||||
raw: bytes,
|
||||
counters: dict,
|
||||
) -> tuple[str, bool]:
|
||||
"""Archive then insert — in that order, and only then may the cursor move past this UID.
|
||||
|
||||
An exception anywhere here leaves the cursor where it was, so the next tick refetches this
|
||||
message: the archive write is append-only and the insert is `ON CONFLICT DO NOTHING`, so
|
||||
the repeat costs one `FileExistsError` and one no-op statement.
|
||||
"""
|
||||
msg = _parse_message(raw)
|
||||
envelope_id = message_id(msg)
|
||||
ts = parse_date(msg)
|
||||
|
||||
headers_entity, used_fallback = parse_headers_resilient(raw)
|
||||
if used_fallback:
|
||||
counters["headers_fallback"] += 1
|
||||
entities = [headers_entity] + parse_attachments(msg)
|
||||
|
||||
try:
|
||||
raw_ref = await save_eml(archive_root, envelope_id, account.name, ts, raw)
|
||||
counters["archived"] += 1
|
||||
except FileExistsError:
|
||||
raw_ref = eml_ref(envelope_id, account.name, ts)
|
||||
counters["archive_exists"] += 1
|
||||
|
||||
envelope = Envelope(id=envelope_id, source=account.name, ts=ts,
|
||||
raw_ref=raw_ref, entities=entities)
|
||||
tag = await insert_envelope(conn, envelope)
|
||||
counters["processed"] += 1
|
||||
|
||||
if rows_affected(tag) == 1:
|
||||
counters["envelopes_inserted"] += 1
|
||||
return envelope_id, True
|
||||
|
||||
existing = await envelope_source(conn, envelope_id)
|
||||
if existing is not None and existing != account.name:
|
||||
_log.info("envelope.conflict_other_source", envelope_id=envelope_id,
|
||||
account=account.name, existing_source=existing)
|
||||
counters["envelopes_conflict_other_source"] += 1
|
||||
else:
|
||||
counters["envelopes_skipped_dup"] += 1
|
||||
return envelope_id, False
|
||||
|
||||
|
||||
async def _sync_account(
|
||||
conn: asyncpg.Connection,
|
||||
account: ImapAccount,
|
||||
archive_root: Path,
|
||||
apply: bool,
|
||||
limit: Optional[int],
|
||||
client_factory: Callable[[ImapAccount], ImapClient],
|
||||
) -> dict:
|
||||
"""All of one account's folders. Failures are contained here: a Gmail login that stopped
|
||||
working must not also stop Fastmail from syncing (same isolation as `cyclic_ingest`)."""
|
||||
counters = _new_counters()
|
||||
folders_report: list[dict] = []
|
||||
error: Optional[str] = None
|
||||
client = client_factory(account)
|
||||
|
||||
try:
|
||||
await asyncio.to_thread(client.connect)
|
||||
folders = await asyncio.to_thread(client.resolve_folders)
|
||||
for folder in folders:
|
||||
try:
|
||||
report = await _sync_folder(
|
||||
conn, client, account, folder, archive_root, apply, limit
|
||||
)
|
||||
folders_report.append(report)
|
||||
_add(counters, report["counters"])
|
||||
except Exception as exc:
|
||||
_log.error("folder_failed", account=account.name, folder=folder,
|
||||
error=str(exc), exc_info=True)
|
||||
counters["folder_errors"] += 1
|
||||
folders_report.append({"account": account.name, "folder": folder,
|
||||
"error": str(exc)})
|
||||
except Exception as exc:
|
||||
_log.error("account_failed", account=account.name, error=str(exc), exc_info=True)
|
||||
error = str(exc)
|
||||
counters["account_errors"] += 1
|
||||
finally:
|
||||
await asyncio.to_thread(client.logout)
|
||||
|
||||
return {
|
||||
"account": account.name,
|
||||
"counters": counters,
|
||||
"folders": folders_report,
|
||||
"error": error,
|
||||
}
|
||||
|
||||
|
||||
async def fetch_last_message_ts(
|
||||
conn: asyncpg.Connection, accounts: list[str]
|
||||
) -> dict[str, float]:
|
||||
"""`max(envelope.ts)` per account — the freshness gauge from recon §3.4.
|
||||
|
||||
Purely observational, so a failure here degrades to "metric absent" rather than failing
|
||||
the run: the DB being down is already loudly reported by every other counter.
|
||||
"""
|
||||
try:
|
||||
rows = await conn.fetch(
|
||||
"SELECT source, max(ts) AS newest FROM envelope "
|
||||
"WHERE source = ANY($1) GROUP BY source",
|
||||
accounts,
|
||||
)
|
||||
except Exception:
|
||||
_log.warning("last_message_ts_failed", exc_info=True)
|
||||
return {}
|
||||
return {
|
||||
r["source"]: r["newest"].timestamp()
|
||||
for r in rows
|
||||
if r["newest"] is not None
|
||||
}
|
||||
|
||||
|
||||
async def run(
|
||||
dsn: str,
|
||||
accounts: list[ImapAccount],
|
||||
archive_root: Path = DEFAULT_ARCHIVE_ROOT,
|
||||
apply: bool = False,
|
||||
limit: Optional[int] = None,
|
||||
client_factory: Optional[Callable[[ImapAccount], ImapClient]] = None,
|
||||
) -> dict:
|
||||
"""One tick over every configured account.
|
||||
|
||||
dry-run (`apply=False`) plans and counts: it connects, resolves folders, reads
|
||||
UIDVALIDITY/UIDNEXT and runs the SEARCH, then stops — no FETCH, no archive write, no DB
|
||||
write, no state update. That is enough to answer "how much would the first real run pull",
|
||||
which is exactly the question before a first sync.
|
||||
"""
|
||||
factory = client_factory or (lambda account: ImapClient(account))
|
||||
totals = _new_counters()
|
||||
per_account: dict[str, dict] = {}
|
||||
reports: list[dict] = []
|
||||
|
||||
conn = await asyncpg.connect(dsn)
|
||||
try:
|
||||
for account in accounts:
|
||||
report = await _sync_account(conn, account, archive_root, apply, limit, factory)
|
||||
reports.append(report)
|
||||
per_account[account.name] = report["counters"]
|
||||
_add(totals, report["counters"])
|
||||
|
||||
last_message_ts = await fetch_last_message_ts(conn, [a.name for a in accounts])
|
||||
finally:
|
||||
await conn.close()
|
||||
|
||||
# Only an apply run has message-level outcomes to balance; a dry run never fetched.
|
||||
problems = balance_errors(totals) if apply else []
|
||||
if problems:
|
||||
_log.error("stats_unbalanced", problems=problems, **totals)
|
||||
|
||||
failed = total_errors(totals) > 0 or bool(problems)
|
||||
result = {
|
||||
"totals": totals,
|
||||
"accounts": per_account,
|
||||
"reports": reports,
|
||||
"last_message_ts": last_message_ts,
|
||||
"balance_problems": problems,
|
||||
"failed": failed,
|
||||
}
|
||||
_log.info("run_complete", apply=apply, failed=failed, **totals)
|
||||
return result
|
||||
|
||||
|
||||
async def measure(
|
||||
accounts: list[ImapAccount],
|
||||
client_factory: Optional[Callable[[ImapAccount], ImapClient]] = None,
|
||||
) -> list[dict]:
|
||||
"""`STATUS (MESSAGES UIDNEXT UIDVALIDITY)` for every configured folder. Read-only, no DB.
|
||||
|
||||
This is the measurement Decyzja (e) leaves open: whether to pull Fastmail's history is a
|
||||
decision about a number nobody has yet, and the recon explicitly refuses to guess it.
|
||||
Run this first, then choose `initial_mode` from what it prints.
|
||||
"""
|
||||
factory = client_factory or (lambda account: ImapClient(account))
|
||||
rows: list[dict] = []
|
||||
for account in accounts:
|
||||
client = factory(account)
|
||||
try:
|
||||
await asyncio.to_thread(client.connect)
|
||||
for folder in await asyncio.to_thread(client.resolve_folders):
|
||||
status = await asyncio.to_thread(client.status, folder)
|
||||
row = {
|
||||
"account": account.name,
|
||||
"folder": folder,
|
||||
"messages": status.messages,
|
||||
"uidnext": status.uidnext,
|
||||
"uidvalidity": status.uidvalidity,
|
||||
}
|
||||
rows.append(row)
|
||||
_log.info("folder.measured", **row)
|
||||
except Exception as exc:
|
||||
_log.error("measure_failed", account=account.name, error=str(exc), exc_info=True)
|
||||
rows.append({"account": account.name, "error": str(exc)})
|
||||
finally:
|
||||
await asyncio.to_thread(client.logout)
|
||||
return rows
|
||||
|
||||
|
||||
def build_metrics(result: dict, now_ts: float, prom_path: Path) -> list[MetricFamily]:
|
||||
"""`run()` result -> the metric families of recon §3.4.
|
||||
|
||||
`kb_mail_sync_last_success_timestamp` carries the previous file's value forward on a
|
||||
failed run. Resetting it to 0 (or to now) would make the staleness alert either fire
|
||||
forever after one blip or never fire at all.
|
||||
"""
|
||||
failed = result["failed"]
|
||||
prev_success = read_prev_value(prom_path, "kb_mail_sync_last_success_timestamp")
|
||||
last_success = now_ts if not failed else (prev_success if prev_success is not None else 0.0)
|
||||
|
||||
def per_account(counter: str) -> list[Sample]:
|
||||
return [
|
||||
Sample(value=counters[counter], labels={"account": name})
|
||||
for name, counters in sorted(result["accounts"].items())
|
||||
]
|
||||
|
||||
return [
|
||||
MetricFamily("kb_mail_sync_last_run_timestamp", "gauge",
|
||||
"Unix timestamp of the last kb-mail-sync run (success or failure).",
|
||||
[Sample(now_ts)]),
|
||||
MetricFamily("kb_mail_sync_last_success_timestamp", "gauge",
|
||||
"Unix timestamp of the last kb-mail-sync run with no hard failure.",
|
||||
[Sample(last_success)]),
|
||||
MetricFamily("kb_mail_sync_last_exit_code", "gauge",
|
||||
"Exit code of the last kb-mail-sync run.",
|
||||
[Sample(1 if failed else 0)]),
|
||||
MetricFamily("kb_mail_sync_envelopes_inserted", "gauge",
|
||||
"New envelope rows inserted in the last run, per account.",
|
||||
per_account("envelopes_inserted")),
|
||||
MetricFamily("kb_mail_sync_envelopes_skipped_dup", "gauge",
|
||||
"Messages already present under the same account (Message-ID dedup).",
|
||||
per_account("envelopes_skipped_dup")),
|
||||
MetricFamily("kb_mail_sync_conflict_other_source", "gauge",
|
||||
"Messages whose Message-ID already belongs to a DIFFERENT account — "
|
||||
"stored once, counted here so the overlap is visible.",
|
||||
per_account("envelopes_conflict_other_source")),
|
||||
MetricFamily("kb_mail_sync_uidvalidity_resets", "gauge",
|
||||
"Folders whose UIDVALIDITY changed in the last run, forcing a full sweep.",
|
||||
per_account("uidvalidity_resets")),
|
||||
MetricFamily("kb_mail_sync_errors", "gauge",
|
||||
"Messages, folders and accounts that failed in the last run.",
|
||||
[Sample(value=total_errors(counters), labels={"account": name})
|
||||
for name, counters in sorted(result["accounts"].items())]),
|
||||
MetricFamily("kb_mail_sync_last_message_ts", "gauge",
|
||||
"Timestamp of the newest envelope in the DB, per account.",
|
||||
[Sample(value=ts, labels={"account": name})
|
||||
for name, ts in sorted(result["last_message_ts"].items())]),
|
||||
]
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Incremental IMAP sync for the KB mail corpus (gmail + fastmail). "
|
||||
"Credentials come from the environment only — see jobs/mail-imap-sync/env.example."
|
||||
)
|
||||
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=Path(os.environ.get("MAIL_ARCHIVE_ROOT", str(DEFAULT_ARCHIVE_ROOT))),
|
||||
help=f"Mail .eml archive root (default: {DEFAULT_ARCHIVE_ROOT})")
|
||||
parser.add_argument("--prom-path", type=Path,
|
||||
default=Path(os.environ.get("KB_MAIL_SYNC_PROM_PATH", str(DEFAULT_PROM_PATH))),
|
||||
help=f"Textfile-collector output path (default: {DEFAULT_PROM_PATH})")
|
||||
parser.add_argument("--limit", type=int, default=None, metavar="N",
|
||||
help="Process at most N messages per folder per tick — a brake for "
|
||||
"the first big sweep; the remainder is picked up next tick")
|
||||
parser.add_argument("--measure", action="store_true",
|
||||
help="Print STATUS (MESSAGES/UIDNEXT/UIDVALIDITY) per folder and exit. "
|
||||
"Read-only, touches neither the DB nor the archive.")
|
||||
parser.add_argument("--apply", action="store_true",
|
||||
help="Actually fetch, archive, insert and advance the sync cursor. "
|
||||
"Default is dry-run: plan + SEARCH counts only.")
|
||||
args = parser.parse_args()
|
||||
|
||||
try:
|
||||
accounts = accounts_from_env()
|
||||
except ConfigError as exc:
|
||||
_log.error("config_error", error=str(exc))
|
||||
sys.exit(EXIT_CONFIG_ERROR)
|
||||
|
||||
if args.measure:
|
||||
rows = asyncio.run(measure(accounts))
|
||||
_log.info("measure_complete", folders=rows)
|
||||
sys.exit(1 if any("error" in r for r in rows) else 0)
|
||||
|
||||
if not args.dsn:
|
||||
_log.error("missing_dsn", hint="pass --dsn or set KB_DSN")
|
||||
sys.exit(EXIT_CONFIG_ERROR)
|
||||
if args.apply and not args.archive_root.is_dir():
|
||||
_log.error("archive_root_not_found", path=str(args.archive_root))
|
||||
sys.exit(EXIT_CONFIG_ERROR)
|
||||
|
||||
result = asyncio.run(
|
||||
run(
|
||||
dsn=args.dsn,
|
||||
accounts=accounts,
|
||||
archive_root=args.archive_root,
|
||||
apply=args.apply,
|
||||
limit=args.limit,
|
||||
)
|
||||
)
|
||||
|
||||
if args.apply:
|
||||
# Only a real run publishes metrics: a dry-run must not touch the success clock the
|
||||
# staleness alert reads.
|
||||
now_ts = time.time()
|
||||
write_prom_atomic(args.prom_path, render_prom(build_metrics(result, now_ts, args.prom_path)))
|
||||
|
||||
_log.info(
|
||||
"summary",
|
||||
mode="APPLY" if args.apply else "DRY-RUN",
|
||||
accounts={name: counters for name, counters in result["accounts"].items()},
|
||||
last_message=_readable_last_message(result["last_message_ts"]),
|
||||
balance_problems=result["balance_problems"],
|
||||
**result["totals"],
|
||||
)
|
||||
sys.exit(1 if result["failed"] else 0)
|
||||
|
||||
|
||||
def _readable_last_message(last_message_ts: dict[str, float]) -> dict[str, str]:
|
||||
return {
|
||||
name: datetime.fromtimestamp(ts, tz=timezone.utc).isoformat()
|
||||
for name, ts in sorted(last_message_ts.items())
|
||||
}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
199
jobs/mail-imap-sync/tests/conftest.py
Normal file
199
jobs/mail-imap-sync/tests/conftest.py
Normal file
|
|
@ -0,0 +1,199 @@
|
|||
"""Fakes shared by the job's tests — no network, no DB, no live mailbox.
|
||||
|
||||
The IMAP fake sits at the `imaplib` level rather than at `ImapClient`, so every test runs the
|
||||
adapter's REAL protocol parsing (LIST attributes, EXAMINE untagged responses, `n:*` range
|
||||
trailers, `(prefix, literal)` FETCH tuples). The DB fake models the two tables the job touches
|
||||
closely enough that dedup, cross-account collisions and cursor persistence are exercised as
|
||||
behaviour rather than asserted as SQL strings.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
|
||||
|
||||
class FakeIMAP:
|
||||
"""Multi-folder `imaplib.IMAP4_SSL` stand-in.
|
||||
|
||||
`folders` maps mailbox name -> {uid: raw bytes}. `uidvalidity` may be a per-folder dict so
|
||||
a test can flip one folder's value to simulate a server-side renumbering.
|
||||
"""
|
||||
|
||||
def __init__(self, folders, *, uidvalidity=42, list_attrs=None, login_error=None,
|
||||
fetch_errors=(), vanished=()):
|
||||
self.folders = folders
|
||||
self._uidvalidity = uidvalidity
|
||||
self.list_attrs = list_attrs or {}
|
||||
self.login_error = login_error
|
||||
self.fetch_errors = set(fetch_errors)
|
||||
self.vanished = set(vanished)
|
||||
self.commands: list[tuple] = []
|
||||
self.selected = None
|
||||
self.logged_out = False
|
||||
|
||||
# -- helpers -----------------------------------------------------------------
|
||||
|
||||
def uidvalidity_of(self, folder: str) -> int:
|
||||
if isinstance(self._uidvalidity, dict):
|
||||
return self._uidvalidity.get(folder, 42)
|
||||
return self._uidvalidity
|
||||
|
||||
def uidnext_of(self, folder: str) -> int:
|
||||
uids = self.folders.get(folder) or {}
|
||||
return (max(uids) + 1) if uids else 1
|
||||
|
||||
def fetch_commands(self):
|
||||
return [c for c in self.commands if c[:2] == ("UID", "FETCH")]
|
||||
|
||||
# -- imaplib surface ---------------------------------------------------------
|
||||
|
||||
def login(self, user, password):
|
||||
self.commands.append(("LOGIN", user))
|
||||
if self.login_error:
|
||||
raise RuntimeError(self.login_error)
|
||||
return ("OK", [b"ok"])
|
||||
|
||||
def logout(self):
|
||||
self.logged_out = True
|
||||
return ("BYE", [b"bye"])
|
||||
|
||||
def list(self, directory='""', pattern="*"):
|
||||
self.commands.append(("LIST",))
|
||||
lines = []
|
||||
for name in self.folders:
|
||||
attrs = " ".join(self.list_attrs.get(name, ["\\HasNoChildren"]))
|
||||
lines.append(f'({attrs}) "/" "{name}"'.encode())
|
||||
return ("OK", lines)
|
||||
|
||||
def select(self, mailbox, readonly=False):
|
||||
name = mailbox.strip('"')
|
||||
self.commands.append(("SELECT", name, readonly))
|
||||
if name not in self.folders:
|
||||
return ("NO", [b"no such mailbox"])
|
||||
self.selected = name
|
||||
return ("OK", [str(len(self.folders[name])).encode()])
|
||||
|
||||
def status(self, mailbox, names):
|
||||
name = mailbox.strip('"')
|
||||
self.commands.append(("STATUS", name, names))
|
||||
line = (
|
||||
f'"{name}" (MESSAGES {len(self.folders.get(name) or {})} '
|
||||
f"UIDNEXT {self.uidnext_of(name)} UIDVALIDITY {self.uidvalidity_of(name)})"
|
||||
).encode()
|
||||
return ("OK", [line])
|
||||
|
||||
def response(self, key):
|
||||
if self.selected is None:
|
||||
return (key, [None])
|
||||
if key == "UIDVALIDITY":
|
||||
return (key, [str(self.uidvalidity_of(self.selected)).encode()])
|
||||
if key == "UIDNEXT":
|
||||
return (key, [str(self.uidnext_of(self.selected)).encode()])
|
||||
return (key, [None])
|
||||
|
||||
def uid(self, command, *args):
|
||||
self.commands.append(("UID", command) + args)
|
||||
uids = sorted(self.folders.get(self.selected) or {})
|
||||
if command == "SEARCH":
|
||||
criteria = args[1:]
|
||||
if criteria and criteria[0] == "UID":
|
||||
start = int(criteria[1].split(":")[0])
|
||||
hits = [u for u in uids if u >= start]
|
||||
if not hits and uids:
|
||||
hits = [uids[-1]] # real servers do this; the client must filter it out
|
||||
elif criteria and criteria[0] == "SINCE":
|
||||
hits = [u for u in uids if u >= getattr(self, "since_uid_floor", 0)]
|
||||
else:
|
||||
hits = uids
|
||||
return ("OK", [" ".join(str(u) for u in hits).encode()])
|
||||
if command == "FETCH":
|
||||
uid = int(args[0])
|
||||
if uid in self.fetch_errors:
|
||||
raise RuntimeError(f"fetch of {uid} exploded")
|
||||
if uid in self.vanished:
|
||||
return ("OK", [None])
|
||||
raw = (self.folders.get(self.selected) or {}).get(uid)
|
||||
if raw is None:
|
||||
return ("OK", [None])
|
||||
return ("OK", [(f"1 (UID {uid} BODY[] {{{len(raw)}}}".encode(), raw), b")"])
|
||||
raise AssertionError(f"unexpected UID command {command}")
|
||||
|
||||
|
||||
class FakeConn:
|
||||
"""In-memory `envelope` + `mail_sync_state`, dispatched on SQL keywords."""
|
||||
|
||||
def __init__(self, envelopes=None, state=None, insert_errors=()):
|
||||
self.envelopes = dict(envelopes or {}) # id -> {"source", "ts", "entities"}
|
||||
self.state = dict(state or {}) # (account, folder) -> (uidvalidity, last_uid)
|
||||
self.insert_errors = set(insert_errors) # envelope ids whose INSERT raises
|
||||
self.queries: list[tuple] = []
|
||||
self.closed = False
|
||||
|
||||
async def fetchrow(self, query, *params):
|
||||
self.queries.append((query, params))
|
||||
if "FROM mail_sync_state" in query:
|
||||
row = self.state.get((params[0], params[1]))
|
||||
if row is None:
|
||||
return None
|
||||
return {"account": params[0], "folder": params[1], "uidvalidity": row[0],
|
||||
"last_uid": row[1], "last_sync_ts": None}
|
||||
raise AssertionError(f"unexpected fetchrow: {query}")
|
||||
|
||||
async def fetchval(self, query, *params):
|
||||
self.queries.append((query, params))
|
||||
if "SELECT source FROM envelope" in query:
|
||||
row = self.envelopes.get(params[0])
|
||||
return row["source"] if row else None
|
||||
raise AssertionError(f"unexpected fetchval: {query}")
|
||||
|
||||
async def fetch(self, query, *params):
|
||||
self.queries.append((query, params))
|
||||
if "max(ts)" in query:
|
||||
newest: dict = {}
|
||||
for row in self.envelopes.values():
|
||||
if row["source"] in params[0]:
|
||||
current = newest.get(row["source"])
|
||||
if current is None or row["ts"] > current:
|
||||
newest[row["source"]] = row["ts"]
|
||||
return [{"source": s, "newest": ts} for s, ts in newest.items()]
|
||||
if "FROM mail_sync_state" in query:
|
||||
return [
|
||||
{"account": a, "folder": f, "uidvalidity": v[0], "last_uid": v[1],
|
||||
"last_sync_ts": None}
|
||||
for (a, f), v in sorted(self.state.items())
|
||||
]
|
||||
raise AssertionError(f"unexpected fetch: {query}")
|
||||
|
||||
async def execute(self, query, *params):
|
||||
self.queries.append((query, params))
|
||||
if "INSERT INTO envelope" in query:
|
||||
envelope_id = params[0]
|
||||
if envelope_id in self.insert_errors:
|
||||
raise RuntimeError(f"insert of {envelope_id} exploded")
|
||||
if envelope_id in self.envelopes:
|
||||
return "INSERT 0 0"
|
||||
self.envelopes[envelope_id] = {"source": params[1], "ts": params[2],
|
||||
"entities": params[5]}
|
||||
return "INSERT 0 1"
|
||||
if "INSERT INTO mail_sync_state" in query:
|
||||
self.state[(params[0], params[1])] = (params[2], params[3])
|
||||
return "INSERT 0 1"
|
||||
raise AssertionError(f"unexpected execute: {query}")
|
||||
|
||||
async def close(self):
|
||||
self.closed = True
|
||||
|
||||
|
||||
def eml(message_id: str, *, subject="Temat", sender="Alice <a@example.com>",
|
||||
date="Mon, 15 Jun 2026 14:00:00 +0200", body="tresc wiadomosci") -> bytes:
|
||||
return (
|
||||
f"Message-ID: <{message_id}>\r\n"
|
||||
f"From: {sender}\r\n"
|
||||
f"To: Oskar <o@example.com>\r\n"
|
||||
f"Subject: {subject}\r\n"
|
||||
f"Date: {date}\r\n"
|
||||
f"Content-Type: text/plain; charset=utf-8\r\n"
|
||||
f"\r\n{body}\r\n"
|
||||
).encode("utf-8")
|
||||
|
||||
|
||||
JUNE_15 = datetime(2026, 6, 15, 12, 0, tzinfo=timezone.utc)
|
||||
139
jobs/mail-imap-sync/tests/test_config.py
Normal file
139
jobs/mail-imap-sync/tests/test_config.py
Normal file
|
|
@ -0,0 +1,139 @@
|
|||
"""Tests for env-only account configuration.
|
||||
|
||||
Two properties matter more than the parsing details: a malformed or missing setting is a
|
||||
startup failure rather than a quiet half-configuration, and the approved folder scope
|
||||
(Decyzja (e)) is the default so a working config is short.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
|
||||
import pytest
|
||||
|
||||
from mail_imap_sync.config import (
|
||||
ConfigError,
|
||||
account_from_env,
|
||||
accounts_from_env,
|
||||
archive_root_from_env,
|
||||
)
|
||||
|
||||
BASE = {
|
||||
"MAIL_ACCOUNTS": "gmail,fastmail",
|
||||
"MAIL_GMAIL_USER": "u@gmail.com",
|
||||
"MAIL_GMAIL_APP_PASSWORD": "pw1",
|
||||
"MAIL_FASTMAIL_USER": "u@fastmail.com",
|
||||
"MAIL_FASTMAIL_APP_PASSWORD": "pw2",
|
||||
}
|
||||
|
||||
|
||||
def env(**overrides) -> dict:
|
||||
merged = dict(BASE)
|
||||
merged.update(overrides)
|
||||
return {k: v for k, v in merged.items() if v is not None}
|
||||
|
||||
|
||||
class TestDefaults:
|
||||
def test_gmail_defaults_to_the_special_use_all_folder(self):
|
||||
account = account_from_env("gmail", env())
|
||||
assert account.special_use == ("\\All",)
|
||||
assert account.folders == ()
|
||||
|
||||
def test_fastmail_defaults_to_inbox_archive_sent(self):
|
||||
account = account_from_env("fastmail", env())
|
||||
assert account.folders == ("INBOX", "Archive", "Sent")
|
||||
assert account.special_use == ()
|
||||
|
||||
def test_spam_trash_and_drafts_are_not_in_any_default_scope(self):
|
||||
for name in ("gmail", "fastmail"):
|
||||
scope = account_from_env(name, env()).folders
|
||||
assert not {"Spam", "Trash", "Drafts"} & set(scope)
|
||||
|
||||
def test_known_hosts_are_defaults_not_requirements(self):
|
||||
assert account_from_env("gmail", env()).host == "imap.gmail.com"
|
||||
assert account_from_env("gmail", env(MAIL_GMAIL_HOST="mail.example.com")).host \
|
||||
== "mail.example.com"
|
||||
|
||||
def test_first_tick_defaults_to_new_only(self):
|
||||
assert account_from_env("fastmail", env()).initial_mode == "new-only"
|
||||
|
||||
|
||||
class TestOverrides:
|
||||
def test_folders_override_replaces_the_default_list(self):
|
||||
account = account_from_env("fastmail", env(MAIL_FASTMAIL_FOLDERS="INBOX,Projects"))
|
||||
assert account.folders == ("INBOX", "Projects")
|
||||
|
||||
def test_empty_override_means_none_not_fall_back_to_default(self):
|
||||
account = account_from_env("gmail", env(MAIL_GMAIL_SPECIAL_USE="",
|
||||
MAIL_GMAIL_FOLDERS="INBOX"))
|
||||
assert account.special_use == ()
|
||||
assert account.folders == ("INBOX",)
|
||||
|
||||
def test_since_mode_parses_the_date(self):
|
||||
account = account_from_env("gmail", env(MAIL_GMAIL_INITIAL_MODE="since",
|
||||
MAIL_GMAIL_INITIAL_SINCE="2026-06-15"))
|
||||
assert account.initial_since == date(2026, 6, 15)
|
||||
|
||||
def test_port_override(self):
|
||||
assert account_from_env("gmail", env(MAIL_GMAIL_PORT="1993")).port == 1993
|
||||
|
||||
def test_archive_root_override(self):
|
||||
assert str(archive_root_from_env({"MAIL_ARCHIVE_ROOT": "/srv/mail"})) == "/srv/mail"
|
||||
|
||||
|
||||
class TestFailsLoudly:
|
||||
def test_missing_accounts_list(self):
|
||||
with pytest.raises(ConfigError, match="MAIL_ACCOUNTS"):
|
||||
accounts_from_env({})
|
||||
|
||||
def test_missing_password(self):
|
||||
with pytest.raises(ConfigError, match="APP_PASSWORD"):
|
||||
account_from_env("gmail", env(MAIL_GMAIL_APP_PASSWORD=None))
|
||||
|
||||
def test_missing_user(self):
|
||||
with pytest.raises(ConfigError, match="USER"):
|
||||
account_from_env("gmail", env(MAIL_GMAIL_USER=None))
|
||||
|
||||
def test_unknown_account_without_a_host(self):
|
||||
with pytest.raises(ConfigError, match="HOST"):
|
||||
account_from_env("mailbox-org", {"MAIL_MAILBOX_ORG_USER": "u",
|
||||
"MAIL_MAILBOX_ORG_APP_PASSWORD": "p"})
|
||||
|
||||
def test_unknown_account_with_a_host_but_no_scope(self):
|
||||
with pytest.raises(ConfigError, match="no scope"):
|
||||
account_from_env("mailbox-org", {"MAIL_MAILBOX_ORG_USER": "u",
|
||||
"MAIL_MAILBOX_ORG_APP_PASSWORD": "p",
|
||||
"MAIL_MAILBOX_ORG_HOST": "imap.mailbox.org"})
|
||||
|
||||
def test_bad_port(self):
|
||||
with pytest.raises(ConfigError, match="PORT"):
|
||||
account_from_env("gmail", env(MAIL_GMAIL_PORT="993x"))
|
||||
|
||||
def test_bad_initial_mode(self):
|
||||
with pytest.raises(ConfigError, match="INITIAL_MODE"):
|
||||
account_from_env("gmail", env(MAIL_GMAIL_INITIAL_MODE="yesterday"))
|
||||
|
||||
def test_since_mode_without_a_date(self):
|
||||
with pytest.raises(ConfigError, match="initial_since"):
|
||||
account_from_env("gmail", env(MAIL_GMAIL_INITIAL_MODE="since"))
|
||||
|
||||
def test_bad_since_date(self):
|
||||
with pytest.raises(ConfigError, match="not a date"):
|
||||
account_from_env("gmail", env(MAIL_GMAIL_INITIAL_MODE="since",
|
||||
MAIL_GMAIL_INITIAL_SINCE="15/06/2026"))
|
||||
|
||||
def test_duplicate_account_name(self):
|
||||
with pytest.raises(ConfigError, match="twice"):
|
||||
accounts_from_env(env(MAIL_ACCOUNTS="gmail,gmail"))
|
||||
|
||||
|
||||
class TestAccountsFromEnv:
|
||||
def test_returns_accounts_in_the_listed_order(self):
|
||||
accounts = accounts_from_env(env(MAIL_ACCOUNTS="fastmail,gmail"))
|
||||
assert [a.name for a in accounts] == ["fastmail", "gmail"]
|
||||
|
||||
def test_a_single_account_is_a_valid_config(self):
|
||||
accounts = accounts_from_env(env(MAIL_ACCOUNTS="gmail"))
|
||||
assert [a.name for a in accounts] == ["gmail"]
|
||||
|
||||
def test_account_name_is_the_envelope_source(self):
|
||||
assert accounts_from_env(env())[0].name == "gmail"
|
||||
85
jobs/mail-imap-sync/tests/test_prom.py
Normal file
85
jobs/mail-imap-sync/tests/test_prom.py
Normal file
|
|
@ -0,0 +1,85 @@
|
|||
"""Tests for the textfile-collector renderer."""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from mail_imap_sync.prom import (
|
||||
MetricFamily,
|
||||
Sample,
|
||||
read_prev_value,
|
||||
render_prom,
|
||||
write_prom_atomic,
|
||||
)
|
||||
|
||||
|
||||
def _family(samples, name="kb_mail_sync_test", mtype="gauge"):
|
||||
return MetricFamily(name, mtype, "a help string", samples)
|
||||
|
||||
|
||||
class TestRender:
|
||||
def test_emits_help_and_type_before_samples(self):
|
||||
text = render_prom([_family([Sample(3)])])
|
||||
assert text.splitlines()[:3] == [
|
||||
"# HELP kb_mail_sync_test a help string",
|
||||
"# TYPE kb_mail_sync_test gauge",
|
||||
"kb_mail_sync_test 3",
|
||||
]
|
||||
|
||||
def test_labels_are_sorted_and_quoted(self):
|
||||
text = render_prom([_family([Sample(1, {"folder": "INBOX", "account": "gmail"})])])
|
||||
assert 'kb_mail_sync_test{account="gmail",folder="INBOX"} 1' in text
|
||||
|
||||
def test_label_values_are_escaped(self):
|
||||
text = render_prom([_family([Sample(1, {"folder": '[Gmail]/"All"\\Mail'})])])
|
||||
assert 'folder="[Gmail]/\\"All\\"\\\\Mail"' in text
|
||||
|
||||
def test_integers_render_without_a_trailing_zero(self):
|
||||
assert "kb_mail_sync_test 3\n" in render_prom([_family([Sample(3.0)])])
|
||||
|
||||
def test_floats_keep_their_precision(self):
|
||||
assert "kb_mail_sync_test 1.5" in render_prom([_family([Sample(1.5)])])
|
||||
|
||||
def test_family_with_no_samples_is_omitted_entirely(self):
|
||||
# "no accounts reported" and "accounts reported zero" are different facts.
|
||||
assert render_prom([_family([])]) == "\n"
|
||||
|
||||
def test_output_ends_with_a_newline(self):
|
||||
assert render_prom([_family([Sample(1)])]).endswith("\n")
|
||||
|
||||
|
||||
class TestReadPrevValue:
|
||||
def test_absent_file_returns_none(self, tmp_path: Path):
|
||||
assert read_prev_value(tmp_path / "nope.prom", "kb_mail_sync_x") is None
|
||||
|
||||
def test_absent_metric_returns_none(self, tmp_path: Path):
|
||||
path = tmp_path / "x.prom"
|
||||
path.write_text("kb_mail_sync_other 5\n")
|
||||
assert read_prev_value(path, "kb_mail_sync_x") is None
|
||||
|
||||
def test_reads_a_scientific_notation_value(self, tmp_path: Path):
|
||||
path = tmp_path / "x.prom"
|
||||
path.write_text("kb_mail_sync_x 1.786e9\n")
|
||||
assert read_prev_value(path, "kb_mail_sync_x") == 1.786e9
|
||||
|
||||
def test_help_lines_are_not_mistaken_for_samples(self, tmp_path: Path):
|
||||
path = tmp_path / "x.prom"
|
||||
path.write_text("# HELP kb_mail_sync_x 42 things\nkb_mail_sync_x 7\n")
|
||||
assert read_prev_value(path, "kb_mail_sync_x") == 7.0
|
||||
|
||||
|
||||
class TestWriteAtomic:
|
||||
def test_creates_the_directory_and_the_file(self, tmp_path: Path):
|
||||
path = tmp_path / "node-exporter" / "kb-mail-sync.prom"
|
||||
write_prom_atomic(path, "kb_mail_sync_x 1\n")
|
||||
assert path.read_text() == "kb_mail_sync_x 1\n"
|
||||
|
||||
def test_leaves_no_temp_file_behind(self, tmp_path: Path):
|
||||
path = tmp_path / "kb-mail-sync.prom"
|
||||
write_prom_atomic(path, "kb_mail_sync_x 1\n")
|
||||
assert [p.name for p in tmp_path.iterdir()] == ["kb-mail-sync.prom"]
|
||||
|
||||
def test_overwrites_in_place(self, tmp_path: Path):
|
||||
path = tmp_path / "kb-mail-sync.prom"
|
||||
write_prom_atomic(path, "kb_mail_sync_x 1\n")
|
||||
write_prom_atomic(path, "kb_mail_sync_x 2\n")
|
||||
assert path.read_text() == "kb_mail_sync_x 2\n"
|
||||
509
jobs/mail-imap-sync/tests/test_sync.py
Normal file
509
jobs/mail-imap-sync/tests/test_sync.py
Normal file
|
|
@ -0,0 +1,509 @@
|
|||
"""Unit tests for the IMAP sync job — no network, no DB, no live mailbox.
|
||||
|
||||
Every test drives the real `ImapClient` over a fake `imaplib` connection and the real
|
||||
`run()` over an in-memory `envelope` / `mail_sync_state` pair, so what is being tested is the
|
||||
job's actual behaviour: what it fetches, what it archives, where the cursor ends up. The
|
||||
scenarios mirror the failure modes the recon named — new mail, UIDVALIDITY invalidation,
|
||||
dedup, resumption after an interrupted run — plus the traps it warned would be silent.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from conftest import FakeConn, FakeIMAP, eml
|
||||
|
||||
from kb_mail.imap import ImapAccount, ImapClient
|
||||
from mail_imap_sync.prom import read_prev_value, render_prom
|
||||
from mail_imap_sync.sync import (
|
||||
balance_errors,
|
||||
build_metrics,
|
||||
measure,
|
||||
run,
|
||||
total_errors,
|
||||
)
|
||||
|
||||
GMAIL_FOLDER = "[Gmail]/Wszystkie"
|
||||
|
||||
|
||||
def gmail_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 fastmail_account(**kwargs) -> ImapAccount:
|
||||
base = dict(name="fastmail", host="imap.example.net", user="u@example.net",
|
||||
password="secret", special_use=(), folders=("INBOX",))
|
||||
base.update(kwargs)
|
||||
return ImapAccount(**base)
|
||||
|
||||
|
||||
def gmail_server(messages, **kwargs) -> FakeIMAP:
|
||||
return FakeIMAP({GMAIL_FOLDER: messages},
|
||||
list_attrs={GMAIL_FOLDER: ["\\HasNoChildren", "\\All"]}, **kwargs)
|
||||
|
||||
|
||||
def factory_for(servers: dict):
|
||||
"""Client factory returning a real ImapClient bound to the named account's fake server."""
|
||||
def _factory(account: ImapAccount) -> ImapClient:
|
||||
return ImapClient(account, connection_factory=lambda acc, timeout: servers[acc.name])
|
||||
return _factory
|
||||
|
||||
|
||||
async def do_run(servers, accounts, conn, archive, *, apply=True, limit=None, monkeypatch=None):
|
||||
async def _connect(_dsn):
|
||||
return conn
|
||||
import asyncpg
|
||||
monkeypatch.setattr(asyncpg, "connect", _connect)
|
||||
return await run(
|
||||
dsn="postgresql://fake",
|
||||
accounts=accounts,
|
||||
archive_root=archive,
|
||||
apply=apply,
|
||||
limit=limit,
|
||||
client_factory=factory_for(servers),
|
||||
)
|
||||
|
||||
|
||||
class TestNewMessages:
|
||||
async def test_fetches_archives_and_inserts(self, tmp_path: Path, monkeypatch):
|
||||
server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
|
||||
conn = FakeConn()
|
||||
result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
|
||||
conn, tmp_path, monkeypatch=monkeypatch)
|
||||
|
||||
assert result["totals"]["envelopes_inserted"] == 2
|
||||
assert set(conn.envelopes) == {"a@x", "b@x"}
|
||||
assert sorted(p.name for p in tmp_path.rglob("*.eml")) == ["a@x.eml", "b@x.eml"]
|
||||
assert not result["failed"]
|
||||
|
||||
async def test_archive_path_is_source_year_month(self, tmp_path: Path, monkeypatch):
|
||||
server = gmail_server({1: eml("a@x")})
|
||||
await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
|
||||
FakeConn(), tmp_path, monkeypatch=monkeypatch)
|
||||
assert (tmp_path / "gmail" / "2026" / "06" / "a@x.eml").exists()
|
||||
|
||||
async def test_cursor_advances_to_the_highest_processed_uid(self, tmp_path: Path, monkeypatch):
|
||||
server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
|
||||
conn = FakeConn()
|
||||
await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
|
||||
conn, tmp_path, monkeypatch=monkeypatch)
|
||||
assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 2)
|
||||
|
||||
async def test_envelope_carries_headers_at_insert_time(self, tmp_path: Path, monkeypatch):
|
||||
# Recon §2.5 (i): without this, mail_body_ingest.build_prefix silently produces
|
||||
# "Temat: (brak tematu) | Od: ?" for every new mail — no error, just worse retrieval.
|
||||
server = gmail_server({1: eml("a@x", subject="Faktura 07/2026")})
|
||||
conn = FakeConn()
|
||||
await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
|
||||
conn, tmp_path, monkeypatch=monkeypatch)
|
||||
entities = json.loads(conn.envelopes["a@x"]["entities"])
|
||||
headers = next(e for e in entities if e["type"] == "headers")
|
||||
assert headers["subject"] == "Faktura 07/2026"
|
||||
assert headers["from"]["address"] == "a@example.com"
|
||||
|
||||
async def test_attachment_manifest_is_appended_next_to_headers(self, tmp_path: Path, monkeypatch):
|
||||
raw = (
|
||||
b"Message-ID: <att@x>\r\nFrom: a@example.com\r\nSubject: s\r\n"
|
||||
b"Date: Mon, 15 Jun 2026 14:00:00 +0200\r\n"
|
||||
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\r\n%PDF\r\n--b--\r\n'
|
||||
)
|
||||
conn = FakeConn()
|
||||
await do_run({"gmail": gmail_server({1: raw})}, [gmail_account(initial_mode="full")],
|
||||
conn, tmp_path, monkeypatch=monkeypatch)
|
||||
types = [e["type"] for e in json.loads(conn.envelopes["att@x"]["entities"])]
|
||||
assert types == ["headers", "attachment"]
|
||||
|
||||
async def test_never_marks_mail_as_read(self, tmp_path: Path, monkeypatch):
|
||||
# A plain SELECT + FETCH RFC822 would set \Seen on the operator's own inbox.
|
||||
server = gmail_server({1: eml("a@x")})
|
||||
await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
|
||||
FakeConn(), tmp_path, monkeypatch=monkeypatch)
|
||||
assert all(cmd[2] is True for cmd in server.commands if cmd[0] == "SELECT")
|
||||
assert all(cmd[-1] == "(BODY.PEEK[])" for cmd in server.fetch_commands())
|
||||
|
||||
|
||||
class TestIncremental:
|
||||
async def test_second_run_fetches_only_what_arrived_since(self, tmp_path: Path, monkeypatch):
|
||||
server = gmail_server({1: eml("a@x")})
|
||||
conn = FakeConn()
|
||||
account = gmail_account(initial_mode="full")
|
||||
await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
|
||||
|
||||
server.folders[GMAIL_FOLDER][2] = eml("b@x")
|
||||
server.commands.clear()
|
||||
result = await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
|
||||
|
||||
assert [int(c[2]) for c in server.fetch_commands()] == [2]
|
||||
assert result["totals"]["envelopes_inserted"] == 1
|
||||
|
||||
async def test_idle_mailbox_reports_zero_new(self, tmp_path: Path, monkeypatch):
|
||||
# The `n:*` range trailer means an unfiltered implementation would report one "new"
|
||||
# mail on every idle tick, forever.
|
||||
server = gmail_server({1: eml("a@x")})
|
||||
conn = FakeConn()
|
||||
account = gmail_account(initial_mode="full")
|
||||
await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
|
||||
server.commands.clear()
|
||||
|
||||
result = await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
|
||||
assert result["totals"]["uids_seen"] == 0
|
||||
assert result["totals"]["envelopes_inserted"] == 0
|
||||
assert server.fetch_commands() == []
|
||||
|
||||
async def test_rerun_of_the_same_uids_is_a_no_op_at_every_layer(self, tmp_path: Path, monkeypatch):
|
||||
# Force a re-fetch by rewinding the cursor: archive says FileExistsError, DB says
|
||||
# ON CONFLICT DO NOTHING, and nothing is double-counted as new.
|
||||
server = gmail_server({1: eml("a@x")})
|
||||
conn = FakeConn()
|
||||
account = gmail_account(initial_mode="full")
|
||||
await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
|
||||
conn.state[("gmail", GMAIL_FOLDER)] = (42, 0)
|
||||
|
||||
result = await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
|
||||
totals = result["totals"]
|
||||
assert totals["envelopes_inserted"] == 0
|
||||
assert totals["envelopes_skipped_dup"] == 1
|
||||
assert totals["archive_exists"] == 1
|
||||
assert totals["archived"] == 0
|
||||
assert len(list(tmp_path.rglob("*.eml"))) == 1
|
||||
|
||||
|
||||
class TestUidvalidityInvalidation:
|
||||
async def test_changed_uidvalidity_triggers_a_full_sweep(self, tmp_path: Path, monkeypatch):
|
||||
server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
|
||||
conn = FakeConn()
|
||||
account = gmail_account(initial_mode="full")
|
||||
await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
|
||||
|
||||
server._uidvalidity = 99
|
||||
server.commands.clear()
|
||||
result = await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
|
||||
|
||||
assert result["totals"]["uidvalidity_resets"] == 1
|
||||
assert sorted(int(c[2]) for c in server.fetch_commands()) == [1, 2]
|
||||
|
||||
async def test_resweep_inserts_nothing_new_thanks_to_message_id_dedup(self, tmp_path, monkeypatch):
|
||||
server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
|
||||
conn = FakeConn()
|
||||
account = gmail_account(initial_mode="full")
|
||||
await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
|
||||
|
||||
server._uidvalidity = 99
|
||||
result = await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
|
||||
assert result["totals"]["envelopes_inserted"] == 0
|
||||
assert result["totals"]["envelopes_skipped_dup"] == 2
|
||||
|
||||
async def test_new_uidvalidity_is_stored_with_the_recomputed_cursor(self, tmp_path, monkeypatch):
|
||||
server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
|
||||
conn = FakeConn()
|
||||
account = gmail_account(initial_mode="full")
|
||||
await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
|
||||
|
||||
server._uidvalidity = 99
|
||||
await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
|
||||
assert conn.state[("gmail", GMAIL_FOLDER)] == (99, 2)
|
||||
|
||||
async def test_renumbered_mailbox_with_lower_uids_is_not_skipped(self, tmp_path, monkeypatch):
|
||||
# The stored cursor (90) is meaningless under a new UIDVALIDITY; treating it as a
|
||||
# floor would skip every renumbered message below it.
|
||||
conn = FakeConn(state={("gmail", GMAIL_FOLDER): (42, 90)})
|
||||
server = gmail_server({1: eml("a@x"), 2: eml("b@x")}, uidvalidity=99)
|
||||
result = await do_run({"gmail": server}, [gmail_account()], conn, tmp_path,
|
||||
monkeypatch=monkeypatch)
|
||||
assert result["totals"]["envelopes_inserted"] == 2
|
||||
assert conn.state[("gmail", GMAIL_FOLDER)] == (99, 2)
|
||||
|
||||
|
||||
class TestResumeAfterInterruption:
|
||||
async def test_cursor_stops_before_a_failed_message(self, tmp_path: Path, monkeypatch):
|
||||
server = gmail_server({1: eml("a@x"), 2: eml("b@x"), 3: eml("c@x")},
|
||||
fetch_errors={2})
|
||||
conn = FakeConn()
|
||||
result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
|
||||
conn, tmp_path, monkeypatch=monkeypatch)
|
||||
|
||||
assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 1)
|
||||
assert result["totals"]["errors"] == 1
|
||||
assert result["failed"] is True
|
||||
|
||||
async def test_messages_after_the_failure_are_still_stored(self, tmp_path: Path, monkeypatch):
|
||||
# Not skipped, not lost: they land in the archive and DB now, and the re-fetch that
|
||||
# the un-advanced cursor forces next tick is absorbed by dedup.
|
||||
server = gmail_server({1: eml("a@x"), 2: eml("b@x"), 3: eml("c@x")}, fetch_errors={2})
|
||||
conn = FakeConn()
|
||||
await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
|
||||
conn, tmp_path, monkeypatch=monkeypatch)
|
||||
assert set(conn.envelopes) == {"a@x", "c@x"}
|
||||
|
||||
async def test_next_run_retries_from_the_stalled_uid(self, tmp_path: Path, monkeypatch):
|
||||
server = gmail_server({1: eml("a@x"), 2: eml("b@x"), 3: eml("c@x")}, fetch_errors={2})
|
||||
conn = FakeConn()
|
||||
account = gmail_account(initial_mode="full")
|
||||
await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
|
||||
|
||||
server.fetch_errors.clear()
|
||||
server.commands.clear()
|
||||
result = await do_run({"gmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
|
||||
|
||||
assert sorted(int(c[2]) for c in server.fetch_commands()) == [2, 3]
|
||||
assert result["totals"]["envelopes_inserted"] == 1 # only b@x was actually missing
|
||||
assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 3)
|
||||
|
||||
async def test_a_failed_db_insert_leaves_the_cursor_behind(self, tmp_path: Path, monkeypatch):
|
||||
# The archive write already happened; the cursor must not move until the row exists.
|
||||
server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
|
||||
conn = FakeConn(insert_errors={"a@x"})
|
||||
result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
|
||||
conn, tmp_path, monkeypatch=monkeypatch)
|
||||
assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 0)
|
||||
assert result["totals"]["errors"] == 1
|
||||
|
||||
async def test_limit_caps_a_tick_and_the_rest_follows_next_time(self, tmp_path, monkeypatch):
|
||||
server = gmail_server({1: eml("a@x"), 2: eml("b@x"), 3: eml("c@x")})
|
||||
conn = FakeConn()
|
||||
account = gmail_account(initial_mode="full")
|
||||
await do_run({"gmail": server}, [account], conn, tmp_path, limit=2, monkeypatch=monkeypatch)
|
||||
assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 2)
|
||||
|
||||
await do_run({"gmail": server}, [account], conn, tmp_path, limit=2, monkeypatch=monkeypatch)
|
||||
assert set(conn.envelopes) == {"a@x", "b@x", "c@x"}
|
||||
|
||||
|
||||
class TestFirstTickModes:
|
||||
async def test_new_only_records_the_high_water_mark_without_fetching(self, tmp_path, monkeypatch):
|
||||
server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
|
||||
conn = FakeConn()
|
||||
result = await do_run({"gmail": server}, [gmail_account(initial_mode="new-only")],
|
||||
conn, tmp_path, monkeypatch=monkeypatch)
|
||||
assert server.fetch_commands() == []
|
||||
assert conn.envelopes == {}
|
||||
assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 2)
|
||||
assert not result["failed"]
|
||||
|
||||
async def test_since_mode_issues_an_imap_date_search(self, tmp_path, monkeypatch):
|
||||
from datetime import date
|
||||
server = gmail_server({1: eml("a@x")})
|
||||
await do_run({"gmail": server},
|
||||
[gmail_account(initial_mode="since", initial_since=date(2026, 6, 15))],
|
||||
FakeConn(), tmp_path, monkeypatch=monkeypatch)
|
||||
search = [c for c in server.commands if c[:2] == ("UID", "SEARCH")][0]
|
||||
assert search[-2:] == ("SINCE", "15-Jun-2026")
|
||||
|
||||
async def test_initial_mode_no_longer_applies_once_a_cursor_exists(self, tmp_path, monkeypatch):
|
||||
conn = FakeConn(state={("gmail", GMAIL_FOLDER): (42, 1)})
|
||||
server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
|
||||
await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
|
||||
conn, tmp_path, monkeypatch=monkeypatch)
|
||||
assert [int(c[2]) for c in server.fetch_commands()] == [2]
|
||||
|
||||
async def test_empty_mailbox_first_tick_stores_a_zero_cursor(self, tmp_path, monkeypatch):
|
||||
server = gmail_server({})
|
||||
conn = FakeConn()
|
||||
await do_run({"gmail": server}, [gmail_account(initial_mode="new-only")],
|
||||
conn, tmp_path, monkeypatch=monkeypatch)
|
||||
assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 0)
|
||||
|
||||
|
||||
class TestCrossAccount:
|
||||
async def test_shared_message_id_counts_as_a_cross_source_conflict(self, tmp_path, monkeypatch):
|
||||
# Recon §2.4: one row, whichever account got there first — but the overlap must be
|
||||
# visible, or "new fastmail mails" reads low for no discoverable reason.
|
||||
servers = {
|
||||
"gmail": gmail_server({1: eml("shared@list")}),
|
||||
"fastmail": FakeIMAP({"INBOX": {1: eml("shared@list")}}),
|
||||
}
|
||||
conn = FakeConn()
|
||||
result = await do_run(servers,
|
||||
[gmail_account(initial_mode="full"),
|
||||
fastmail_account(initial_mode="full")],
|
||||
conn, tmp_path, monkeypatch=monkeypatch)
|
||||
|
||||
assert result["accounts"]["gmail"]["envelopes_inserted"] == 1
|
||||
assert result["accounts"]["fastmail"]["envelopes_inserted"] == 0
|
||||
assert result["accounts"]["fastmail"]["envelopes_conflict_other_source"] == 1
|
||||
assert conn.envelopes["shared@list"]["source"] == "gmail"
|
||||
|
||||
async def test_both_accounts_keep_their_own_archive_copy(self, tmp_path, monkeypatch):
|
||||
servers = {
|
||||
"gmail": gmail_server({1: eml("shared@list")}),
|
||||
"fastmail": FakeIMAP({"INBOX": {1: eml("shared@list")}}),
|
||||
}
|
||||
await do_run(servers,
|
||||
[gmail_account(initial_mode="full"), fastmail_account(initial_mode="full")],
|
||||
FakeConn(), tmp_path, monkeypatch=monkeypatch)
|
||||
assert (tmp_path / "gmail" / "2026" / "06" / "shared@list.eml").exists()
|
||||
assert (tmp_path / "fastmail" / "2026" / "06" / "shared@list.eml").exists()
|
||||
|
||||
async def test_a_broken_account_does_not_stop_the_other(self, tmp_path, monkeypatch):
|
||||
servers = {
|
||||
"gmail": gmail_server({1: eml("a@x")}, login_error="AUTHENTICATIONFAILED"),
|
||||
"fastmail": FakeIMAP({"INBOX": {1: eml("b@x")}}),
|
||||
}
|
||||
conn = FakeConn()
|
||||
result = await do_run(servers,
|
||||
[gmail_account(initial_mode="full"),
|
||||
fastmail_account(initial_mode="full")],
|
||||
conn, tmp_path, monkeypatch=monkeypatch)
|
||||
|
||||
assert set(conn.envelopes) == {"b@x"}
|
||||
assert result["accounts"]["gmail"]["account_errors"] == 1
|
||||
assert result["failed"] is True
|
||||
|
||||
async def test_fastmail_syncs_its_three_configured_folders(self, tmp_path, monkeypatch):
|
||||
server = FakeIMAP({"INBOX": {1: eml("i@x")}, "Archive": {1: eml("ar@x")},
|
||||
"Sent": {1: eml("s@x")}})
|
||||
conn = FakeConn()
|
||||
account = fastmail_account(folders=("INBOX", "Archive", "Sent"), initial_mode="full")
|
||||
await do_run({"fastmail": server}, [account], conn, tmp_path, monkeypatch=monkeypatch)
|
||||
assert set(conn.envelopes) == {"i@x", "ar@x", "s@x"}
|
||||
assert len(conn.state) == 3
|
||||
|
||||
|
||||
class TestVanishedAndErrors:
|
||||
async def test_expunged_uid_is_not_an_error_and_does_not_block_the_cursor(self, tmp_path, monkeypatch):
|
||||
server = gmail_server({1: eml("a@x"), 2: eml("b@x")}, vanished={1})
|
||||
conn = FakeConn()
|
||||
result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
|
||||
conn, tmp_path, monkeypatch=monkeypatch)
|
||||
assert result["totals"]["vanished"] == 1
|
||||
assert result["totals"]["errors"] == 0
|
||||
assert conn.state[("gmail", GMAIL_FOLDER)] == (42, 2)
|
||||
assert not result["failed"]
|
||||
|
||||
async def test_missing_special_use_folder_fails_the_account_loudly(self, tmp_path, monkeypatch):
|
||||
# Rather than "synced 0 mails, all good" every hour.
|
||||
server = FakeIMAP({"INBOX": {}})
|
||||
result = await do_run({"gmail": server}, [gmail_account()], FakeConn(), tmp_path,
|
||||
monkeypatch=monkeypatch)
|
||||
assert result["failed"] is True
|
||||
assert result["accounts"]["gmail"]["account_errors"] == 1
|
||||
|
||||
|
||||
class TestBalance:
|
||||
async def test_counters_balance_on_a_mixed_run(self, tmp_path, monkeypatch):
|
||||
server = gmail_server({1: eml("a@x"), 2: eml("b@x"), 3: eml("c@x"), 4: eml("d@x")},
|
||||
fetch_errors={3}, vanished={2})
|
||||
result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
|
||||
FakeConn(), tmp_path, monkeypatch=monkeypatch)
|
||||
assert balance_errors(result["totals"]) == []
|
||||
assert result["balance_problems"] == []
|
||||
|
||||
def test_balance_errors_names_each_broken_identity(self):
|
||||
broken = {"uids_seen": 5, "processed": 1, "vanished": 0, "errors": 0,
|
||||
"archived": 0, "archive_exists": 0, "envelopes_inserted": 0,
|
||||
"envelopes_skipped_dup": 0, "envelopes_conflict_other_source": 0}
|
||||
assert len(balance_errors(broken)) == 3
|
||||
|
||||
def test_total_errors_sums_all_three_levels(self):
|
||||
assert total_errors({"errors": 1, "folder_errors": 2, "account_errors": 3}) == 6
|
||||
|
||||
|
||||
class TestDryRun:
|
||||
async def test_dry_run_fetches_nothing_and_writes_nothing(self, tmp_path, monkeypatch):
|
||||
server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
|
||||
conn = FakeConn()
|
||||
result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
|
||||
conn, tmp_path, apply=False, monkeypatch=monkeypatch)
|
||||
|
||||
assert server.fetch_commands() == []
|
||||
assert conn.envelopes == {}
|
||||
assert conn.state == {}
|
||||
assert list(tmp_path.rglob("*.eml")) == []
|
||||
assert not result["failed"]
|
||||
|
||||
async def test_dry_run_reports_how_much_the_real_run_would_pull(self, tmp_path, monkeypatch):
|
||||
server = gmail_server({1: eml("a@x"), 2: eml("b@x")})
|
||||
result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
|
||||
FakeConn(), tmp_path, apply=False, monkeypatch=monkeypatch)
|
||||
folder = result["reports"][0]["folders"][0]
|
||||
assert folder["candidates"] == 2
|
||||
assert folder["mode"] == "initial-full"
|
||||
|
||||
async def test_dry_run_is_not_flagged_unbalanced(self, tmp_path, monkeypatch):
|
||||
# It never fetched, so there are no message-level outcomes to balance.
|
||||
server = gmail_server({1: eml("a@x")})
|
||||
result = await do_run({"gmail": server}, [gmail_account(initial_mode="full")],
|
||||
FakeConn(), tmp_path, apply=False, monkeypatch=monkeypatch)
|
||||
assert result["balance_problems"] == []
|
||||
|
||||
|
||||
class TestMeasure:
|
||||
async def test_reports_message_counts_per_folder(self):
|
||||
server = FakeIMAP({"INBOX": {1: eml("a@x"), 2: eml("b@x")}, "Archive": {1: eml("c@x")}})
|
||||
rows = await measure([fastmail_account(folders=("INBOX", "Archive"))],
|
||||
client_factory=factory_for({"fastmail": server}))
|
||||
assert [(r["folder"], r["messages"]) for r in rows] == [("INBOX", 2), ("Archive", 1)]
|
||||
|
||||
async def test_touches_neither_archive_nor_mail_flags(self):
|
||||
server = FakeIMAP({"INBOX": {1: eml("a@x")}})
|
||||
await measure([fastmail_account()], client_factory=factory_for({"fastmail": server}))
|
||||
assert server.fetch_commands() == []
|
||||
|
||||
async def test_a_failing_account_is_reported_not_raised(self):
|
||||
server = FakeIMAP({"INBOX": {}}, login_error="nope")
|
||||
rows = await measure([fastmail_account()], client_factory=factory_for({"fastmail": server}))
|
||||
assert "error" in rows[0]
|
||||
|
||||
|
||||
class TestMetrics:
|
||||
def _result(self, *, failed=False, inserted=3):
|
||||
return {
|
||||
"totals": {},
|
||||
"accounts": {
|
||||
"gmail": {"envelopes_inserted": inserted, "envelopes_skipped_dup": 1,
|
||||
"envelopes_conflict_other_source": 0, "uidvalidity_resets": 0,
|
||||
"errors": 0, "folder_errors": 0, "account_errors": 0},
|
||||
"fastmail": {"envelopes_inserted": 0, "envelopes_skipped_dup": 0,
|
||||
"envelopes_conflict_other_source": 2, "uidvalidity_resets": 1,
|
||||
"errors": 0, "folder_errors": 0, "account_errors": 0},
|
||||
},
|
||||
"last_message_ts": {"gmail": 1786000000.0},
|
||||
"failed": failed,
|
||||
}
|
||||
|
||||
def test_counters_are_labelled_per_account(self, tmp_path: Path):
|
||||
text = render_prom(build_metrics(self._result(), 1786000123.0,
|
||||
tmp_path / "kb-mail-sync.prom"))
|
||||
assert 'kb_mail_sync_envelopes_inserted{account="gmail"} 3' in text
|
||||
assert 'kb_mail_sync_conflict_other_source{account="fastmail"} 2' in text
|
||||
assert 'kb_mail_sync_uidvalidity_resets{account="fastmail"} 1' in text
|
||||
|
||||
def test_success_timestamp_is_set_on_a_clean_run(self, tmp_path: Path):
|
||||
text = render_prom(build_metrics(self._result(), 1786000123.0, tmp_path / "x.prom"))
|
||||
assert "kb_mail_sync_last_success_timestamp 1786000123" in text
|
||||
assert "kb_mail_sync_last_exit_code 0" in text
|
||||
|
||||
def test_success_timestamp_carries_forward_across_a_failure(self, tmp_path: Path):
|
||||
path = tmp_path / "kb-mail-sync.prom"
|
||||
path.write_text(render_prom(build_metrics(self._result(), 1786000000.0, path)))
|
||||
text = render_prom(build_metrics(self._result(failed=True), 1786009999.0, path))
|
||||
# The staleness alert measures time since the last GOOD tick — a transient failure
|
||||
# must not reset its clock, and must not pretend this run succeeded either.
|
||||
assert "kb_mail_sync_last_success_timestamp 1786000000" in text
|
||||
assert "kb_mail_sync_last_run_timestamp 1786009999" in text
|
||||
assert "kb_mail_sync_last_exit_code 1" in text
|
||||
|
||||
def test_first_ever_run_failing_leaves_the_clock_at_zero(self, tmp_path: Path):
|
||||
text = render_prom(build_metrics(self._result(failed=True), 1786000123.0,
|
||||
tmp_path / "absent.prom"))
|
||||
assert "kb_mail_sync_last_success_timestamp 0" in text
|
||||
|
||||
def test_last_message_ts_is_omitted_for_an_account_with_no_rows(self, tmp_path: Path):
|
||||
text = render_prom(build_metrics(self._result(), 1786000123.0, tmp_path / "x.prom"))
|
||||
assert 'kb_mail_sync_last_message_ts{account="gmail"}' in text
|
||||
assert 'kb_mail_sync_last_message_ts{account="fastmail"}' not in text
|
||||
|
||||
def test_read_prev_value_ignores_labelled_series(self, tmp_path: Path):
|
||||
path = tmp_path / "x.prom"
|
||||
path.write_text('kb_mail_sync_envelopes_inserted{account="gmail"} 3\n'
|
||||
"kb_mail_sync_last_success_timestamp 1786000000\n")
|
||||
assert read_prev_value(path, "kb_mail_sync_last_success_timestamp") == 1786000000.0
|
||||
assert read_prev_value(path, "kb_mail_sync_envelopes_inserted") is None
|
||||
|
|
@ -40,7 +40,12 @@ DEFAULT_SUMMARY_MODEL = "claude-haiku-4-5" # plan §2 D3 resolution 2026-07-17:
|
|||
DEFAULT_EMBED_MODEL = "bge-m3"
|
||||
DEFAULT_N = 10 # plan §6.1 start value
|
||||
DEFAULT_K = 5 # plan §6.1 start value
|
||||
DEFAULT_SUMMARYLESS_SOURCES = ("gmail",) # faza mailowa plan §6 Krok 3: sources with no summaries
|
||||
# faza mailowa plan §6 Krok 3: sources with no summaries. `fastmail` added 2026-08-06 in the
|
||||
# same commit that introduced the source (recon Decyzja (g)) — mail reaches /search results
|
||||
# ONLY through the hybrid path's summaryless leg, which filters `WHERE e.source = ANY($2)`.
|
||||
# Left out, fastmail envelopes would archive, chunk and embed correctly and then be invisible
|
||||
# to every query, with nothing anywhere reporting an error.
|
||||
DEFAULT_SUMMARYLESS_SOURCES = ("gmail", "fastmail")
|
||||
|
||||
|
||||
async def flat_retrieve(conn: asyncpg.Connection, query_vector: str, k: int = DEFAULT_K) -> list[dict]:
|
||||
|
|
|
|||
Loading…
Reference in a new issue