PacketStation/server/packetstation/feed/writes.py

"""Packet writes by the worker: the pipeline's packet_write for the system principal, with no session.

`system_write` locks the packet row (FOR UPDATE SKIP LOCKED: a row a person is writing right now is skipped and
retried next run, `Busy`), builds the policy's PacketState from the same child rows the API uses, asks the policy
as `Principal(user_id=None, system="visual-feed")` (`Refused` when it says no), yields a `SysWrite`, then bumps
the version, writes the packet_log row (by NULL, by_name 'VISUAL feed', role 'system'), the audit row
(`packet.<action>`, role_snapshot 'system') and the NOTIFY. Everything runs in a savepoint (or a transaction when
none is open), so a refusal or an error inside the body leaves nothing behind.

VIS_AUTO_RELEASE carries the SEND_TO_WAREHOUSE side effects here, locally: status warehouse, status_at,
wh_skipped false, wh_sent_at, a packet_wh incoming row (by NULL, auto true) and a packet_transitions row (by NULL,
action 'vis_auto_release').
"""

from __future__ import annotations

from collections.abc import Iterator, Mapping
from contextlib import contextmanager
from dataclasses import dataclass, field
from typing import Any

from psycopg.rows import dict_row

from .. import audit, events, policy
from ..domain import packets
from ..text import LIMITS, clean_opt

A = policy.Action
SYSTEM = policy.Principal(user_id=None, system="visual-feed")
BY_NAME = "VISUAL feed"
ROLE = "system"

LABELS: dict[policy.Action, str] = {
    A.VIS_UPDATE_DRAFT: "Updated from VISUAL",
    A.VIS_OPEN_CO: "Change from VISUAL",
    A.VIS_AUTO_RELEASE: "Sent to warehouse automatically",
    A.VIS_RETIRE_DRAFT: "Retired: work orders moved to another packet",
}

_SELECT = (
    "SELECT *, status::text AS status, source_kind::text AS source_kind, routed_to::text AS routed_to,"
    " drawing_state::text AS drawing_state FROM packets WHERE id = %s FOR UPDATE SKIP LOCKED"
)

EMPTY: Mapping[str, Any] = {}


class Busy(Exception):
    """Someone else holds the packet row right now. Retry next run."""


class Missing(LookupError):
    """No such packet."""


class Refused(Exception):
    """The policy said no."""

    def __init__(self, decision: policy.Decision) -> None:
        super().__init__(f"{decision.code}: {decision.why}")
        self.decision = decision


@dataclass
class SysWrite:
    conn: Any
    packet: dict[str, Any]  # the packets row, locked
    state: policy.PacketState
    action: policy.Action
    label: str
    note: str | None = None
    detail: dict[str, Any] = field(default_factory=dict)
    fields: list[str] = field(default_factory=list)
    noop: bool = False
    version: int = 0
    transition_id: int | None = None

    @property
    def id(self) -> int:
        return int(self.packet["id"])

    def changed(self, *names: str) -> None:
        for n in names:
            if n not in self.fields:
                self.fields.append(n)

    def log(self, label: str, note: str | None = None) -> None:
        """Override the packet_log label (and note)."""
        self.label = label
        if note is not None:
            self.note = note

    def skip(self) -> None:
        """Nothing changed: no version bump, no log, audit or event."""
        self.noop = True


def lock(conn: Any, packet_id: int) -> dict[str, Any]:
    """The packets row locked FOR UPDATE SKIP LOCKED. Busy when another transaction holds it."""
    with conn.cursor(row_factory=dict_row) as cur:
        row = cur.execute(_SELECT, (packet_id,)).fetchone()
    if row is not None:
        return dict(row)
    if conn.execute("SELECT 1 FROM packets WHERE id = %s", (packet_id,)).fetchone() is None:
        raise Missing(packet_id)
    raise Busy(packet_id)


@contextmanager
def system_write(
    conn: Any,
    packet_id: int,
    action: policy.Action,
    *,
    label: str | None = None,
    note: str | None = None,
    detail: dict[str, Any] | None = None,
    extra: Mapping[str, Any] = EMPTY,
) -> Iterator[SysWrite]:
    with conn.transaction():
        row = lock(conn, packet_id)
        ch = packets.load_children(conn, row, full=False)
        state = packets.state_from(row, ch)
        decision = policy.check(SYSTEM, state, action, extra=extra)
        if not decision.ok:
            raise Refused(decision)
        sw = SysWrite(
            conn=conn,
            packet=row,
            state=state,
            action=action,
            label=label or LABELS.get(action, str(action)),
            note=note,
            detail=dict(detail or {}),
            version=int(row["version"]),
        )
        yield sw
        _finish(sw)


def _finish(sw: SysWrite) -> None:
    if sw.noop:
        return
    conn = sw.conn
    from_status = str(sw.packet["status"])
    from_dept = str(sw.packet["dept_code"])
    if sw.action is A.VIS_AUTO_RELEASE:
        _auto_release(sw)
    sw.version = int(
        conn.execute("UPDATE packets SET version = version + 1 WHERE id = %s RETURNING version", (sw.id,)).fetchone()[0]
    )
    after = conn.execute("SELECT status::text, dept_code FROM packets WHERE id = %s", (sw.id,)).fetchone()
    status, dept = str(after[0]), str(after[1])
    note = clean_opt(sw.note, LIMITS["note"]) if sw.note else None
    conn.execute(
        "INSERT INTO packet_log (packet_id, by, by_name, role, action, note, transition_id)"
        " VALUES (%s, NULL, %s, %s, %s, %s, %s)",
        (sw.id, BY_NAME, ROLE, sw.label, note, sw.transition_id),
    )
    detail: dict[str, Any] = {"from": from_status, "to": status, "dept": dept}
    if from_dept != dept:
        detail["from_dept"] = from_dept
    if sw.fields:
        detail["fields"] = sw.fields
    if note:
        detail["note"] = note
    detail.update(sw.detail)
    audit.append(
        conn,
        action=f"packet.{sw.action}",
        role_snapshot=ROLE,
        subject_type="packet",
        subject_id=sw.id,
        packet_id=sw.id,
        detail=packets_jsonable(detail),
    )
    events.publish(
        conn, {"event": "changed", "packet_id": sw.id, "version": sw.version, "status": status, "dept": dept}
    )


def _auto_release(sw: SysWrite) -> None:
    conn = sw.conn
    conn.execute(
        "UPDATE packets SET status = 'warehouse', status_at = now(), wh_skipped = false, wh_sent_at = now(),"
        " return_note = NULL WHERE id = %s",
        (sw.id,),
    )
    conn.execute("UPDATE packet_wh SET closed_at = now() WHERE packet_id = %s AND closed_at IS NULL", (sw.id,))
    conn.execute(
        "INSERT INTO packet_wh (packet_id, stage, by, note, auto) VALUES (%s, 'incoming', NULL, %s, true)",
        (sw.id, "Sent automatically: no flags"),
    )
    dept = str(sw.packet["dept_code"])
    tid = conn.execute(
        "INSERT INTO packet_transitions (packet_id, by, action, from_status, to_status, from_dept, to_dept, reason)"
        " VALUES (%s, NULL, 'vis_auto_release', %s::packet_status, 'warehouse', %s, %s, %s) RETURNING id",
        (sw.id, str(sw.packet["status"]), dept, dept, sw.note),
    ).fetchone()
    sw.transition_id = int(tid[0])
    sw.changed("status")


def packets_jsonable(value: Any) -> Any:
    """Audit detail as plain JSON (dates and Decimals as text)."""
    from fastapi.encoders import jsonable_encoder

    return jsonable_encoder(value)


def log_only(conn: Any, packet_id: int, label: str, note: str | None = None) -> None:
    """A packet_log row by the feed with no version bump (for observations that change nothing a user edits)."""
    conn.execute(
        "INSERT INTO packet_log (packet_id, by, by_name, role, action, note) VALUES (%s, NULL, %s, %s, %s, %s)",
        (packet_id, BY_NAME, ROLE, label, clean_opt(note, LIMITS["note"]) if note else None),
    )