Invariance

Author: xormania

src/mboxer/classify.py

@@ -6,6 +6,7 @@
 from typing import Any
 
 from .naming import normalize_category_path
+from .records import decode_address_fields
 
 
 def _load_rules(config: dict[str, Any]) -> list[dict[str, Any]]:
@@ -16,10 +17,7 @@ def _match_rule(rule: dict[str, Any], record: dict[str, Any]) -> bool:
     match = rule.get("match", {})
     sender = (record.get("sender") or "").lower()
     subject = (record.get("subject") or "").lower()
-    try:
-        recipients = json.loads(record.get("recipients_json") or "[]")
-    except Exception:
-        recipients = []
+    recipients = record["recipients"]
     all_addrs = [sender] + [r.lower() for r in recipients]
 
     for domain in match.get("from_domain", []):
@@ -112,7 +110,7 @@ def _build_thread_input(
 ) -> dict[str, Any]:
     """Aggregate thread messages into a single record suitable for rule matching."""
     if not messages:
-        return {"thread_key": thread_key, "subject": "", "sender": "", "recipients_json": "[]"}
+        return {"thread_key": thread_key, "subject": "", "sender": "", "recipients": []}
 
     subject = ""
     for m in messages:
@@ -128,17 +126,14 @@ def _build_thread_input(
             break
 
     # Collect all unique participant addresses across every message in the thread.
-    # Putting all of them into recipients_json ensures _match_rule's all_addrs covers
-    # every sender domain, not just the first message's sender.
+    # Putting all non-sender participants into recipients ensures _match_rule's
+    # all_addrs covers every sender domain, not just the first message's sender.
     all_addrs: set[str] = set()
     for m in messages:
         if m.get("sender"):
             all_addrs.add(m["sender"].lower())
-        try:
-            for r in json.loads(m.get("recipients_json") or "[]"):
-                all_addrs.add(r.lower())
-        except Exception:
-            pass
+        for r in m["recipients"]:
+            all_addrs.add(r.lower())
 
     sender = messages[0].get("sender") or ""
     other_addrs = sorted(all_addrs - {sender.lower()})
@@ -151,7 +146,7 @@ def _build_thread_input(
         "thread_key": thread_key,
         "subject": subject,
         "sender": sender,
-        "recipients_json": json.dumps(other_addrs),
+        "recipients": other_addrs,
         "_participants": sorted(all_addrs),
         "_excerpts": excerpts,
         "_message_count": len(messages),
@@ -330,8 +325,8 @@ def _run_rule_classification_thread(
         if account_id is not None:
             msg_rows = conn.execute(
                 """
-                SELECT id, subject, sender, recipients_json, date_utc, body_text,
-                       thread_key, account_id
+                SELECT id, subject, sender, recipients_json, cc_json, bcc_json,
+                       date_utc, body_text, thread_key, account_id
                 FROM messages
                 WHERE thread_key = ? AND account_id = ?
                 ORDER BY date_utc NULLS LAST, id
@@ -341,8 +336,8 @@ def _run_rule_classification_thread(
         else:
             msg_rows = conn.execute(
                 """
-                SELECT id, subject, sender, recipients_json, date_utc, body_text,
-                       thread_key, account_id
+                SELECT id, subject, sender, recipients_json, cc_json, bcc_json,
+                       date_utc, body_text, thread_key, account_id
                 FROM messages
                 WHERE thread_key = ?
                 ORDER BY date_utc NULLS LAST, id
@@ -351,10 +346,10 @@ def _run_rule_classification_thread(
             ).fetchall()
 
         cols = [
-            "id", "subject", "sender", "recipients_json", "date_utc",
-            "body_text", "thread_key", "account_id",
+            "id", "subject", "sender", "recipients_json", "cc_json", "bcc_json",
+            "date_utc", "body_text", "thread_key", "account_id",
         ]
-        messages = [dict(zip(cols, row)) for row in msg_rows]
+        messages = [decode_address_fields(dict(zip(cols, row))) for row in msg_rows]
         if not messages:
             continue
 
@@ -404,7 +399,7 @@ def run_rule_classification(
 
     query = """
         SELECT m.id, m.account_id, m.source_id, m.mbox_key, m.message_id, m.thread_key,
-               m.subject, m.sender, m.recipients_json, m.date_utc
+               m.subject, m.sender, m.recipients_json, m.cc_json, m.bcc_json, m.date_utc
         FROM messages m
         LEFT JOIN classifications c
           ON c.message_db_id = m.id AND c.classifier_type IN ('rule', 'rule_hint')
@@ -418,9 +413,9 @@ def run_rule_classification(
     rows = conn.execute(query, params).fetchall()
     cols = [
         "id", "account_id", "source_id", "mbox_key", "message_id", "thread_key",
-        "subject", "sender", "recipients_json", "date_utc",
+        "subject", "sender", "recipients_json", "cc_json", "bcc_json", "date_utc",
     ]
-    records = [dict(zip(cols, row)) for row in rows]
+    records = [decode_address_fields(dict(zip(cols, row))) for row in rows]
 
     classified = 0
     skipped = 0

src/mboxer/db/migrations/003_address_invariant.sql

@@ -0,0 +1,96 @@
+-- Migration 003: enforce JSON-array invariants for message address columns.
+-- SQLite cannot add NOT NULL/CHECK constraints to existing columns in place,
+-- so rebuild messages and recreate its indexes.
+
+PRAGMA foreign_keys = OFF;
+
+CREATE TABLE messages_new (
+    id INTEGER PRIMARY KEY,
+    source_id INTEGER NOT NULL,
+    mbox_key TEXT NOT NULL,
+    message_id TEXT,
+    thread_key TEXT,
+    subject TEXT,
+    sender TEXT,
+    recipients_json TEXT NOT NULL DEFAULT '[]' CHECK (json_type(recipients_json) = 'array'),
+    cc_json TEXT NOT NULL DEFAULT '[]' CHECK (json_type(cc_json) = 'array'),
+    bcc_json TEXT NOT NULL DEFAULT '[]' CHECK (json_type(bcc_json) = 'array'),
+    date_header TEXT,
+    date_utc TEXT,
+    body_text TEXT,
+    body_html TEXT,
+    body_hash TEXT,
+    body_chars INTEGER DEFAULT 0,
+    body_word_count INTEGER DEFAULT 0,
+    attachment_count INTEGER DEFAULT 0,
+    raw_headers_json TEXT,
+    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    account_id INTEGER REFERENCES accounts(id),
+    FOREIGN KEY(source_id) REFERENCES mbox_sources(id),
+    UNIQUE(source_id, mbox_key)
+);
+
+INSERT INTO messages_new (
+    id,
+    source_id,
+    mbox_key,
+    message_id,
+    thread_key,
+    subject,
+    sender,
+    recipients_json,
+    cc_json,
+    bcc_json,
+    date_header,
+    date_utc,
+    body_text,
+    body_html,
+    body_hash,
+    body_chars,
+    body_word_count,
+    attachment_count,
+    raw_headers_json,
+    created_at,
+    updated_at,
+    account_id
+)
+SELECT
+    id,
+    source_id,
+    mbox_key,
+    message_id,
+    thread_key,
+    subject,
+    sender,
+    recipients_json,
+    cc_json,
+    bcc_json,
+    date_header,
+    date_utc,
+    body_text,
+    body_html,
+    body_hash,
+    body_chars,
+    body_word_count,
+    attachment_count,
+    raw_headers_json,
+    created_at,
+    updated_at,
+    account_id
+FROM messages;
+
+DROP TABLE messages;
+ALTER TABLE messages_new RENAME TO messages;
+
+CREATE INDEX IF NOT EXISTS idx_messages_message_id ON messages(message_id);
+CREATE INDEX IF NOT EXISTS idx_messages_thread_key ON messages(thread_key);
+CREATE INDEX IF NOT EXISTS idx_messages_date_utc ON messages(date_utc);
+CREATE INDEX IF NOT EXISTS idx_messages_body_hash ON messages(body_hash);
+CREATE INDEX IF NOT EXISTS idx_messages_account ON messages(account_id);
+CREATE INDEX IF NOT EXISTS idx_messages_account_message_id ON messages(account_id, message_id);
+CREATE INDEX IF NOT EXISTS idx_messages_account_thread_key ON messages(account_id, thread_key);
+CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_account_source_mbox_key
+ON messages(account_id, source_id, mbox_key) WHERE account_id IS NOT NULL;
+
+PRAGMA foreign_keys = ON;

src/mboxer/db/schema.sql

@@ -74,9 +74,9 @@ CREATE TABLE IF NOT EXISTS messages (
     thread_key TEXT,
     subject TEXT,
     sender TEXT,
-    recipients_json TEXT,
-    cc_json TEXT,
-    bcc_json TEXT,
+    recipients_json TEXT NOT NULL DEFAULT '[]' CHECK (json_type(recipients_json) = 'array'),
+    cc_json TEXT NOT NULL DEFAULT '[]' CHECK (json_type(cc_json) = 'array'),
+    bcc_json TEXT NOT NULL DEFAULT '[]' CHECK (json_type(bcc_json) = 'array'),
     date_header TEXT,
     date_utc TEXT,
     body_text TEXT,

src/mboxer/exporters/jsonl.py

@@ -6,6 +6,7 @@
 from pathlib import Path
 from typing import Any
 
+from ..records import decode_address_fields
 from ..security.findings import ResidualFindingsBlocked, merge_counts
 from ..security.policy import default_export_profile, resolve_export_profile, resolve_findings_policy
 from .projection import prepare_projection
@@ -37,7 +38,7 @@ def export_jsonl(
         rows = conn.execute(
             """
             SELECT m.id, m.message_id, m.thread_key, m.subject, m.sender,
-                   m.recipients_json, m.cc_json, m.date_utc,
+                   m.recipients_json, m.cc_json, m.bcc_json, m.date_utc,
                    m.body_text, m.body_hash, m.body_chars, m.body_word_count,
                    m.attachment_count, s.source_name, s.source_slug
             FROM messages m
@@ -51,7 +52,7 @@ def export_jsonl(
         rows = conn.execute(
             """
             SELECT m.id, m.message_id, m.thread_key, m.subject, m.sender,
-                   m.recipients_json, m.cc_json, m.date_utc,
+                   m.recipients_json, m.cc_json, m.bcc_json, m.date_utc,
                    m.body_text, m.body_hash, m.body_chars, m.body_word_count,
                    m.attachment_count, s.source_name, s.source_slug
             FROM messages m
@@ -62,7 +63,7 @@ def export_jsonl(
 
     cols = [
         "id", "message_id", "thread_key", "subject", "sender",
-        "recipients_json", "cc_json", "date_utc",
+        "recipients_json", "cc_json", "bcc_json", "date_utc",
         "body_text", "body_hash", "body_chars", "body_word_count",
         "attachment_count", "source_name", "source_slug",
     ]
@@ -118,12 +119,7 @@ def export_jsonl(
             any_scrubbed = True
 
         record["account_key"] = account_key
-        try:
-            record["recipients"] = json.loads(record.pop("recipients_json") or "[]")
-            record["cc"] = json.loads(record.pop("cc_json") or "[]")
-        except Exception:
-            record["recipients"] = []
-            record["cc"] = []
+        record = decode_address_fields(record)
 
         if include_classification and record["id"] in classifications:
             record["classification"] = classifications[record["id"]]

src/mboxer/ingest.py

@@ -13,6 +13,7 @@
 from .db import init_db
 from .naming import slugify
 from .normalize import normalize_message
+from .records import loads_address_list
 
 
 class SourceIdentityError(RuntimeError):
@@ -409,7 +410,7 @@ def ingest_mbox(
                             raise RuntimeError("INSERT succeeded but cursor.lastrowid is None")
 
                         if record.get("thread_key"):
-                            participants = json.loads(record.get("recipients_json") or "[]")
+                            participants = loads_address_list(record["recipients_json"])
                             if record.get("sender"):
                                 participants = [record["sender"]] + participants
                             _upsert_thread(

src/mboxer/normalize.py

@@ -26,6 +26,7 @@ def _decode_header_value(value: str | None) -> str:
 
 
 def _parse_address_list(header_value: str | None) -> list[str]:
+    """Return bare lowercased addresses for one address-list header, preserving order."""
     if not header_value:
         return []
     addrs: list[str] = []

src/mboxer/records.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+_ADDRESS_FIELDS = (
+    ("recipients_json", "recipients"),
+    ("cc_json", "cc"),
+    ("bcc_json", "bcc"),
+)
+
+
+def loads_address_list(value: str) -> list[str]:
+    """Decode a DB-backed address-list JSON value into a strict list of strings."""
+    parsed = json.loads(value)
+    if not isinstance(parsed, list):
+        raise ValueError("address-list JSON must be an array")
+    if not all(isinstance(item, str) for item in parsed):
+        raise ValueError("address-list JSON elements must be strings")
+    return parsed
+
+
+def decode_address_fields(record: dict[str, Any]) -> dict[str, Any]:
+    """Replace address-list JSON fields on a materialized row with typed lists."""
+    for json_field, list_field in _ADDRESS_FIELDS:
+        if json_field in record:
+            record[list_field] = loads_address_list(record.pop(json_field))
+    return record