685 lines
42 KiB
Python
685 lines
42 KiB
Python
"""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]
|