"""Updater restore must publish snapshot rows without replaying the old WAL."""

import shutil
import sqlite3
from pathlib import Path

import pytest

from hermes_cli.update_cmd import (
    _restore_state_db_from_snapshot,
)

OLD_ROWS = 201
SNAPSHOT_ROWS = 400


def _sidecar(db_path: Path, suffix: str) -> Path:
    return db_path.with_name(db_path.name + suffix)


@pytest.fixture()
def live_db_with_hot_wal(tmp_path):
    """A WAL-mode database whose committed rows live in an UNCHECKPOINTED
    ``-wal``, exactly as a force-killed writer leaves them on disk.

    A clean ``close()`` checkpoints and unlinks the WAL, so the on-disk trio is
    copied aside while the connection is still open and put back afterwards.
    """
    live = tmp_path / "state.db"
    holding = tmp_path / "_killed_writer_state"
    holding.mkdir()

    conn = sqlite3.connect(live)
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA wal_autocheckpoint=0")
    conn.execute("CREATE TABLE sessions (id INTEGER PRIMARY KEY, name TEXT)")
    conn.executemany(
        "INSERT INTO sessions (name) VALUES (?)",
        [(f"old{i}",) for i in range(OLD_ROWS)],
    )
    conn.commit()
    for suffix in ("", "-wal", "-shm"):
        src = _sidecar(live, suffix)
        if src.exists():
            shutil.copy2(src, holding / (live.name + suffix))
    conn.close()

    for suffix in ("", "-wal", "-shm"):
        held = holding / (live.name + suffix)
        if held.exists():
            shutil.copy2(held, _sidecar(live, suffix))

    assert _sidecar(live, "-wal").exists(), "fixture failed to leave a hot WAL"
    return live


@pytest.fixture()
def snapshot_db(tmp_path):
    """A consistent, checkpointed image owning no WAL — what
    ``backup._safe_copy_db`` produces through ``sqlite3.backup()``."""
    source_path = tmp_path / "_snapshot_source.db"
    snapshot = tmp_path / "state-snapshots" / "20260804-pre-update" / "state.db"
    snapshot.parent.mkdir(parents=True)

    source = sqlite3.connect(source_path)
    source.execute("CREATE TABLE sessions (id INTEGER PRIMARY KEY, name TEXT)")
    source.executemany(
        "INSERT INTO sessions (name) VALUES (?)",
        [(f"snap{i}",) for i in range(SNAPSHOT_ROWS)],
    )
    source.commit()
    destination = sqlite3.connect(snapshot)
    source.backup(destination)
    destination.close()
    source.close()
    source_path.unlink()

    assert not _sidecar(snapshot, "-wal").exists()
    return snapshot


@pytest.fixture(autouse=True)
def _isolate_database_holders(monkeypatch):
    # These fixtures own all DB connections. Do not scan other users' /proc FDs.
    monkeypatch.setattr("hermes_cli.backup_restore._foreign_db_holder_pids", lambda path: [])


def _row_count(db_path: Path) -> int:
    conn = sqlite3.connect(db_path)
    try:
        return conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0]
    finally:
        conn.close()


def test_torn_restore_is_what_the_guard_prevents(live_db_with_hot_wal, snapshot_db):
    """Pin the SQLite behaviour that makes the guard necessary.

    Copying the snapshot over a database that still owns a hot WAL yields a file
    that reports ``integrity_check`` clean — so the CLI's ``_restored_ok`` test
    passes and it prints success — while serving the OLD row set.
    """
    from hermes_cli.backup import verify_sqlite_integrity

    shutil.copy2(snapshot_db, live_db_with_hot_wal)  # no sidecar clearing

    assert (
        verify_sqlite_integrity(
            live_db_with_hot_wal, check_header=True, run_pragma=True
        ).get("valid")
        is True
    )
    assert _row_count(live_db_with_hot_wal) == OLD_ROWS
    assert _row_count(live_db_with_hot_wal) != SNAPSHOT_ROWS


def test_restore_helper_serves_snapshot_rows_over_a_hot_wal(
    live_db_with_hot_wal, snapshot_db
):
    """The shared restore helper is what both update paths call.

    It must restore and verify as one unit: after it returns, the database
    has to hold the SNAPSHOT's rows even though the destination still owned a
    hot WAL from the corrupt database.
    """
    assert _restore_state_db_from_snapshot(live_db_with_hot_wal, snapshot_db) is True

    assert _row_count(live_db_with_hot_wal) == SNAPSHOT_ROWS
    for suffix in ("-wal", "-shm", "-journal"):
        assert not _sidecar(live_db_with_hot_wal, suffix).exists()


def test_restore_helper_reports_failure_when_the_restored_copy_is_corrupt(tmp_path):
    """A snapshot that does not survive the copy must return False, so the
    caller prints the failure branch instead of claiming success."""
    state_path = tmp_path / "state.db"
    state_path.write_bytes(b"whatever")
    bad_snapshot = tmp_path / "bad-snapshot.db"
    bad_snapshot.write_bytes(b"\x00" * 4096)

    assert _restore_state_db_from_snapshot(state_path, bad_snapshot) is False


def test_restore_helper_propagates_copy_errors(tmp_path):
    """A missing snapshot raises OSError, which both call sites already catch
    and report as 'Auto-restore file copy failed'."""
    state_path = tmp_path / "state.db"
    state_path.write_bytes(b"whatever")

    with pytest.raises(OSError):
        _restore_state_db_from_snapshot(state_path, tmp_path / "does-not-exist.db")


# ── Multi-profile coverage (#97994) ─────────────────────────────────────


def _make_valid_db(path: Path, rows: int) -> None:
    conn = sqlite3.connect(path)
    conn.execute("CREATE TABLE sessions (id INTEGER PRIMARY KEY, name TEXT)")
    conn.executemany(
        "INSERT INTO sessions (name) VALUES (?)",
        [(str(i),) for i in range(rows)],
    )
    conn.commit()
    conn.close()


def _make_valid_snapshot(home: Path, snap_id: str, rows: int) -> None:
    snap_dir = home / "state-snapshots" / snap_id
    snap_dir.mkdir(parents=True)
    _make_valid_db(snap_dir / "state.db", rows)


def test_post_update_guard_covers_sibling_profiles(tmp_path, monkeypatch, capsys):
    """#97994: the guard must verify + auto-restore EVERY profile's state.db,
    not just the root home's. Pre-update snapshots already cover siblings
    (#66140); the guard was the missing half."""
    from hermes_cli import update_cmd
    from hermes_cli.backup import _sibling_profile_homes

    root_home = tmp_path / "default-home"
    root_home.mkdir()
    sibling_home = tmp_path / "profiles" / "work"
    sibling_home.mkdir(parents=True)

    # Root DB: valid, with its own snapshot — must be left untouched.
    _make_valid_db(root_home / "state.db", 10)
    root_before = (root_home / "state.db").read_bytes()

    # Corrupt only the sessions B-tree; the canonical epoch must remain readable.
    _make_valid_snapshot(sibling_home, "20260901-pre-update", 25)
    state = sibling_home / "state.db"
    _make_valid_db(state, 3)
    with sqlite3.connect(state) as conn:
        page = conn.execute("SELECT rootpage FROM sqlite_master WHERE name='sessions'").fetchone()[0]
        page_size = conn.execute("PRAGMA page_size").fetchone()[0]
    conn.close()
    with state.open("r+b") as stream:
        stream.seek((page - 1) * page_size)
        stream.write(b"\x00")

    monkeypatch.setattr(update_cmd, "get_hermes_home", lambda: root_home)
    monkeypatch.setattr(
        "hermes_cli.backup._sibling_profile_homes",
        lambda invoking_home: [("work", sibling_home)],
    )

    update_cmd._verify_and_restore_state_dbs_post_update()

    # Sibling restored from ITS snapshot (snap rows, not zeroed bytes).
    assert _row_count(sibling_home / "state.db") == 25
    # Root DB byte-identical — untouched.
    assert (root_home / "state.db").read_bytes() == root_before
    # Operator-visible restore message mentions the profile.
    out = capsys.readouterr().out
    assert "profile work" in out
    assert "✓ Auto-restored from snapshot" in out


def test_post_update_guard_leaves_valid_sibling_dbs_alone(tmp_path, monkeypatch, capsys):
    """A healthy sibling profile must not be touched — the guard only acts
    on corruption."""
    from hermes_cli import update_cmd

    root_home = tmp_path / "default-home"
    root_home.mkdir()
    sibling_home = tmp_path / "profiles" / "work"
    sibling_home.mkdir(parents=True)

    _make_valid_db(root_home / "state.db", 10)
    _make_valid_db(sibling_home / "state.db", 7)
    sibling_before = (sibling_home / "state.db").read_bytes()

    monkeypatch.setattr(update_cmd, "get_hermes_home", lambda: root_home)
    monkeypatch.setattr(
        "hermes_cli.backup._sibling_profile_homes",
        lambda invoking_home: [("work", sibling_home)],
    )

    update_cmd._verify_and_restore_state_dbs_post_update()

    assert (sibling_home / "state.db").read_bytes() == sibling_before
    out = capsys.readouterr().out
    assert "corrupted" not in out


def test_post_update_guard_survives_missing_sibling_snapshot(tmp_path, monkeypatch, capsys):
    """Corrupt sibling with NO snapshot: guard must report and continue,
    never raise into the update tail."""
    from hermes_cli import update_cmd

    root_home = tmp_path / "default-home"
    root_home.mkdir()
    sibling_home = tmp_path / "profiles" / "work"
    sibling_home.mkdir(parents=True)

    _make_valid_db(root_home / "state.db", 10)
    (sibling_home / "state.db").write_bytes(b"\x00" * 4096)

    monkeypatch.setattr(update_cmd, "get_hermes_home", lambda: root_home)
    monkeypatch.setattr(
        "hermes_cli.backup._sibling_profile_homes",
        lambda invoking_home: [("work", sibling_home)],
    )

    # Must not raise even though no snapshot exists to restore from.
    update_cmd._verify_and_restore_state_dbs_post_update()

    out = capsys.readouterr().out
    assert "corrupted" in out
    # Still corrupt (no snapshot) — but the guard completed cleanly.
    assert (sibling_home / "state.db").read_bytes() == b"\x00" * 4096