"""Tests for revision numbers: runtime counter, snapshot ``r`` field, CLI rN."""

import asyncio
import json
from datetime import UTC, datetime, timedelta

import pytest

from kanta.exceptions import DataIntegrityError
from kanta.serialization import JsonSerializer
from kanta.structs import META_FORMAT, ChangeRecord, Snapshot
from tests.support import Data, make_kanta, read_last_snapshot
from tests.test_rotation import make_clock, read_all, write_history

T0 = datetime(2026, 1, 1, tzinfo=UTC)

# The CLI reads files with its own JSON Kanta instance unless -k is given.
JSON = ("json", JsonSerializer)


@pytest.mark.asyncio
async def test_revision_requires_open(tmp_path, format_config):
    kanta = make_kanta(tmp_path / "data.kantadb", Data, format_config)
    with pytest.raises(DataIntegrityError):
        kanta.rev


@pytest.mark.asyncio
async def test_revision_counts_change_records(tmp_path, format_config):
    path = tmp_path / "data.kantadb"
    kanta = make_kanta(path, Data, format_config)
    await kanta.open(log=False)
    # The bootstrap record is the first diff: state is at revision 1.
    assert kanta.rev == 1

    with kanta.transaction("inc", log=False) as data:
        data.counter += 1
    assert kanta.rev == 2

    # A no-op transaction queues no record and does not advance the revision.
    with kanta.transaction("noop", log=False):
        pass
    assert kanta.rev == 2
    await kanta.close()


@pytest.mark.asyncio
async def test_revision_restored_on_reopen(tmp_path, format_config):
    path = tmp_path / "data.kantadb"
    kanta = make_kanta(path, Data, format_config)
    await kanta.open(log=False)
    for _ in range(3):
        with kanta.transaction("inc", log=False) as data:
            data.counter += 1
    await kanta.close()

    kanta2 = make_kanta(path, Data, format_config)
    await kanta2.open(log=False)
    assert kanta2.rev == 4
    await kanta2.close()


@pytest.mark.asyncio
async def test_snapshot_carries_revision(tmp_path, format_config):
    path = tmp_path / "data.kantadb"
    kanta = make_kanta(path, Data, format_config)
    await kanta.open(log=False)
    for _ in range(2):
        with kanta.transaction("inc", log=False) as data:
            data.counter += 1
    await kanta.flush()
    kanta.request_snapshot()
    kanta._impl.maybe_snapshot()
    await kanta.close()

    snap = read_last_snapshot(path, format_config)
    assert snap is not None
    assert snap.r == 3
    assert snap.k == META_FORMAT

    # Reopening from the snapshot keeps the absolute revision.
    kanta2 = make_kanta(path, Data, format_config)
    await kanta2.open(log=False)
    assert kanta2.rev == 3
    with kanta2.transaction("inc", log=False) as data:
        data.counter += 1
    assert kanta2.rev == 4
    await kanta2.close()


@pytest.mark.asyncio
async def test_legacy_snapshot_recounted_and_rewritten(tmp_path, format_config):
    """Meta migration k1 rewrites a k0 file in full (see kanta/metamigr.py)."""
    path = tmp_path / "data.kantadb"
    kanta = make_kanta(path, Data, format_config)
    await kanta.open(log=False)
    for _ in range(2):
        with kanta.transaction("inc", log=False) as data:
            data.counter += 1
    await kanta.flush()
    kanta.request_snapshot()
    kanta._impl.maybe_snapshot()
    await kanta.close()

    # Simulate a pre-revision database: rewrite the last snapshot without k,
    # with a wrong r, then append one more change record.
    _, serializer_cls = format_config
    serializer = serializer_cls()
    framer = serializer.framer_cls()
    records = read_all(path, format_config)
    out = bytearray()
    for rec in records:
        if isinstance(rec, Snapshot):
            rec = Snapshot(ts=rec.ts, v=rec.v, state=rec.state, m=rec.m, r=0)
            out += framer.frame_snapshot(
                serializer.encode(rec), record_offset=len(out)
            )
        else:
            out += framer.frame_change(
                serializer.encode(rec), record_offset=len(out)
            )
    extra = ChangeRecord(ts=T0, a="inc", v=0, diff={"counter": 3})
    out += framer.frame_change(serializer.encode(extra), record_offset=len(out))
    path.write_bytes(bytes(out))

    kanta2 = make_kanta(path, Data, format_config)
    await kanta2.open(log=False)
    # bootstrap + 2 changes + trailing change = revision 4, despite r=0 above.
    assert kanta2.rev == 4
    assert kanta2.data.counter == 3
    await kanta2.close()

    # The file was rewritten in full: all change records preserved in order,
    # the legacy intermediate snapshot dropped, and a k1 closing snapshot
    # carrying the total revision appended.
    records = read_all(path, format_config)
    changes = [r for r in records if isinstance(r, ChangeRecord)]
    assert [c.a for c in changes] == ["bootstrap", "inc", "inc", "inc"]
    snapshots = [r for r in records if isinstance(r, Snapshot)]
    assert len(snapshots) == 1
    assert isinstance(records[-1], Snapshot)
    closing = snapshots[0]
    assert closing.k == META_FORMAT
    assert closing.r == 4
    assert closing.state["counter"] == 3

    # The next open trusts the stored r (file unchanged this time).
    content = path.read_bytes()
    kanta3 = make_kanta(path, Data, format_config)
    await kanta3.open(log=False)
    assert kanta3.rev == 4
    await kanta3.close()
    assert path.read_bytes() == content


@pytest.mark.asyncio
async def test_legacy_migration_preserves_leading_snapshot(tmp_path, format_config):
    """A legacy file starting with a snapshot keeps it, re-stamped k1."""
    path = tmp_path / "data.kantadb"
    _, serializer_cls = format_config
    serializer = serializer_cls()
    framer = serializer.framer_cls()
    snap_ts = T0 - timedelta(days=40)
    snap_m = T0 - timedelta(days=41)
    leading = Snapshot(ts=snap_ts, v=3, state={"users": {}, "counter": 7}, m=snap_m)
    out = bytearray()
    out += framer.frame_snapshot(serializer.encode(leading), record_offset=0)
    for i in range(2):
        rec = ChangeRecord(ts=T0, a=f"inc{i}", v=3, diff={"counter": 8 + i})
        out += framer.frame_change(serializer.encode(rec), record_offset=len(out))
    path.write_bytes(bytes(out))

    kanta = make_kanta(path, Data, format_config)
    await kanta.open(log=False)
    # Only the in-file history is countable: 2 change records.
    assert kanta.rev == 2
    assert kanta.data.counter == 9
    await kanta.close()

    records = read_all(path, format_config)
    assert isinstance(records[0], Snapshot)
    assert records[0].k == META_FORMAT
    assert records[0].r == 0
    assert records[0].ts == snap_ts
    assert records[0].m == snap_m
    assert records[0].v == 3
    assert records[0].state == {"users": {}, "counter": 7}
    changes = [r for r in records if isinstance(r, ChangeRecord)]
    assert [c.a for c in changes] == ["inc0", "inc1"]
    assert isinstance(records[-1], Snapshot)
    assert records[-1].k == META_FORMAT
    assert records[-1].r == 2
    assert records[-1].state["counter"] == 9


@pytest.mark.asyncio
async def test_rotation_of_legacy_file_migrates(tmp_path, format_config):
    """Rotation and meta migration coincide: rotation already produces k1
    snapshots with revisions counted from pre-rotation record positions."""
    path = tmp_path / "data.kantadb"
    _, serializer_cls = format_config
    serializer = serializer_cls()
    framer = serializer.framer_cls()

    def change(action, day, counter):
        return ChangeRecord(
            ts=T0 + day * timedelta(days=1), a=action, v=0, diff={"counter": counter}
        )

    legacy_snap = Snapshot(
        ts=T0 - 40 * timedelta(days=1), v=0, state={"users": {}, "counter": 1}, r=0
    )
    out = bytearray()
    out += framer.frame_change(
        serializer.encode(change("bootstrap", -41, 0)), record_offset=len(out)
    )
    out += framer.frame_change(
        serializer.encode(change("day-40", -40, 1)), record_offset=len(out)
    )
    out += framer.frame_snapshot(
        serializer.encode(legacy_snap), record_offset=len(out)
    )
    out += framer.frame_change(
        serializer.encode(change("day-20", -20, 2)), record_offset=len(out)
    )
    out += framer.frame_change(
        serializer.encode(change("day-5", -5, 3)), record_offset=len(out)
    )
    path.write_bytes(bytes(out))

    cell = [T0]
    kanta = make_kanta(path, Data, format_config, retention=30 * timedelta(days=1))
    kanta.clock(make_clock(cell))
    await kanta.open(log=False)
    assert kanta.rev == 4
    assert kanta.data.counter == 3
    await kanta.close()

    # The rotation's leading snapshot is k1 with the pre-rotation revision.
    records = read_all(path, format_config)
    assert isinstance(records[0], Snapshot)
    assert records[0].k == META_FORMAT
    assert records[0].r == 2
    assert [r.a for r in records if isinstance(r, ChangeRecord)] == [
        "day-20",
        "day-5",
    ]

    # No further rewriting on the next open.
    content = path.read_bytes()
    kanta2 = make_kanta(path, Data, format_config)
    await kanta2.open(log=False)
    assert kanta2.rev == 4
    await kanta2.close()
    assert path.read_bytes() == content


@pytest.mark.asyncio
async def test_legacy_snapshot_readonly_recounts_without_writing(
    tmp_path, format_config
):
    path = tmp_path / "data.kantadb"
    kanta = make_kanta(path, Data, format_config)
    await kanta.open(log=False)
    with kanta.transaction("inc", log=False) as data:
        data.counter += 1
    await kanta.close()

    # Inject a legacy snapshot (no k marker) at the start; the whole file is
    # re-framed with a single framer so binary checksums stay valid.
    _, serializer_cls = format_config
    serializer = serializer_cls()
    framer = serializer.framer_cls()
    legacy = Snapshot(ts=T0, v=0, state={"users": {}, "counter": 0}, r=0)
    out = bytearray()
    out += framer.frame_snapshot(serializer.encode(legacy), record_offset=0)
    for rec in read_all(path, format_config):
        out += framer.frame_change(serializer.encode(rec), record_offset=len(out))
    path.write_bytes(bytes(out))
    content = path.read_bytes()

    kanta2 = make_kanta(path, Data, format_config)
    await kanta2.open(readonly=True, log=False)
    assert kanta2.rev == 2
    await kanta2.close()
    assert path.read_bytes() == content


@pytest.mark.asyncio
async def test_rotation_preserves_revisions(tmp_path, format_config):
    path = tmp_path / "data.kantadb"
    await write_history(path, format_config, days=[-40, -20, -5])

    cell = [T0]
    kanta = make_kanta(path, Data, format_config, retention=30 * timedelta(days=1))
    kanta.clock(make_clock(cell))
    await kanta.open(log=False)
    # bootstrap + 3 changes = revision 4 before and after the rotation.
    assert kanta.rev == 4
    await kanta.close()

    records = read_all(path, format_config)
    leading = records[0]
    assert isinstance(leading, Snapshot)
    # One change was dropped (bootstrap + day-40 are before the cutoff
    # position... bootstrap is a change record too: 2 records before).
    assert leading.r == 2
    assert leading.k == META_FORMAT

    kanta2 = make_kanta(path, Data, format_config)
    await kanta2.open(log=False)
    assert kanta2.rev == 4
    with kanta2.transaction("inc", log=False) as data:
        data.counter += 1
    assert kanta2.rev == 5
    await kanta2.close()


def test_cli_revision_snapshot_base_file(tmp_path, capsys):
    """File = snapshot {r: 2} + one change; the two producing changes are gone."""
    path = tmp_path / "data.kantadb"
    serializer = JsonSerializer()
    framer = serializer.framer_cls()
    snap = Snapshot(
        ts=T0, v=0, state={"foo": 1, "bar": 1}, r=2, k=META_FORMAT
    )
    out = bytearray()
    out += framer.frame_snapshot(serializer.encode(snap), record_offset=0)
    change = ChangeRecord(ts=T0, a="add_baz", v=0, diff={"baz": 1})
    out += framer.frame_change(serializer.encode(change), record_offset=len(out))
    path.write_bytes(bytes(out))
    capsys.readouterr()

    # r1's producing change is gone and no snapshot holds state 1: error.
    assert _run_cli(path, "-r", "r1") == 2
    assert "r1" in capsys.readouterr().err

    # r2 prints the snapshot line (it literally carries r: 2) and its state;
    # revisions are shown on change records, not on snapshot indicators.
    assert _run_cli(path, "-r", "r2", "-o", "-") == 0
    out = capsys.readouterr()
    assert "snapshot s0" in out.err
    assert "r2" not in out.err
    assert json.loads(out.out) == {"foo": 1, "bar": 1}

    # r3 is the change, replayed on the snapshot base; state includes baz.
    assert _run_cli(path, "-r", "r3", "-o", "-") == 0
    out = capsys.readouterr()
    assert "add_baz" in out.err
    assert " r3" in out.err
    assert json.loads(out.out) == {"foo": 1, "bar": 1, "baz": 1}

    # Single index and half-open range agree.
    assert _run_cli(path, "-r", "r2") == 0
    single = capsys.readouterr().err
    assert _run_cli(path, "-r", "r2:r3") == 0
    assert capsys.readouterr().err == single


def _run_cli(path, *args):
    from kanta.__main__ import main

    return main([str(path), *args])


def test_cli_revision_ranges(tmp_path, capsys):
    path = tmp_path / "data.kantadb"

    async def build():
        kanta = make_kanta(path, Data, JSON)
        await kanta.open(log=False)
        for i in range(3):
            with kanta.transaction(f"inc{i}", log=False) as data:
                data.counter += 1
        await kanta.close()

    asyncio.run(build())
    capsys.readouterr()

    # A revision selects the record that produced that state, with the
    # state: r1 and r1:r2 both show only the bootstrap and output state 1.
    assert _run_cli(path, "-r", "r1") == 0
    single = capsys.readouterr().err
    assert "bootstrap" in single
    # Change record headers carry the revision of the state they produce.
    assert " r1" in single
    assert "inc0" not in single

    assert _run_cli(path, "-r", "r1:r2") == 0
    assert capsys.readouterr().err == single

    assert _run_cli(path, "-r", "r1", "-o", "-") == 0
    assert json.loads(capsys.readouterr().out) == {"users": {}, "counter": 0}

    # r1..r2 (inclusive end) adds the next change and outputs state 2.
    assert _run_cli(path, "-r", "r1..r2", "-o", "-") == 0
    out = capsys.readouterr()
    assert "inc0" in out.err
    assert json.loads(out.out) == {"users": {}, "counter": 1}

    # r0 is not meaningful: no record produces the empty initial state.
    assert _run_cli(path, "-r", "r0") == 2
    assert "r0" in capsys.readouterr().err

    # Range by revision.
    assert _run_cli(path, "-r", "r2:r4") == 0
    err = capsys.readouterr().err
    assert "inc0" in err
    assert "inc1" in err
    assert "inc2" not in err

    # Negative revisions are rejected.
    assert _run_cli(path, "-r", "r-1") == 1
    assert "negative" in capsys.readouterr().err

    # Out-of-range single revision is a range error (exit 2).
    assert _run_cli(path, "-r", "r99") == 2
    assert "r99" in capsys.readouterr().err

    # The same bounds as a range truncate Python-style instead: an empty
    # display and the closest known state, not an error.
    assert _run_cli(path, "-r", "r99:r100", "-o", "-") == 0
    out = capsys.readouterr()
    assert "bootstrap" not in out.err
    assert "inc" not in out.err
    assert json.loads(out.out) == {"users": {}, "counter": 3}

    # Only the end beyond the last revision: truncated to end of file.
    assert _run_cli(path, "-r", "r3:r99", "-o", "-") == 0
    out = capsys.readouterr()
    assert "inc1" in out.err
    assert "inc2" in out.err
    assert json.loads(out.out) == {"users": {}, "counter": 3}


def test_cli_revision_after_rotation(tmp_path, capsys):
    path = tmp_path / "data.kantadb"

    async def build():
        await write_history(path, JSON, days=[-40, -20, -5])
        cell = [T0]
        kanta = make_kanta(
            path, Data, JSON, retention=30 * timedelta(days=1)
        )
        kanta.clock(make_clock(cell))
        await kanta.open(log=False)
        await kanta.close()

    asyncio.run(build())
    capsys.readouterr()

    # The leading snapshot holds revision 2 (bootstrap + day-40).  Since the
    # change that produced state 2 is gone, r2 shows the snapshot itself and
    # outputs the snapshot state.
    assert _run_cli(path, "-r", "r2", "-o", "-") == 0
    out = capsys.readouterr()
    assert "snapshot s0" in out.err
    assert "r2" not in out.err
    assert "day-20" not in out.err
    assert json.loads(out.out) == {"users": {}, "counter": 1}

    # r1 has no record in this file: a single index does not truncate.
    assert _run_cli(path, "-r", "r1") == 2
    assert "r1" in capsys.readouterr().err

    # A range truncates like Python slicing instead: r1:r3 starts at the
    # beginning of the file and stops before the r3 record, showing just the
    # leading snapshot.
    assert _run_cli(path, "-r", "r1:r3") == 0
    err = capsys.readouterr().err
    assert "snapshot s0" in err
    assert "r2" not in err
    assert "day-20" not in err

    # Absolute revisions reach the retained change records.
    assert _run_cli(path, "-r", "r3:r4") == 0
    err = capsys.readouterr().err
    assert "day-20" in err
    assert "snapshot" not in err
    assert "day-5" not in err