#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""DM exchange status through the canonical ``scitex_dev.status`` primitive.
The DM row and the exchange row answer different questions. ``dm_messages``
proves that the content is durable; ``status_exchanges`` says whether the
accepted delivery exchange ever reached a later, final observation. The
immediate status is HTTP 202: accepted and persisted, not displayed or
acknowledged.
"""
from __future__ import annotations
import socket
import uuid
from datetime import datetime, timezone
from psycopg import Error as PostgresError
from scitex_dev.status import (
LEDGER_TABLE,
StatusCode,
ledger_record,
new_exchange_id,
)
from scitex_dev.store import NEW_RECORD, Store, plugin_for, resolve_target
__all__ = [
"DmExchangeError",
"accepted_status",
"failure_status",
"get_exchange",
"new_exchange",
"new_client_request_id",
"open_exchange_store",
"persist_dm",
"record_exchange",
]
[docs]
class DmExchangeError(RuntimeError):
"""A DM exchange failed, retaining its safe canonical wire facts."""
def __init__(self, exchange_id, status, *, message_id=None):
super().__init__(status.message)
self.exchange_id = exchange_id
self.status = status
self.message_id = message_id
def new_client_request_id() -> str:
"""Mint the caller-owned retry identity, distinct from responder xch_."""
return f"req_{uuid.uuid4().hex}"
def new_exchange() -> tuple[str, str]:
"""Mint the exchange id and its immutable opening timestamp."""
return new_exchange_id(), datetime.now(timezone.utc).isoformat()
def open_exchange_store(sender: str) -> Store:
"""Open scitex-dev's canonical exchange ledger, with no local fallback."""
plugin = plugin_for(LEDGER_TABLE)
return Store(
resolve_target(plugin),
plugin.schema,
node=socket.gethostname(),
writer_policy=plugin.writer_policy,
actor=sender,
)
def accepted_status(exchange_id: str, message_id: str) -> StatusCode:
"""The immediate, non-final status after the DM row commits."""
return StatusCode(
kind="http",
code=202,
message=(
f"OBSERVED: DM {message_id} and its recipient notification "
f"persisted in canonical PostgreSQL as exchange {exchange_id}. "
"NOT ESTABLISHED: live visibility or "
"recipient acknowledgement. NEXT: poll "
f"`scitex-cards dm get-status {exchange_id} --json`."
),
)
def failure_status(
exc: Exception,
exchange_id: str | None,
*,
persistence_known_absent: bool = False,
stage: str = "persist",
) -> StatusCode:
"""Map observed failures onto native HTTP codes without exposing secrets."""
from scitex_dev.store import StoreError
from ._backend import BackendUnavailableError
from ._messaging import AgentIdentityUnresolved
from ._store_target import StoreTargetNotConfigured
if isinstance(exc, AgentIdentityUnresolved):
code = 400
persistence_known = True
observation = "sender identity was unresolved; the DM API was not called"
action = (
"set SCITEX_CARDS_AGENT_ID=<your-agent> or pass --sender, then rerun "
"`scitex-cards dm send RECIPIENT MESSAGE`"
)
elif isinstance(exc, StoreTargetNotConfigured):
code = 503
persistence_known = True
observation = "no canonical PostgreSQL target was configured"
action = (
"set SCITEX_CARDS_DB, then inspect it with "
"`scitex-cards resolve-store` before retrying"
)
elif isinstance(exc, ModuleNotFoundError):
code = 503
persistence_known = True
observation = "the PostgreSQL driver was unavailable before the write"
action = "install `scitex-cards[all]`, then retry the command"
elif isinstance(exc, (StoreError, PostgresError)):
code = 503
persistence_known = persistence_known_absent
detail = str(exc).lower()
refused_config = any(
marker in detail
for marker in ("auth", "password", "pgbouncer", "configuration")
)
if stage == "config":
observation = (
"PostgreSQL authentication/configuration refused the shared "
"status-ledger connection before DM persistence"
if refused_config
else "the shared PostgreSQL status ledger could not be opened"
)
action = (
"run `psql \"$SCITEX_STORE_DSN\" -c 'select "
"current_database(), pg_is_in_recovery()'`; correct the "
"DSN/auth route, then run `scitex-cards health`"
)
else:
observation = (
"the canonical PostgreSQL DM/notification write returned no "
"durable result"
)
action = (
"run `scitex-cards health`; inspect dm_messages and "
"notifications by the quoted exchange before sending again"
)
elif isinstance(exc, BackendUnavailableError):
detail = str(exc).lower()
if "token" in detail or "auth" in detail or "bearer" in detail:
code = 401
persistence_known = True
observation = "the configured hub rejected or lacked credentials"
action = (
"run `scitex-cards hub doctor`, re-provision the token, then retry"
)
elif "identity" in detail or "x-scitex-agent" in detail:
code = 400
persistence_known = True
observation = "the configured hub received no sender identity"
action = (
"set SCITEX_CARDS_AGENT_ID=<your-agent> or pass --sender, then retry"
)
else:
code = 503
persistence_known = False
observation = "the configured canonical hub route returned no DM result"
action = (
"run `scitex-cards hub doctor`; inspect dm_messages before retrying"
)
elif isinstance(exc, ValueError):
code = 400
persistence_known = True
observation = "sender, recipient, or message validation was refused"
action = "supply non-empty values, then rerun `scitex-cards dm send --help`"
else:
code = 500
persistence_known = persistence_known_absent
observation = (
"the shared PostgreSQL ledger could not be opened"
if persistence_known_absent
else "the DM API returned no durable message id"
)
action = (
"verify SCITEX_CARDS_DB and SCITEX_STORE_DSN name the writable "
"canonical PostgreSQL, then run `scitex-cards health`"
if persistence_known_absent
else "run `scitex-cards health`; inspect dm_messages before retrying"
)
exchange_label = f" for exchange {exchange_id}" if exchange_id else ""
persistence = (
f"No DM was persisted{exchange_label}."
if persistence_known
else (
"NOT ESTABLISHED: whether the request produced a durable DM"
f"{exchange_label}."
)
)
return StatusCode(
kind="http",
code=code,
message=(
f"OBSERVED: {observation}. {persistence} NEXT: {action}."
),
)
def record_exchange(
store: Store,
*,
exchange_id: str,
opened_at: str,
sender: str,
recipient: str,
status: StatusCode,
) -> None:
"""Append the exchange's first status to scitex-dev's shared ledger."""
row = ledger_record(
exchange_id=exchange_id,
initiator=sender,
responder=recipient,
operation="cards.dm.delivery",
status=status,
opened_at=opened_at,
)
store.put(row, expected_revision=NEW_RECORD, owner=recipient, actor=sender)
def persist_dm(
sender: str,
recipient: str,
body: str,
*,
store=None,
client_request_id: str,
) -> dict:
"""Persist one DM, its notification exchange, and the canonical HTTP 202."""
from . import _threads
if not isinstance(client_request_id, str) or not client_request_id.strip():
raise ValueError("dm_send requires a non-empty client_request_id")
exchange_id, opened_at = new_exchange()
try:
ledger = open_exchange_store(sender)
except Exception as exc:
status = failure_status(
exc,
exchange_id,
persistence_known_absent=True,
stage="config",
)
raise DmExchangeError(exchange_id, status) from None
record = None
try:
with ledger:
try:
record = _threads.append_message(
sender,
recipient,
body,
store=store,
exchange_id=exchange_id,
client_request_id=client_request_id,
)
except Exception as exc:
status = failure_status(exc, exchange_id, stage="persist")
_record_failure(
ledger,
exchange_id,
opened_at,
sender,
recipient,
status,
)
raise DmExchangeError(exchange_id, status) from None
if record.get("idempotent_replay"):
original_exchange = record["exchange_id"]
row = ledger.get({"exchange_id": original_exchange})
if row is None:
status = persisted_without_exchange_status(
original_exchange, record["id"]
)
raise DmExchangeError(
original_exchange, status, message_id=record["id"]
)
from ._inbox_postgres import notification_for_exchange
notification = notification_for_exchange(
original_exchange, store=store
)
if notification is None:
status = failure_status(
RuntimeError("notification absent"), original_exchange
)
raise DmExchangeError(
original_exchange, status, message_id=record["id"]
)
payload = _exchange_payload(row)
return {
**record,
"notification_id": notification["id"],
"exchange_id": original_exchange,
"client_request_id": client_request_id,
"status": payload["status"],
}
status = accepted_status(exchange_id, record["id"])
try:
record_exchange(
ledger,
exchange_id=exchange_id,
opened_at=opened_at,
sender=sender,
recipient=recipient,
status=status,
)
except Exception:
status = persisted_without_exchange_status(
exchange_id, record["id"]
)
raise DmExchangeError(
exchange_id, status, message_id=record["id"]
) from None
except DmExchangeError:
raise
except Exception as exc:
if record is None:
status = failure_status(
exc,
exchange_id,
persistence_known_absent=True,
stage="config",
)
raise DmExchangeError(exchange_id, status) from None
status = persisted_without_exchange_status(exchange_id, record["id"])
raise DmExchangeError(
exchange_id, status, message_id=record["id"]
) from None
return {
**record,
"exchange_id": exchange_id,
"client_request_id": client_request_id,
"status": status.to_dict(),
}
def _record_failure(ledger, exchange_id, opened_at, sender, recipient, status):
"""Preserve the primary failure even if its ledger append also fails."""
try:
record_exchange(
ledger,
exchange_id=exchange_id,
opened_at=opened_at,
sender=sender,
recipient=recipient,
status=status,
)
except Exception:
return
def persisted_without_exchange_status(exchange_id: str, message_id: str):
"""Status for a durable DM whose ledger write was not observed."""
return StatusCode(
kind="http",
code=500,
message=(
f"OBSERVED: DM {message_id} persisted, but exchange {exchange_id} "
"was not observed in the status ledger. NOT ESTABLISHED: live "
"visibility or acknowledgement. NEXT: run `scitex-cards health`, "
"inspect dm_messages by the quoted message id, and repair the "
"status-ledger route before sending again."
),
)
def get_exchange(exchange_id: str, *, sender: str) -> dict | None:
"""Read one exchange as a plain protocol payload."""
with open_exchange_store(sender) as store:
row = store.get({"exchange_id": exchange_id})
if row is None:
return None
return _exchange_payload(row)
def _exchange_payload(row) -> dict:
values = dict(row.values)
return {
"exchange_id": values["exchange_id"],
"initiator": values["initiator"],
"responder": values["responder"],
"operation": values["operation"],
"status": {
"kind": values["kind"],
"code": values["code"],
"message": values["message"],
},
"final": values["final"],
"opened_at": values["opened_at"],
"updated_at": values["updated_at"],
}
# EOF