Implement document reconciliation v2
This commit is contained in:
plx
684
app/document_reconciliation_service.py
Normal file
684
app/document_reconciliation_service.py
Normal file
@@ -0,0 +1,684 @@
|
||||
"""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]
|
||||
Reference in New Issue
Block a user