PacketStation/server/packetstation/feed/apply.py

"""The apply: VISUAL candidates onto projects, units and packets, in one transaction.

Steps, in order (phase3-design "The apply"): projects and units with their order-derived fields; match candidates
to packets by visual_key; new drafts; drafts updated in place; change orders on packets past draft; drafts whose
work orders moved retired; work orders gone from VISUAL flagged after two misses; closed and cancelled observed;
auto-release; one `feed.run` audit row and a `feed` event.

Every packet write goes through `writes.system_write` as the system principal, so the policy decides. A packet row
someone else holds is skipped this run (FOR UPDATE SKIP LOCKED) and counted in `skipped_busy`. A candidate that
fails is rolled back to its savepoint, counted in `errors` and the run is degraded; the others carry on.
"""

from __future__ import annotations

import logging
from collections.abc import Callable, Collection, Iterable, Mapping
from datetime import UTC, date, datetime
from decimal import Decimal
from typing import Any

import psycopg.errors as pgerr
from psycopg.rows import dict_row
from psycopg.types.json import Jsonb

from .. import audit, events, policy
from ..domain import packets as packets_mod
from ..domain import projects as projects_domain
from ..text import LIMITS, RX_PROJECT, clean_opt
from . import changes, materials
from . import orders as orders_mod
from .candidates import NO_DRAWING, Built, Candidate, project_of, unit_number
from .fingerprint import Fingerprint, bom_sig, classify, compute, drawing_key, merge
from .source import redact
from .types import Order, Row, WoKey
from .writes import Busy, Missing, Refused, SysWrite, packets_jsonable, system_write

log = logging.getLogger(__name__)

A = policy.Action

COUNT_KEYS = (
    "wos_window",
    "wos_tracked",
    "rows",
    "materials",
    "orders",
    "labor_legs",
    "new_drafts",
    "drafts_updated",
    "cos_opened",
    "cos_reaimed",
    "cos_withdrawn",
    "cos_unchanged",
    "retired",
    "wo_gone_flagged",
    "wo_back",
    "closed_observed",
    "cancelled_flagged",
    "ignored_golive",
    "projects_created",
    "units_created",
    "auto_released",
    "unmapped_resources",
    "skipped_removed",
    "skipped_closed",
    "skipped_lots",
    "skipped_busy",
    "skipped_conflict",
    "skipped_hand",
    "refused",
    "errors",
    "suspect",
)

# Every flag the drawing side sets. A drawing or rev change on a draft clears them and starts over.
DRAWING_FLAGS = (
    "drawing_needed",
    "rev_mismatch",
    "not_released",
    "not_found",
    "lookup_failed",
    "bad_rev",
    "no_drawing",
    "hand_drawing",
)
FIELDS = orders_mod.FIELDS  # customer, po, order_date, ship_date
TOP = 30

Counts = dict[str, Any]
LaborMap = Mapping[tuple[WoKey, int], Decimal]


def _now() -> datetime:
    return datetime.now(UTC)


def _iso(v: Any) -> Any:
    return v.isoformat() if isinstance(v, date | datetime) else v


def _top(d: Mapping[str, int]) -> dict[str, int]:
    return dict(sorted(d.items(), key=lambda kv: (-kv[1], kv[0]))[:TOP])


class _Run:
    """The apply's working state: counts, errors and the context."""

    def __init__(self, conn: Any, ctx: Any, batch: Any, feed_settings: Any, today: date) -> None:
        self.conn = conn
        self.ctx = ctx
        self.batch = batch
        self.fs = feed_settings
        self.today = today
        self.counts: Counts = dict.fromkeys(COUNT_KEYS, 0)
        self.now = _now()

    @property
    def batch_id(self) -> int | None:
        return getattr(self.batch, "id", None)

    def error(self, what: str, e: BaseException | str) -> None:
        msg = f"{what}: {redact(e)}"
        log.warning("feed apply: %s", msg)
        self.counts["errors"] += 1
        if self.ctx is None:
            return
        if callable(getattr(self.ctx, "error", None)):
            self.ctx.error(msg)
            return
        self.ctx.degraded = True
        errs = getattr(self.ctx, "errors", None)
        if isinstance(errs, list) and len(errs) < 20:
            errs.append(msg[:500])

    def guarded(self, what: str, fn: Callable[..., Any], *args: Any) -> Any:
        """Run fn(*args) in a savepoint. Busy and lock timeouts are skipped; refusals and errors counted."""
        try:
            with self.conn.transaction():
                return fn(*args)
        except (Busy, pgerr.LockNotAvailable):
            self.counts["skipped_busy"] += 1
        except Missing:
            pass
        except Refused as e:
            self.counts["refused"] += 1
            log.info("feed apply: %s refused: %s", what, e)
        except Exception as e:
            self.error(what, e)
        return None


# ---------------------------------------------------------------- 1. projects and units


def _audit_system(conn: Any, action: str, subject_type: str, subject_id: Any, detail: Any) -> None:
    audit.append(
        conn,
        action=action,
        role_snapshot="system",
        subject_type=subject_type,
        subject_id=subject_id,
        detail=packets_jsonable(detail),
    )


def _publish_project(conn: Any, project_id: int, unit_id: int | None, action: str) -> None:
    payload: dict[str, Any] = {"event": "changed", "kind": "project", "project_id": project_id, "action": action}
    if unit_id is not None:
        payload["unit_id"] = unit_id
    events.publish(conn, payload)


def _one(conn: Any, sql: str, params: Any) -> dict[str, Any] | None:
    with conn.cursor(row_factory=dict_row) as cur:
        r = cur.execute(sql, params).fetchone()
    return dict(r) if r else None


def _ensure_project(run: _Run, number: str, due: date | None) -> dict[str, Any] | None:
    conn = run.conn
    found = projects_domain.lookup(conn, number)
    if found is not None and found["kind"] != "project":
        run.counts["skipped_conflict"] += 1
        return None
    if found is None:
        conn.execute("SELECT pg_advisory_xact_lock(7312)")
        found = projects_domain.lookup(conn, number)
    if found is None:
        pid = int(
            conn.execute(
                "INSERT INTO projects (number, name, source, needs_details, created_by, due)"
                " VALUES (%s, NULL, 'visual', true, NULL, %s) RETURNING id",
                (number, due),
            ).fetchone()[0]
        )
        conn.execute(
            "INSERT INTO project_aliases (alias, project_id, unit_id, kind) VALUES (%s, %s, NULL, 'import')"
            " ON CONFLICT (alias) DO NOTHING",
            (number.upper(), pid),
        )
        _audit_system(conn, "project.created", "project", pid, {"number": number, "source": "visual", "due": due})
        _publish_project(conn, pid, None, "project.created")
        run.counts["projects_created"] += 1
    else:
        pid = int(found["project_id"])
    return _one(conn, "SELECT * FROM projects WHERE id = %s", (pid,))


def _ensure_unit(run: _Run, project: Mapping[str, Any], unit_number: str, lot: str, due: date | None) -> Any:
    conn = run.conn
    found = projects_domain.lookup(conn, unit_number)
    if found is not None and found["kind"] != "unit":
        run.counts["skipped_conflict"] += 1
        return None
    if found is None:
        conn.execute("SELECT pg_advisory_xact_lock(7312)")
        found = projects_domain.lookup(conn, unit_number)
    if found is None:
        uid = int(
            conn.execute(
                "INSERT INTO units (project_id, lot, due, created_by) VALUES (%s, %s, %s, NULL) RETURNING id",
                (project["id"], lot, due),
            ).fetchone()[0]
        )
        conn.execute(
            "INSERT INTO project_aliases (alias, project_id, unit_id, kind) VALUES (%s, %s, %s, 'import')"
            " ON CONFLICT (alias) DO NOTHING",
            (unit_number.upper(), project["id"], uid),
        )
        _audit_system(
            conn,
            "unit.created",
            "unit",
            uid,
            {"number": unit_number, "project_id": project["id"], "lot": lot, "due": due, "source": "visual"},
        )
        _publish_project(conn, int(project["id"]), uid, "unit.created")
        run.counts["units_created"] += 1
    else:
        uid = int(found["id"])
    return _one(conn, "SELECT * FROM units WHERE id = %s", (uid,))


def _derive(run: _Run, table: str, row: Mapping[str, Any], chosen: orders_mod.Chosen | None) -> dict[str, Any] | None:
    """Write the order-derived fields with provenance. Returns the order_ref now on the row."""
    current_ref = row.get("order_ref")
    if chosen is None:
        return current_ref
    values = orders_mod.derive(chosen.order)
    limits = {"customer": LIMITS["customer"], "po": LIMITS["po"]}
    sets: list[str] = []
    params: list[Any] = []
    changed: list[str] = []
    for f in FIELDS:
        v = values[f]
        if f in limits:
            v = clean_opt(v, limits[f])
        if row.get(f"{f}_visual") != v:
            sets.append(f"{f}_visual = %s")
            params.append(v)
            changed.append(f"{f}_visual")
        src = row.get(f"{f}_src")
        if src in (None, "visual") and (row.get(f) != v or (v is not None and src is None)):
            sets.append(f"{f} = %s")
            params.append(v)
            if v is not None:
                sets.append(f"{f}_src = 'visual'")
            changed.append(f)
    ref = orders_mod.order_ref(chosen, run.now)
    if not orders_mod.same_ref(current_ref, ref):
        sets.append("order_ref = %s")
        params.append(Jsonb(ref))
        changed.append("order_ref")
    else:
        ref = current_ref or ref
    if not sets:
        return ref
    sets.append("version = version + 1")
    sets.append("updated_at = now()")
    run.conn.execute(
        f"UPDATE {table} SET {', '.join(sets)} WHERE id = %s",  # noqa: S608 (fixed table and column names)
        (*params, row["id"]),
    )
    kind = "project" if table == "projects" else "unit"
    if any(c != "order_ref" for c in changed):
        _audit_system(run.conn, f"{kind}.updated", kind, row["id"], {"fields": changed, "src": "visual"})
    return ref


# ---------------------------------------------------------------- shapes for the packet rows


def _visual_json(c: Candidate, fp: Fingerprint) -> dict[str, Any]:
    return {
        "folded": changes.folded_json(c),
        "partspecs": list(c.partspecs),
        "fp_parts": dict(fp.parts),
        "mat_known": c.bom is not None,
    }


def _legs_state(c: Candidate) -> list[dict[str, Any]]:
    return [{"key": str(lg.key), "seq": lg.seq, "st": c.statuses.get(lg.key, lg.st)} for lg in c.legs]


def _hours(labor: LaborMap | None, legs: Iterable[tuple[WoKey, int]]) -> Decimal | None:
    if labor is None:
        return None
    total = Decimal(0)
    seen = False
    for k in legs:
        h = labor.get(k)
        if h is not None:
            total += h
            seen = True
    return total if seen else Decimal(0)


def _insert_legs(conn: Any, packet_id: int, c: Candidate) -> None:
    for sort, lg in enumerate(c.legs):
        conn.execute(
            "INSERT INTO packet_work_orders (packet_id, wo_base, lot, split, sub, seq, qty, target_hrs, op,"
            " description, visual_status, partspecs, dwg, rev, sort)"
            " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
            (
                packet_id,
                lg.key.base,
                lg.key.lot,
                lg.key.split,
                lg.key.sub,
                lg.seq,
                lg.qty,
                lg.tgt,
                clean_opt(lg.op, LIMITS["op"]),
                clean_opt(c.title, LIMITS["description"]),
                lg.st or None,
                clean_opt(lg.partspecs, LIMITS["notes"]),
                clean_opt(lg.dwg, LIMITS["drawing"]),
                clean_opt(lg.rev, LIMITS["rev"]),
                sort,
            ),
        )


def _replace_legs(conn: Any, packet_id: int, c: Candidate) -> None:
    """Keep rows whose key and seq still match (their ids and missing_runs), update their fields, drop the rest,
    add the new ones."""
    with conn.cursor(row_factory=dict_row) as cur:
        old = cur.execute("SELECT * FROM packet_work_orders WHERE packet_id = %s", (packet_id,)).fetchall()
    by = {(r["wo_base"], r["lot"], r["split"], r["sub"], r["seq"]): r for r in old}
    keep: set[int] = set()
    for sort, lg in enumerate(c.legs):
        r = by.get((lg.key.base, lg.key.lot, lg.key.split, lg.key.sub, lg.seq))
        vals = (
            lg.qty,
            lg.tgt,
            clean_opt(lg.op, LIMITS["op"]),
            clean_opt(c.title, LIMITS["description"]),
            lg.st or None,
            clean_opt(lg.partspecs, LIMITS["notes"]),
            clean_opt(lg.dwg, LIMITS["drawing"]),
            clean_opt(lg.rev, LIMITS["rev"]),
            sort,
        )
        if r is not None:
            keep.add(int(r["id"]))
            conn.execute(
                "UPDATE packet_work_orders SET qty = %s, target_hrs = %s, op = %s, description = %s,"
                " visual_status = %s, partspecs = %s, dwg = %s, rev = %s, sort = %s, row_version = row_version + 1"
                " WHERE id = %s",
                (*vals, r["id"]),
            )
    gone = [int(r["id"]) for r in old if int(r["id"]) not in keep]
    if gone:
        conn.execute("DELETE FROM packet_work_orders WHERE id = ANY(%s)", (gone,))
    for sort, lg in enumerate(c.legs):
        if (lg.key.base, lg.key.lot, lg.key.split, lg.key.sub, lg.seq) in by:
            continue
        conn.execute(
            "INSERT INTO packet_work_orders (packet_id, wo_base, lot, split, sub, seq, qty, target_hrs, op,"
            " description, visual_status, partspecs, dwg, rev, sort)"
            " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
            (
                packet_id,
                lg.key.base,
                lg.key.lot,
                lg.key.split,
                lg.key.sub,
                lg.seq,
                lg.qty,
                lg.tgt,
                clean_opt(lg.op, LIMITS["op"]),
                clean_opt(c.title, LIMITS["description"]),
                lg.st or None,
                clean_opt(lg.partspecs, LIMITS["notes"]),
                clean_opt(lg.dwg, LIMITS["drawing"]),
                clean_opt(lg.rev, LIMITS["rev"]),
                sort,
            ),
        )


def _visual_state(conn: Any, packet_id: int, legs: list[dict[str, Any]], hours: Decimal | None, ref: Any) -> None:
    conn.execute(
        "INSERT INTO packet_visual_state (packet_id, legs, ticket_hours, order_ref, observed_at)"
        " VALUES (%s, %s, %s, %s, now()) ON CONFLICT (packet_id) DO UPDATE SET legs = EXCLUDED.legs,"
        " ticket_hours = coalesce(EXCLUDED.ticket_hours, packet_visual_state.ticket_hours),"
        " order_ref = coalesce(EXCLUDED.order_ref, packet_visual_state.order_ref), observed_at = now()",
        (packet_id, Jsonb(legs), hours, None if ref is None else Jsonb(packets_jsonable(ref))),
    )


# ---------------------------------------------------------------- 3. new drafts


def _create(
    run: _Run, c: Candidate, unit: Mapping[str, Any], project: Mapping[str, Any], ref: Any, labor: LaborMap | None
) -> int:
    conn = run.conn
    fp = compute(c)
    has_dwg = bool(c.drawing)
    flags = ["drawing_needed"] if has_dwg else ["no_drawing"]
    pid = int(
        conn.execute(
            "INSERT INTO packets (unit_id, dept_code, op, title, customer, drawing, rev, source_kind, source,"
            " visual_key, visual_rel, visual_want, drawing_state, flags, visual_fp_applied, visual, created_by)"
            " VALUES (%s, %s, %s, %s, %s, %s, %s, 'visual', %s, %s, %s, %s, %s::drawing_state, %s, %s, %s, NULL)"
            " RETURNING id",
            (
                unit["id"],
                c.dept,
                clean_opt(c.op, LIMITS["op"]),
                clean_opt(c.title, LIMITS["title"]),
                clean_opt(project.get("customer"), LIMITS["customer"]),
                clean_opt(c.drawing, LIMITS["drawing"]),
                clean_opt(c.rev, LIMITS["rev"]),
                Jsonb({"kind": "visual", "batch": run.batch_id}),
                c.key,
                c.rel,
                c.want,
                "needed" if has_dwg else "none",
                flags,
                fp.full,
                Jsonb(packets_jsonable(_visual_json(c, fp))),
            ),
        ).fetchone()[0]
    )
    _insert_legs(conn, pid, c)
    if c.bom is not None:
        materials.insert_bom(conn, pid, c.bom)
    _visual_state(conn, pid, _legs_state(c), _hours(labor, ((lg.key, lg.seq) for lg in c.legs)), ref)
    n = len(c.legs)
    note = f"{n} work order{'' if n == 1 else 's'} · {c.drawing or NO_DRAWING}" + (f" rev {c.rev}" if c.rev else "")
    conn.execute(
        "INSERT INTO packet_log (packet_id, by, by_name, role, action, note) VALUES (%s, NULL, 'VISUAL feed',"
        " 'system', 'Packet created from VISUAL', %s)",
        (pid, note),
    )
    _audit_system_packet(conn, "packet.created", pid, {"visual_key": c.key, "dept": c.dept, "batch": run.batch_id})
    events.publish(conn, {"event": "changed", "packet_id": pid, "version": 1, "status": "draft", "dept": c.dept})
    run.counts["new_drafts"] += 1
    return pid


def _audit_system_packet(conn: Any, action: str, packet_id: int, detail: Any) -> None:
    audit.append(
        conn,
        action=action,
        role_snapshot="system",
        subject_type="packet",
        subject_id=packet_id,
        packet_id=packet_id,
        detail=packets_jsonable(detail),
    )


# ---------------------------------------------------------------- 4. drafts updated in place


def _update_draft(run: _Run, packet: Mapping[str, Any], c: Candidate) -> None:
    visual = dict(packet.get("visual") or {})
    old_parts = visual.get("fp_parts")
    fp = merge(old_parts, compute(c))
    if fp.full == packet.get("visual_fp_applied"):
        return
    with system_write(run.conn, int(packet["id"]), A.VIS_UPDATE_DRAFT) as sw:
        conn = sw.conn
        before = changes.packet_view(conn, sw.id)
        diff = changes.diff_text(before, c)
        row = sw.packet
        visual = dict(row.get("visual") or {})
        _replace_legs(conn, sw.id, c)
        sw.changed("work_orders")
        if c.bom is not None and not visual.get("manual_bom") and bom_sig(changes.view_bom(before)) != bom_sig(c.bom):
            conn.execute("DELETE FROM bom_lines WHERE packet_id = %s", (sw.id,))
            materials.insert_bom(conn, sw.id, c.bom)
            sw.changed("bom")
        moved = str(row.get("rev") or "") != str(c.rev or "") or drawing_key(row.get("drawing")) != drawing_key(
            c.drawing
        )
        flags = list(row.get("flags") or [])
        state = str(row.get("drawing_state"))
        if moved:
            conn.execute("DELETE FROM packet_pages WHERE packet_id = %s AND type = 'drawing'", (sw.id,))
            for k in ("rev_ok", "manual_dwg", "vault", "appr", "called"):
                visual.pop(k, None)
            visual["rev_confirmed"] = False
            flags = [f for f in flags if f not in DRAWING_FLAGS]
            flags.append("drawing_needed" if c.drawing else "no_drawing")
            state = "needed" if c.drawing else "none"
            sw.changed("drawing", "rev", "drawing_state", "flags", "pages")
        visual.update(_visual_json(c, fp))
        conn.execute(
            "UPDATE packets SET drawing = %s, rev = %s, drawing_state = %s::drawing_state, flags = %s,"
            " visual_fp_applied = %s, visual = %s, visual_rel = %s, visual_want = %s WHERE id = %s",
            (
                clean_opt(c.drawing, LIMITS["drawing"]),
                clean_opt(c.rev, LIMITS["rev"]),
                state,
                flags,
                fp.full,
                Jsonb(packets_jsonable(visual)),
                c.rel,
                c.want,
                sw.id,
            ),
        )
        label = "Updated from VISUAL"
        if moved and before.get("rev") and c.rev and str(before.get("rev")) != str(c.rev):
            label = f"Updated from VISUAL: drawing rev {before.get('rev')} taken off, waiting on rev {c.rev}"
        sw.log(label, "; ".join(diff) or None)
    run.counts["drafts_updated"] += 1


# ---------------------------------------------------------------- 5. change orders past draft


def _past_draft(run: _Run, packet: Mapping[str, Any], c: Candidate) -> None:
    visual = dict(packet.get("visual") or {})
    old_parts = visual.get("fp_parts")
    fp = merge(old_parts, compute(c))
    applied = packet.get("visual_fp_applied")
    pending = changes.pending_visual(run.conn, int(packet["id"]))
    if fp.full == applied and pending is None:
        return
    if pending is not None and pending.get("target_fp") == fp.full:
        run.counts["cos_unchanged"] += 1
        return
    with system_write(run.conn, int(packet["id"]), A.VIS_OPEN_CO) as sw:
        conn = sw.conn
        pend = changes.pending_visual(conn, sw.id)
        if fp.full == applied:
            # VISUAL went back to what the packet holds: withdraw a pending VISUAL change order.
            if pend is not None and pend["origin"] == "visual":
                changes.withdraw(sw, int(pend["id"]), "VISUAL went back to what the packet already holds.")
                if not sw.noop:
                    run.counts["cos_withdrawn"] += 1
            else:
                sw.skip()
            return
        change = classify(old_parts, fp.parts)
        if not change.changed:
            sw.skip()
            return
        last = changes.last_feed_co(conn, sw.id)
        if (
            pend is None
            and last is not None
            and last.get("target_fp") == fp.full
            and last["review"]
            in (
                "rejected",
                "approved",
            )
        ):
            sw.skip()  # Clerical rejected exactly this, or already approved it
            return
        before = changes.packet_view(conn, sw.id)
        diff = changes.diff_text(before, c)
        vis = changes.vis_for(c, fp.parts, before, diff, fp.full)
        waiting = (
            {"dwg": vis["to_dwg"], "rev": vis["to_rev"], "since": run.now.isoformat(timespec="seconds")}
            if vis["dwg_change"]
            else None
        )
        res = changes.open_or_reaim(
            sw,
            origin="visual",
            reason=changes.co_reason(vis),
            vis=vis,
            waiting=waiting,
            staged=None,
            target_fp=fp.full,
            minor=change.is_minor,
        )
        sw.detail["changes"] = diff[:20]
    if res.fresh:
        run.counts["cos_opened"] += 1
    elif res.reaimed:
        run.counts["cos_reaimed"] += 1


# ---------------------------------------------------------------- flag writes (steps 7 and 8)


def _flag_write(
    run: _Run, packet_id: int, status: str, label: str, note: str | None, body: Callable[[SysWrite], bool]
) -> bool:
    """A flag or observation change on a packet: VIS_UPDATE_DRAFT on a draft, VIS_OPEN_CO past draft (the only
    system action open there). `body` returns False to write nothing."""
    action = A.VIS_UPDATE_DRAFT if status == "draft" else A.VIS_OPEN_CO
    with system_write(run.conn, packet_id, action, label=label, note=note) as sw:
        if not body(sw):
            sw.skip()
            return False
    return True


def _set_flags(
    conn: Any, packet_id: int, add: Collection[str] = (), remove: Collection[str] = (), extra_sql: str = ""
) -> None:
    row = conn.execute("SELECT flags FROM packets WHERE id = %s", (packet_id,)).fetchone()
    flags = [f for f in (row[0] or []) if f not in remove]
    for f in add:
        if f not in flags:
            flags.append(f)
    conn.execute(
        f"UPDATE packets SET flags = %s{extra_sql} WHERE id = %s",  # noqa: S608 (fixed fragments)
        (flags, packet_id),
    )


# ---------------------------------------------------------------- the apply


def apply(
    conn: Any,
    *,
    ctx: Any,
    batch: Any,
    feed_settings: Any,
    built: Built,
    orders: tuple[Mapping[str, list[Order]], Mapping[str, list[Order]]],
    labor: LaborMap | None,
    today: date,
    seen: Mapping[WoKey, str] | None = None,
    missing: Collection[WoKey] = (),
    paused: bool | None = None,
) -> Counts:
    """Apply one build. `seen` is every work order key this read returned with its status code (closed,
    cancelled, back in VISUAL); `missing` the tracked keys VISUAL lacks (work order gone)."""
    run = _Run(conn, ctx, batch, feed_settings, today)
    suspect = bool(getattr(batch, "suspect", False))
    run.counts["suspect"] = int(suspect)
    run.counts["unmapped_resources"] = sum(built.unmapped.values())
    if built.unmapped:
        run.counts["unmapped"] = _top(built.unmapped)
    run.counts["skipped_lots"] = sum(built.skipped_lots.values())
    if built.skipped_lots:
        run.counts["skipped_lots_detail"] = _top(built.skipped_lots)
    seen = dict(seen or {})
    with conn.transaction():
        conn.execute("SET LOCAL lock_timeout = '5s'")
        _apply(run, built, orders, labor, seen, set(missing), suspect, paused)
        _audit_system(conn, "feed.run", "job_run", getattr(ctx, "run_id", None), {"batch": run.batch_id, **run.counts})
        events.publish(conn, {"event": "feed", "run_id": getattr(ctx, "run_id", None), "batch": run.batch_id})
    return run.counts


def _apply(
    run: _Run,
    built: Built,
    orders: tuple[Mapping[str, list[Order]], Mapping[str, list[Order]]],
    labor: LaborMap | None,
    seen: Mapping[WoKey, str],
    missing: set[WoKey],
    suspect: bool,
    paused: bool | None,
) -> None:
    conn = run.conn
    by_base, by_project = orders
    golive = getattr(run.fs, "golive_date", None)
    depts = {r[0] for r in conn.execute("SELECT code FROM departments")}

    # Go-live boundary: rows the build left out, and any candidate with no tracked leg released before go-live.
    tracked_keys = {
        (r[0], r[1], r[2], r[3])
        for r in conn.execute(
            "SELECT w.wo_base, w.lot, w.split, w.sub FROM packet_work_orders w JOIN packets p ON p.id = w.packet_id"
            " WHERE p.removed_at IS NULL AND p.status <> 'filed'"
        )
    }
    pulled = {
        (r[0], r[1], r[2], r[3])
        for r in conn.execute("SELECT wo_base, lot, split, sub FROM visual_ignored_wos WHERE pulled_in_at IS NOT NULL")
    }
    tracked_keys |= pulled
    ignored_rows: list[Row] = list(built.ignored)
    cands: list[Candidate] = []
    for c in built.candidates.values():
        if (
            golive is not None
            and c.rel is not None
            and c.rel < golive
            and not any(lg.key.as_tuple() in tracked_keys for lg in c.legs)
        ):
            ignored_rows.extend(
                Row(
                    lg.key.base,
                    lg.key.lot,
                    lg.key.split,
                    lg.key.sub,
                    lg.seq,
                    lg.op,
                    lg.dwg,
                    lg.rev,
                    "",
                    "",
                    lg.qty,
                    lg.tgt,
                    c.rel,
                    c.want,
                    lg.st,
                )
                for lg in c.legs
            )
            continue
        if c.dept not in depts:
            run.error(f"candidate {c.key}", f"department {c.dept} does not exist")
            continue
        cands.append(c)
    if ignored_rows:
        run.counts["ignored_golive"] = run.guarded("go-live list", _upsert_ignored, run, ignored_rows) or 0

    # 1. Projects and units, then 2. match, 3-5 per candidate.
    by_proj: dict[str, list[Candidate]] = {}
    for c in cands:
        by_proj.setdefault(c.project, []).append(c)
    live_keys: set[str] = set()
    for number, group in sorted(by_proj.items()):
        if not RX_PROJECT.match(number):
            run.counts["skipped_conflict"] += len(group)
            continue
        due = max((c.want for c in group if c.want), default=None)
        project = run.guarded(f"project {number}", _ensure_project, run, number, due)
        if project is None:
            continue
        if str(project["status"]) == "closed":
            run.counts["skipped_closed"] += len(group)
            continue
        pchosen = orders_mod.choose(
            unit_number="",
            base=number,
            by_base=by_base,
            by_project=by_project,
            project=number,
            pin=project.get("order_pin"),
        )
        run.guarded(f"project {number} order", _derive, run, "projects", project, pchosen)
        project = _one(conn, "SELECT * FROM projects WHERE id = %s", (project["id"],)) or project
        by_unit: dict[str, list[Candidate]] = {}
        for c in group:
            by_unit.setdefault(c.unit, []).append(c)
        for unit_no, ucands in sorted(by_unit.items()):
            udue = max((c.want for c in ucands if c.want), default=None)
            unit = run.guarded(
                f"unit {unit_no}",
                _ensure_unit,
                run,
                project,
                unit_no,
                ucands[0].lot,
                udue,
            )
            if unit is None:
                continue
            if str(unit["status"]) == "closed":
                run.counts["skipped_closed"] += len(ucands)
                continue
            uchosen = orders_mod.choose(
                unit_number=unit_no,
                base=ucands[0].base,
                by_base=by_base,
                by_project=by_project,
                project=number,
                pin=unit.get("order_pin"),
            )
            ref = run.guarded(f"unit {unit_no} order", _derive, run, "units", unit, uchosen)
            if ref is None:
                ref = unit.get("order_ref")
            _match_and_apply(run, project, unit, ucands, ref, labor, live_keys)

    # 6. Retire drafts whose work orders moved.
    _retire_moved(run, built, live_keys)
    # 7. Work orders gone.
    if not suspect:
        _wo_gone(run, missing)
    _wo_back(run, seen)
    # 8. Closed and cancelled observed.
    _observe(run, seen, labor)
    # 9. Auto-release.
    is_paused = bool(getattr(run.fs, "paused", False)) if paused is None else paused
    if getattr(run.fs, "auto_release", False) and not is_paused:
        _auto_release(run)


def _upsert_ignored(run: _Run, rows: list[Row]) -> int:
    keys: dict[WoKey, Row] = {}
    for r in rows:
        keys.setdefault(r.key, r)
    for k, r in keys.items():
        project = project_of(r.wo)
        run.conn.execute(
            "INSERT INTO visual_ignored_wos (wo_base, lot, split, sub, rel, project, unit_number, res, dwg,"
            " first_seen_batch) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
            " ON CONFLICT (wo_base, lot, split, sub) DO UPDATE SET rel = EXCLUDED.rel, res = EXCLUDED.res,"
            " dwg = EXCLUDED.dwg, last_seen_at = now()",
            (k.base, k.lot, k.split, k.sub, r.rel, project, unit_number(project, r.lot), r.res, r.dwg, run.batch_id),
        )
    return len(keys)


def _match_and_apply(
    run: _Run,
    project: Mapping[str, Any],
    unit: Mapping[str, Any],
    cands: list[Candidate],
    ref: Any,
    labor: LaborMap | None,
    live_keys: set[str],
) -> None:
    conn = run.conn
    with conn.cursor(row_factory=dict_row) as cur:
        found = cur.execute(
            "SELECT id, visual_key, status::text AS status, removed_at, visual, visual_fp_applied, source_kind::text"
            " AS source_kind FROM packets WHERE visual_key = ANY(%s) ORDER BY id",
            ([c.key for c in cands],),
        ).fetchall()
    live: dict[str, dict[str, Any]] = {}
    removed: set[str] = set()
    for r in found:
        if r["removed_at"] is None:
            live[r["visual_key"]] = dict(r)
        else:
            removed.add(r["visual_key"])
    for c in cands:
        p = live.get(c.key)
        if p is None:
            if c.key in removed:
                run.counts["skipped_removed"] += 1
                continue
            if _on_hand_packet(run.conn, c):
                run.counts["skipped_hand"] += 1
                continue
            pid = run.guarded(f"new draft {c.key}", _create, run, c, unit, project, ref, labor)
            if pid is not None:
                live_keys.add(c.key)
            continue
        live_keys.add(c.key)
        if p["source_kind"] != "visual" or p["status"] == "filed":
            continue
        if p["status"] == "draft":
            run.guarded(f"update draft {c.key}", _update_draft, run, p, c)
        else:
            run.guarded(f"change order {c.key}", _past_draft, run, p, c)


def _on_hand_packet(conn: Any, c: Candidate) -> bool:
    """The old visMatch "exists" rule: every leg's work order (base, lot, split, sub) already sits, with the same
    operation, on one live packet made by hand. Such a candidate is not created."""
    keys = sorted({lg.key.as_tuple() for lg in c.legs})
    if not keys:
        return False
    row = conn.execute(
        "SELECT p.id FROM packets p JOIN packet_work_orders w ON w.packet_id = p.id"
        " JOIN unnest(%s::text[], %s::text[], %s::text[], %s::text[]) AS k(b, l, sp, su)"
        " ON w.wo_base = k.b AND w.lot = k.l AND w.split = k.sp AND w.sub = k.su"
        " WHERE p.removed_at IS NULL AND p.source_kind = 'hand' AND upper(coalesce(w.op, p.op, '')) = upper(%s)"
        " GROUP BY p.id HAVING count(DISTINCT (w.wo_base, w.lot, w.split, w.sub)) = %s LIMIT 1",
        ([k[0] for k in keys], [k[1] for k in keys], [k[2] for k in keys], [k[3] for k in keys], c.op, len(keys)),
    ).fetchone()
    return row is not None


def _retire_moved(run: _Run, built: Built, live_keys: set[str]) -> None:
    owner: dict[tuple[str, str, str, str, int], str] = {}
    for c in built.candidates.values():
        if c.key not in live_keys:
            continue
        for lg in c.legs:
            owner[(*lg.key.as_tuple(), lg.seq)] = c.key
    if not owner:
        return
    with run.conn.cursor(row_factory=dict_row) as cur:
        drafts = cur.execute(
            "SELECT p.id, p.visual_key, p.visual FROM packets p WHERE p.removed_at IS NULL AND p.status = 'draft'"
            " AND p.source_kind = 'visual' AND NOT (p.visual_key = ANY(%s))"
            " AND NOT (p.visual ? 'manual_dwg') AND NOT (p.visual ? 'manual_bom') AND NOT (p.visual ? 'moved')"
            " AND NOT EXISTS (SELECT 1 FROM packet_pages g WHERE g.packet_id = p.id AND g.type = 'drawing')"
            " AND EXISTS (SELECT 1 FROM packet_work_orders w WHERE w.packet_id = p.id)",
            (sorted(built.candidates),),
        ).fetchall()
    for d in drafts:
        with run.conn.cursor(row_factory=dict_row) as cur:
            legs = cur.execute(
                "SELECT wo_base, lot, split, sub, seq FROM packet_work_orders WHERE packet_id = %s", (d["id"],)
            ).fetchall()
        to = [owner.get((w["wo_base"], w["lot"], w["split"], w["sub"], w["seq"])) for w in legs]
        if any(t is None or t == d["visual_key"] for t in to):
            continue
        dest = sorted({t for t in to if t})
        drawings = sorted({t.split("|", 2)[2] for t in dest})

        def retire(
            d: Mapping[str, Any] = d,
            dest: list[str] = dest,
            drawings: list[str] = drawings,
            legs: list[dict[str, Any]] = legs,
        ) -> None:
            note = (
                f"{', '.join(sorted({w['wo_base'] for w in legs}))} now on drawing {', '.join(drawings)} in VISUAL"
                f" ({', '.join(dest)}). Kept for the record, hidden from the boards."
            )
            with system_write(run.conn, int(d["id"]), A.VIS_RETIRE_DRAFT, note=note) as sw:
                visual = dict(sw.packet.get("visual") or {})
                visual["moved"] = {"to": dest, "drawing": ", ".join(drawings), "at": run.now.isoformat()}
                sw.conn.execute(
                    "UPDATE packets SET removed_at = now(), removed_by = NULL, removed_reason = 'moved', visual = %s"
                    " WHERE id = %s",
                    (Jsonb(packets_jsonable(visual)), sw.id),
                )
                sw.changed("removed_at", "visual")
            run.counts["retired"] += 1

        run.guarded(f"retire {d['visual_key']}", retire)


def _key_arrays(keys: Iterable[WoKey]) -> tuple[list[str], list[str], list[str], list[str]]:
    ks = list(dict.fromkeys(keys))
    return [k.base for k in ks], [k.lot for k in ks], [k.split for k in ks], [k.sub for k in ks]


_UNNEST = "unnest(%s::text[], %s::text[], %s::text[], %s::text[]) AS k(b, l, sp, su)"
_MATCH = "w.wo_base = k.b AND w.lot = k.l AND w.split = k.sp AND w.sub = k.su"
_LIVE = "p.removed_at IS NULL AND p.status <> 'filed' AND p.source_kind = 'visual'"


def _wo_gone(run: _Run, missing: set[WoKey]) -> None:
    if not missing:
        return
    conn = run.conn
    with conn.cursor(row_factory=dict_row) as cur:
        hit = cur.execute(
            f"UPDATE packet_work_orders w SET missing_runs = missing_runs + 1 FROM packets p, {_UNNEST}"  # noqa: S608
            f" WHERE p.id = w.packet_id AND {_LIVE} AND {_MATCH}"
            " RETURNING w.id, w.packet_id, w.wo_base, w.lot, w.split, w.sub, w.missing_runs, w.gone_at,"
            " p.status::text AS status",
            _key_arrays(missing),
        ).fetchall()
    newly: dict[int, list[dict[str, Any]]] = {}
    for h in hit:
        if h["missing_runs"] >= 2 and h["gone_at"] is None:
            newly.setdefault(int(h["packet_id"]), []).append(dict(h))
    for pid, legs in newly.items():
        keys = sorted({f"{h['wo_base']}/{h['lot']}/{h['split']}/{h['sub']}" for h in legs})

        def body(sw: SysWrite, legs: list[dict[str, Any]] = legs) -> bool:
            sw.conn.execute(
                "UPDATE packet_work_orders SET gone_at = now() WHERE id = ANY(%s) AND gone_at IS NULL",
                ([h["id"] for h in legs],),
            )
            _set_flags(sw.conn, sw.id, add=("wo_gone",), extra_sql=", wo_gone_at = coalesce(wo_gone_at, now())")
            sw.changed("flags")
            return True

        label = f"Work order {', '.join(keys)} no longer in VISUAL"
        if run.guarded(
            f"wo gone {pid}",
            _flag_write,
            run,
            pid,
            legs[0]["status"],
            label,
            None,
            body,
        ):
            run.counts["wo_gone_flagged"] += 1


def _wo_back(run: _Run, seen: Mapping[WoKey, str]) -> None:
    if not seen:
        return
    conn = run.conn
    with conn.cursor(row_factory=dict_row) as cur:
        back = cur.execute(
            f"WITH b AS (SELECT w.id, w.gone_at AS was_gone FROM packet_work_orders w, {_UNNEST}"  # noqa: S608
            f" WHERE {_MATCH} AND (w.missing_runs > 0 OR w.gone_at IS NOT NULL) FOR UPDATE OF w)"
            " UPDATE packet_work_orders w SET missing_runs = 0, gone_at = NULL FROM b WHERE w.id = b.id"
            " RETURNING w.packet_id, b.was_gone",
            _key_arrays(seen),
        ).fetchall()
    packets_back = sorted({int(r["packet_id"]) for r in back if r["was_gone"] is not None})
    for pid in packets_back:
        row = _one(
            conn,
            "SELECT p.status::text AS status, p.flags, p.removed_at,"
            " EXISTS (SELECT 1 FROM packet_work_orders w WHERE w.packet_id = p.id AND w.gone_at IS NOT NULL) AS still"
            " FROM packets p WHERE p.id = %s",
            (pid,),
        )
        if row is None or row["removed_at"] is not None or row["status"] == "filed" or row["still"]:
            continue
        if "wo_gone" not in (row["flags"] or []):
            continue

        def body(sw: SysWrite) -> bool:
            _set_flags(sw.conn, sw.id, remove=("wo_gone",), extra_sql=", wo_gone_at = NULL")
            sw.changed("flags")
            return True

        if run.guarded(
            f"wo back {pid}",
            _flag_write,
            run,
            pid,
            row["status"],
            "Work order back in VISUAL",
            None,
            body,
        ):
            run.counts["wo_back"] += 1


def _observe(run: _Run, seen: Mapping[WoKey, str], labor: LaborMap | None) -> None:
    """Per leg visual_status, packet_visual_state, closed in VISUAL, cancelled flag."""
    if not seen:
        return
    conn = run.conn
    with conn.cursor(row_factory=dict_row) as cur:
        legs = cur.execute(
            "SELECT w.id, w.packet_id, w.wo_base, w.lot, w.split, w.sub, w.seq, w.visual_status,"  # noqa: S608
            " p.status::text AS status, p.flags, p.visual_closed_at, p.vclosed_at FROM packet_work_orders w"
            f" JOIN packets p ON p.id = w.packet_id WHERE {_LIVE} AND w.packet_id IN (SELECT w2.packet_id"
            f" FROM packet_work_orders w2, {_UNNEST.replace('k(', 'k2(')} WHERE w2.wo_base = k2.b AND w2.lot = k2.l"
            " AND w2.split = k2.sp AND w2.sub = k2.su) ORDER BY w.packet_id, w.sort, w.id",
            _key_arrays(seen),
        ).fetchall()
    per: dict[int, list[dict[str, Any]]] = {}
    for lg in legs:
        per.setdefault(int(lg["packet_id"]), []).append(dict(lg))
    for pid, pl in per.items():
        sts: list[str] = []
        state_legs: list[dict[str, Any]] = []
        for lg in pl:
            k = WoKey(lg["wo_base"], lg["lot"], lg["split"], lg["sub"])
            st = seen.get(k, lg["visual_status"] or "")
            sts.append(st)
            state_legs.append({"key": str(k), "seq": lg["seq"], "st": st})
            if st and st != (lg["visual_status"] or ""):
                conn.execute("UPDATE packet_work_orders SET visual_status = %s WHERE id = %s", (st, lg["id"]))
        hours = _hours(
            labor,
            (
                (WoKey(lg["wo_base"], lg["lot"], lg["split"], lg["sub"]), int(lg["seq"]))
                for lg in pl
                if lg["seq"] is not None
            ),
        )
        run.guarded(f"observe {pid}", _visual_state, conn, pid, state_legs, hours, None)
        head = pl[0]
        status = head["status"]
        flags = list(head["flags"] or [])
        # C5: the feed is the one writer of vclosed on a VISUAL packet. Every leg closed (or cancelled) sets it, and
        # the Clerical board moves the packet to "Ready for data book"; a leg seen open again clears it.
        done = bool(sts) and all(s in packets_mod.LEG_DONE for s in sts)
        all_closed = bool(sts) and all(s == "C" for s in sts)
        if (done and head["vclosed_at"] is None) or (all_closed and head["visual_closed_at"] is None):

            def closed(sw: SysWrite, all_closed: bool = all_closed) -> bool:
                sw.conn.execute(
                    "UPDATE packets SET vclosed_at = coalesce(vclosed_at, now()), vclosed_by = CASE WHEN vclosed_at"
                    " IS NULL THEN NULL ELSE vclosed_by END, visual_closed_at = CASE WHEN %s THEN"
                    " coalesce(visual_closed_at, now()) ELSE visual_closed_at END WHERE id = %s",
                    (all_closed, sw.id),
                )
                sw.changed("vclosed", "visual_closed_at")
                return True

            if run.guarded(
                f"closed {pid}",
                _flag_write,
                run,
                pid,
                status,
                "Closed in VISUAL (observed)",
                None,
                closed,
            ):
                run.counts["closed_observed"] += 1
        elif not done and any(s and s not in packets_mod.LEG_DONE for s in sts) and head["vclosed_at"] is not None:

            def reopened(sw: SysWrite) -> bool:
                sw.conn.execute(
                    "UPDATE packets SET vclosed_at = NULL, vclosed_by = NULL, visual_closed_at = NULL WHERE id = %s",
                    (sw.id,),
                )
                sw.changed("vclosed", "visual_closed_at")
                return True

            run.guarded(
                f"reopened {pid}",
                _flag_write,
                run,
                pid,
                status,
                "Open again in VISUAL (observed)",
                ", ".join(packets_mod.open_wos({"legs": state_legs})) or None,
                reopened,
            )
        cancelled = any(s == "X" for s in sts)
        if cancelled and "cancelled" not in flags:

            def cancel(sw: SysWrite) -> bool:
                _set_flags(
                    sw.conn,
                    sw.id,
                    add=("cancelled",),
                    extra_sql=", visual_cancelled_at = coalesce(visual_cancelled_at, now())",
                )
                sw.changed("flags")
                return True

            if run.guarded(
                f"cancelled {pid}",
                _flag_write,
                run,
                pid,
                status,
                "Work order cancelled in VISUAL",
                None,
                cancel,
            ):
                run.counts["cancelled_flagged"] += 1
        elif not cancelled and "cancelled" in flags:

            def uncancel(sw: SysWrite) -> bool:
                _set_flags(sw.conn, sw.id, remove=("cancelled",), extra_sql=", visual_cancelled_at = NULL")
                sw.changed("flags")
                return True

            run.guarded(
                f"uncancelled {pid}",
                _flag_write,
                run,
                pid,
                status,
                "Work order no longer cancelled in VISUAL",
                None,
                uncancel,
            )


HEAT_SQL = (
    "EXISTS (SELECT 1 FROM bom_lines b WHERE b.packet_id = packets.id AND coalesce(b.sub, '') <> 'Y' AND NOT b.na"
    " AND coalesce(b.cons, '') <> 'Y')"
)


def _auto_release(run: _Run) -> None:
    limit = int(getattr(run.fs, "auto_release_max", 25) or 0)
    if limit <= 0:
        return
    ids = [
        int(r[0])
        for r in run.conn.execute(
            "SELECT id FROM packets WHERE removed_at IS NULL AND status = 'draft' AND source_kind = 'visual'"
            " AND drawing_state = 'attached' AND flags = '{}' AND coalesce((visual->>'rev_confirmed')::boolean, false)"
            " ORDER BY id LIMIT %s",
            (limit,),
        )
    ]
    for pid in ids:

        def release(pid: int = pid) -> None:
            conn = run.conn
            hm = conn.execute(
                f"UPDATE packets SET heat_map = coalesce(heat_map, {HEAT_SQL}) WHERE id = %s"  # noqa: S608
                " RETURNING heat_map, drawing, rev",
                (pid,),
            ).fetchone()
            if hm is None:
                return
            note = f"no flags · drawing {hm[1] or '-'} rev {hm[2] or '-'} confirmed in M-Files · heat map " + (
                "required (material lines)" if hm[0] else "not required (sub-assemblies only)"
            )
            with system_write(conn, pid, A.VIS_AUTO_RELEASE, note=note):
                pass
            run.counts["auto_released"] += 1

        run.guarded(f"auto-release {pid}", release)


__all__ = ["COUNT_KEYS", "DRAWING_FLAGS", "HEAT_SQL", "apply"]