"""Rename journal persistence, fencing, idempotency, and restart durability
(US04-02).

The journal is what makes a rename recoverable, so these tests care about what
survives: intent recorded before any mutation, terminal transitions that can be
replayed, ownership that a superseded worker cannot steal back, and classification
that reads real files on disk after a simulated restart.
"""

import json
import uuid
from datetime import datetime, timezone

import pytest

from photo_pipeline.config import Config
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
from photo_pipeline.models import RenameOperation, RenamePlan
from photo_pipeline.services.rename_journal import (
    MANUAL,
    RESUMABLE,
    ROLLBACK_SAFE,
    InvalidTransition,
    JournalConflict,
    JournalError,
    JournalState,
    RenameJournal,
)

NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)


def _factory(tmp_path):
    (tmp_path / "data").mkdir()
    config = Config.from_env({"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data")})
    run_migrations(config.database_url)
    return config, create_session_factory(create_db_engine(config.database_url))


def _plan(sf, tmp_path, *, count=1):
    """A persisted plan whose operations point at real folders on disk."""
    plan_id = str(uuid.uuid4())
    lib = tmp_path / "lib"
    lib.mkdir(exist_ok=True)
    ids = []
    with sf() as session:
        session.add(
            RenamePlan(id=plan_id, state="validated", schema_version=1, operation_count=count)
        )
        session.flush()
        for index in range(count):
            source = lib / f"album{index}"
            source.mkdir(exist_ok=True)
            (source / "a.jpg").write_bytes(b"x" * 8)
            operation_id = str(uuid.uuid4())
            ids.append(operation_id)
            session.add(
                RenameOperation(
                    id=operation_id,
                    plan_id=plan_id,
                    sequence=index,
                    operation="move_folder",
                    album=f"album{index}",
                    source_path=str(source),
                    destination_path=str(lib / f"Renamed {index}"),
                    asset_ids=json.dumps([f"asset{index}"]),
                    asset_count=1,
                )
            )
        session.commit()
    return plan_id, ids


# ── intent ───────────────────────────────────────────────────────────────────


def test_begin_records_intent_ownership_and_attempt(tmp_path):
    _, sf = _factory(tmp_path)
    _, (operation_id,) = _plan(sf, tmp_path)
    journal = RenameJournal(sf)

    started = journal.begin(operation_id, worker_id="w1", fencing_token=1)
    assert started["journal_state"] == JournalState.MOVING
    assert started["worker_id"] == "w1" and started["fencing_token"] == 1
    assert started["attempt_count"] == 1


def test_begin_is_replayable_without_inflating_attempts(tmp_path):
    _, sf = _factory(tmp_path)
    _, (operation_id,) = _plan(sf, tmp_path)
    journal = RenameJournal(sf)

    journal.begin(operation_id, worker_id="w1", fencing_token=1)
    again = journal.begin(operation_id, worker_id="w1", fencing_token=1)
    # Re-entering `moving` is the same attempt, not a new one.
    assert again["attempt_count"] == 1


def test_a_retry_after_failure_counts_a_new_attempt(tmp_path):
    _, sf = _factory(tmp_path)
    _, (operation_id,) = _plan(sf, tmp_path)
    journal = RenameJournal(sf)

    journal.begin(operation_id, worker_id="w1", fencing_token=1)
    journal.transition(operation_id, JournalState.FAILED, error=("io", "disk hiccup"))
    journal.transition(operation_id, JournalState.PLANNED)
    retried = journal.begin(operation_id, worker_id="w1", fencing_token=2)
    assert retried["attempt_count"] == 2 and retried["error_code"] is None


def test_a_superseded_worker_cannot_take_the_operation_back(tmp_path):
    _, sf = _factory(tmp_path)
    _, (operation_id,) = _plan(sf, tmp_path)
    journal = RenameJournal(sf)

    journal.begin(operation_id, worker_id="w1", fencing_token=1)
    journal.begin(operation_id, worker_id="w2", fencing_token=5)  # takeover

    with pytest.raises(JournalConflict):
        journal.begin(operation_id, worker_id="w1", fencing_token=1)
    with pytest.raises(JournalConflict):
        journal.transition(operation_id, JournalState.MOVED, fencing_token=1)
    assert journal.get(operation_id)["worker_id"] == "w2"


def test_begin_refuses_a_terminal_operation(tmp_path):
    _, sf = _factory(tmp_path)
    _, (operation_id,) = _plan(sf, tmp_path)
    journal = RenameJournal(sf)
    _walk_to_complete(journal, operation_id)

    with pytest.raises(InvalidTransition):
        journal.begin(operation_id, worker_id="w1", fencing_token=9)


def test_unknown_operation_is_reported(tmp_path):
    _, sf = _factory(tmp_path)
    journal = RenameJournal(sf)
    with pytest.raises(JournalError):
        journal.begin("nope", worker_id="w1", fencing_token=1)
    with pytest.raises(JournalError):
        journal.classify("nope")


# ── transitions ──────────────────────────────────────────────────────────────


def _walk_to_complete(journal, operation_id):
    journal.begin(operation_id, worker_id="w1", fencing_token=1)
    for state in (
        JournalState.MOVED,
        JournalState.DATABASE_UPDATED,
        JournalState.VERIFIED,
        JournalState.COMPLETE,
    ):
        journal.transition(operation_id, state)


def test_full_lifecycle_is_recorded_with_verification_evidence(tmp_path):
    _, sf = _factory(tmp_path)
    _, (operation_id,) = _plan(sf, tmp_path)
    journal = RenameJournal(sf)

    _walk_to_complete(journal, operation_id)
    final = journal.get(operation_id)
    assert final["journal_state"] == JournalState.COMPLETE
    assert final["verified_at"] is not None, "verification evidence must be durable"


def test_terminal_transitions_are_idempotent(tmp_path):
    _, sf = _factory(tmp_path)
    _, (operation_id,) = _plan(sf, tmp_path)
    journal = RenameJournal(sf)
    _walk_to_complete(journal, operation_id)

    # Replaying the terminal transition is a no-op, so recovery can run repeatedly.
    for _ in range(3):
        replayed = journal.transition(operation_id, JournalState.COMPLETE)
        assert replayed["journal_state"] == JournalState.COMPLETE


def test_invalid_transition_is_refused_without_changing_state(tmp_path):
    _, sf = _factory(tmp_path)
    _, (operation_id,) = _plan(sf, tmp_path)
    journal = RenameJournal(sf)

    with pytest.raises(InvalidTransition):
        journal.transition(operation_id, JournalState.COMPLETE)  # planned -> complete
    assert journal.get(operation_id)["journal_state"] == JournalState.PLANNED


def test_failure_records_a_structured_error(tmp_path):
    _, sf = _factory(tmp_path)
    _, (operation_id,) = _plan(sf, tmp_path)
    journal = RenameJournal(sf)
    journal.begin(operation_id, worker_id="w1", fencing_token=1)

    failed = journal.transition(
        operation_id, JournalState.FAILED, error=("permission_denied", "read-only target")
    )
    assert failed["error_code"] == "permission_denied"
    assert "read-only" in failed["error_message"]


# ── classification against real files ────────────────────────────────────────


def test_moving_with_untouched_source_classifies_resumable(tmp_path):
    _, sf = _factory(tmp_path)
    _, (operation_id,) = _plan(sf, tmp_path)
    journal = RenameJournal(sf)
    journal.begin(operation_id, worker_id="w1", fencing_token=1)

    verdict = journal.classify(operation_id)
    assert verdict["classification"] == RESUMABLE
    assert verdict["source_exists"] is True and verdict["destination_exists"] is False


def test_moving_after_the_move_happened_classifies_rollback_safe(tmp_path):
    _, sf = _factory(tmp_path)
    _, (operation_id,) = _plan(sf, tmp_path)
    journal = RenameJournal(sf)
    row = journal.get(operation_id)
    journal.begin(operation_id, worker_id="w1", fencing_token=1)

    # Simulate the crash window: the move completed, the journal never advanced.
    from pathlib import Path

    Path(row["source_path"]).rename(row["destination_path"])

    verdict = journal.classify(operation_id)
    assert verdict["classification"] == ROLLBACK_SAFE


def test_an_unexpected_occupant_at_the_destination_requires_manual_recovery(tmp_path):
    _, sf = _factory(tmp_path)
    _, (operation_id,) = _plan(sf, tmp_path)
    journal = RenameJournal(sf)
    row = journal.get(operation_id)
    journal.begin(operation_id, worker_id="w1", fencing_token=1)

    from pathlib import Path

    Path(row["destination_path"]).mkdir()  # both paths now exist

    verdict = journal.classify(operation_id)
    assert verdict["classification"] == MANUAL
    assert verdict["reason"]


def test_classification_is_stable_across_repeated_runs(tmp_path):
    _, sf = _factory(tmp_path)
    _, (operation_id,) = _plan(sf, tmp_path)
    journal = RenameJournal(sf)
    journal.begin(operation_id, worker_id="w1", fencing_token=1)

    first = journal.classify(operation_id)
    for _ in range(3):
        assert journal.classify(operation_id) == first


# ── incomplete work and mutation blocking ────────────────────────────────────


def test_incomplete_lists_only_unsettled_operations(tmp_path):
    _, sf = _factory(tmp_path)
    _, ids = _plan(sf, tmp_path, count=3)
    journal = RenameJournal(sf)

    _walk_to_complete(journal, ids[0])  # settled
    journal.begin(ids[1], worker_id="w1", fencing_token=1)  # in flight
    # ids[2] stays `planned` — not started, so not something a restart must resolve.

    incomplete = {row["id"] for row in journal.incomplete()}
    assert incomplete == {ids[1]}


def test_in_flight_work_blocks_unrelated_mutations(tmp_path):
    _, sf = _factory(tmp_path)
    _, ids = _plan(sf, tmp_path, count=2)
    journal = RenameJournal(sf)
    assert journal.blocks_mutation() is False

    journal.begin(ids[0], worker_id="w1", fencing_token=1)
    assert journal.blocks_mutation() is True, "a half-renamed library must block"

    for state in (
        JournalState.MOVED,
        JournalState.DATABASE_UPDATED,
        JournalState.VERIFIED,
        JournalState.COMPLETE,
    ):
        journal.transition(ids[0], state)
    assert journal.blocks_mutation() is False


def test_classify_all_covers_every_incomplete_operation(tmp_path):
    _, sf = _factory(tmp_path)
    _, ids = _plan(sf, tmp_path, count=2)
    journal = RenameJournal(sf)
    journal.begin(ids[0], worker_id="w1", fencing_token=1)
    journal.begin(ids[1], worker_id="w1", fencing_token=1)

    verdicts = journal.classify_all()
    assert {v["operation_id"] for v in verdicts} == set(ids)


# ── plan-level derivation ────────────────────────────────────────────────────


def test_plan_state_is_derived_from_its_operations(tmp_path):
    _, sf = _factory(tmp_path)
    plan_id, ids = _plan(sf, tmp_path, count=2)
    journal = RenameJournal(sf)
    assert journal.plan_state(plan_id) == "planned"

    journal.begin(ids[0], worker_id="w1", fencing_token=1)
    assert journal.plan_state(plan_id) == "applying"
    assert journal.sync_plan_state(plan_id) == "applying"

    _finish = (
        JournalState.MOVED,
        JournalState.DATABASE_UPDATED,
        JournalState.VERIFIED,
        JournalState.COMPLETE,
    )
    for state in _finish:
        journal.transition(ids[0], state)
    _walk_to_complete(journal, ids[1])
    assert journal.sync_plan_state(plan_id) == "applied"

    with sf() as session:
        plan = session.get(RenamePlan, plan_id)
        assert plan.state == "applied" and plan.applied_at is not None


def test_a_failed_operation_makes_the_plan_failed(tmp_path):
    _, sf = _factory(tmp_path)
    plan_id, ids = _plan(sf, tmp_path, count=2)
    journal = RenameJournal(sf)
    journal.begin(ids[0], worker_id="w1", fencing_token=1)
    journal.transition(ids[0], JournalState.FAILED, error=("io", "nope"))
    assert journal.sync_plan_state(plan_id) == "failed"


def test_sync_plan_state_rejects_an_unknown_plan(tmp_path):
    _, sf = _factory(tmp_path)
    journal = RenameJournal(sf)
    with pytest.raises(JournalError):
        journal.sync_plan_state("nope")


# ── durability ───────────────────────────────────────────────────────────────


def test_journal_survives_a_process_restart(tmp_path):
    config, sf = _factory(tmp_path)
    _, (operation_id,) = _plan(sf, tmp_path)
    RenameJournal(sf).begin(operation_id, worker_id="w1", fencing_token=7)

    # Drop every in-process handle, then reopen the database like a fresh process.
    reopened = create_session_factory(create_db_engine(config.database_url))
    journal = RenameJournal(reopened)
    row = journal.get(operation_id)
    assert row["journal_state"] == JournalState.MOVING
    assert row["fencing_token"] == 7 and row["worker_id"] == "w1"
    assert journal.classify(operation_id)["classification"] == RESUMABLE
    # And the stale worker still cannot commit after the restart.
    with pytest.raises(JournalConflict):
        journal.transition(operation_id, JournalState.MOVED, fencing_token=1)