"""Canonical Document Reconciliation v2 domain service.

All relationship mutations and their audit event share one database
transaction.  Callers provide the authenticated actor; HTTP handlers must
derive it from ``request.state`` and never from submitted form data.
"""
from __future__ import annotations

import json
import hashlib
import uuid
from contextlib import nullcontext
from typing import Any, Dict, List, Optional

from sqlalchemy import text
from sqlalchemy.exc import IntegrityError

from app.db import engine

RELATIONSHIPS = frozenset({
    "PRIMARY", "SECONDARY", "HISTORICAL", "IGNORED", "REMOVED",
    "REASSIGNED", "REVIEW_REQUIRED",
})
NON_PROMOTABLE_STATUSES = frozenset({"CANCELLED", "CANCELED"})


class DocumentReconciliationError(ValueError):
    pass


class DocumentConflictError(DocumentReconciliationError):
    pass


def _json(value: Any) -> str:
    return json.dumps(value or {}, ensure_ascii=False, default=str)


def _row(row: Any) -> Optional[Dict[str, Any]]:
    return dict(row) if row else None


def _actor(value: str) -> str:
    value = str(value or "").strip()
    if not value:
        raise DocumentReconciliationError("authenticated actor is required")
    return value


def document_reconciliation_v2_available(conn: Any = None) -> bool:
    """Return whether every v2 object required by readers is available.

    ``to_regclass`` is safe before migration 007 and avoids making application
    startup depend on the migration having already run.
    """
    def _check(connection: Any) -> bool:
        return bool(connection.execute(text("""
            SELECT to_regclass('public.opportunity_document_links') IS NOT NULL
               AND to_regclass('public.opportunity_document_link_events') IS NOT NULL
        """)).scalar())
    try:
        if conn is not None:
            return _check(conn)
        with engine.begin() as connection:
            return _check(connection)
    except Exception:
        return False


def _legacy_relationship(row: Dict[str, Any]) -> str:
    role = str(row.get("role") or "current").lower()
    if row.get("is_active") is False or role == "detached":
        return "REMOVED"
    if bool(row.get("is_primary")) and role in {"current", "accepted"}:
        return "PRIMARY"
    if role in {"historical", "history", "superseded"}:
        return "HISTORICAL"
    return "SECONDARY"


def resolve_document_links(opportunity_id: str, *, include_ended: bool = False,
                           conn: Any = None) -> List[Dict[str, Any]]:
    """Resolve complete groups from v2, falling back group-wise to legacy.

    A group becomes v2-authoritative only after every legacy document in that
    ``(opportunity_id, document_kind)`` group has a current v2 link.  This is
    the sole controlled legacy read used during phased rollout.
    """
    def _resolve(connection: Any) -> List[Dict[str, Any]]:
        legacy = [dict(r) for r in connection.execute(text("""
            SELECT d.id::text, d.customer_id::text, d.opportunity_id::text,
                   d.document_kind, d.external_id, d.document_number, d.status AS jasmin_status,
                   d.status, d.total_amount, d.amount, d.tax_amount, d.currency, d.document_date,
                   d.due_date, d.external_url, d.company, d.document_type, d.serie, d.series_number,
                   d.customer_party_key, d.parent_document_id::text, d.version_number, d.payload,
                   d.system, d.role, d.is_primary, d.is_active, d.created_at, d.updated_at
            FROM commercial_documents d
            WHERE d.opportunity_id=CAST(:opportunity_id AS UUID)
            ORDER BY d.document_kind, d.created_at, d.id
        """), {"opportunity_id": opportunity_id}).mappings().all()]
        if not document_reconciliation_v2_available(connection):
            return [row | {"document_id": row["id"], "link_id": None,
                    "relationship": _legacy_relationship(row),
                    "resolution_source": "legacy_rollout"} for row in legacy]
        v2 = [dict(r) for r in connection.execute(text(_LINK_SELECT + """
            WHERE l.opportunity_id=CAST(:opportunity_id AS UUID)
              AND (:include_ended OR l.ended_at IS NULL)
            ORDER BY l.document_kind, l.updated_at DESC
        """), {"opportunity_id": opportunity_id, "include_ended": include_ended}).mappings().all()]
        legacy_groups: Dict[str, List[Dict[str, Any]]] = {}
        v2_groups: Dict[str, List[Dict[str, Any]]] = {}
        for row in legacy:
            legacy_groups.setdefault(str(row.get("document_kind") or ""), []).append(row)
        for row in v2:
            v2_groups.setdefault(str(row.get("document_kind") or ""), []).append(row)
        resolved: List[Dict[str, Any]] = []
        for kind in sorted(set(legacy_groups) | set(v2_groups)):
            legacy_group, v2_group = legacy_groups.get(kind, []), v2_groups.get(kind, [])
            current_ids = {str(row["document_id"]) for row in v2_group if not row.get("ended_at")}
            legacy_ids = {str(row["id"]) for row in legacy_group}
            if legacy_ids and legacy_ids.issubset(current_ids):
                resolved.extend(row | {"resolution_source": "v2"} for row in v2_group)
            else:
                resolved.extend(row | {"document_id": row["id"], "link_id": None,
                    "relationship": _legacy_relationship(row), "resolution_source": "legacy_rollout"}
                    for row in legacy_group)
        return resolved
    if conn is not None:
        return _resolve(conn)
    with engine.begin() as connection:
        return _resolve(connection)


def normalize_jasmin_status(value: Any) -> str:
    value = str(value or "").strip().upper().replace(" ", "_")
    aliases = {"CANCELED": "CANCELLED", "ANULADO": "CANCELLED", "FECHADO": "CLOSED", "ABERTO": "OPEN"}
    return aliases.get(value, value or "UNKNOWN")


def select_valid_primary(rows: List[Dict[str, Any]], document_kind: Optional[str] = None) -> Optional[Dict[str, Any]]:
    """Never infer a primary from ordering or from a non-primary relationship."""
    for row in rows:
        if document_kind and row.get("document_kind") != document_kind:
            continue
        if row.get("relationship") == "PRIMARY" and normalize_jasmin_status(
                row.get("jasmin_status") or row.get("status")) not in NON_PROMOTABLE_STATUSES:
            return row
    return None


_LINK_SELECT = """
SELECT d.id::text AS id, d.id::text AS document_id, l.id::text AS link_id,
       l.opportunity_id::text, l.document_kind,
       l.relationship, l.is_manual, l.decision_reason, l.decided_by, l.decided_at,
       l.source, l.origin_opportunity_id::text, l.destination_opportunity_id::text,
       l.correlation_id, l.request_id, l.restored_from_link_id::text,
       l.created_at AS link_created_at, l.updated_at AS link_updated_at,
       l.ended_at, l.metadata, l.version,
       d.customer_id::text, d.company, d.document_type, d.serie, d.series_number,
       d.document_number, d.version_number, d.external_id, d.external_url,
       d.role, d.is_primary, d.is_active, d.status AS jasmin_status, d.status,
       d.amount, d.tax_amount, d.total_amount, d.currency, d.document_date,
       d.due_date, d.payload, d.system, d.customer_party_key,
       d.parent_document_id::text, d.created_at, d.updated_at
FROM opportunity_document_links l
JOIN commercial_documents d ON d.id = l.document_id
"""


def list_document_links(opportunity_id: str, *, include_ended: bool = False) -> List[Dict[str, Any]]:
    rows = resolve_document_links(opportunity_id, include_ended=include_ended)
    order = {"REVIEW_REQUIRED": 0, "PRIMARY": 1, "SECONDARY": 2, "HISTORICAL": 3}
    return sorted(rows, key=lambda row: (order.get(str(row.get("relationship")), 4),
                                         str(row.get("updated_at") or "")), reverse=False)


def prepare_effective_document_links(conn: Any) -> str:
    """Materialize rollout-safe effective links for set-oriented consumers.

    Forecast/operations queries need a relational input rather than N Python
    result sets.  The temporary table is session-local and is populated solely
    through the group-aware resolver.
    """
    opportunity_ids = {str(row[0]) for row in conn.execute(text("""
        SELECT DISTINCT opportunity_id::text FROM commercial_documents
        WHERE opportunity_id IS NOT NULL
    """)).all()}
    if document_reconciliation_v2_available(conn):
        opportunity_ids.update(str(row[0]) for row in conn.execute(text("""
            SELECT DISTINCT opportunity_id::text FROM opportunity_document_links
            WHERE ended_at IS NULL
        """)).all())
    conn.execute(text("""CREATE TEMP TABLE IF NOT EXISTS _effective_document_links (
        opportunity_id UUID NOT NULL, document_id UUID NOT NULL, document_kind TEXT NOT NULL,
        relationship TEXT NOT NULL, ended_at TIMESTAMPTZ) ON COMMIT DROP"""))
    conn.execute(text("TRUNCATE _effective_document_links"))
    for opportunity_id in sorted(opportunity_ids):
        for row in resolve_document_links(opportunity_id, conn=conn):
            conn.execute(text("""INSERT INTO _effective_document_links
                (opportunity_id,document_id,document_kind,relationship,ended_at)
                VALUES(CAST(:oid AS UUID),CAST(:did AS UUID),:kind,:relationship,:ended_at)"""),
                {"oid": opportunity_id, "did": row["document_id"],
                 "kind": row.get("document_kind") or "unknown",
                 "relationship": row.get("relationship") or "SECONDARY",
                 "ended_at": row.get("ended_at")})
    return "_effective_document_links"


def get_document_link(opportunity_id: str, document_id: str, *, include_ended: bool = False) -> Optional[Dict[str, Any]]:
    with engine.begin() as conn:
        row = conn.execute(text(_LINK_SELECT + """
            WHERE l.opportunity_id = CAST(:opportunity_id AS UUID)
              AND l.document_id = CAST(:document_id AS UUID)
              AND (:include_ended OR l.ended_at IS NULL)
            ORDER BY l.ended_at NULLS FIRST, l.updated_at DESC LIMIT 1
        """), {"opportunity_id": opportunity_id, "document_id": document_id, "include_ended": include_ended}).mappings().first()
    return _row(row)


def resolve_document_opportunity(document_id: str) -> Optional[str]:
    """Resolve ownership server-side; form-submitted opportunity ids are untrusted."""
    with engine.begin() as conn:
        if document_reconciliation_v2_available(conn):
            owner = conn.execute(text("""SELECT opportunity_id::text FROM opportunity_document_links
                WHERE document_id=CAST(:did AS UUID) AND ended_at IS NULL
                ORDER BY updated_at DESC LIMIT 1"""), {"did": document_id}).scalar()
            if owner:
                return str(owner)
        # Controlled rollout compatibility only.
        owner = conn.execute(text("SELECT opportunity_id::text FROM commercial_documents WHERE id=CAST(:did AS UUID)"),
                             {"did": document_id}).scalar()
        return str(owner) if owner else None


def get_primary_document(opportunity_id: str, document_kind: str) -> Optional[Dict[str, Any]]:
    return select_valid_primary(resolve_document_links(opportunity_id), document_kind)


def list_primary_document_lines(opportunity_id: str, document_kind: str) -> List[Dict[str, Any]]:
    with engine.begin() as conn:
        primary = select_valid_primary(resolve_document_links(opportunity_id, conn=conn), document_kind)
        if not primary:
            return []
        has_canonical = bool(conn.execute(text("""SELECT EXISTS (SELECT 1 FROM information_schema.columns
            WHERE table_schema='public' AND table_name='commercial_document_lines'
              AND column_name='commercial_document_id')""")).scalar())
        id_column = "commercial_document_id" if has_canonical else "document_id"
        rows = conn.execute(text(f"""SELECT dl.*, dl.id::text, dl.document_id::text
            FROM commercial_document_lines dl
            WHERE dl.{id_column}=CAST(:document_id AS UUID)
            ORDER BY dl.line_index, dl.created_at"""),
            {"document_id": primary["document_id"]}).mappings().all()
    return [dict(r) for r in rows]


def list_effective_document_lines(opportunity_id: str) -> List[Dict[str, Any]]:
    """Document lines only; manual/Odoo operational lines remain separate."""
    with engine.begin() as conn:
        links = [row for row in resolve_document_links(opportunity_id, conn=conn)
                 if row.get("relationship") == "PRIMARY"
                 and normalize_jasmin_status(row.get("jasmin_status") or row.get("status")) not in NON_PROMOTABLE_STATUSES]
        if not links:
            return []
        has_canonical = bool(conn.execute(text("""SELECT EXISTS (SELECT 1 FROM information_schema.columns
            WHERE table_schema='public' AND table_name='commercial_document_lines'
              AND column_name='commercial_document_id')""")).scalar())
        id_column = "commercial_document_id" if has_canonical else "document_id"
        ids = [str(row["document_id"]) for row in links]
        kinds = {str(row["document_id"]): row.get("document_kind") for row in links}
        rows = conn.execute(text(f"""SELECT dl.*, dl.id::text, dl.document_id::text
            FROM commercial_document_lines dl WHERE dl.{id_column}=ANY(CAST(:ids AS UUID[]))
            ORDER BY dl.line_index, dl.created_at"""), {"ids": ids}).mappings().all()
    return [dict(r) | {"document_kind": kinds.get(str(r[id_column]))} for r in rows]


def _event(conn: Any, *, link_id: Optional[str], opportunity_id: str, document_id: str,
           event_type: str, actor: str, reason: Optional[str], old_relationship: Optional[str],
           new_relationship: Optional[str], correlation_id: Optional[str], request_id: Optional[str],
           idempotency_key: Optional[str], old_opportunity_id: Optional[str] = None,
           new_opportunity_id: Optional[str] = None, old_jasmin_status: Optional[str] = None,
           new_jasmin_status: Optional[str] = None, payload: Optional[Dict[str, Any]] = None) -> None:
    conn.execute(text("""
      INSERT INTO opportunity_document_link_events (
        link_id, opportunity_id, document_id, event_type, actor, reason,
        old_relationship, new_relationship, old_opportunity_id, new_opportunity_id,
        old_jasmin_status, new_jasmin_status, correlation_id, request_id,
        idempotency_key, payload)
      VALUES (CAST(:link_id AS UUID), CAST(:opportunity_id AS UUID), CAST(:document_id AS UUID),
        :event_type, :actor, :reason, :old_relationship, :new_relationship,
        CAST(:old_opportunity_id AS UUID), CAST(:new_opportunity_id AS UUID),
        :old_jasmin_status, :new_jasmin_status, :correlation_id, :request_id,
        :idempotency_key, CAST(:payload AS JSONB))
      ON CONFLICT (idempotency_key) WHERE idempotency_key IS NOT NULL DO NOTHING
    """), locals() | {"payload": _json(payload)})


def _lock_context(conn: Any, opportunity_id: str, document_id: str) -> Dict[str, Any]:
    # Opportunity row serializes competing primary/reassignment decisions.
    if not conn.execute(text("SELECT 1 FROM opportunities WHERE id=CAST(:id AS UUID) FOR UPDATE"), {"id": opportunity_id}).scalar():
        raise DocumentReconciliationError("opportunity not found")
    doc = conn.execute(text("""
        SELECT id::text, document_kind, status, opportunity_id::text
        FROM commercial_documents WHERE id=CAST(:id AS UUID) FOR UPDATE
    """), {"id": document_id}).mappings().first()
    if not doc:
        raise DocumentReconciliationError("document not found")
    return dict(doc)


def _dual_write(conn: Any, document_id: str, opportunity_id: str, relationship: str) -> None:
    # Rollout compatibility only: v2 remains authoritative after group completion.
    mapping = {
        "PRIMARY": ("current", True, True), "SECONDARY": ("related", False, True),
        "HISTORICAL": ("historical", False, False), "IGNORED": ("detached", False, False),
        "REMOVED": ("detached", False, False), "REVIEW_REQUIRED": ("review_required", False, True),
        "REASSIGNED": ("detached", False, False),
    }
    role, primary, active = mapping[relationship]
    conn.execute(text("""UPDATE commercial_documents SET opportunity_id=CAST(:opportunity_id AS UUID),
        role=:role, is_primary=:primary, is_active=:active, updated_at=now()
        WHERE id=CAST(:document_id AS UUID)"""), {"opportunity_id": opportunity_id, "document_id": document_id,
        "role": role, "primary": primary, "active": active})


def _command_fingerprint(*, opportunity_id: str, document_id: str, target_relationship: str,
                         origin_opportunity_id: Optional[str] = None,
                         destination_opportunity_id: Optional[str] = None,
                         reason: Optional[str] = None, is_manual: Optional[bool] = None,
                         source: Optional[str] = None, event_type: Optional[str] = None,
                         metadata: Optional[Dict[str, Any]] = None, actor: Optional[str] = None,
                         document_kind: Optional[str] = None,
                         correlation_id: Optional[str] = None,
                         request_id: Optional[str] = None) -> tuple[str, Dict[str, Any]]:
    command = {"opportunity_id": str(opportunity_id), "document_id": str(document_id),
        "target_relationship": str(target_relationship),
        "origin_opportunity_id": str(origin_opportunity_id or ""),
        "destination_opportunity_id": str(destination_opportunity_id or ""),
        "reason": str(reason or "").strip(), "is_manual": is_manual,
        "source": str(source or ""), "event_type": str(event_type or ""),
        "metadata": metadata or {}, "actor": str(actor or ""),
        "document_kind": str(document_kind or ""),
        "correlation_id": str(correlation_id or ""),
        "request_id": str(request_id or "")}
    canonical = json.dumps(command, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest(), command


def _claim_idempotency(conn: Any, key: Optional[str], fingerprint: str,
                       command: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    if not key:
        return None
    inserted = conn.execute(text("""
        INSERT INTO document_reconciliation_commands(idempotency_key, fingerprint, command_payload)
        VALUES (:key, :fingerprint, CAST(:command AS JSONB))
        ON CONFLICT (idempotency_key) DO NOTHING
        RETURNING idempotency_key
    """), {"key": key, "fingerprint": fingerprint, "command": _json(command)}).scalar()
    if inserted:
        return None
    existing = conn.execute(text("""SELECT fingerprint, result_payload
        FROM document_reconciliation_commands WHERE idempotency_key=:key FOR UPDATE
    """), {"key": key}).mappings().first()
    if not existing or existing["fingerprint"] != fingerprint:
        raise DocumentConflictError("idempotency key was already used with a different command")
    result = existing.get("result_payload")
    return dict(result) if isinstance(result, dict) else {}


def _complete_idempotency(conn: Any, key: Optional[str], result: Dict[str, Any]) -> None:
    if key:
        conn.execute(text("""UPDATE document_reconciliation_commands
            SET result_payload=CAST(:result AS JSONB), completed_at=now()
            WHERE idempotency_key=:key"""), {"key": key, "result": _json(result)})


def _validate_membership(conn: Any, opportunity_id: str, document_id: str,
                         *, allow_legacy: bool = True) -> Optional[Dict[str, Any]]:
    current = conn.execute(text("""SELECT * FROM opportunity_document_links
        WHERE opportunity_id=CAST(:oid AS UUID) AND document_id=CAST(:did AS UUID)
          AND ended_at IS NULL FOR UPDATE"""), {"oid": opportunity_id, "did": document_id}).mappings().first()
    historical = conn.execute(text("""SELECT 1 FROM opportunity_document_links
        WHERE opportunity_id=CAST(:oid AS UUID) AND document_id=CAST(:did AS UUID) LIMIT 1"""),
        {"oid": opportunity_id, "did": document_id}).scalar()
    legacy = conn.execute(text("""SELECT opportunity_id::text FROM commercial_documents
        WHERE id=CAST(:did AS UUID)"""), {"did": document_id}).scalar()
    if current:
        return dict(current)
    if historical:
        return None
    if allow_legacy and str(legacy or "") == str(opportunity_id):
        return None
    raise DocumentReconciliationError("document does not belong to this opportunity")


def set_document_relationship(opportunity_id: str, document_id: str, relationship: str, *, actor: str,
        reason: Optional[str] = None, is_manual: bool = True, source: str = "domain",
        correlation_id: Optional[str] = None, request_id: Optional[str] = None,
        idempotency_key: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None,
        event_type: str = "RELATIONSHIP_CHANGED", _conn: Any = None) -> Dict[str, Any]:
    relationship = str(relationship or "").upper()
    if relationship not in RELATIONSHIPS or relationship == "REASSIGNED":
        raise DocumentReconciliationError("unsupported direct relationship")
    actor = _actor(actor)
    if is_manual and not str(reason or "").strip():
        raise DocumentReconciliationError("reason is required for every manual relationship change")
    if not is_manual and not str(reason or "").strip():
        reason = f"automatic {source} classification to {relationship}"
    fingerprint, command = _command_fingerprint(opportunity_id=opportunity_id,
        document_id=document_id, target_relationship=relationship, reason=reason,
        is_manual=is_manual, source=source, event_type=event_type, metadata=metadata,
        actor=actor, correlation_id=correlation_id, request_id=request_id)
    try:
        with (nullcontext(_conn) if _conn is not None else engine.begin()) as conn:
            replay = _claim_idempotency(conn, idempotency_key, fingerprint, command)
            if replay is not None:
                return replay
            doc = _lock_context(conn, opportunity_id, document_id)
            existing = _validate_membership(conn, opportunity_id, document_id)
            old = existing.get("relationship") if existing else None
            if old == relationship:
                result = dict(existing)
                _complete_idempotency(conn, idempotency_key, result)
                return result
            if relationship == "PRIMARY":
                if old in {"IGNORED", "REMOVED"} or normalize_jasmin_status(doc.get("status")) in NON_PROMOTABLE_STATUSES:
                    raise DocumentReconciliationError("ignored, removed or cancelled documents cannot be promoted")
                previous = conn.execute(text("""SELECT * FROM opportunity_document_links
                  WHERE opportunity_id=CAST(:opportunity_id AS UUID) AND document_kind=:kind
                  AND relationship='PRIMARY' AND ended_at IS NULL FOR UPDATE"""),
                  {"opportunity_id": opportunity_id, "kind": doc["document_kind"]}).mappings().first()
                if previous and str(previous["document_id"]) != document_id:
                    conn.execute(text("""UPDATE opportunity_document_links SET relationship='HISTORICAL',
                      decision_reason=:reason, decided_by=:actor, decided_at=now(), updated_at=now(), version=version+1
                      WHERE id=:id"""), {"reason": reason or "superseded primary", "actor": actor, "id": previous["id"]})
                    _dual_write(conn, str(previous["document_id"]), opportunity_id, "HISTORICAL")
                    _event(conn, link_id=str(previous["id"]), opportunity_id=opportunity_id,
                      document_id=str(previous["document_id"]), event_type="PRIMARY_DEMOTED", actor=actor,
                      reason=reason, old_relationship="PRIMARY", new_relationship="HISTORICAL",
                      correlation_id=correlation_id, request_id=request_id,
                      idempotency_key=(idempotency_key + ":demote") if idempotency_key else None)
            params = {"opportunity_id": opportunity_id, "document_id": document_id,
                "kind": doc["document_kind"], "relationship": relationship, "manual": is_manual,
                "reason": reason, "actor": actor, "source": source, "correlation_id": correlation_id,
                "request_id": request_id, "metadata": _json(metadata)}
            if existing:
                link = conn.execute(text("""UPDATE opportunity_document_links SET relationship=:relationship,
                  is_manual=:manual, decision_reason=:reason, decided_by=:actor, decided_at=now(), source=:source,
                  correlation_id=:correlation_id, request_id=:request_id,
                  metadata=metadata || CAST(:metadata AS JSONB), updated_at=now(), version=version+1
                  WHERE id=:id RETURNING *"""), params | {"id": existing["id"]}).mappings().first()
            else:
                link = conn.execute(text("""INSERT INTO opportunity_document_links
                  (opportunity_id,document_id,document_kind,relationship,is_manual,decision_reason,decided_by,
                   decided_at,source,correlation_id,request_id,metadata)
                  VALUES(CAST(:opportunity_id AS UUID),CAST(:document_id AS UUID),:kind,:relationship,:manual,
                   :reason,:actor,now(),:source,:correlation_id,:request_id,CAST(:metadata AS JSONB)) RETURNING *"""), params).mappings().first()
            _dual_write(conn, document_id, opportunity_id, relationship)
            conn.execute(text("UPDATE commercial_document_lines SET opportunity_document_link_id=:link_id WHERE commercial_document_id=CAST(:document_id AS UUID)"), {"link_id": link["id"], "document_id": document_id})
            _event(conn, link_id=str(link["id"]), opportunity_id=opportunity_id, document_id=document_id,
              event_type=event_type, actor=actor, reason=reason, old_relationship=old,
              new_relationship=relationship, correlation_id=correlation_id, request_id=request_id,
              idempotency_key=idempotency_key, payload=metadata)
            result = dict(link)
            _complete_idempotency(conn, idempotency_key, result)
            return result
    except IntegrityError as exc:
        raise DocumentConflictError("concurrent document reconciliation conflict") from exc


def set_primary_document(opportunity_id: str, document_id: str, **kwargs: Any) -> Dict[str, Any]:
    return set_document_relationship(opportunity_id, document_id, "PRIMARY", **kwargs)


def ignore_document(opportunity_id: str, document_id: str, **kwargs: Any) -> Dict[str, Any]:
    return set_document_relationship(opportunity_id, document_id, "IGNORED", **kwargs)


def remove_document(opportunity_id: str, document_id: str, **kwargs: Any) -> Dict[str, Any]:
    return set_document_relationship(opportunity_id, document_id, "REMOVED", **kwargs)


def reassign_document(origin_opportunity_id: str, destination_opportunity_id: str, document_id: str, *,
        actor: str, reason: str, correlation_id: Optional[str] = None, request_id: Optional[str] = None,
        idempotency_key: Optional[str] = None) -> Dict[str, Any]:
    actor = _actor(actor)
    if origin_opportunity_id == destination_opportunity_id:
        raise DocumentReconciliationError("origin and destination must differ")
    if not str(reason or "").strip():
        raise DocumentReconciliationError("reason is required")
    fingerprint, command = _command_fingerprint(opportunity_id=origin_opportunity_id,
        document_id=document_id, target_relationship="REASSIGNED",
        origin_opportunity_id=origin_opportunity_id,
        destination_opportunity_id=destination_opportunity_id, reason=reason,
        is_manual=True, source="reassignment", event_type="REASSIGNED", actor=actor,
        correlation_id=correlation_id, request_id=request_id)
    with engine.begin() as conn:
        replay = _claim_idempotency(conn, idempotency_key, fingerprint, command)
        if replay is not None:
            return replay
        # Stable lock order avoids deadlocks between opposite reassignments.
        for oid in sorted([origin_opportunity_id, destination_opportunity_id]):
            if not conn.execute(text("SELECT 1 FROM opportunities WHERE id=CAST(:id AS UUID) FOR UPDATE"), {"id": oid}).scalar():
                raise DocumentReconciliationError("opportunity not found")
        doc = conn.execute(text("SELECT document_kind FROM commercial_documents WHERE id=CAST(:id AS UUID) FOR UPDATE"), {"id": document_id}).mappings().first()
        old = _validate_membership(conn, origin_opportunity_id, document_id)
        if doc and not old:
            # Explicit reassignment may safely materialize its validated legacy
            # source relationship during rollout; ordinary transitions cannot
            # invent ownership.
            old = conn.execute(text("""INSERT INTO opportunity_document_links
                (opportunity_id,document_id,document_kind,relationship,is_manual,decision_reason,
                 decided_by,decided_at,source,metadata)
                SELECT CAST(:oid AS UUID),id,document_kind,'SECONDARY',TRUE,:reason,:actor,now(),
                       'reassignment_legacy_materialization','{}'::jsonb
                FROM commercial_documents WHERE id=CAST(:did AS UUID)
                  AND opportunity_id=CAST(:oid AS UUID) RETURNING *"""),
                {"oid": origin_opportunity_id, "did": document_id, "reason": reason, "actor": actor}).mappings().first()
            if old:
                _event(conn, link_id=str(old["id"]), opportunity_id=origin_opportunity_id,
                    document_id=document_id, event_type="LEGACY_SOURCE_MATERIALIZED", actor=actor,
                    reason=reason, old_relationship=None, new_relationship="SECONDARY",
                    correlation_id=correlation_id, request_id=request_id,
                    idempotency_key=(idempotency_key + ":legacy-source") if idempotency_key else None)
        if not doc or not old:
            raise DocumentReconciliationError("document does not belong to the source opportunity")
        existing_destination = conn.execute(text("""SELECT * FROM opportunity_document_links WHERE opportunity_id=CAST(:oid AS UUID)
          AND document_id=CAST(:did AS UUID) AND ended_at IS NULL"""), {"oid": destination_opportunity_id, "did": document_id}).mappings().first()
        if existing_destination:
            result = dict(existing_destination)
            _complete_idempotency(conn, idempotency_key, result)
            return result
        conn.execute(text("""UPDATE opportunity_document_links SET relationship='REASSIGNED',
          origin_opportunity_id=CAST(:origin AS UUID), destination_opportunity_id=CAST(:destination AS UUID),
          decision_reason=:reason, decided_by=:actor, decided_at=now(), ended_at=now(), updated_at=now(), version=version+1
          WHERE id=:id"""), {"origin": origin_opportunity_id, "destination": destination_opportunity_id,
          "reason": reason, "actor": actor, "id": old["id"]})
        link = conn.execute(text("""INSERT INTO opportunity_document_links
          (opportunity_id,document_id,document_kind,relationship,is_manual,decision_reason,decided_by,decided_at,
           source,origin_opportunity_id,destination_opportunity_id,correlation_id,request_id,metadata)
          VALUES(CAST(:destination AS UUID),CAST(:document_id AS UUID),:kind,'SECONDARY',TRUE,:reason,:actor,now(),
           'reassignment',CAST(:origin AS UUID),CAST(:destination AS UUID),:correlation_id,:request_id,
           jsonb_build_object('reassigned_from_link_id',:old_link)) RETURNING *"""), {
          "destination": destination_opportunity_id, "document_id": document_id, "kind": doc["document_kind"],
          "reason": reason, "actor": actor, "origin": origin_opportunity_id, "correlation_id": correlation_id,
          "request_id": request_id, "old_link": str(old["id"])}).mappings().first()
        _dual_write(conn, document_id, destination_opportunity_id, "SECONDARY")
        conn.execute(text("UPDATE commercial_document_lines SET opportunity_document_link_id=:link WHERE commercial_document_id=CAST(:doc AS UUID)"), {"link": link["id"], "doc": document_id})
        _event(conn, link_id=str(old["id"]), opportunity_id=origin_opportunity_id, document_id=document_id,
          event_type="REASSIGNED", actor=actor, reason=reason, old_relationship=old["relationship"],
          new_relationship="REASSIGNED", old_opportunity_id=origin_opportunity_id,
          new_opportunity_id=destination_opportunity_id, correlation_id=correlation_id,
          request_id=request_id, idempotency_key=idempotency_key)
        _event(conn, link_id=str(link["id"]), opportunity_id=destination_opportunity_id, document_id=document_id,
          event_type="REASSIGNMENT_RECEIVED", actor=actor, reason=reason, old_relationship=None,
          new_relationship="SECONDARY", old_opportunity_id=origin_opportunity_id,
          new_opportunity_id=destination_opportunity_id, correlation_id=correlation_id,
          request_id=request_id, idempotency_key=(idempotency_key + ":destination") if idempotency_key else None)
        result = dict(link)
        _complete_idempotency(conn, idempotency_key, result)
        return result


def restore_document(opportunity_id: str, document_id: str, *, actor: str, reason: str, **kwargs: Any) -> Dict[str, Any]:
    if not str(reason or "").strip():
        raise DocumentReconciliationError("reason is required")
    # Restoration is deliberately SECONDARY, never PRIMARY.
    return set_document_relationship(opportunity_id, document_id, "SECONDARY", actor=actor, reason=reason, **kwargs)


def classify_sync_document(conn: Any, opportunity_id: str, document_id: str, document_kind: str, *,
                           status: Any, old_status: Any = None, actor: str = "jasmin_sync") -> Dict[str, Any]:
    """Canonical in-transaction classification used by external document sync."""
    actor = _actor(actor)
    legacy_owner = conn.execute(text("SELECT opportunity_id::text FROM commercial_documents WHERE id=CAST(:id AS UUID)"),
                                {"id": document_id}).scalar()
    if str(legacy_owner or "") != str(opportunity_id):
        raise DocumentReconciliationError("synced document does not belong to this opportunity")
    link = conn.execute(text("""SELECT * FROM opportunity_document_links
        WHERE opportunity_id=CAST(:oid AS UUID) AND document_id=CAST(:did AS UUID)
          AND ended_at IS NULL FOR UPDATE"""), {"oid": opportunity_id, "did": document_id}).mappings().first()
    cancelled = normalize_jasmin_status(status) == "CANCELLED"
    if link:
        result = dict(link)
        if cancelled and normalize_jasmin_status(old_status) != "CANCELLED" and link["relationship"] == "PRIMARY":
            reason = "primary document cancelled in Jasmin"
            conn.execute(text("""UPDATE opportunity_document_links SET relationship='REVIEW_REQUIRED',
                decision_reason=:reason, source='jasmin_sync', updated_at=now(), version=version+1 WHERE id=:id"""),
                {"reason": reason, "id": link["id"]})
            _dual_write(conn, document_id, opportunity_id, "REVIEW_REQUIRED")
            _event(conn, link_id=str(link["id"]), opportunity_id=opportunity_id, document_id=document_id,
                event_type="PRIMARY_CANCELLED_REVIEW_REQUIRED", actor=actor, reason=reason,
                old_relationship="PRIMARY", new_relationship="REVIEW_REQUIRED",
                old_jasmin_status=normalize_jasmin_status(old_status), new_jasmin_status="CANCELLED",
                correlation_id=None, request_id=None,
                idempotency_key=f"jasmin-primary-cancelled:{opportunity_id}:{document_id}")
            result["relationship"] = "REVIEW_REQUIRED"
        return result
    candidates = conn.execute(text("""SELECT * FROM opportunity_document_links
        WHERE opportunity_id=CAST(:oid AS UUID) AND document_kind=:kind AND ended_at IS NULL FOR UPDATE"""),
        {"oid": opportunity_id, "kind": document_kind}).mappings().all()
    manual_primary = next((row for row in candidates if row["relationship"] == "PRIMARY" and row["is_manual"]), None)
    relationship = "SECONDARY" if manual_primary else ("PRIMARY" if not candidates and not cancelled else "REVIEW_REQUIRED")
    if candidates and not manual_primary:
        reason = "multiple Jasmin candidates require review"
        for previous in candidates:
            if previous["relationship"] == "PRIMARY" and not previous["is_manual"]:
                conn.execute(text("""UPDATE opportunity_document_links SET relationship='REVIEW_REQUIRED',
                    decision_reason=:reason, source='jasmin_sync', updated_at=now(), version=version+1 WHERE id=:id"""),
                    {"reason": reason, "id": previous["id"]})
                _dual_write(conn, str(previous["document_id"]), opportunity_id, "REVIEW_REQUIRED")
                _event(conn, link_id=str(previous["id"]), opportunity_id=opportunity_id,
                    document_id=str(previous["document_id"]), event_type="SYNC_AMBIGUITY_DEMOTED",
                    actor=actor, reason=reason, old_relationship="PRIMARY", new_relationship="REVIEW_REQUIRED",
                    correlation_id=None, request_id=None,
                    idempotency_key=f"jasmin-sync-ambiguity:{opportunity_id}:{previous['document_id']}:{document_id}")
        relationship = "REVIEW_REQUIRED"
    reason = "single unambiguous Jasmin candidate" if relationship == "PRIMARY" else "Jasmin candidate classification"
    new_link = conn.execute(text("""INSERT INTO opportunity_document_links
        (opportunity_id,document_id,document_kind,relationship,is_manual,decision_reason,decided_by,decided_at,source,metadata)
        VALUES(CAST(:oid AS UUID),CAST(:did AS UUID),:kind,:relationship,FALSE,:reason,:actor,now(),'jasmin_sync','{}'::jsonb)
        RETURNING *"""), {"oid": opportunity_id, "did": document_id, "kind": document_kind,
        "relationship": relationship, "reason": reason, "actor": actor}).mappings().first()
    _dual_write(conn, document_id, opportunity_id, relationship)
    _event(conn, link_id=str(new_link["id"]), opportunity_id=opportunity_id, document_id=document_id,
        event_type="SYNC_LINK_CLASSIFIED", actor=actor, reason=reason, old_relationship=None,
        new_relationship=relationship, correlation_id=None, request_id=None,
        idempotency_key=f"jasmin-sync-link:{opportunity_id}:{document_id}")
    return dict(new_link)


def register_created_document(conn: Any, opportunity_id: str, document_id: str,
                              document_kind: str, *, status: Any = None,
                              actor: str = "commercial_document_writer") -> Optional[Dict[str, Any]]:
    """Register a newly inserted legacy document in v2 in the same transaction.

    Before migration 007 this intentionally does nothing.  Afterwards the
    canonical classifier creates the current link, preserves manual primaries,
    and turns competing automatic candidates into REVIEW_REQUIRED instead of
    silently choosing a replacement.
    """
    if not document_reconciliation_v2_available(conn):
        return None
    return classify_sync_document(
        conn, opportunity_id, document_id, document_kind,
        status=status, actor=actor,
    )


def record_jasmin_status_change(opportunity_id: str, document_id: str, old_status: Any, new_status: Any, *,
        actor: str = "jasmin_sync", correlation_id: Optional[str] = None, request_id: Optional[str] = None,
        idempotency_key: Optional[str] = None) -> None:
    old_normalized, new_normalized = normalize_jasmin_status(old_status), normalize_jasmin_status(new_status)
    with engine.begin() as conn:
        link = conn.execute(text("""SELECT * FROM opportunity_document_links WHERE opportunity_id=CAST(:oid AS UUID)
          AND document_id=CAST(:did AS UUID) AND ended_at IS NULL FOR UPDATE"""), {"oid": opportunity_id, "did": document_id}).mappings().first()
        if not link or old_normalized == new_normalized:
            return
        _event(conn, link_id=str(link["id"]), opportunity_id=opportunity_id, document_id=document_id,
          event_type="JASMIN_STATUS_CHANGED", actor=_actor(actor), reason=None,
          old_relationship=link["relationship"], new_relationship=link["relationship"],
          old_jasmin_status=old_normalized, new_jasmin_status=new_normalized,
          correlation_id=correlation_id, request_id=request_id, idempotency_key=idempotency_key)
        if new_normalized == "CANCELLED" and link["relationship"] == "PRIMARY":
            conn.execute(text("""UPDATE opportunity_document_links SET relationship='REVIEW_REQUIRED',
              decision_reason='primary document cancelled in Jasmin', source='jasmin_sync', updated_at=now(), version=version+1
              WHERE id=:id"""), {"id": link["id"]})
            _dual_write(conn, document_id, opportunity_id, "REVIEW_REQUIRED")
            _event(conn, link_id=str(link["id"]), opportunity_id=opportunity_id, document_id=document_id,
              event_type="PRIMARY_CANCELLED_REVIEW_REQUIRED", actor=_actor(actor),
              reason="primary document cancelled in Jasmin", old_relationship="PRIMARY",
              new_relationship="REVIEW_REQUIRED", old_jasmin_status=old_normalized,
              new_jasmin_status=new_normalized, correlation_id=correlation_id, request_id=request_id,
              idempotency_key=(idempotency_key + ":review") if idempotency_key else None)


def list_document_link_events(opportunity_id: str, *, document_id: Optional[str] = None, limit: int = 200) -> List[Dict[str, Any]]:
    with engine.begin() as conn:
        rows = conn.execute(text("""SELECT id::text, link_id::text, opportunity_id::text, document_id::text,
          event_type, actor, reason, old_relationship, new_relationship, old_opportunity_id::text,
          new_opportunity_id::text, old_jasmin_status, new_jasmin_status, correlation_id, request_id,
          idempotency_key, created_at, payload FROM opportunity_document_link_events
          WHERE opportunity_id=CAST(:oid AS UUID) AND (:did IS NULL OR document_id=CAST(:did AS UUID))
          ORDER BY created_at DESC LIMIT :limit"""), {"oid": opportunity_id, "did": document_id, "limit": int(limit)}).mappings().all()
    return [dict(r) for r in rows]
