from __future__ import annotations

from collections import Counter, deque
from dataclasses import dataclass, field
from datetime import UTC, datetime
import hashlib
import json
import re
from typing import Any, Mapping

from sqlalchemy import func, or_, select
from sqlalchemy.orm import Session

from govoplan_core.auth import ApiPrincipal
from govoplan_core.db.base import utcnow
from govoplan_risk_compliance.backend.db.models import (
    RiskAssuranceEdge,
    RiskAssuranceNode,
    RiskScreeningRun,
)
from govoplan_risk_compliance.backend.permissions import (
    ADMIN_SCOPE,
    READ_SCOPE,
    WRITE_SCOPE,
)


ASSURANCE_NODE_KINDS = (
    "obligation",
    "governed_object",
    "risk",
    "control",
    "evidence",
    "finding",
    "corrective_measure",
    "effectiveness_review",
)
ASSURANCE_EDGE_RELATIONS: Mapping[str, tuple[str, str]] = {
    "applies_to": ("obligation", "governed_object"),
    "exposes_risk": ("governed_object", "risk"),
    "mitigated_by": ("risk", "control"),
    "evidenced_by": ("control", "evidence"),
    "results_in": ("evidence", "finding"),
    "addressed_by": ("finding", "corrective_measure"),
    "reviewed_by": ("corrective_measure", "effectiveness_review"),
}
ASSURANCE_EDGE_STATES = frozenset({"active", "suspended", "retired"})
ASSURANCE_NODE_STATES: Mapping[str, frozenset[str]] = {
    "obligation": frozenset({"active", "suspended", "retired"}),
    "governed_object": frozenset({"active", "inactive", "retired"}),
    "risk": frozenset({"identified", "assessed", "accepted", "mitigated", "closed"}),
    "control": frozenset({"designed", "implemented", "effective", "failed", "suspended", "retired"}),
    "evidence": frozenset({"current", "stale", "invalid", "superseded"}),
    "finding": frozenset({"open", "accepted", "exception", "remediating", "resolved"}),
    "corrective_measure": frozenset({"planned", "in_progress", "completed", "cancelled"}),
    "effectiveness_review": frozenset({"pending", "effective", "ineffective", "inconclusive"}),
}

MAX_GRAPH_DEPTH = 8
MAX_GRAPH_ITEMS = 500
MAX_PROVENANCE_BYTES = 100_000
_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:/-]{0,254}$")


class RiskAssuranceError(ValueError):
    pass


class RiskAssuranceAccessError(RiskAssuranceError):
    pass


class RiskAssuranceNotFoundError(RiskAssuranceError):
    pass


class RiskAssuranceConflictError(RiskAssuranceError):
    pass


@dataclass(frozen=True, slots=True)
class AssuranceNodeInput:
    stable_id: str
    kind: str
    label: str
    state: str
    owner_ref: str
    valid_from: datetime
    description: str | None = None
    scope_ref: str | None = None
    governed_object_ref: str | None = None
    valid_to: datetime | None = None
    provenance: Mapping[str, Any] = field(default_factory=dict)
    legal_basis_refs: tuple[str, ...] = ()
    policy_refs: tuple[str, ...] = ()
    evidence_refs: tuple[str, ...] = ()
    classification: str = "internal"

    def __post_init__(self) -> None:
        _validate_node_input(self)


@dataclass(frozen=True, slots=True)
class AssuranceEdgeInput:
    stable_id: str
    source_node_ref: str
    target_node_ref: str
    relation: str
    owner_ref: str
    valid_from: datetime
    state: str = "active"
    scope_ref: str | None = None
    valid_to: datetime | None = None
    provenance: Mapping[str, Any] = field(default_factory=dict)
    legal_basis_refs: tuple[str, ...] = ()
    policy_refs: tuple[str, ...] = ()
    evidence_refs: tuple[str, ...] = ()

    def __post_init__(self) -> None:
        _validate_edge_input(self)


@dataclass(frozen=True, slots=True)
class AssuranceGraph:
    root_ref: str
    nodes: tuple[RiskAssuranceNode, ...]
    edges: tuple[RiskAssuranceEdge, ...]
    truncated: bool


def create_assurance_node(
    session: Session,
    principal: ApiPrincipal,
    *,
    value: AssuranceNodeInput,
) -> RiskAssuranceNode:
    _require_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
    if _current_node(session, principal.tenant_id, value.stable_id) is not None:
        raise RiskAssuranceConflictError(
            f"Assurance node {value.stable_id!r} already exists."
        )
    return _insert_node(
        session,
        tenant_id=principal.tenant_id,
        value=value,
        revision=1,
        previous=None,
        actor_id=_actor_id(principal),
    )


def revise_assurance_node(
    session: Session,
    principal: ApiPrincipal,
    *,
    stable_id: str,
    expected_revision: int,
    value: AssuranceNodeInput,
) -> RiskAssuranceNode:
    _require_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
    if value.stable_id != stable_id:
        raise RiskAssuranceConflictError(
            "The assurance node identity cannot change during revision."
        )
    current = _locked_current_node(session, principal.tenant_id, stable_id)
    if current is None:
        raise RiskAssuranceNotFoundError("Assurance node not found.")
    if current.revision != expected_revision:
        raise RiskAssuranceConflictError(
            f"Expected revision {expected_revision}, current revision is {current.revision}."
        )
    if current.kind != value.kind:
        raise RiskAssuranceConflictError(
            "The assurance node kind cannot change during revision."
        )
    return _insert_node(
        session,
        tenant_id=principal.tenant_id,
        value=value,
        revision=current.revision + 1,
        previous=current,
        actor_id=_actor_id(principal),
    )


def create_assurance_edge(
    session: Session,
    principal: ApiPrincipal,
    *,
    value: AssuranceEdgeInput,
) -> RiskAssuranceEdge:
    _require_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
    if _current_edge(session, principal.tenant_id, value.stable_id) is not None:
        raise RiskAssuranceConflictError(
            f"Assurance edge {value.stable_id!r} already exists."
        )
    _validate_edge_endpoints(session, principal.tenant_id, value)
    return _insert_edge(
        session,
        tenant_id=principal.tenant_id,
        value=value,
        revision=1,
        previous=None,
        actor_id=_actor_id(principal),
    )


def revise_assurance_edge(
    session: Session,
    principal: ApiPrincipal,
    *,
    stable_id: str,
    expected_revision: int,
    value: AssuranceEdgeInput,
) -> RiskAssuranceEdge:
    _require_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
    if value.stable_id != stable_id:
        raise RiskAssuranceConflictError(
            "The assurance edge identity cannot change during revision."
        )
    current = _locked_current_edge(session, principal.tenant_id, stable_id)
    if current is None:
        raise RiskAssuranceNotFoundError("Assurance edge not found.")
    if current.revision != expected_revision:
        raise RiskAssuranceConflictError(
            f"Expected revision {expected_revision}, current revision is {current.revision}."
        )
    identity = (
        value.source_node_ref,
        value.target_node_ref,
        value.relation,
    )
    if identity != (
        current.source_node_ref,
        current.target_node_ref,
        current.relation,
    ):
        raise RiskAssuranceConflictError(
            "Edge endpoints and relation cannot change during revision."
        )
    _validate_edge_endpoints(session, principal.tenant_id, value)
    return _insert_edge(
        session,
        tenant_id=principal.tenant_id,
        value=value,
        revision=current.revision + 1,
        previous=current,
        actor_id=_actor_id(principal),
    )


def get_assurance_node(
    session: Session,
    principal: ApiPrincipal,
    *,
    stable_id: str,
) -> RiskAssuranceNode:
    _require_scope(principal, READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE)
    item = _current_node(session, principal.tenant_id, stable_id)
    if item is None:
        raise RiskAssuranceNotFoundError("Assurance node not found.")
    return item


def list_assurance_node_history(
    session: Session,
    principal: ApiPrincipal,
    *,
    stable_id: str,
    limit: int = 100,
) -> tuple[RiskAssuranceNode, ...]:
    _require_scope(principal, READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE)
    bounded_limit = _bounded_limit(limit)
    return tuple(
        session.scalars(
            select(RiskAssuranceNode)
            .where(
                RiskAssuranceNode.tenant_id == principal.tenant_id,
                RiskAssuranceNode.stable_id == stable_id,
            )
            .order_by(RiskAssuranceNode.revision.desc())
            .limit(bounded_limit)
        )
    )


def list_assurance_edge_history(
    session: Session,
    principal: ApiPrincipal,
    *,
    stable_id: str,
    limit: int = 100,
) -> tuple[RiskAssuranceEdge, ...]:
    _require_scope(principal, READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE)
    bounded_limit = _bounded_limit(limit)
    return tuple(
        session.scalars(
            select(RiskAssuranceEdge)
            .where(
                RiskAssuranceEdge.tenant_id == principal.tenant_id,
                RiskAssuranceEdge.stable_id == stable_id,
            )
            .order_by(RiskAssuranceEdge.revision.desc())
            .limit(bounded_limit)
        )
    )


def list_assurance_nodes(
    session: Session,
    principal: ApiPrincipal,
    *,
    kind: str | None = None,
    state: str | None = None,
    governed_object_ref: str | None = None,
    query: str | None = None,
    limit: int = 100,
) -> tuple[RiskAssuranceNode, ...]:
    _require_scope(principal, READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE)
    statement = select(RiskAssuranceNode).where(
        RiskAssuranceNode.tenant_id == principal.tenant_id,
        RiskAssuranceNode.superseded_at.is_(None),
    )
    if kind:
        _validate_kind(kind)
        statement = statement.where(RiskAssuranceNode.kind == kind)
    if state:
        statement = statement.where(RiskAssuranceNode.state == state)
    if governed_object_ref:
        statement = statement.where(
            RiskAssuranceNode.governed_object_ref == governed_object_ref
        )
    search = str(query or "").strip()
    if search:
        pattern = f"%{_escape_like(search)}%"
        statement = statement.where(
            or_(
                RiskAssuranceNode.label.ilike(pattern, escape="\\"),
                RiskAssuranceNode.description.ilike(pattern, escape="\\"),
                RiskAssuranceNode.stable_id.ilike(pattern, escape="\\"),
            )
        )
    return tuple(
        session.scalars(
            statement.order_by(
                RiskAssuranceNode.kind,
                RiskAssuranceNode.label,
                RiskAssuranceNode.stable_id,
            ).limit(_bounded_limit(limit))
        )
    )


def assurance_graph(
    session: Session,
    principal: ApiPrincipal,
    *,
    root_ref: str,
    max_depth: int = 4,
    limit: int = 200,
) -> AssuranceGraph:
    _require_scope(principal, READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE)
    if not 0 <= max_depth <= MAX_GRAPH_DEPTH:
        raise RiskAssuranceConflictError(
            f"Graph depth must be between 0 and {MAX_GRAPH_DEPTH}."
        )
    bounded_limit = _bounded_limit(limit)
    root = _current_node(session, principal.tenant_id, root_ref)
    if root is None:
        raise RiskAssuranceNotFoundError("Assurance graph root not found.")

    nodes: dict[str, RiskAssuranceNode] = {root.stable_id: root}
    edges: dict[str, RiskAssuranceEdge] = {}
    queue: deque[tuple[str, int]] = deque(((root.stable_id, 0),))
    expanded: set[str] = set()
    truncated = False
    while queue:
        node_ref, depth = queue.popleft()
        if node_ref in expanded or depth >= max_depth:
            continue
        expanded.add(node_ref)
        remaining = bounded_limit - len(edges)
        if remaining <= 0:
            truncated = True
            break
        connected = tuple(
            session.scalars(
                select(RiskAssuranceEdge)
                .where(
                    RiskAssuranceEdge.tenant_id == principal.tenant_id,
                    RiskAssuranceEdge.superseded_at.is_(None),
                    RiskAssuranceEdge.state == "active",
                    or_(
                        RiskAssuranceEdge.source_node_ref == node_ref,
                        RiskAssuranceEdge.target_node_ref == node_ref,
                    ),
                )
                .order_by(RiskAssuranceEdge.stable_id)
                .limit(remaining + 1)
            )
        )
        if len(connected) > remaining:
            connected = connected[:remaining]
            truncated = True
        neighbor_refs: set[str] = set()
        for edge in connected:
            edges[edge.stable_id] = edge
            neighbor_refs.update(
                (edge.source_node_ref, edge.target_node_ref)
            )
        missing = neighbor_refs - nodes.keys()
        if missing:
            for item in session.scalars(
                select(RiskAssuranceNode).where(
                    RiskAssuranceNode.tenant_id == principal.tenant_id,
                    RiskAssuranceNode.superseded_at.is_(None),
                    RiskAssuranceNode.stable_id.in_(missing),
                )
            ):
                nodes[item.stable_id] = item
        for neighbor in sorted(neighbor_refs):
            if neighbor not in expanded:
                queue.append((neighbor, depth + 1))
        if len(nodes) >= bounded_limit:
            truncated = truncated or bool(queue)
            break
    selected_nodes = tuple(
        sorted(nodes.values(), key=lambda item: (item.kind, item.stable_id))
    )[:bounded_limit]
    selected_refs = {item.stable_id for item in selected_nodes}
    selected_edges = tuple(
        edge
        for edge in sorted(edges.values(), key=lambda item: item.stable_id)
        if edge.source_node_ref in selected_refs
        and edge.target_node_ref in selected_refs
    )[:bounded_limit]
    return AssuranceGraph(
        root_ref=root_ref,
        nodes=selected_nodes,
        edges=selected_edges,
        truncated=truncated,
    )


def assurance_summary(
    session: Session,
    principal: ApiPrincipal,
) -> dict[str, object]:
    _require_scope(principal, READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE)
    rows = session.execute(
        select(
            RiskAssuranceNode.kind,
            RiskAssuranceNode.state,
            func.count(RiskAssuranceNode.id),
        )
        .where(
            RiskAssuranceNode.tenant_id == principal.tenant_id,
            RiskAssuranceNode.superseded_at.is_(None),
        )
        .group_by(RiskAssuranceNode.kind, RiskAssuranceNode.state)
    )
    by_kind: Counter[str] = Counter()
    by_state: Counter[str] = Counter()
    for kind, state, count in rows:
        by_kind[str(kind)] += int(count)
        by_state[str(state)] += int(count)
    return {
        "node_count": sum(by_kind.values()),
        "edge_count": int(
            session.scalar(
                select(func.count(RiskAssuranceEdge.id)).where(
                    RiskAssuranceEdge.tenant_id == principal.tenant_id,
                    RiskAssuranceEdge.superseded_at.is_(None),
                )
            )
            or 0
        ),
        "by_kind": dict(sorted(by_kind.items())),
        "by_state": dict(sorted(by_state.items())),
    }


def record_sanctions_assurance(
    session: Session,
    principal: ApiPrincipal,
    *,
    run: RiskScreeningRun,
) -> tuple[RiskAssuranceNode, ...]:
    """Project a completed sanctions run into the horizontal assurance graph."""
    if run.tenant_id != principal.tenant_id:
        raise RiskAssuranceAccessError("Screening assurance cannot cross tenants.")
    if run.status != "completed":
        raise RiskAssuranceConflictError(
            "Only completed screenings can become assurance evidence."
        )
    source = run.list_snapshot
    subject = run.subject_snapshot
    now = _as_aware(run.completed_at or utcnow())
    started_at = _as_aware(run.started_at)
    source_effective_at = _as_aware(
        source.effective_at or source.acquired_at
    )
    actor_id = _actor_id(principal)
    legal_ref = (
        f"sanctions-list:{source.jurisdiction}:{source.source_id}:"
        f"{source.source_version}"
    )
    evidence_ref = f"risk-compliance:sanctions-screening:{run.id}"
    owner_ref = "risk-compliance:sanctions"
    inputs = (
        AssuranceNodeInput(
            stable_id=_bounded_id(
                f"sanctions-obligation:{source.jurisdiction}:{source.list_type}"
            ),
            kind="obligation",
            label=f"{source.jurisdiction} sanctions screening obligation",
            state="active",
            owner_ref=owner_ref,
            valid_from=source_effective_at,
            provenance={
                "provider_id": source.provider_id,
                "source_version": source.source_version,
            },
            legal_basis_refs=(legal_ref,),
        ),
        AssuranceNodeInput(
            stable_id=f"sanctions-subject:{subject.fingerprint}",
            kind="governed_object",
            label="Screened party",
            state="active",
            owner_ref=owner_ref,
            valid_from=started_at,
            governed_object_ref=(
                subject.subject_ref
                or f"risk-compliance:screening-subject:{subject.id}"
            ),
            provenance={"subject_fingerprint": subject.fingerprint},
            classification="confidential",
        ),
        AssuranceNodeInput(
            stable_id=f"sanctions-risk:{subject.fingerprint}",
            kind="risk",
            label="Sanctions exposure",
            state=("identified" if run.outcome == "potential" else "mitigated"),
            owner_ref=owner_ref,
            valid_from=started_at,
            provenance={"screening_outcome": run.outcome},
            legal_basis_refs=(legal_ref,),
            classification="confidential",
        ),
        AssuranceNodeInput(
            stable_id=_bounded_id(f"sanctions-control:{source.provider_id}"),
            kind="control",
            label="Version-pinned sanctions screening",
            state=(
                "failed"
                if run.outcome in {"unavailable", "stale", "insufficient"}
                else "effective"
            ),
            owner_ref=owner_ref,
            valid_from=started_at,
            provenance={
                "matcher_version": run.matcher_version,
                "normalization_version": run.normalization_version,
                "policy_version": run.policy_version,
            },
            policy_refs=(f"sanctions-policy:{run.policy_version}",),
        ),
        AssuranceNodeInput(
            stable_id=f"sanctions-evidence:{run.id}",
            kind="evidence",
            label="Sanctions screening evidence",
            state=(
                "stale"
                if run.outcome == "stale"
                else "invalid"
                if run.outcome in {"unavailable", "insufficient"}
                else "current"
            ),
            owner_ref=owner_ref,
            valid_from=now,
            provenance={
                "list_snapshot_id": run.list_snapshot_id,
                "list_source_version": source.source_version,
                "request_hash": run.request_hash,
            },
            evidence_refs=(evidence_ref,),
            classification="confidential",
        ),
        AssuranceNodeInput(
            stable_id=f"sanctions-finding:{run.id}",
            kind="finding",
            label=(
                "Potential sanctions match"
                if run.outcome == "potential"
                else "Sanctions screening result"
            ),
            state="open" if run.outcome == "potential" else "resolved",
            owner_ref=owner_ref,
            valid_from=now,
            provenance={
                "outcome": run.outcome,
                "candidate_count": run.candidate_count,
            },
            legal_basis_refs=(legal_ref,),
            evidence_refs=(evidence_ref,),
            classification="confidential",
        ),
    )
    nodes = tuple(
        _ensure_node(
            session,
            tenant_id=principal.tenant_id,
            value=value,
            actor_id=actor_id,
        )
        for value in inputs
    )
    edge_inputs = (
        (nodes[0], "applies_to", nodes[1]),
        (nodes[1], "exposes_risk", nodes[2]),
        (nodes[2], "mitigated_by", nodes[3]),
        (nodes[3], "evidenced_by", nodes[4]),
        (nodes[4], "results_in", nodes[5]),
    )
    for source_node, relation, target_node in edge_inputs:
        _ensure_edge(
            session,
            tenant_id=principal.tenant_id,
            value=AssuranceEdgeInput(
                stable_id=_edge_id(
                    source_node.stable_id,
                    relation,
                    target_node.stable_id,
                ),
                source_node_ref=source_node.stable_id,
                target_node_ref=target_node.stable_id,
                relation=relation,
                owner_ref=owner_ref,
                valid_from=now,
                provenance={"screening_run_id": run.id},
                evidence_refs=(evidence_ref,),
            ),
            actor_id=actor_id,
        )
    session.flush()
    return nodes


def _insert_node(
    session: Session,
    *,
    tenant_id: str,
    value: AssuranceNodeInput,
    revision: int,
    previous: RiskAssuranceNode | None,
    actor_id: str | None,
) -> RiskAssuranceNode:
    recorded_at = utcnow()
    if previous is not None:
        previous.superseded_at = recorded_at
    item = RiskAssuranceNode(
        tenant_id=tenant_id,
        stable_id=value.stable_id,
        kind=value.kind,
        revision=revision,
        previous_revision_id=previous.id if previous is not None else None,
        label=value.label.strip(),
        description=_optional_text(value.description),
        state=value.state,
        owner_ref=value.owner_ref.strip(),
        scope_ref=_optional_text(value.scope_ref),
        governed_object_ref=_optional_text(value.governed_object_ref),
        valid_from=value.valid_from,
        valid_to=value.valid_to,
        recorded_at=recorded_at,
        provenance=dict(value.provenance),
        legal_basis_refs=list(value.legal_basis_refs),
        policy_refs=list(value.policy_refs),
        evidence_refs=list(value.evidence_refs),
        classification=value.classification.strip(),
        created_by=actor_id,
    )
    session.add(item)
    session.flush()
    return item


def _insert_edge(
    session: Session,
    *,
    tenant_id: str,
    value: AssuranceEdgeInput,
    revision: int,
    previous: RiskAssuranceEdge | None,
    actor_id: str | None,
) -> RiskAssuranceEdge:
    recorded_at = utcnow()
    if previous is not None:
        previous.superseded_at = recorded_at
    item = RiskAssuranceEdge(
        tenant_id=tenant_id,
        stable_id=value.stable_id,
        revision=revision,
        previous_revision_id=previous.id if previous is not None else None,
        source_node_ref=value.source_node_ref,
        target_node_ref=value.target_node_ref,
        relation=value.relation,
        state=value.state,
        owner_ref=value.owner_ref.strip(),
        scope_ref=_optional_text(value.scope_ref),
        valid_from=value.valid_from,
        valid_to=value.valid_to,
        recorded_at=recorded_at,
        provenance=dict(value.provenance),
        legal_basis_refs=list(value.legal_basis_refs),
        policy_refs=list(value.policy_refs),
        evidence_refs=list(value.evidence_refs),
        created_by=actor_id,
    )
    session.add(item)
    session.flush()
    return item


def _ensure_node(
    session: Session,
    *,
    tenant_id: str,
    value: AssuranceNodeInput,
    actor_id: str | None,
) -> RiskAssuranceNode:
    current = _locked_current_node(session, tenant_id, value.stable_id)
    if current is None:
        return _insert_node(
            session,
            tenant_id=tenant_id,
            value=value,
            revision=1,
            previous=None,
            actor_id=actor_id,
        )
    if _node_matches(current, value):
        return current
    if current.kind != value.kind:
        raise RiskAssuranceConflictError(
            f"Assurance node {value.stable_id!r} changed kind."
        )
    return _insert_node(
        session,
        tenant_id=tenant_id,
        value=value,
        revision=current.revision + 1,
        previous=current,
        actor_id=actor_id,
    )


def _ensure_edge(
    session: Session,
    *,
    tenant_id: str,
    value: AssuranceEdgeInput,
    actor_id: str | None,
) -> RiskAssuranceEdge:
    current = _locked_current_edge(session, tenant_id, value.stable_id)
    if current is None:
        _validate_edge_endpoints(session, tenant_id, value)
        return _insert_edge(
            session,
            tenant_id=tenant_id,
            value=value,
            revision=1,
            previous=None,
            actor_id=actor_id,
        )
    if _edge_matches(current, value):
        return current
    identity = (
        value.source_node_ref,
        value.target_node_ref,
        value.relation,
    )
    if identity != (
        current.source_node_ref,
        current.target_node_ref,
        current.relation,
    ):
        raise RiskAssuranceConflictError(
            f"Assurance edge {value.stable_id!r} changed identity."
        )
    return _insert_edge(
        session,
        tenant_id=tenant_id,
        value=value,
        revision=current.revision + 1,
        previous=current,
        actor_id=actor_id,
    )


def _current_node(
    session: Session,
    tenant_id: str,
    stable_id: str,
) -> RiskAssuranceNode | None:
    return session.scalar(
        select(RiskAssuranceNode).where(
            RiskAssuranceNode.tenant_id == tenant_id,
            RiskAssuranceNode.stable_id == stable_id,
            RiskAssuranceNode.superseded_at.is_(None),
        )
    )


def _locked_current_node(
    session: Session,
    tenant_id: str,
    stable_id: str,
) -> RiskAssuranceNode | None:
    return session.scalar(
        select(RiskAssuranceNode)
        .where(
            RiskAssuranceNode.tenant_id == tenant_id,
            RiskAssuranceNode.stable_id == stable_id,
            RiskAssuranceNode.superseded_at.is_(None),
        )
        .with_for_update()
    )


def _current_edge(
    session: Session,
    tenant_id: str,
    stable_id: str,
) -> RiskAssuranceEdge | None:
    return session.scalar(
        select(RiskAssuranceEdge).where(
            RiskAssuranceEdge.tenant_id == tenant_id,
            RiskAssuranceEdge.stable_id == stable_id,
            RiskAssuranceEdge.superseded_at.is_(None),
        )
    )


def _locked_current_edge(
    session: Session,
    tenant_id: str,
    stable_id: str,
) -> RiskAssuranceEdge | None:
    return session.scalar(
        select(RiskAssuranceEdge)
        .where(
            RiskAssuranceEdge.tenant_id == tenant_id,
            RiskAssuranceEdge.stable_id == stable_id,
            RiskAssuranceEdge.superseded_at.is_(None),
        )
        .with_for_update()
    )


def _validate_edge_endpoints(
    session: Session,
    tenant_id: str,
    value: AssuranceEdgeInput,
) -> None:
    source = _current_node(session, tenant_id, value.source_node_ref)
    target = _current_node(session, tenant_id, value.target_node_ref)
    if source is None or target is None:
        raise RiskAssuranceNotFoundError(
            "Both assurance edge endpoints must exist in the current tenant."
        )
    expected = ASSURANCE_EDGE_RELATIONS[value.relation]
    if (source.kind, target.kind) != expected:
        raise RiskAssuranceConflictError(
            f"Relation {value.relation!r} requires {expected[0]} -> {expected[1]}."
        )


def _validate_node_input(value: AssuranceNodeInput) -> None:
    _validate_id(value.stable_id, "Assurance node stable id")
    _validate_kind(value.kind)
    _require_text(value.label, "Assurance node label", 500)
    _require_text(value.owner_ref, "Assurance node owner", 500)
    _validate_optional_text(value.description, "Assurance node description", 20_000)
    _validate_optional_text(value.scope_ref, "Assurance node scope", 500)
    _validate_optional_text(
        value.governed_object_ref,
        "Assurance governed object reference",
        1_000,
    )
    if value.state not in ASSURANCE_NODE_STATES[value.kind]:
        raise RiskAssuranceError(
            f"Unsupported {value.kind} state: {value.state!r}."
        )
    if value.kind == "governed_object" and not _optional_text(
        value.governed_object_ref
    ):
        raise RiskAssuranceError(
            "Governed-object nodes require an opaque governed object reference."
        )
    _validate_period(value.valid_from, value.valid_to)
    _validate_reference_values(value.legal_basis_refs, "legal basis")
    _validate_reference_values(value.policy_refs, "policy")
    _validate_reference_values(value.evidence_refs, "evidence")
    _validate_provenance(value.provenance)
    _require_text(value.classification, "Assurance classification", 50)


def _validate_edge_input(value: AssuranceEdgeInput) -> None:
    _validate_id(value.stable_id, "Assurance edge stable id")
    _validate_id(value.source_node_ref, "Assurance edge source")
    _validate_id(value.target_node_ref, "Assurance edge target")
    if value.source_node_ref == value.target_node_ref:
        raise RiskAssuranceError("Assurance edges cannot reference themselves.")
    if value.relation not in ASSURANCE_EDGE_RELATIONS:
        raise RiskAssuranceError(
            f"Unsupported assurance relation: {value.relation!r}."
        )
    if value.state not in ASSURANCE_EDGE_STATES:
        raise RiskAssuranceError(
            f"Unsupported assurance edge state: {value.state!r}."
        )
    _require_text(value.owner_ref, "Assurance edge owner", 500)
    _validate_optional_text(value.scope_ref, "Assurance edge scope", 500)
    _validate_period(value.valid_from, value.valid_to)
    _validate_reference_values(value.legal_basis_refs, "legal basis")
    _validate_reference_values(value.policy_refs, "policy")
    _validate_reference_values(value.evidence_refs, "evidence")
    _validate_provenance(value.provenance)


def _validate_kind(value: str) -> None:
    if value not in ASSURANCE_NODE_STATES:
        raise RiskAssuranceError(f"Unsupported assurance node kind: {value!r}.")


def _validate_period(valid_from: datetime, valid_to: datetime | None) -> None:
    for value in (valid_from, valid_to):
        if value is not None and (
            value.tzinfo is None or value.utcoffset() is None
        ):
            raise RiskAssuranceError(
                "Assurance effective times must include a timezone."
            )
    if valid_to is not None and valid_to <= valid_from:
        raise RiskAssuranceError(
            "Assurance valid_to must be later than valid_from."
        )


def _validate_reference_values(values: tuple[str, ...], label: str) -> None:
    if len(values) > 100:
        raise RiskAssuranceError(
            f"Assurance {label} references are limited to 100 values."
        )
    if len(values) != len(set(values)):
        raise RiskAssuranceError(
            f"Assurance {label} references cannot contain duplicates."
        )
    for value in values:
        _require_text(value, f"Assurance {label} reference", 1000)


def _node_matches(item: RiskAssuranceNode, value: AssuranceNodeInput) -> bool:
    return (
        item.kind,
        item.label,
        item.description,
        item.state,
        item.owner_ref,
        item.scope_ref,
        item.governed_object_ref,
        _time_key(item.valid_from),
        _time_key(item.valid_to),
        dict(item.provenance),
        tuple(item.legal_basis_refs),
        tuple(item.policy_refs),
        tuple(item.evidence_refs),
        item.classification,
    ) == (
        value.kind,
        value.label.strip(),
        _optional_text(value.description),
        value.state,
        value.owner_ref.strip(),
        _optional_text(value.scope_ref),
        _optional_text(value.governed_object_ref),
        _time_key(value.valid_from),
        _time_key(value.valid_to),
        dict(value.provenance),
        value.legal_basis_refs,
        value.policy_refs,
        value.evidence_refs,
        value.classification.strip(),
    )


def _edge_matches(item: RiskAssuranceEdge, value: AssuranceEdgeInput) -> bool:
    return (
        item.source_node_ref,
        item.target_node_ref,
        item.relation,
        item.state,
        item.owner_ref,
        item.scope_ref,
        _time_key(item.valid_from),
        _time_key(item.valid_to),
        dict(item.provenance),
        tuple(item.legal_basis_refs),
        tuple(item.policy_refs),
        tuple(item.evidence_refs),
    ) == (
        value.source_node_ref,
        value.target_node_ref,
        value.relation,
        value.state,
        value.owner_ref.strip(),
        _optional_text(value.scope_ref),
        _time_key(value.valid_from),
        _time_key(value.valid_to),
        dict(value.provenance),
        value.legal_basis_refs,
        value.policy_refs,
        value.evidence_refs,
    )


def _edge_id(source_ref: str, relation: str, target_ref: str) -> str:
    digest = hashlib.sha256(
        f"{source_ref}\0{relation}\0{target_ref}".encode("utf-8")
    ).hexdigest()[:32]
    return f"assurance-edge:{digest}"


def _as_aware(value: datetime) -> datetime:
    return value.replace(tzinfo=UTC) if value.tzinfo is None else value


def _time_key(value: datetime | None) -> datetime | None:
    return _as_aware(value).astimezone(UTC) if value is not None else None


def _bounded_id(value: str) -> str:
    if len(value) <= 255 and _ID_PATTERN.fullmatch(value):
        return value
    digest = hashlib.sha256(value.encode("utf-8")).hexdigest()[:32]
    prefix = re.sub(r"[^A-Za-z0-9_.:/-]", "-", value[:200]).rstrip("-")
    return f"{prefix}:{digest}"[:255]


def _validate_id(value: str, label: str) -> None:
    if not _ID_PATTERN.fullmatch(str(value or "")):
        raise RiskAssuranceError(f"{label} is invalid.")


def _require_text(value: str, label: str, limit: int) -> None:
    cleaned = str(value or "").strip()
    if not cleaned:
        raise RiskAssuranceError(f"{label} is required.")
    if len(cleaned) > limit:
        raise RiskAssuranceError(f"{label} is limited to {limit} characters.")


def _optional_text(value: str | None) -> str | None:
    cleaned = str(value or "").strip()
    return cleaned or None


def _validate_optional_text(value: str | None, label: str, limit: int) -> None:
    cleaned = _optional_text(value)
    if cleaned is not None and len(cleaned) > limit:
        raise RiskAssuranceError(f"{label} is limited to {limit} characters.")


def _validate_provenance(value: Mapping[str, Any]) -> None:
    try:
        encoded = json.dumps(
            dict(value),
            ensure_ascii=True,
            separators=(",", ":"),
            sort_keys=True,
        ).encode("utf-8")
    except (TypeError, ValueError) as exc:
        raise RiskAssuranceError(
            "Assurance provenance must contain JSON-compatible values."
        ) from exc
    if len(encoded) > MAX_PROVENANCE_BYTES:
        raise RiskAssuranceError(
            f"Assurance provenance is limited to {MAX_PROVENANCE_BYTES} bytes."
        )


def _bounded_limit(value: int) -> int:
    if not 1 <= value <= MAX_GRAPH_ITEMS:
        raise RiskAssuranceConflictError(
            f"Assurance result limits must be between 1 and {MAX_GRAPH_ITEMS}."
        )
    return value


def _escape_like(value: str) -> str:
    return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")


def _require_scope(principal: ApiPrincipal, *scopes: str) -> None:
    if not any(principal.has(scope) for scope in scopes):
        raise RiskAssuranceAccessError("Risk assurance access is not permitted.")


def _actor_id(principal: ApiPrincipal) -> str | None:
    return getattr(principal.user, "id", None) or getattr(
        principal.account,
        "id",
        None,
    )


__all__ = [
    "ASSURANCE_EDGE_RELATIONS",
    "ASSURANCE_NODE_KINDS",
    "AssuranceEdgeInput",
    "AssuranceGraph",
    "AssuranceNodeInput",
    "RiskAssuranceAccessError",
    "RiskAssuranceConflictError",
    "RiskAssuranceError",
    "RiskAssuranceNotFoundError",
    "assurance_graph",
    "assurance_summary",
    "create_assurance_edge",
    "create_assurance_node",
    "get_assurance_node",
    "list_assurance_edge_history",
    "list_assurance_node_history",
    "list_assurance_nodes",
    "record_sanctions_assurance",
    "revise_assurance_edge",
    "revise_assurance_node",
]