DataOps-platform/app/core/system/tenant_lifecycle.py

"""Deterministic lifecycle and fence contract used before tenant activation."""

from __future__ import annotations

import hashlib
import json
import re
from dataclasses import dataclass, field
from typing import Any


class TenantLifecycleError(ValueError):
    pass


_DIGEST = re.compile(r"^[0-9a-f]{64}$")
_TRANSITIONS = {
    ("provisioning", "activate"): "active",
    ("active", "freeze"): "frozen",
    ("frozen", "begin_recovery"): "recovering",
    ("recovering", "activate"): "active",
    ("active", "mark_deletion_candidate"): "deletion_candidate",
    ("frozen", "mark_deletion_candidate"): "deletion_candidate",
    ("deletion_candidate", "delete"): "deleted",
    ("deletion_candidate", "rollback"): "frozen",
}


@dataclass(frozen=True)
class TenantLifecycleRecord:
    tenant_id: str
    state: str
    fence: int


@dataclass
class TenantLifecycle:
    tenant_id: str
    state: str = "provisioning"
    fence: int = 0
    _replays: dict[str, tuple[str, TenantLifecycleRecord]] = field(default_factory=dict)

    @classmethod
    def provision(cls, tenant_id: str) -> TenantLifecycle:
        return cls(tenant_id=tenant_id)

    def _digest(self, action: str, expected_fence: int, values: dict[str, Any]) -> str:
        payload = json.dumps(
            {"action": action, "expected_fence": expected_fence, **values},
            sort_keys=True,
            separators=(",", ":"),
        )
        return hashlib.sha256(payload.encode("utf-8")).hexdigest()

    def transition(
        self,
        action: str,
        *,
        expected_fence: int,
        idempotency_key: str,
        approval_ref: str | None = None,
        backup_digest: str | None = None,
        retention_seconds: int | None = None,
        hold_active: bool = False,
    ) -> TenantLifecycleRecord:
        if not isinstance(expected_fence, int) or isinstance(expected_fence, bool):
            raise TenantLifecycleError("tenant_fence_conflict")
        if not isinstance(idempotency_key, str) or not idempotency_key:
            raise TenantLifecycleError("tenant_replay_conflict")
        values = {
            "approval_ref": approval_ref,
            "backup_digest": backup_digest,
            "retention_seconds": retention_seconds,
            "hold_active": hold_active,
        }
        request_digest = self._digest(action, expected_fence, values)
        replay = self._replays.get(idempotency_key)
        if replay is not None:
            if replay[0] != request_digest:
                raise TenantLifecycleError("tenant_replay_conflict")
            return replay[1]
        if expected_fence != self.fence:
            raise TenantLifecycleError("tenant_fence_conflict")
        target = _TRANSITIONS.get((self.state, action))
        if target is None:
            raise TenantLifecycleError("tenant_transition_denied")
        if action in {"begin_recovery", "activate"} and self.state == "recovering" and not approval_ref:
            raise TenantLifecycleError("tenant_recovery_approval_required")
        if action in {"mark_deletion_candidate", "delete"}:
            if not approval_ref:
                raise TenantLifecycleError("tenant_deletion_approval_required")
            if hold_active:
                raise TenantLifecycleError("tenant_hold_active")
            if not isinstance(backup_digest, str) or not _DIGEST.fullmatch(backup_digest):
                raise TenantLifecycleError("tenant_backup_required")
            if (
                not isinstance(retention_seconds, int)
                or isinstance(retention_seconds, bool)
                or retention_seconds < 1
            ):
                raise TenantLifecycleError("tenant_retention_required")
        self.state = target
        self.fence += 1
        result = TenantLifecycleRecord(self.tenant_id, self.state, self.fence)
        self._replays[idempotency_key] = (request_digest, result)
        return result

    def require_active_task_scope(self, *, expected_fence: int) -> TenantLifecycleRecord:
        if expected_fence != self.fence:
            raise TenantLifecycleError("tenant_fence_conflict")
        if self.state != "active":
            raise TenantLifecycleError("tenant_not_active")
        return TenantLifecycleRecord(self.tenant_id, self.state, self.fence)