govoplan-records/src/govoplan_records/backend/recovery.py

from __future__ import annotations

from dataclasses import dataclass
from contextvars import ContextVar, Token
import hashlib
import json
from typing import Any

from sqlalchemy.orm import Session, sessionmaker

from govoplan_core.core.recovery import (
    RecoveryGuaranteeError,
    RecoveryMode,
    RecoveryPlan,
)
from govoplan_core.core.recovery_runtime import (
    DurableRecoveryOperation,
    RecoveryOperationBusy,
    RecoveryOperationStateConflict,
    begin_durable_recovery_operation,
)
from govoplan_core.core.runtime_coordination import process_runtime_identity


class RecordRecoveryError(RuntimeError):
    pass


_recovery_operation_id: ContextVar[str | None] = ContextVar(
    "records_recovery_operation_id", default=None
)


def bind_record_recovery_operation(operation_id: str) -> Token[str | None]:
    return _recovery_operation_id.set(operation_id)


def current_record_recovery_operation() -> str | None:
    return _recovery_operation_id.get()


def reset_record_recovery_operation(token: Token[str | None]) -> None:
    _recovery_operation_id.reset(token)


def record_session_factory(session: Session) -> sessionmaker[Session]:
    bind = session.get_bind()
    if bind is None:
        raise RecordRecoveryError("Records recovery requires a bound database session.")
    return sessionmaker(bind=bind, expire_on_commit=False)


@dataclass(slots=True)
class RecordAtomicRecovery:
    operation: DurableRecoveryOperation | None
    operation_id: str
    replayed: bool

    def commit_success(
        self,
        session: Session,
        *,
        result: object,
        resource_id: str,
    ) -> None:
        evidence = {
            "verified": True,
            "resource_id": resource_id,
            "result_sha256": _canonical_sha256(result),
            "domain_and_checkpoint_atomic": True,
            "checks": {
                "domain_result_digest_recorded": True,
                "domain_and_checkpoint_atomic": True,
            },
        }
        if self.operation is None:
            session.commit()
            return
        self.operation.commit_atomic_success(session, evidence=evidence)

    def reject(self, *, summary: str, error_type: str) -> None:
        if self.operation is None:
            return
        self.operation.reject(
            summary=summary,
            evidence={
                "verified": True,
                "domain_mutation_committed": False,
                "error_type": error_type,
                "checks": {"definitive_domain_rejection": True},
            },
        )

    def fail(self, *, summary: str, error_type: str) -> None:
        if self.operation is None:
            return
        self.operation.fail(
            summary=summary,
            evidence={
                "verified": True,
                "domain_mutation_committed": False,
                "error_type": error_type,
            },
        )


def begin_record_atomic_recovery(
    session: Session,
    *,
    tenant_id: str,
    operation_type: str,
    idempotency_key: str,
    request: dict[str, Any],
    resource_type: str,
    resource_id: str,
) -> RecordAtomicRecovery:
    operation_key = hashlib.sha256(
        f"{tenant_id}:{operation_type}:{idempotency_key}".encode("utf-8")
    ).hexdigest()
    lease_id = hashlib.sha256(
        f"{tenant_id}:{resource_type}:{resource_id}".encode("utf-8")
    ).hexdigest()
    try:
        started = begin_durable_recovery_operation(
            record_session_factory(session),
            identity=process_runtime_identity(),
            module_id="records",
            operation_type=operation_type,
            idempotency_key=f"records:{operation_key}",
            request={"tenant_id": tenant_id, **request},
            recovery_plan=RecoveryPlan(
                mode=RecoveryMode.ATOMIC,
                preconditions=(
                    "the current actor is authorized for the Records mutation",
                    "the expected revision and idempotency key are present",
                    "the target resource has no unresolved recovery operation",
                ),
                verification_steps=(
                    "commit the immutable domain revision and chronology entry",
                    "commit the terminal recovery checkpoint in the same transaction",
                    "verify the result digest and recovery evidence chain",
                ),
            ),
            precondition_evidence={
                "tenant_id": tenant_id,
                "resource_type": resource_type,
                "resource_id_sha256": hashlib.sha256(
                    resource_id.encode("utf-8")
                ).hexdigest(),
                "request_sha256": _canonical_sha256(request),
                "external_effect": False,
            },
            lease_resource_key=f"records:{tenant_id}:{lease_id}",
            lease_ttl_seconds=5 * 60,
            resource_type=resource_type,
            resource_id=resource_id,
            metadata={
                "resources": ["postgresql"],
                "external_effect": False,
                "recovery_declaration": "records-atomic-revision",
            },
            block_unresolved_resource=True,
        )
    except RecoveryOperationBusy as exc:
        raise RecordRecoveryError(
            "Another runtime is changing this Records resource."
        ) from exc
    except RecoveryOperationStateConflict as exc:
        raise RecordRecoveryError(
            "This Records resource has an active or unresolved recovery operation."
        ) from exc
    except (RecoveryGuaranteeError, RuntimeError) as exc:
        raise RecordRecoveryError(
            "The recovery ledger is unavailable; the Records mutation was not started."
        ) from exc
    return RecordAtomicRecovery(
        operation=started.operation,
        operation_id=started.operation_id,
        replayed=started.replayed,
    )


def _canonical_sha256(value: object) -> str:
    return hashlib.sha256(
        json.dumps(
            value,
            sort_keys=True,
            separators=(",", ":"),
            ensure_ascii=False,
            default=str,
        ).encode("utf-8")
    ).hexdigest()


__all__ = [
    "RecordAtomicRecovery",
    "RecordRecoveryError",
    "begin_record_atomic_recovery",
    "bind_record_recovery_operation",
    "current_record_recovery_operation",
    "record_session_factory",
    "reset_record_recovery_operation",
]