"""Rename journal — the durable, crash-safe record of every rename transition
(US04-02).

The filesystem and SQLite cannot share a transaction, so the journal is the only
thing that makes a rename recoverable: **intent is written before the mutation it
describes**, and the recorded state plus the real filesystem evidence is what a later
restart reasons about. This module owns the state machine and the durable writes; it
performs no filesystem mutation itself (that is US04-03).

Per-operation state machine (concept §7):

```
planned → moving → moved → database_updated → verified → complete
              ↘ failed          ↘ rollback_required → rolled_back
```

``moving`` is the dangerous window: it means "we have committed to touching the disk
and may already have done so". Anything found in ``moving`` after a restart needs
evidence from the filesystem, never a blind retry.

Terminal transitions are idempotent: re-completing an already-``complete`` operation
is a no-op rather than an error, so a recovery pass may run repeatedly. Ownership uses
the same fencing-token discipline as durable jobs — a worker whose token has been
superseded cannot commit.

Startup classification (``classify``) inspects the journal together with the current
source/destination on disk and labels each incomplete operation:

- ``resumable``     — the move has not happened; the source is intact and the
                      destination is free, so applying again is safe.
- ``rollback_safe`` — the move happened but the database was not updated; the
                      destination holds the expected content and the source is gone,
                      so the operation can be completed or reversed deterministically.
- ``manual``        — the evidence is ambiguous or contradicts the journal (both or
                      neither path present, or an unexpected occupant). Nothing is
                      guessed; the operation blocks until a human decides.
"""

from __future__ import annotations

import json
from datetime import datetime, timezone
from pathlib import Path

from sqlalchemy import select
from sqlalchemy.orm import sessionmaker

from photo_pipeline.models import RenameOperation, RenamePlan


class JournalState:
    PLANNED = "planned"
    MOVING = "moving"
    MOVED = "moved"
    DATABASE_UPDATED = "database_updated"
    VERIFIED = "verified"
    COMPLETE = "complete"
    FAILED = "failed"
    ROLLBACK_REQUIRED = "rollback_required"
    ROLLED_BACK = "rolled_back"


ALLOWED_TRANSITIONS = {
    JournalState.PLANNED: {JournalState.MOVING, JournalState.FAILED},
    # From `moving` the outcome is genuinely unknown until evidence is gathered,
    # so it may resolve forward, back to planned (proven untouched), or to failure.
    JournalState.MOVING: {
        JournalState.MOVED,
        JournalState.PLANNED,
        JournalState.FAILED,
        JournalState.ROLLBACK_REQUIRED,
    },
    JournalState.MOVED: {JournalState.DATABASE_UPDATED, JournalState.ROLLBACK_REQUIRED},
    JournalState.DATABASE_UPDATED: {JournalState.VERIFIED, JournalState.ROLLBACK_REQUIRED},
    JournalState.VERIFIED: {JournalState.COMPLETE, JournalState.ROLLBACK_REQUIRED},
    JournalState.COMPLETE: set(),
    JournalState.FAILED: {JournalState.PLANNED, JournalState.ROLLBACK_REQUIRED},
    JournalState.ROLLBACK_REQUIRED: {JournalState.ROLLED_BACK, JournalState.FAILED},
    JournalState.ROLLED_BACK: set(),
}

TERMINAL_STATES = frozenset({JournalState.COMPLETE, JournalState.ROLLED_BACK})
# States where the disk may already have been touched by this operation.
UNSAFE_STATES = frozenset({JournalState.MOVING, JournalState.MOVED, JournalState.DATABASE_UPDATED})

RESUMABLE = "resumable"
ROLLBACK_SAFE = "rollback_safe"
MANUAL = "manual"


def can_transition(current: str, target: str) -> bool:
    return target in ALLOWED_TRANSITIONS.get(current, set())


class JournalError(RuntimeError):
    pass


class InvalidTransition(JournalError):
    pass


class JournalConflict(JournalError):
    """Fencing check failed; a newer owner has taken over this operation."""


def _now() -> datetime:
    return datetime.now(timezone.utc)


class RenameJournal:
    def __init__(self, session_factory: sessionmaker) -> None:
        self._session_factory = session_factory

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

    def begin(self, operation_id: str, *, worker_id: str, fencing_token: int) -> dict:
        """Record the intent to mutate **before** touching the filesystem.

        Claims the operation for ``worker_id`` with ``fencing_token`` and moves it to
        ``moving``. A token lower than the one already recorded loses the race and
        raises ``JournalConflict`` without changing anything.
        """
        with self._session_factory() as session:
            row = self._require(session, operation_id)
            if row.fencing_token is not None and fencing_token < row.fencing_token:
                raise JournalConflict(
                    f"stale fencing token {fencing_token} (current {row.fencing_token})"
                )
            if row.journal_state in TERMINAL_STATES:
                raise InvalidTransition(f"{row.journal_state} is terminal")
            if row.journal_state != JournalState.MOVING and not can_transition(
                row.journal_state, JournalState.MOVING
            ):
                raise InvalidTransition(f"{row.journal_state} -> {JournalState.MOVING}")
            if row.journal_state != JournalState.MOVING:
                row.attempt_count += 1
            row.journal_state = JournalState.MOVING
            row.worker_id = worker_id
            row.fencing_token = fencing_token
            row.error_code = row.error_message = None
            row.updated_at = _now()
            session.commit()
            return _operation_dict(row)

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

    def transition(
        self,
        operation_id: str,
        target: str,
        *,
        fencing_token: int | None = None,
        error: tuple[str, str] | None = None,
    ) -> dict:
        """Move one operation to ``target``, enforcing the state machine.

        Re-entering the state an operation already holds is a no-op, which is what
        makes recovery idempotent across repeated restarts.
        """
        with self._session_factory() as session:
            row = self._require(session, operation_id)
            if fencing_token is not None and row.fencing_token is not None:
                if fencing_token < row.fencing_token:
                    raise JournalConflict(
                        f"stale fencing token {fencing_token} (current {row.fencing_token})"
                    )
            if row.journal_state == target:
                return _operation_dict(row)  # idempotent
            if not can_transition(row.journal_state, target):
                raise InvalidTransition(f"{row.journal_state} -> {target}")

            row.journal_state = target
            row.updated_at = _now()
            if target == JournalState.VERIFIED:
                row.verified_at = _now()
            if error:
                row.error_code, row.error_message = error[0], error[1][:500]
            elif target not in (JournalState.FAILED, JournalState.ROLLBACK_REQUIRED):
                row.error_code = row.error_message = None
            session.commit()
            return _operation_dict(row)

    # ── reads ─────────────────────────────────────────────────────────────────

    def get(self, operation_id: str) -> dict | None:
        with self._session_factory() as session:
            row = session.get(RenameOperation, operation_id)
            return _operation_dict(row) if row else None

    def operations(self, plan_id: str) -> list[dict]:
        with self._session_factory() as session:
            rows = session.scalars(
                select(RenameOperation)
                .where(RenameOperation.plan_id == plan_id)
                .order_by(RenameOperation.sequence)
            )
            return [_operation_dict(row) for row in rows]

    def incomplete(self) -> list[dict]:
        """Every operation left in a non-terminal, non-planned state — the work a
        restart has to reason about."""
        with self._session_factory() as session:
            rows = session.scalars(
                select(RenameOperation)
                .where(
                    RenameOperation.journal_state.not_in([*TERMINAL_STATES, JournalState.PLANNED])
                )
                .order_by(RenameOperation.plan_id, RenameOperation.sequence)
            )
            return [_operation_dict(row) for row in rows]

    # ── startup classification ────────────────────────────────────────────────

    def classify(self, operation_id: str) -> dict:
        """Classify one incomplete operation from the journal plus disk evidence.

        Returns ``{operation_id, journal_state, classification, reason, source_exists,
        destination_exists}``. Never mutates anything — deciding is US04-04's job.
        """
        row = self.get(operation_id)
        if row is None:
            raise JournalError(f"unknown rename operation {operation_id!r}")
        source = Path(row["source_path"])
        destination = Path(row["destination_path"])
        source_exists = source.exists()
        destination_exists = destination.exists()

        classification, reason = _classify(row["journal_state"], source_exists, destination_exists)
        return {
            "operation_id": operation_id,
            "plan_id": row["plan_id"],
            "album": row["album"],
            # The paths travel with the verdict so a reviewer can see which folder is
            # unresolved without correlating an opaque operation id by hand.
            "source_path": row["source_path"],
            "destination_path": row["destination_path"],
            "journal_state": row["journal_state"],
            "classification": classification,
            "reason": reason,
            "source_exists": source_exists,
            "destination_exists": destination_exists,
        }

    def classify_all(self) -> list[dict]:
        return [self.classify(row["id"]) for row in self.incomplete()]

    def blocks_mutation(self) -> bool:
        """True when any operation is in a state where the library may be
        half-renamed. Unrelated mutations must not start until it is resolved."""
        return any(row["journal_state"] in UNSAFE_STATES for row in self.incomplete())

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

    def plan_state(self, plan_id: str) -> str:
        """Derive the plan's state from its operations, so the summary can never
        disagree with the journal."""
        states = {row["journal_state"] for row in self.operations(plan_id)}
        if not states:
            return "planned"
        if states <= {JournalState.COMPLETE}:
            return "applied"
        if states <= {JournalState.ROLLED_BACK, JournalState.COMPLETE}:
            return "rolled_back"
        if states & {JournalState.FAILED, JournalState.ROLLBACK_REQUIRED}:
            return "failed"
        if states & UNSAFE_STATES:
            return "applying"
        return "planned"

    def sync_plan_state(self, plan_id: str) -> str:
        state = self.plan_state(plan_id)
        with self._session_factory() as session:
            plan = session.get(RenamePlan, plan_id)
            if plan is None:
                raise JournalError(f"unknown rename plan {plan_id!r}")
            if plan.state != state:
                plan.state = state
                plan.version += 1
                plan.updated_at = _now()
                if state == "applied" and plan.applied_at is None:
                    plan.applied_at = _now()
                session.commit()
        return state

    @staticmethod
    def _require(session, operation_id: str) -> RenameOperation:
        row = session.get(RenameOperation, operation_id)
        if row is None:
            raise JournalError(f"unknown rename operation {operation_id!r}")
        return row


def _classify(state: str, source_exists: bool, destination_exists: bool) -> tuple[str, str]:
    """The evidence table. Kept a pure function so every combination is testable."""
    if state == JournalState.MOVING:
        # Intent was recorded; the move may or may not have happened.
        if source_exists and not destination_exists:
            return RESUMABLE, "source intact and destination free; the move did not happen"
        if destination_exists and not source_exists:
            return ROLLBACK_SAFE, "the move completed before the journal was updated"
        if source_exists and destination_exists:
            return MANUAL, "both source and destination exist; cannot tell them apart safely"
        return MANUAL, "neither source nor destination exists"

    if state in (JournalState.MOVED, JournalState.DATABASE_UPDATED, JournalState.VERIFIED):
        if destination_exists and not source_exists:
            return (
                ROLLBACK_SAFE,
                "destination holds the content; finish or reverse deterministically",
            )
        if source_exists and not destination_exists:
            return MANUAL, f"journal says {state} but the content is still at the source"
        if source_exists and destination_exists:
            return MANUAL, "both paths exist after a recorded move"
        return MANUAL, "the recorded destination is missing"

    if state == JournalState.ROLLBACK_REQUIRED:
        if source_exists and not destination_exists:
            return RESUMABLE, "already back at the source; the rollback is effectively done"
        if destination_exists and not source_exists:
            return ROLLBACK_SAFE, "content is at the destination and can be moved back"
        return MANUAL, "ambiguous paths for a rollback"

    if state == JournalState.FAILED:
        if source_exists and not destination_exists:
            return RESUMABLE, "the failure left the source untouched"
        return MANUAL, "a failed operation with unexpected paths on disk"

    return MANUAL, f"unhandled journal state {state}"


def _operation_dict(row: RenameOperation) -> dict:
    return {
        "id": row.id,
        "plan_id": row.plan_id,
        "sequence": row.sequence,
        "album": row.album,
        "source_path": row.source_path,
        "destination_path": row.destination_path,
        "asset_ids": json.loads(row.asset_ids or "[]"),
        "expected_sha256": json.loads(row.expected_sha256 or "{}"),
        "case_only": row.case_only,
        "same_filesystem": row.same_filesystem,
        "journal_state": row.journal_state,
        "attempt_count": row.attempt_count,
        "fencing_token": row.fencing_token,
        "worker_id": row.worker_id,
        "verified_at": row.verified_at.isoformat() if row.verified_at else None,
        "error_code": row.error_code,
        "error_message": row.error_message,
    }