"""Append-only weekly day-status sidecar contract tests."""

from __future__ import annotations

import stat
from concurrent.futures import ThreadPoolExecutor
from dataclasses import replace
from datetime import date, datetime
from pathlib import Path
from threading import Barrier
from unittest import TestCase

from checkin_cli.models import Provenance, build_satisfaction_event
from checkin_cli.store import EventStore
from checkin_cli.weekly_operations import (
    CanonicalPin,
    CustomerKey,
    DayState,
    ReminderIdentity,
    SourceLineage,
    WEEKLY_OPERATIONS_SCHEMA,
    WeeklyOperationInput, customer_identity_digest,
    WeeklyOperationsConflict,
    WeeklyOperationsCorruption,
    WeeklyOperationsInputError,
)
from checkin_cli.weekly_operations_authority import AuthorityId, issue_repair_authority, open_authority_root
from checkin_cli.weekly_operations_layout import customer_data_name, customer_lock_name
from checkin_cli.weekly_operations_store import WeeklyOperationsStore

from tests._weekly_operations_support import initialize_at

CUSTOMER = CustomerKey("client_001")
DAY = date(2026, 8, 17)
ASSERTIONS = TestCase()
AUTHORITY_ID = AuthorityId("9" * 64)


def _operation() -> WeeklyOperationInput:
    return WeeklyOperationInput.for_customer(CUSTOMER, DAY, DayState.SUBMITTED, CanonicalPin(1, "a" * 64), datetime.fromisoformat("2026-08-17T20:00:00+09:00"))


def _store(root: Path, customer_key: CustomerKey = CUSTOMER) -> WeeklyOperationsStore:
    authority_path = root / "profile-authority"
    authority_path.mkdir(parents=True, mode=0o700)
    authority_path.chmod(0o700)
    return WeeklyOperationsStore.for_authority(initialize_at(authority_path, AUTHORITY_ID), customer_key)


def _data_path(store: WeeklyOperationsStore) -> Path:
    return store.authority.observed_path / customer_data_name(store.customer_identity_digest)


def _lock_path(store: WeeklyOperationsStore) -> Path:
    return _data_path(store).with_name(customer_lock_name(store.customer_identity_digest))


def test_canonical_reader_bytes_are_unchanged_without_weekly_sidecar(tmp_path: Path) -> None:
    # Given: one canonical event persisted through the v1.3.2-compatible store.
    home = tmp_path / "customer"
    store = EventStore.for_standalone(home)
    event = build_satisfaction_event("client_001", score=8, collected_on=DAY, provenance=Provenance(source_type="fixture", source_ref="client_001", content_sha256="1" * 64))
    expected = (event.model_dump_json(exclude_none=True) + "\n").encode()

    # When: the canonical event is appended and read through the legacy surface.
    stored = store.append_wizard_event(event)
    reopened = EventStore.for_standalone(home).load_wizard_event(event.event_id)

    # Then: canonical bytes/read behavior remain exact and no sidecar is inferred.
    assert stored.event_id == event.event_id
    assert (home / "events.jsonl").read_bytes() == expected
    assert reopened == event
    assert not (home / "nutrition-plans" / "weekly-operations.jsonl").exists()


def test_append_read_replay_and_schema_are_deterministic(tmp_path: Path) -> None:
    # Given: an empty customer-bound sidecar.
    store = _store(tmp_path)
    operation = replace(_operation(), reminder=ReminderIdentity("reservation-1", "audit-1"))

    # When: the same submitted operation is appended twice.
    first = store.append(operation)
    replay = store.append(operation)

    # Then: replay is a no-op and the canonical hash row is stable.
    assert first.appended is True
    assert replay.appended is False
    assert store.read() == (first.row,)
    assert first.row.schema_version == WEEKLY_OPERATIONS_SCHEMA
    assert first.row.predecessor_row_digest == "0" * 64


def test_missed_advances_once_to_late_submitted(tmp_path: Path) -> None:
    # Given: a day closed as missed.
    store = _store(tmp_path)
    missed = store.append(replace(_operation(), state=DayState.MISSED, occurred_at=datetime.fromisoformat("2026-08-17T23:00:00+09:00"))).row

    # When: a later canonical source submits after cutoff.
    late = store.append(replace(_operation(), state=DayState.LATE_SUBMITTED, canonical=CanonicalPin(2, "b" * 64), occurred_at=datetime.fromisoformat("2026-08-17T23:00:00+09:00"), source=SourceLineage("event-late", "c" * 64))).row

    # Then: only the legal transition is chained.
    assert tuple(row.state for row in store.read()) == (DayState.MISSED, DayState.LATE_SUBMITTED)
    assert late.predecessor_row_digest == missed.row_digest


def test_correction_advances_source_without_changing_timeliness(tmp_path: Path) -> None:
    # Given: an on-time submission with a source lineage.
    store = _store(tmp_path)
    _ = store.append(replace(_operation(), source=SourceLineage("event-root", "1" * 64)))

    # When: a correction advances the canonical source.
    _ = store.append(replace(_operation(), canonical=CanonicalPin(2, "b" * 64), source=SourceLineage("event-correction", "2" * 64)))

    # Then: timeliness stays submitted across immutable rows.
    assert tuple(row.state for row in store.read()) == (DayState.SUBMITTED, DayState.SUBMITTED)


def _assert_illegal_transition(root: Path, state: DayState) -> None:
    store = _store(root)
    _ = store.append(_operation())
    with ASSERTIONS.assertRaises(WeeklyOperationsConflict):
        _ = store.append(replace(_operation(), state=state, canonical=CanonicalPin(2, "b" * 64), source=SourceLineage("event-next", "2" * 64)))
    assert len(store.read()) == 1


def test_submitted_to_missed_transition_fails_closed(tmp_path: Path) -> None:
    # Given/When/Then: a submitted day cannot be reversed to missed.
    _assert_illegal_transition(tmp_path, DayState.MISSED)


def test_submitted_to_late_transition_fails_closed(tmp_path: Path) -> None:
    # Given/When/Then: a submitted day cannot be changed to late.
    _assert_illegal_transition(tmp_path, DayState.LATE_SUBMITTED)


def test_late_first_and_duplicate_conflict_fail_closed(tmp_path: Path) -> None:
    # Given: one empty and one submitted sidecar.
    empty = _store(tmp_path / "empty")
    store = _store(tmp_path / "used")
    _ = store.append(_operation())

    # When/Then: late-first and changed terminal replay are rejected.
    with ASSERTIONS.assertRaises(WeeklyOperationsConflict):
        _ = empty.append(replace(_operation(), state=DayState.LATE_SUBMITTED, source=SourceLineage("event-late", "2" * 64)))
    with ASSERTIONS.assertRaisesRegex(WeeklyOperationsConflict, "same-state"):
        _ = store.append(replace(_operation(), canonical=CanonicalPin(2, "b" * 64)))


def test_canonical_pin_drift_fails_closed(tmp_path: Path) -> None:
    # Given: canonical sequence one pinned for one day.
    store = _store(tmp_path)
    _ = store.append(_operation())

    # When/Then: sequence one cannot name another digest on another day.
    with ASSERTIONS.assertRaisesRegex(WeeklyOperationsConflict, "canonical pin drift"):
        _ = store.append(replace(_operation(), kst_day=date(2026, 8, 18), canonical=CanonicalPin(1, "b" * 64), occurred_at=datetime.fromisoformat("2026-08-18T20:00:00+09:00")))
    assert len(store.read()) == 1


def test_wrong_customer_and_day_fail_before_mutation(tmp_path: Path) -> None:
    # Given: a customer-bound empty sidecar.
    store = _store(tmp_path)

    # When/Then: wrong identity/day are rejected before sidecar creation.
    with ASSERTIONS.assertRaisesRegex(WeeklyOperationsInputError, "wrong customer"):
        _ = store.append(replace(_operation(), customer_identity_digest=customer_identity_digest(CustomerKey("client_002"))))
    with ASSERTIONS.assertRaisesRegex(WeeklyOperationsInputError, "KST day"):
        _ = WeeklyOperationInput.for_customer(CUSTOMER, DAY, DayState.SUBMITTED, CanonicalPin(1, "a" * 64), datetime.fromisoformat("2026-08-18T20:00:00+09:00"))
    assert not _data_path(store).exists()


def test_empty_sidecar_initialization_is_durable_and_idempotent(
    tmp_path: Path,
) -> None:
    store = _store(tmp_path)

    first = store.ensure_initialized()
    second = store.ensure_initialized()

    path = _data_path(store)
    assert first is True and second is False
    assert store.read() == ()
    assert path.read_bytes() == b""
    assert stat.S_IMODE(path.stat().st_mode) == 0o600
    assert path.stat().st_nlink == 1


def test_files_are_private_and_restart_needs_no_index(tmp_path: Path) -> None:
    # Given: one durable sidecar append.
    store = _store(tmp_path)
    row = store.append(_operation()).row

    # When: a fresh instance reopens only the JSONL data surface.
    reopened_authority = open_authority_root(store.authority.parent, store.authority.binding)
    reopened = WeeklyOperationsStore.for_authority(reopened_authority, CUSTOMER)

    # Then: mode is 0600 and restart reconstructs without a derived index.
    assert stat.S_IMODE(_data_path(store).stat().st_mode) == 0o600
    assert stat.S_IMODE(_lock_path(store).stat().st_mode) == 0o600
    assert reopened.read() == (row,)
    assert not tuple(_data_path(store).parent.glob("*index*"))


def test_concurrent_writers_serialize_with_a_barrier(tmp_path: Path) -> None:
    # Given: two store instances synchronized before append.
    authority_path = tmp_path / "profile-authority"
    authority_path.mkdir(mode=0o700)
    authority = initialize_at(authority_path, AUTHORITY_ID)
    stores = (WeeklyOperationsStore.for_authority(authority, CUSTOMER), WeeklyOperationsStore.for_authority(authority, CUSTOMER))
    barrier = Barrier(2)

    def append(store: WeeklyOperationsStore) -> bool:
        _ = barrier.wait(timeout=5)
        return store.append(_operation()).appended

    # When: both writers race through the real filesystem lock.
    with ThreadPoolExecutor(max_workers=2) as executor:
        results = tuple(executor.map(append, stores))

    # Then: one append wins and one replay is a no-op.
    assert sorted(results) == [False, True]
    assert len(stores[0].read()) == 1


def test_torn_tail_requires_explicit_repair(tmp_path: Path) -> None:
    # Given: one durable row and an incomplete final fragment.
    store = _store(tmp_path)
    row = store.append(_operation()).row
    with _data_path(store).open("ab") as handle:
        _ = handle.write(b'{"schema_version"')

    # When: ordinary read fails and explicit repair is invoked.
    with ASSERTIONS.assertRaisesRegex(WeeklyOperationsCorruption, "torn tail"):
        _ = store.read()
    repair_authority = issue_repair_authority(store.authority, store.customer_identity_digest)
    repair_store = WeeklyOperationsStore.for_authority(repair_authority, CUSTOMER)
    repaired = repair_store.repair_torn_tail()
    repair_authority.close()

    # Then: only the incomplete final bytes are removed.
    assert repaired.retained_rows == 1
    assert repaired.removed_bytes > 0
    assert store.read() == (row,)


def test_interior_byte_flip_fails_before_read_repair_or_append(tmp_path: Path) -> None:
    # Given: two rows whose first canonical digest is flipped in place.
    store = _store(tmp_path)
    _ = store.append(_operation())
    _ = store.append(replace(_operation(), kst_day=date(2026, 8, 18), canonical=CanonicalPin(2, "b" * 64), occurred_at=datetime.fromisoformat("2026-08-18T20:00:00+09:00")))
    corrupted = _data_path(store).read_bytes().replace(b'"canonical_digest":"a', b'"canonical_digest":"f', 1)
    _ = _data_path(store).write_bytes(corrupted)
    _data_path(store).chmod(0o600)

    # When/Then: every mutating/recovery surface fails closed on interior corruption.
    with ASSERTIONS.assertRaisesRegex(WeeklyOperationsCorruption, "hash chain"):
        _ = store.read()
    with ASSERTIONS.assertRaisesRegex(WeeklyOperationsCorruption, "hash chain"):
        _ = issue_repair_authority(store.authority, store.customer_identity_digest)
    with ASSERTIONS.assertRaisesRegex(WeeklyOperationsCorruption, "hash chain"):
        _ = store.append(replace(_operation(), kst_day=date(2026, 8, 19), canonical=CanonicalPin(3, "c" * 64), occurred_at=datetime.fromisoformat("2026-08-19T20:00:00+09:00")))
    assert _data_path(store).read_bytes() == corrupted


def test_sidecar_coexists_with_old_reader_and_unchanged_canonical_bytes(tmp_path: Path) -> None:
    # Given: canonical bytes captured before a separate sidecar append.
    home = tmp_path / "customer"
    canonical = EventStore.for_standalone(home)
    event = build_satisfaction_event("client_001", score=8, collected_on=DAY)
    _ = canonical.append_wizard_event(event)
    before = (home / "events.jsonl").read_bytes()

    # When: status is appended and the v1.3.2 reader reopens canonical events.
    _ = _store(tmp_path / "weekly-profile").append(_operation())
    old_row = EventStore.for_standalone(home).load_wizard_event(event.event_id)

    # Then: the canonical stream is byte-identical and old reader-compatible.
    assert (home / "events.jsonl").read_bytes() == before
    assert old_row == event