"""SQLite teaching knowledge base: versions, outbox, publication pointers, ACLs, and deletion propagation.

This implementation demonstrates offline lifecycle contracts. It does not include
vector retrieval, identity providers, distributed concurrency control, backup
erasure, or compliance-grade physical deletion.
"""

from __future__ import annotations

from contextlib import contextmanager
from dataclasses import asdict, dataclass
from hashlib import sha256
import json
import sqlite3
import sys
from typing import Any, Iterator, Sequence
import unicodedata


SCHEMA_VERSION = "1.0"
MAX_CONTENT_BYTES = 100_000
MAX_ACL_GROUPS = 128
MAX_SUBJECT_GROUPS = 128
MAX_SEARCH_RESULTS = 100
DELETE_REASONS = {
    "access_revoked",
    "retention_expired",
    "source_deleted",
    "user_request",
}

SCHEMA = """
CREATE TABLE IF NOT EXISTS documents (
    tenant_id TEXT NOT NULL,
    document_id TEXT NOT NULL,
    last_source_sequence INTEGER NOT NULL CHECK (last_source_sequence >= 0),
    current_revision_id INTEGER,
    published_revision_id INTEGER,
    current_event_version INTEGER NOT NULL CHECK (current_event_version >= 0),
    deleted INTEGER NOT NULL DEFAULT 0 CHECK (deleted IN (0, 1)),
    access_blocked INTEGER NOT NULL DEFAULT 1 CHECK (access_blocked IN (0, 1)),
    PRIMARY KEY (tenant_id, document_id)
);

CREATE TABLE IF NOT EXISTS revisions (
    revision_id INTEGER PRIMARY KEY AUTOINCREMENT,
    tenant_id TEXT NOT NULL,
    document_id TEXT NOT NULL,
    revision_number INTEGER NOT NULL CHECK (revision_number > 0),
    source_sequence INTEGER NOT NULL CHECK (source_sequence > 0),
    source_uri TEXT NOT NULL,
    source_version TEXT NOT NULL,
    pipeline_version TEXT NOT NULL,
    content TEXT,
    content_hash TEXT NOT NULL,
    source_state_hash TEXT NOT NULL,
    build_state_hash TEXT NOT NULL,
    run_id TEXT NOT NULL,
    content_purged INTEGER NOT NULL DEFAULT 0 CHECK (content_purged IN (0, 1)),
    UNIQUE (tenant_id, document_id, revision_number),
    FOREIGN KEY (tenant_id, document_id)
        REFERENCES documents(tenant_id, document_id)
);

CREATE TABLE IF NOT EXISTS revision_acl (
    revision_id INTEGER NOT NULL,
    group_id TEXT NOT NULL,
    PRIMARY KEY (revision_id, group_id),
    FOREIGN KEY (revision_id) REFERENCES revisions(revision_id) ON DELETE CASCADE
);

CREATE TABLE IF NOT EXISTS outbox (
    outbox_id INTEGER PRIMARY KEY AUTOINCREMENT,
    event_id TEXT NOT NULL UNIQUE,
    tenant_id TEXT NOT NULL,
    document_id TEXT NOT NULL,
    event_version INTEGER NOT NULL CHECK (event_version > 0),
    event_kind TEXT NOT NULL CHECK (event_kind IN ('document.upserted', 'document.deleted')),
    revision_id INTEGER,
    processed INTEGER NOT NULL DEFAULT 0 CHECK (processed IN (0, 1)),
    attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0),
    last_error TEXT,
    UNIQUE (tenant_id, document_id, event_version),
    FOREIGN KEY (revision_id) REFERENCES revisions(revision_id)
);

CREATE TABLE IF NOT EXISTS tombstones (
    tenant_id TEXT NOT NULL,
    document_id TEXT NOT NULL,
    event_version INTEGER NOT NULL,
    source_sequence INTEGER NOT NULL,
    reason_code TEXT NOT NULL,
    run_id TEXT NOT NULL,
    canonical_content_purged INTEGER NOT NULL DEFAULT 0
        CHECK (canonical_content_purged IN (0, 1)),
    PRIMARY KEY (tenant_id, document_id, event_version)
);

CREATE TABLE IF NOT EXISTS search_revisions (
    revision_id INTEGER PRIMARY KEY,
    tenant_id TEXT NOT NULL,
    document_id TEXT NOT NULL,
    revision_number INTEGER NOT NULL,
    content TEXT NOT NULL,
    content_hash TEXT NOT NULL,
    FOREIGN KEY (revision_id) REFERENCES revisions(revision_id) ON DELETE CASCADE
);

CREATE TABLE IF NOT EXISTS search_acl (
    revision_id INTEGER NOT NULL,
    group_id TEXT NOT NULL,
    PRIMARY KEY (revision_id, group_id),
    FOREIGN KEY (revision_id) REFERENCES search_revisions(revision_id) ON DELETE CASCADE
);

CREATE INDEX IF NOT EXISTS idx_outbox_pending ON outbox(processed, outbox_id);
CREATE INDEX IF NOT EXISTS idx_search_document
    ON search_revisions(tenant_id, document_id);
"""


class StoreError(ValueError):
    """Knowledge-base input, ordering, or state conflict."""


@dataclass(frozen=True)
class SourceRecord:
    tenant_id: str
    document_id: str
    source_sequence: int
    source_uri: str
    source_version: str
    content: str
    allowed_groups: tuple[str, ...]


@dataclass(frozen=True)
class BuildConfig:
    pipeline_version: str


@dataclass(frozen=True)
class ChangeResult:
    action: str
    event_id: str | None
    revision_number: int | None


@dataclass(frozen=True)
class ProjectionResult:
    action: str
    event_id: str | None


def _canonical_json(value: Any) -> str:
    return json.dumps(
        value,
        ensure_ascii=False,
        allow_nan=False,
        sort_keys=True,
        separators=(",", ":"),
    )


def _digest_text(value: str) -> str:
    return sha256(value.encode("utf-8")).hexdigest()


def _normalise_content(value: str) -> str:
    return unicodedata.normalize(
        "NFC", value.replace("\r\n", "\n").replace("\r", "\n")
    )


def _validate_token(name: str, value: str, *, maximum: int = 200) -> str:
    if not isinstance(value, str):
        raise StoreError(f"{name} must be a string")
    if not value or value != value.strip():
        raise StoreError(f"{name} must not be empty or have leading/trailing whitespace")
    if len(value) > maximum:
        raise StoreError(f"{name} must not exceed {maximum} characters")
    if any(ord(character) < 32 for character in value):
        raise StoreError(f"{name} must not contain control characters")
    return value


def _validate_sequence(value: int) -> int:
    if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
        raise StoreError("source_sequence must be a positive integer")
    return value


def _normalise_record(record: SourceRecord) -> SourceRecord:
    tenant_id = _validate_token("tenant_id", record.tenant_id)
    document_id = _validate_token("document_id", record.document_id)
    source_uri = _validate_token("source_uri", record.source_uri, maximum=2_000)
    source_version = _validate_token("source_version", record.source_version)
    source_sequence = _validate_sequence(record.source_sequence)
    if not isinstance(record.content, str):
        raise StoreError("content must be a string")
    content = _normalise_content(record.content)
    if not content.strip():
        raise StoreError("content must not be empty")
    if len(content.encode("utf-8")) > MAX_CONTENT_BYTES:
        raise StoreError(f"content must not exceed {MAX_CONTENT_BYTES} UTF-8 bytes")
    if not isinstance(record.allowed_groups, tuple) or not record.allowed_groups:
        raise StoreError("allowed_groups must be a nonempty tuple; public content still requires explicit groups")
    if len(record.allowed_groups) > MAX_ACL_GROUPS:
        raise StoreError(f"allowed_groups must not exceed {MAX_ACL_GROUPS} entries")
    validated_groups = tuple(
        _validate_token("group_id", group) for group in record.allowed_groups
    )
    if len(set(validated_groups)) != len(validated_groups):
        raise StoreError("allowed_groups must not contain duplicates")
    groups = tuple(sorted(validated_groups))
    return SourceRecord(
        tenant_id,
        document_id,
        source_sequence,
        source_uri,
        source_version,
        content,
        groups,
    )


def _validate_config(config: BuildConfig) -> BuildConfig:
    return BuildConfig(_validate_token("pipeline_version", config.pipeline_version))


def _source_state(record: SourceRecord) -> dict[str, Any]:
    return {
        "allowed_groups": list(record.allowed_groups),
        "content": record.content,
        "source_uri": record.source_uri,
        "source_version": record.source_version,
    }


def _state_hashes(record: SourceRecord, config: BuildConfig) -> tuple[str, str, str]:
    content_hash = _digest_text(record.content)
    source_state_hash = _digest_text(_canonical_json(_source_state(record)))
    build_state_hash = _digest_text(
        _canonical_json(
            {
                "pipeline_version": config.pipeline_version,
                "source_state_hash": source_state_hash,
            }
        )
    )
    return content_hash, source_state_hash, build_state_hash


def _persisted_revision_hash_errors(
    revision: sqlite3.Row, groups: tuple[str, ...]
) -> tuple[str, ...]:
    """Recompute canonical hashes from persisted values instead of trusting labels."""

    content = revision["content"]
    if not isinstance(content, str):
        return ("canonical_content_missing",)
    content_hash = _digest_text(content)
    source_state_hash = _digest_text(
        _canonical_json(
            {
                "allowed_groups": list(groups),
                "content": content,
                "source_uri": revision["source_uri"],
                "source_version": revision["source_version"],
            }
        )
    )
    build_state_hash = _digest_text(
        _canonical_json(
            {
                "pipeline_version": revision["pipeline_version"],
                "source_state_hash": source_state_hash,
            }
        )
    )
    errors: list[str] = []
    if revision["content_hash"] != content_hash:
        errors.append("canonical_content_hash_mismatch")
    if revision["source_state_hash"] != source_state_hash:
        errors.append("canonical_source_state_hash_mismatch")
    if revision["build_state_hash"] != build_state_hash:
        errors.append("canonical_build_state_hash_mismatch")
    return tuple(errors)


def connect(path: str = ":memory:") -> sqlite3.Connection:
    connection = sqlite3.connect(path, isolation_level=None)
    connection.row_factory = sqlite3.Row
    connection.execute("PRAGMA foreign_keys = ON")
    connection.executescript(SCHEMA)
    return connection


@contextmanager
def _transaction(connection: sqlite3.Connection) -> Iterator[None]:
    if connection.in_transaction:
        raise RuntimeError("nested transactions are not supported in this teaching implementation; production code must define a savepoint policy")
    connection.execute("BEGIN IMMEDIATE")
    try:
        yield
    except BaseException:
        connection.rollback()
        raise
    else:
        connection.commit()


def _document(
    connection: sqlite3.Connection, tenant_id: str, document_id: str
) -> sqlite3.Row | None:
    return connection.execute(
        """
        SELECT * FROM documents
        WHERE tenant_id = ? AND document_id = ?
        """,
        (tenant_id, document_id),
    ).fetchone()


def _revision(
    connection: sqlite3.Connection, revision_id: int | None
) -> sqlite3.Row | None:
    if revision_id is None:
        return None
    return connection.execute(
        "SELECT * FROM revisions WHERE revision_id = ?", (revision_id,)
    ).fetchone()


def _revision_groups(connection: sqlite3.Connection, revision_id: int) -> tuple[str, ...]:
    rows = connection.execute(
        "SELECT group_id FROM revision_acl WHERE revision_id = ? ORDER BY group_id",
        (revision_id,),
    ).fetchall()
    return tuple(row["group_id"] for row in rows)


def _event_id(
    tenant_id: str, document_id: str, event_version: int, event_kind: str
) -> str:
    payload = _canonical_json(
        {
            "document_id": document_id,
            "event_kind": event_kind,
            "event_version": event_version,
            "tenant_id": tenant_id,
        }
    )
    return f"evt_{_digest_text(payload)}"


def upsert_record(
    connection: sqlite3.Connection,
    record: SourceRecord,
    config: BuildConfig,
    *,
    run_id: str,
) -> ChangeResult:
    record = _normalise_record(record)
    config = _validate_config(config)
    run_id = _validate_token("run_id", run_id)
    content_hash, source_state_hash, build_state_hash = _state_hashes(record, config)

    with _transaction(connection):
        document = _document(connection, record.tenant_id, record.document_id)
        if document is None:
            connection.execute(
                """
                INSERT INTO documents(
                    tenant_id, document_id, last_source_sequence,
                    current_event_version, deleted, access_blocked
                ) VALUES (?, ?, 0, 0, 0, 1)
                """,
                (record.tenant_id, record.document_id),
            )
            document = _document(connection, record.tenant_id, record.document_id)
        if document is None:
            raise RuntimeError("failed to read the documents row after creation")

        last_sequence = int(document["last_source_sequence"])
        current = _revision(connection, document["current_revision_id"])
        if record.source_sequence < last_sequence:
            return ChangeResult("stale_ignored", None, None)

        if record.source_sequence == last_sequence:
            if bool(document["deleted"]):
                raise StoreError("the same source_sequence cannot resurrect a deleted document")
            if current is None:
                raise StoreError("the current document has no revision; cannot validate the same-sequence event")
            if current["source_state_hash"] != source_state_hash:
                raise StoreError("the same source_sequence carries a different source state")
            if current["build_state_hash"] == build_state_hash:
                return ChangeResult("noop", None, int(current["revision_number"]))
            action = "reprocessed"
        else:
            if (
                current is not None
                and not bool(document["deleted"])
                and current["source_state_hash"] == source_state_hash
                and current["build_state_hash"] == build_state_hash
            ):
                connection.execute(
                    """
                    UPDATE documents SET last_source_sequence = ?
                    WHERE tenant_id = ? AND document_id = ?
                    """,
                    (record.source_sequence, record.tenant_id, record.document_id),
                )
                return ChangeResult(
                    "checkpoint_advanced", None, int(current["revision_number"])
                )
            if current is None:
                action = "inserted"
            elif bool(document["deleted"]):
                action = "resurrected"
            else:
                action = "updated"

        previous_groups = () if current is None else _revision_groups(
            connection, int(current["revision_id"])
        )
        acl_changed = previous_groups != record.allowed_groups
        revision_number = 1 if current is None else int(current["revision_number"]) + 1
        cursor = connection.execute(
            """
            INSERT INTO revisions(
                tenant_id, document_id, revision_number, source_sequence,
                source_uri, source_version, pipeline_version, content,
                content_hash, source_state_hash, build_state_hash, run_id
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                record.tenant_id,
                record.document_id,
                revision_number,
                record.source_sequence,
                record.source_uri,
                record.source_version,
                config.pipeline_version,
                record.content,
                content_hash,
                source_state_hash,
                build_state_hash,
                run_id,
            ),
        )
        revision_id = int(cursor.lastrowid)
        connection.executemany(
            "INSERT INTO revision_acl(revision_id, group_id) VALUES (?, ?)",
            [(revision_id, group) for group in record.allowed_groups],
        )

        event_version = int(document["current_event_version"]) + 1
        event_kind = "document.upserted"
        event_id = _event_id(
            record.tenant_id, record.document_id, event_version, event_kind
        )
        must_block = (
            bool(document["deleted"])
            or acl_changed
            or document["published_revision_id"] is None
            or bool(document["access_blocked"])
        )
        connection.execute(
            """
            UPDATE documents SET
                last_source_sequence = ?,
                current_revision_id = ?,
                current_event_version = ?,
                deleted = 0,
                access_blocked = ?
            WHERE tenant_id = ? AND document_id = ?
            """,
            (
                record.source_sequence,
                revision_id,
                event_version,
                int(must_block),
                record.tenant_id,
                record.document_id,
            ),
        )
        connection.execute(
            """
            INSERT INTO outbox(
                event_id, tenant_id, document_id, event_version,
                event_kind, revision_id
            ) VALUES (?, ?, ?, ?, ?, ?)
            """,
            (
                event_id,
                record.tenant_id,
                record.document_id,
                event_version,
                event_kind,
                revision_id,
            ),
        )
        return ChangeResult(action, event_id, revision_number)


def delete_document(
    connection: sqlite3.Connection,
    tenant_id: str,
    document_id: str,
    source_sequence: int,
    *,
    reason_code: str,
    run_id: str,
) -> ChangeResult:
    tenant_id = _validate_token("tenant_id", tenant_id)
    document_id = _validate_token("document_id", document_id)
    source_sequence = _validate_sequence(source_sequence)
    run_id = _validate_token("run_id", run_id)
    if reason_code not in DELETE_REASONS:
        raise StoreError(f"reason_code must be one of: {', '.join(sorted(DELETE_REASONS))}")

    with _transaction(connection):
        document = _document(connection, tenant_id, document_id)
        if document is None:
            connection.execute(
                """
                INSERT INTO documents(
                    tenant_id, document_id, last_source_sequence,
                    current_event_version, deleted, access_blocked
                ) VALUES (?, ?, 0, 0, 0, 1)
                """,
                (tenant_id, document_id),
            )
            document = _document(connection, tenant_id, document_id)
        if document is None:
            raise RuntimeError("failed to read the documents row after creation")

        last_sequence = int(document["last_source_sequence"])
        if source_sequence < last_sequence:
            return ChangeResult("stale_ignored", None, None)
        if source_sequence == last_sequence:
            if bool(document["deleted"]):
                tombstone = connection.execute(
                    """
                    SELECT reason_code FROM tombstones
                    WHERE tenant_id = ? AND document_id = ? AND event_version = ?
                    """,
                    (tenant_id, document_id, document["current_event_version"]),
                ).fetchone()
                if tombstone is None:
                    raise StoreError("the deleted state lacks its current tombstone; cannot confirm idempotent replay")
                if tombstone["reason_code"] != reason_code:
                    raise StoreError("delete reason conflicts at the same source_sequence")
                return ChangeResult("noop", None, None)
            raise StoreError("the same source_sequence cannot represent both upsert and delete")

        event_version = int(document["current_event_version"]) + 1
        event_kind = "document.deleted"
        event_id = _event_id(tenant_id, document_id, event_version, event_kind)
        connection.execute(
            """
            UPDATE documents SET
                last_source_sequence = ?,
                current_event_version = ?,
                deleted = 1,
                access_blocked = 1
            WHERE tenant_id = ? AND document_id = ?
            """,
            (source_sequence, event_version, tenant_id, document_id),
        )
        connection.execute(
            """
            INSERT INTO tombstones(
                tenant_id, document_id, event_version, source_sequence,
                reason_code, run_id
            ) VALUES (?, ?, ?, ?, ?, ?)
            """,
            (
                tenant_id,
                document_id,
                event_version,
                source_sequence,
                reason_code,
                run_id,
            ),
        )
        connection.execute(
            """
            INSERT INTO outbox(
                event_id, tenant_id, document_id, event_version, event_kind
            ) VALUES (?, ?, ?, ?, ?)
            """,
            (event_id, tenant_id, document_id, event_version, event_kind),
        )
        return ChangeResult("deleted", event_id, None)


def next_pending_event(connection: sqlite3.Connection) -> sqlite3.Row | None:
    return connection.execute(
        """
        SELECT * FROM outbox
        WHERE processed = 0
        ORDER BY outbox_id
        LIMIT 1
        """
    ).fetchone()


def project_next_event(
    connection: sqlite3.Connection, *, fail_event_id: str | None = None
) -> ProjectionResult:
    event = next_pending_event(connection)
    if event is None:
        return ProjectionResult("empty", None)
    event_id = str(event["event_id"])
    if fail_event_id == event_id:
        with _transaction(connection):
            connection.execute(
                """
                UPDATE outbox SET attempts = attempts + 1, last_error = ?
                WHERE event_id = ? AND processed = 0
                """,
                ("simulated_projection_failure", event_id),
            )
        return ProjectionResult("failed", event_id)

    with _transaction(connection):
        event = connection.execute(
            "SELECT * FROM outbox WHERE event_id = ? AND processed = 0",
            (event_id,),
        ).fetchone()
        if event is None:
            return ProjectionResult("already_processed", event_id)
        document = _document(
            connection, str(event["tenant_id"]), str(event["document_id"])
        )
        if document is None:
            raise RuntimeError("outbox references a nonexistent documents row")

        if event["event_kind"] == "document.upserted":
            if event["revision_id"] is None:
                raise RuntimeError("upsert event lacks a revision pointer")
            revision = _revision(connection, int(event["revision_id"]))
            if revision is None or revision["content"] is None:
                raise RuntimeError("upsert event lacks projectable revision content")
            if (
                revision["tenant_id"] != event["tenant_id"]
                or revision["document_id"] != event["document_id"]
            ):
                raise RuntimeError("outbox upsert revision does not match the event identity")
            is_current = (
                not bool(document["deleted"])
                and int(document["current_event_version"]) == int(event["event_version"])
                and document["current_revision_id"] is not None
                and int(document["current_revision_id"]) == int(revision["revision_id"])
            )
            # Superseded revisions must not be materialized, or they could survive a later deletion or expose content that was never publishable.
            if is_current:
                connection.execute(
                    """
                    INSERT INTO search_revisions(
                        revision_id, tenant_id, document_id, revision_number,
                        content, content_hash
                    ) VALUES (?, ?, ?, ?, ?, ?)
                    ON CONFLICT(revision_id) DO UPDATE SET
                        content = excluded.content,
                        content_hash = excluded.content_hash
                    """,
                    (
                        revision["revision_id"],
                        revision["tenant_id"],
                        revision["document_id"],
                        revision["revision_number"],
                        revision["content"],
                        revision["content_hash"],
                    ),
                )
                connection.execute(
                    "DELETE FROM search_acl WHERE revision_id = ?",
                    (revision["revision_id"],),
                )
                groups = _revision_groups(connection, int(revision["revision_id"]))
                connection.executemany(
                    "INSERT INTO search_acl(revision_id, group_id) VALUES (?, ?)",
                    [(revision["revision_id"], group) for group in groups],
                )
                connection.execute(
                    """
                    UPDATE documents SET
                        published_revision_id = ?, access_blocked = 0
                    WHERE tenant_id = ? AND document_id = ?
                    """,
                    (
                        revision["revision_id"],
                        event["tenant_id"],
                        event["document_id"],
                    ),
                )
        else:
            is_current_delete = (
                bool(document["deleted"])
                and int(document["current_event_version"]) == int(event["event_version"])
            )
            if is_current_delete:
                revision_ids = connection.execute(
                    """
                    SELECT revision_id FROM search_revisions
                    WHERE tenant_id = ? AND document_id = ?
                    """,
                    (event["tenant_id"], event["document_id"]),
                ).fetchall()
                connection.executemany(
                    "DELETE FROM search_revisions WHERE revision_id = ?",
                    [(row["revision_id"],) for row in revision_ids],
                )
                connection.execute(
                    """
                    UPDATE documents SET published_revision_id = NULL
                    WHERE tenant_id = ? AND document_id = ?
                    """,
                    (event["tenant_id"], event["document_id"]),
                )

        connection.execute(
            """
            UPDATE outbox SET
                processed = 1, attempts = attempts + 1, last_error = NULL
            WHERE event_id = ?
            """,
            (event_id,),
        )
    return ProjectionResult("projected", event_id)


def drain_outbox(connection: sqlite3.Connection) -> list[str]:
    event_ids: list[str] = []
    while True:
        result = project_next_event(connection)
        if result.action == "empty":
            return event_ids
        if result.action != "projected" or result.event_id is None:
            raise RuntimeError(f"unable to drain outbox: {result.action}")
        event_ids.append(result.event_id)


def search_visible(
    connection: sqlite3.Connection,
    *,
    tenant_id: str,
    subject_groups: Sequence[str],
    term: str,
    limit: int = MAX_SEARCH_RESULTS,
) -> list[dict[str, Any]]:
    tenant_id = _validate_token("tenant_id", tenant_id)
    term = _validate_token("term", term, maximum=500)
    if isinstance(subject_groups, (str, bytes)) or not isinstance(
        subject_groups, Sequence
    ):
        raise StoreError("subject_groups must be a sequence of strings")
    if len(subject_groups) > MAX_SUBJECT_GROUPS:
        raise StoreError(f"subject_groups must not exceed {MAX_SUBJECT_GROUPS} entries")
    if (
        not isinstance(limit, int)
        or isinstance(limit, bool)
        or not 1 <= limit <= MAX_SEARCH_RESULTS
    ):
        raise StoreError(f"limit must be an integer in 1..{MAX_SEARCH_RESULTS}")
    groups = tuple(
        sorted({_validate_token("group_id", group) for group in subject_groups})
    )
    if not groups:
        return []
    placeholders = ",".join("?" for _ in groups)
    rows = connection.execute(
        f"""
        SELECT
            d.document_id,
            s.revision_number,
            s.revision_id,
            s.content AS search_content,
            s.content_hash AS search_content_hash,
            r.content,
            r.content_hash,
            r.source_uri,
            r.source_version,
            r.pipeline_version,
            r.source_state_hash,
            r.build_state_hash
        FROM documents AS d
        JOIN search_revisions AS s
          ON s.revision_id = d.published_revision_id
         AND s.tenant_id = d.tenant_id
         AND s.document_id = d.document_id
        JOIN revisions AS r
          ON r.revision_id = d.published_revision_id
         AND r.tenant_id = d.tenant_id
         AND r.document_id = d.document_id
        WHERE d.tenant_id = ?
          AND d.deleted = 0
          AND d.access_blocked = 0
          AND EXISTS (
              SELECT 1 FROM revision_acl AS a
              WHERE a.revision_id = r.revision_id
                AND a.group_id IN ({placeholders})
          )
          AND instr(lower(s.content), lower(?)) > 0
        ORDER BY d.document_id
        LIMIT ?
        """,
        (tenant_id, *groups, term, limit + 1),
    ).fetchall()
    if len(rows) > limit:
        raise StoreError("candidate window exceeds limit; narrow the query or paginate explicitly")
    result: list[dict[str, Any]] = []
    requested_groups = set(groups)
    for row in rows:
        revision_id = int(row["revision_id"])
        canonical_groups = _revision_groups(connection, revision_id)
        search_groups = tuple(
            item["group_id"]
            for item in connection.execute(
                "SELECT group_id FROM search_acl WHERE revision_id = ? ORDER BY group_id",
                (revision_id,),
            ).fetchall()
        )
        integrity_errors = list(
            _persisted_revision_hash_errors(row, canonical_groups)
        )
        search_content = row["search_content"]
        if (
            not isinstance(search_content, str)
            or _digest_text(search_content) != row["search_content_hash"]
            or row["search_content_hash"] != row["content_hash"]
        ):
            integrity_errors.append("search_content_hash_mismatch")
        if search_groups != canonical_groups:
            integrity_errors.append("published_acl_mismatch")
        if integrity_errors:
            raise StoreError(
                "search candidate integrity check failed: "
                + ",".join(sorted(set(integrity_errors)))
            )
        if not requested_groups.intersection(canonical_groups):
            continue
        result.append(
            {
                "document_id": row["document_id"],
                "revision_number": row["revision_number"],
                "content": search_content,
            }
        )
    return result


def reconcile(connection: sqlite3.Connection) -> dict[str, int]:
    scalar_queries = {
        "active_documents": "SELECT count(*) FROM documents WHERE deleted = 0",
        "blocked_active_documents": (
            "SELECT count(*) FROM documents WHERE deleted = 0 AND access_blocked = 1"
        ),
        "pending_events": "SELECT count(*) FROM outbox WHERE processed = 0",
        "revisions": "SELECT count(*) FROM revisions",
        "search_revisions": "SELECT count(*) FROM search_revisions",
        "tombstones": "SELECT count(*) FROM tombstones",
        "unpublished_active_documents": """
            SELECT count(*) FROM documents
            WHERE deleted = 0 AND published_revision_id IS NULL
        """,
        "published_not_current_documents": """
            SELECT count(*) FROM documents
            WHERE deleted = 0
              AND published_revision_id IS NOT NULL
              AND published_revision_id <> current_revision_id
        """,
        "missing_published_projection": """
            SELECT count(*) FROM documents AS d
            WHERE d.published_revision_id IS NOT NULL
              AND NOT EXISTS (
                  SELECT 1 FROM search_revisions AS s
                  WHERE s.revision_id = d.published_revision_id
              )
        """,
        "orphan_search_acl": """
            SELECT count(*) FROM search_acl AS a
            WHERE NOT EXISTS (
                SELECT 1 FROM search_revisions AS s
                WHERE s.revision_id = a.revision_id
            )
        """,
        "deleted_projection_rows": """
            SELECT count(*) FROM search_revisions AS s
            JOIN documents AS d
              ON d.tenant_id = s.tenant_id AND d.document_id = s.document_id
            WHERE d.deleted = 1
        """,
        "published_cross_identity": """
            SELECT count(*) FROM documents AS d
            JOIN search_revisions AS s
              ON s.revision_id = d.published_revision_id
            WHERE s.tenant_id <> d.tenant_id
               OR s.document_id <> d.document_id
        """,
        "published_hash_mismatch": """
            SELECT count(*) FROM documents AS d
            JOIN search_revisions AS s
              ON s.revision_id = d.published_revision_id
            JOIN revisions AS r
              ON r.revision_id = d.published_revision_id
            WHERE s.content_hash <> r.content_hash
        """,
        "published_acl_mismatch": """
            SELECT count(*) FROM documents AS d
            WHERE d.published_revision_id IS NOT NULL
              AND (
                EXISTS (
                    SELECT 1 FROM revision_acl AS r
                    WHERE r.revision_id = d.published_revision_id
                      AND NOT EXISTS (
                          SELECT 1 FROM search_acl AS s
                          WHERE s.revision_id = r.revision_id
                            AND s.group_id = r.group_id
                      )
                )
                OR EXISTS (
                    SELECT 1 FROM search_acl AS s
                    WHERE s.revision_id = d.published_revision_id
                      AND NOT EXISTS (
                          SELECT 1 FROM revision_acl AS r
                          WHERE r.revision_id = s.revision_id
                            AND r.group_id = s.group_id
                      )
                )
              )
        """,
    }
    result: dict[str, int] = {}
    for name, query in scalar_queries.items():
        row = connection.execute(query).fetchone()
        result[name] = int(row[0])
    body_hash_tables = {
        "canonical_content_hash_mismatch": "revisions",
        "search_content_hash_mismatch": "search_revisions",
    }
    for name, table_name in body_hash_tables.items():
        rows = connection.execute(
            f"SELECT content, content_hash FROM {table_name} WHERE content IS NOT NULL"
        ).fetchall()
        result[name] = sum(
            1
            for row in rows
            if not isinstance(row["content"], str)
            or _digest_text(row["content"]) != row["content_hash"]
        )
    canonical_rows = connection.execute(
        "SELECT * FROM revisions WHERE content IS NOT NULL"
    ).fetchall()
    canonical_hash_errors = {
        "canonical_source_state_hash_mismatch": 0,
        "canonical_build_state_hash_mismatch": 0,
    }
    for row in canonical_rows:
        errors = _persisted_revision_hash_errors(
            row, _revision_groups(connection, int(row["revision_id"]))
        )
        for name in canonical_hash_errors:
            canonical_hash_errors[name] += int(name in errors)
    result.update(canonical_hash_errors)
    return result


def require_reconciled(
    connection: sqlite3.Connection, *, require_empty_outbox: bool = True
) -> dict[str, int]:
    report = reconcile(connection)
    failures = {
        name: report[name]
        for name in (
            "missing_published_projection",
            "orphan_search_acl",
            "deleted_projection_rows",
            "published_cross_identity",
            "published_hash_mismatch",
            "canonical_content_hash_mismatch",
            "canonical_source_state_hash_mismatch",
            "canonical_build_state_hash_mismatch",
            "search_content_hash_mismatch",
            "published_acl_mismatch",
        )
        if report[name]
    }
    if require_empty_outbox and report["pending_events"]:
        failures["pending_events"] = report["pending_events"]
    if require_empty_outbox:
        for name in (
            "unpublished_active_documents",
            "published_not_current_documents",
            "blocked_active_documents",
        ):
            if report[name]:
                failures[name] = report[name]
    if failures:
        raise StoreError(f"reconciliation failed: {_canonical_json(failures)}")
    return report


def purge_deleted_canonical_content(
    connection: sqlite3.Connection, *, tenant_id: str, document_id: str
) -> int:
    tenant_id = _validate_token("tenant_id", tenant_id)
    document_id = _validate_token("document_id", document_id)
    with _transaction(connection):
        document = _document(connection, tenant_id, document_id)
        if document is None or not bool(document["deleted"]):
            raise StoreError("only deleted documents may receive canonical content purge")
        pending = connection.execute(
            """
            SELECT count(*) FROM outbox
            WHERE tenant_id = ? AND document_id = ? AND processed = 0
            """,
            (tenant_id, document_id),
        ).fetchone()[0]
        projected = connection.execute(
            """
            SELECT count(*) FROM search_revisions
            WHERE tenant_id = ? AND document_id = ?
            """,
            (tenant_id, document_id),
        ).fetchone()[0]
        if pending or projected:
            raise StoreError("delete-event propagation has not completed; cannot purge canonical content")
        cursor = connection.execute(
            """
            UPDATE revisions SET content = NULL, content_purged = 1
            WHERE tenant_id = ? AND document_id = ? AND content_purged = 0
            """,
            (tenant_id, document_id),
        )
        connection.execute(
            """
            UPDATE tombstones SET canonical_content_purged = 1
            WHERE tenant_id = ? AND document_id = ?
            """,
            (tenant_id, document_id),
        )
        return int(cursor.rowcount)


def main() -> None:
    reconfigure = getattr(sys.stdout, "reconfigure", None)
    if callable(reconfigure):
        reconfigure(encoding="utf-8", newline="\n")
    connection = connect()
    config = BuildConfig("pipeline-v1")
    actions: list[dict[str, Any]] = []
    public = SourceRecord(
        "tenant-a",
        "policy:leave",
        1,
        "https://kb.example.invalid/policy/leave",
        "v1",
        "Leave policy requires advance submission.",
        ("employees",),
    )
    secret = SourceRecord(
        "tenant-a",
        "policy:security",
        1,
        "https://kb.example.invalid/policy/security",
        "v1",
        "Security policy requires administrator review.",
        ("admins",),
    )
    actions.append(asdict(upsert_record(connection, public, config, run_id="run-1")))
    actions.append(asdict(upsert_record(connection, secret, config, run_id="run-1")))
    drain_outbox(connection)

    updated = SourceRecord(
        "tenant-a",
        "policy:leave",
        2,
        public.source_uri,
        "v2",
        "Leave policy requires submission two days in advance.",
        public.allowed_groups,
    )
    update_result = upsert_record(connection, updated, config, run_id="run-2")
    actions.append(asdict(update_result))
    before_publish = search_visible(
        connection,
        tenant_id="tenant-a",
        subject_groups=("employees",),
        term="leave",
    )
    failed = project_next_event(connection, fail_event_id=update_result.event_id)
    actions.append(asdict(failed))
    project_next_event(connection)
    after_publish = search_visible(
        connection,
        tenant_id="tenant-a",
        subject_groups=("employees",),
        term="leave",
    )

    deletion = delete_document(
        connection,
        "tenant-a",
        "policy:leave",
        3,
        reason_code="source_deleted",
        run_id="run-3",
    )
    actions.append(asdict(deletion))
    drain_outbox(connection)
    purged_revisions = purge_deleted_canonical_content(
        connection, tenant_id="tenant-a", document_id="policy:leave"
    )
    report = require_reconciled(connection)
    output = {
        "schema_version": SCHEMA_VERSION,
        "actions": actions,
        "before_publish": before_publish,
        "after_publish": after_publish,
        "purged_revisions": purged_revisions,
        "reconciliation": report,
    }
    print(json.dumps(output, ensure_ascii=False, allow_nan=False, indent=2))
    connection.close()


if __name__ == "__main__":
    main()