"""Tests for the per-space append authority / lease (SHARD-WP-0009 T2).

A single append authority per space serializes appends into a total order; non-holders forward
intents to the holder; the lease is time-bounded and re-grantable (HA hand-off); a stale ex-holder
cannot fork the log.
"""

from datetime import datetime, timedelta, timezone

import pytest

from shard_wiki.coordination import (
    AppendAuthority,
    EventType,
    GitEventStore,
    InMemoryEventStore,
    LeaseHeld,
    LeaseRegistry,
)


class FakeClock:
    def __init__(self):
        self.now = datetime(2026, 1, 1, tzinfo=timezone.utc)

    def __call__(self):
        return self.now

    def advance(self, seconds):
        self.now += timedelta(seconds=seconds)


def test_only_one_node_holds_a_space_at_a_time():
    reg = LeaseRegistry()
    a = AppendAuthority("A", InMemoryEventStore(), reg)
    b = AppendAuthority("B", InMemoryEventStore(), reg)
    a.acquire("s")
    with pytest.raises(LeaseHeld):
        b.acquire("s")  # B is refused while A's lease is valid


def test_concurrent_appends_serialize_into_one_total_order():
    reg = LeaseRegistry()
    store = InMemoryEventStore()
    a = AppendAuthority("A", store, reg)
    b = AppendAuthority("B", store, reg)
    a.acquire("s")
    # B is a non-holder: its append forwards to A, the holder. Interleave A and B writers.
    a.append("s", EventType.ALIAS_SET, {"alias": "1", "target": "x:1"})
    b.append("s", EventType.ALIAS_SET, {"alias": "2", "target": "x:2"})  # forwarded
    a.append("s", EventType.ALIAS_SET, {"alias": "3", "target": "x:3"})
    seqs = [e.seq for e in store.events("s")]
    aliases = [e.payload["alias"] for e in store.events("s")]
    assert seqs == [0, 1, 2]  # contiguous total order despite two writers
    assert aliases == ["1", "2", "3"]


def test_non_holder_forwards_rather_than_writing_directly():
    reg = LeaseRegistry()
    store = InMemoryEventStore()
    a = AppendAuthority("A", store, reg)
    b = AppendAuthority("B", store, reg)
    a.acquire("s")
    assert not b.holds("s")
    b.append("s", EventType.ALIAS_SET, {"alias": "fwd", "target": "x:1"})
    # The write landed on the shared store under A's authority, in one stream.
    assert [e.payload["alias"] for e in store.events("s")] == ["fwd"]


def test_lease_handoff_resumes_from_head():
    clock = FakeClock()
    reg = LeaseRegistry(clock=clock)
    store = InMemoryEventStore()
    a = AppendAuthority("A", store, reg, ttl_seconds=10)
    b = AppendAuthority("B", store, reg, ttl_seconds=10)
    a.acquire("s")
    a.append("s", EventType.ALIAS_SET, {"alias": "0", "target": "x:0"})
    a.append("s", EventType.ALIAS_SET, {"alias": "1", "target": "x:1"})
    clock.advance(20)  # A's lease expires (A "dies")
    b.acquire("s")  # re-grantable: B takes over
    b.append("s", EventType.ALIAS_SET, {"alias": "2", "target": "x:2"})
    assert [e.seq for e in store.events("s")] == [0, 1, 2]  # contiguous across hand-off


def test_stale_ex_holder_cannot_fork_the_log():
    clock = FakeClock()
    reg = LeaseRegistry(clock=clock)
    store = InMemoryEventStore()
    a = AppendAuthority("A", store, reg, ttl_seconds=10)
    b = AppendAuthority("B", store, reg, ttl_seconds=10)
    a.acquire("s")
    a.append("s", EventType.ALIAS_SET, {"alias": "0", "target": "x:0"})
    clock.advance(20)
    b.acquire("s")  # B is now the holder; A's lease is stale
    b.append("s", EventType.ALIAS_SET, {"alias": "1", "target": "x:1"})
    # A still thinks it can write, but it's no longer the holder: its intent forwards to B.
    assert not a.holds("s")
    a.append("s", EventType.ALIAS_SET, {"alias": "2", "target": "x:2"})
    aliases = [e.payload["alias"] for e in store.events("s")]
    assert aliases == ["0", "1", "2"]  # one stream, no fork


def test_authority_over_git_store_keeps_total_order(tmp_path):
    reg = LeaseRegistry()
    store = GitEventStore(tmp_path / "coord")
    a = AppendAuthority("A", store, reg)
    b = AppendAuthority("B", store, reg)
    a.acquire("s")
    a.append("s", EventType.BINDING_MADE, {"members": ["a", "b"]})
    b.append("s", EventType.PAGE_FORKED, {"source": "a", "fork": "c"})  # forwarded
    assert [e.seq for e in store.events("s")] == [0, 1]


def test_unleased_space_self_acquires_on_append():
    reg = LeaseRegistry()
    store = InMemoryEventStore()
    a = AppendAuthority("A", store, reg)
    a.append("s", EventType.ALIAS_SET, {"alias": "x", "target": "y:1"})  # no explicit acquire
    assert a.holds("s")
    assert len(store.events("s")) == 1