"""Projects and units: numbers, lots, aliases, renumbering, field CAS, soft delete and packet counts.

Ports the old projects.js rules. A unit's number is its project's number plus its lot. Every old number
of a project or unit stays in project_aliases, so a lookup by any number still finds it. Packets point at
units by id, so a renumber never touches a packet. Numbers and lots are text: never cast them.
"""

from __future__ import annotations

import re
from collections.abc import Iterable, Mapping
from dataclasses import dataclass
from datetime import UTC, date, datetime
from typing import Any

from psycopg import errors as pgerr
from psycopg.rows import dict_row

from .. import audit, events
from ..api.errors import ApiError
from ..text import LIMITS, RX_LOT, RX_PROJECT, clean_opt, clean_text

STATUSES = ("open", "hold", "closed")
LIST_FILTERS = ("active", "open", "hold", "closed", "all")
SHOP_STATUSES = ("ready", "assigned", "in_progress", "review")
COUNT_KEYS = ("draft", "warehouse", "shop", "qc", "complete", "filed", "total")
MAX_UNITS_AT_ONCE = 200
MAX_WORDS = 8
LIST_LIMIT = 500

PROJECT_FIELDS = ("name", "customer", "po", "due", "status", "notes", "order_pin", "order_pin_reason")
UNIT_FIELDS = ("name", "customer", "po", "due", "status", "notes", "order_pin", "order_pin_reason")
# Order-derived fields: the value in use, where it came from (visual | hand) and what VISUAL last said.
ORDER_FIELDS = ("customer", "po", "order_date", "ship_date")
RX_ORDER_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9 ./_-]{0,39}$")

_DIGITS = re.compile(r"^[0-9]+$")
_ISO_DATE = re.compile(r"^\d{4}-\d{2}-\d{2}$")

COUNTS_SQL = (
    "count(*) FILTER (WHERE k.status = 'draft') AS c_draft,"
    " count(*) FILTER (WHERE k.status = 'warehouse') AS c_warehouse,"
    " count(*) FILTER (WHERE k.status IN ('ready', 'assigned', 'in_progress', 'review')) AS c_shop,"
    " count(*) FILTER (WHERE k.status = 'qc') AS c_qc,"
    " count(*) FILTER (WHERE k.status = 'complete') AS c_complete,"
    " count(*) FILTER (WHERE k.status = 'filed') AS c_filed,"
    " count(*) AS c_total"
)

PROJECT_SELECT = (
    "SELECT p.*,"  # noqa: S608
    " (SELECT count(*) FROM units u WHERE u.project_id = p.id AND u.removed_at IS NULL) AS unit_count,"
    " c.*"
    " FROM projects p"
    " LEFT JOIN LATERAL (SELECT " + COUNTS_SQL + " FROM packets k JOIN units u ON u.id = k.unit_id"
    " WHERE u.project_id = p.id AND k.removed_at IS NULL) c ON true"
)

UNIT_SELECT = (
    "SELECT v.*, c.* FROM v_units v"  # noqa: S608
    " LEFT JOIN LATERAL (SELECT " + COUNTS_SQL + " FROM packets k"
    " WHERE k.unit_id = v.id AND k.removed_at IS NULL) c ON true"
)


@dataclass(frozen=True)
class Who:
    """Who is writing, for the audit row. it_reason is set only when an admin acts outside their roles."""

    user_id: Any
    session_id: Any = None
    role_snapshot: str | None = None
    ip: str | None = None
    it_reason: str | None = None


# ---------------------------------------------------------------- normalisation


def normalize_number(raw: Any) -> str:
    """A project number: cleaned, upper-cased, RX_PROJECT. 422 invalid_number otherwise."""
    n = clean_text(raw, 200).upper()
    if not RX_PROJECT.match(n):
        raise ApiError(
            422,
            "invalid_number",
            "A project number is 1 to 20 letters, digits or dashes, starting with a letter or digit.",
        )
    return n


def normalize_lot(raw: Any) -> str:
    """A unit lot: upper-cased, digits zero-padded to 3. `"7"` -> `"007"`, `"7a"` -> `"7A"`. 422 invalid_lot."""
    lot = clean_text(raw, 40).upper()
    if _DIGITS.match(lot) and len(lot) < 3:
        lot = lot.zfill(3)
    if not RX_LOT.match(lot) or (_DIGITS.match(lot) and len(lot) != 3):
        raise ApiError(422, "invalid_lot", f"Lot {lot or '(blank)'}: 1 to 3 letters or digits, numbers as 001 to 999.")
    return lot


def lot_input(project_number: str, raw: Any) -> str:
    """A lot as typed: a full unit number that starts with the project number is cut down to its lot."""
    text = clean_text(raw, 40).upper()
    pn = project_number.upper()
    if len(text) > 3 and text.startswith(pn) and len(text) > len(pn):
        text = text[len(pn) :]
    return normalize_lot(text)


def parse_due(raw: Any) -> date | None:
    if raw is None or (isinstance(raw, str) and not raw.strip()):
        return None
    if isinstance(raw, date):
        return raw
    text = str(raw).strip()
    try:
        if not _ISO_DATE.match(text):
            raise ValueError(text)
        return date.fromisoformat(text)
    except ValueError:
        raise ApiError(422, "invalid_date", "Give the due date as YYYY-MM-DD.") from None


def parse_status(raw: Any) -> str:
    s = clean_text(raw, 20).lower()
    if s not in STATUSES:
        raise ApiError(422, "invalid_status", "Status is open, hold or closed.")
    return s


def clean_reason(raw: Any) -> str:
    text = clean_text(raw, LIMITS["reason"])
    if not text:
        raise ApiError(400, "reason_required", "Give a reason.")
    return text


def _like(word: str) -> str:
    return "%" + word.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + "%"


def _words(q: str | None) -> list[str]:
    return [w for w in clean_text(q, 200).split() if w][:MAX_WORDS]


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


def _counts(r: Mapping[str, Any]) -> dict[str, int]:
    return {k: int(r.get("c_" + k) or 0) for k in COUNT_KEYS}


def _removed(r: Mapping[str, Any]) -> dict[str, Any] | None:
    if r.get("removed_at") is None:
        return None
    by = r.get("removed_by")
    return {"at": r["removed_at"], "by": str(by) if by else None, "reason": r.get("removed_reason")}


# ---------------------------------------------------------------- shapes


def _order_out(r: Mapping[str, Any]) -> dict[str, Any]:
    """The order-derived columns with their provenance, and the order pin. Shared by projects and units."""
    pin_by = r.get("order_pin_by")
    return {
        "customer_src": r.get("customer_src"),
        "customer_visual": r.get("customer_visual"),
        "po_src": r.get("po_src"),
        "po_visual": r.get("po_visual"),
        "order_date": r.get("order_date"),
        "order_date_src": r.get("order_date_src"),
        "order_date_visual": r.get("order_date_visual"),
        "ship_date": r.get("ship_date"),
        "ship_date_src": r.get("ship_date_src"),
        "ship_date_visual": r.get("ship_date_visual"),
        "order_ref": r.get("order_ref"),
        "order_pin": r.get("order_pin"),
        "order_pin_by": str(pin_by) if pin_by else None,
        "order_pin_at": r.get("order_pin_at"),
        "order_pin_reason": r.get("order_pin_reason"),
    }


def project_out(r: Mapping[str, Any]) -> dict[str, Any]:
    return {
        "id": r["id"],
        "number": r["number"],
        "name": r["name"],
        "customer": r["customer"],
        "po": r["po"],
        "due": r["due"],
        "status": r["status"],
        "notes": r["notes"],
        "source": r["source"],
        "needs_details": r["needs_details"],
        **_order_out(r),
        "units": int(r.get("unit_count") or 0),
        "packets": _counts(r),
        "version": r["version"],
        "created_at": r["created_at"],
        "updated_at": r["updated_at"],
        "removed": _removed(r),
    }


def unit_out(r: Mapping[str, Any]) -> dict[str, Any]:
    status = r["status"]
    return {
        "id": r["id"],
        "project_id": r["project_id"],
        "number": r["number"],
        "lot": r["lot"],
        "name": r["name"],
        "due": r["due"],
        "due_effective": r["due_effective"],
        "status": status,
        "status_effective": "closed" if r["project_status"] == "closed" else status,
        "notes": r["notes"],
        "customer": r.get("customer"),
        "po": r.get("po"),
        **_order_out(r),
        "packets": _counts(r),
        "version": r["version"],
        "created_at": r["created_at"],
        "updated_at": r["updated_at"],
        "removed": _removed(r),
    }


# ---------------------------------------------------------------- lookups


def owners(conn: Any, number: str) -> list[dict[str, Any]]:
    """Everything that answers to `number`: live project and unit numbers first, then aliases of live ones."""
    n = clean_text(number, 40).upper()
    if not n:
        return []
    with conn.cursor(row_factory=dict_row) as cur:
        rows: list[dict[str, Any]] = cur.execute(
            "SELECT * FROM ("
            " SELECT 0 AS rank, 'project' AS kind, p.id, p.id AS project_id, p.number, false AS alias"
            "  FROM projects p WHERE upper(p.number) = %(n)s AND p.removed_at IS NULL"
            " UNION ALL"
            " SELECT 1, 'unit', u.id, u.project_id, p.number || u.lot, false"
            "  FROM units u JOIN projects p ON p.id = u.project_id"
            "  WHERE upper(p.number || u.lot) = %(n)s AND u.removed_at IS NULL AND p.removed_at IS NULL"
            " UNION ALL"
            " SELECT 2, CASE WHEN a.unit_id IS NULL THEN 'project' ELSE 'unit' END, coalesce(a.unit_id, a.project_id),"
            "  a.project_id, CASE WHEN a.unit_id IS NULL THEN p.number ELSE p.number || u.lot END, true"
            "  FROM project_aliases a JOIN projects p ON p.id = a.project_id AND p.removed_at IS NULL"
            "  LEFT JOIN units u ON u.id = a.unit_id"
            "  WHERE a.alias = %(n)s AND (a.unit_id IS NULL OR u.removed_at IS NULL)"
            ") x ORDER BY rank, id",
            {"n": n},
        ).fetchall()
    return rows


def lookup(conn: Any, number: str) -> dict[str, Any] | None:
    """`{kind, id, project_id, number, alias}` for a current or old number, or None."""
    found = owners(conn, number)
    if not found:
        return None
    o = found[0]
    return {"kind": o["kind"], "id": o["id"], "project_id": o["project_id"], "number": o["number"], "alias": o["alias"]}


def _taken(o: Mapping[str, Any]) -> ApiError:
    if o["alias"]:
        msg = f"That number is an old number of {o['number']}. Pick another number."
    elif o["kind"] == "unit":
        msg = f"{o['number']} already exists as a unit."
    else:
        msg = f"{o['number']} already exists as a project."
    return ApiError(409, "number_taken", msg, extra={"taken": {"kind": o["kind"], "id": o["id"], "alias": o["alias"]}})


def assert_free(conn: Any, number: str, *, own_project: int | None = None, own_unit: int | None = None) -> None:
    """409 number_taken unless nothing answers to `number`, or only the entity's own old number does."""
    for o in owners(conn, number):
        if o["alias"] and own_unit is not None and o["kind"] == "unit" and o["id"] == own_unit:
            continue
        if o["alias"] and own_project is not None and o["kind"] == "project" and o["id"] == own_project:
            continue
        raise _taken(o)


def next_lot(conn: Any, project_id: int, number: str) -> str | None:
    """The old nextUnitNumber: highest numeric lot plus one, skipping numbers already used (old ones too)."""
    rows = conn.execute(
        "SELECT lot FROM units WHERE project_id = %s AND removed_at IS NULL AND lot ~ '^[0-9]{3}$'", (project_id,)
    ).fetchall()
    n = max((int(r[0]) for r in rows), default=0) + 1
    while n <= 999:
        lot = f"{n:03d}"
        if not owners(conn, number + lot):
            return lot
        n += 1
    return None


def _unique_taken(e: Exception) -> ApiError:
    return ApiError(409, "number_taken", "That number was just taken. Pick another number.")


# ---------------------------------------------------------------- reads


def get_project(conn: Any, project_id: int, *, removed: bool = False) -> dict[str, Any] | None:
    with conn.cursor(row_factory=dict_row) as cur:
        r = cur.execute(PROJECT_SELECT + " WHERE p.id = %s", (project_id,)).fetchone()
    if r is None or (r["removed_at"] is not None and not removed):
        return None
    return project_out(r)


def _project_or_404(conn: Any, project_id: int) -> dict[str, Any]:
    out = get_project(conn, project_id)
    if out is None:
        raise ApiError(404, "not_found", "No such project.")
    return out


def aliases_of(conn: Any, *, project_id: int | None = None, unit_id: int | None = None) -> list[dict[str, Any]]:
    with conn.cursor(row_factory=dict_row) as cur:
        if unit_id is not None:
            rows = cur.execute(
                "SELECT alias, kind, unit_id, created_at FROM project_aliases WHERE unit_id = %s"
                " ORDER BY created_at, alias",
                (unit_id,),
            ).fetchall()
        else:
            rows = cur.execute(
                "SELECT alias, kind, unit_id, created_at FROM project_aliases WHERE project_id = %s"
                " ORDER BY created_at, alias",
                (project_id,),
            ).fetchall()
    return [dict(r) for r in rows]


def project_detail(conn: Any, project_id: int) -> dict[str, Any]:
    out = _project_or_404(conn, project_id)
    with conn.cursor(row_factory=dict_row) as cur:
        rows = cur.execute(
            UNIT_SELECT + " WHERE v.project_id = %s AND v.removed_at IS NULL ORDER BY v.lot", (project_id,)
        ).fetchall()
    out["units"] = [unit_out(r) for r in rows]
    out["unit_count"] = len(rows)
    out["aliases"] = aliases_of(conn, project_id=project_id)
    out["next_lot"] = next_lot(conn, project_id, out["number"])
    out["orders"] = cached_orders(conn, out["number"], _project_bases(conn, project_id))
    return out


def list_projects(conn: Any, *, q: str | None = None, status: str = "active") -> list[dict[str, Any]]:
    if status not in LIST_FILTERS:
        raise ApiError(422, "invalid_status", "Status filter is active, open, hold, closed or all.")
    where = ["p.removed_at IS NULL"]
    params: list[Any] = []
    if status == "active":
        where.append("p.status <> 'closed'")
    elif status != "all":
        where.append("p.status = %s")
        params.append(status)
    for w in _words(q):
        where.append(
            "(p.number ILIKE %s OR p.name ILIKE %s OR p.customer ILIKE %s OR p.po ILIKE %s"
            " OR EXISTS (SELECT 1 FROM project_aliases a WHERE a.project_id = p.id AND a.alias ILIKE %s))"
        )
        pat = _like(w)
        params.extend([pat, pat, pat, pat, pat])
    sql = PROJECT_SELECT + " WHERE " + " AND ".join(where) + f" ORDER BY p.number LIMIT {LIST_LIMIT}"
    with conn.cursor(row_factory=dict_row) as cur:
        rows = cur.execute(sql, params).fetchall()
    return [project_out(r) for r in rows]


def get_unit(conn: Any, unit_id: int, *, removed: bool = False) -> dict[str, Any] | None:
    with conn.cursor(row_factory=dict_row) as cur:
        r = cur.execute(UNIT_SELECT + " WHERE v.id = %s", (unit_id,)).fetchone()
    if r is None or (r["removed_at"] is not None and not removed):
        return None
    return unit_out(r)


def _unit_or_404(conn: Any, unit_id: int) -> dict[str, Any]:
    out = get_unit(conn, unit_id)
    if out is None:
        raise ApiError(404, "not_found", "No such unit.")
    return out


def unit_detail(conn: Any, unit_id: int) -> dict[str, Any]:
    out = _unit_or_404(conn, unit_id)
    p = get_project(conn, out["project_id"])
    if p is None:
        raise ApiError(404, "not_found", "No such unit.")
    out["project"] = {k: p[k] for k in ("id", "number", "name", "customer", "po", "due", "status", "version")}
    out["aliases"] = aliases_of(conn, unit_id=unit_id)
    out["orders"] = cached_orders(conn, p["number"], _unit_bases(conn, unit_id))
    return out


def list_units(
    conn: Any, *, project_id: int | None = None, q: str | None = None, status: str = "active"
) -> list[dict[str, Any]]:
    """Units for the picker's second step. Status applies to the effective status (a closed project closes all)."""
    if status not in LIST_FILTERS:
        raise ApiError(422, "invalid_status", "Status filter is active, open, hold, closed or all.")
    eff = "(CASE WHEN v.project_status = 'closed' THEN 'closed' ELSE v.status::text END)"
    where = [
        "v.removed_at IS NULL",
        "EXISTS (SELECT 1 FROM projects p WHERE p.id = v.project_id AND p.removed_at IS NULL)",
    ]
    params: list[Any] = []
    if project_id is not None:
        where.append("v.project_id = %s")
        params.append(project_id)
    if status == "active":
        where.append(f"{eff} <> 'closed'")
    elif status != "all":
        where.append(f"{eff} = %s")
        params.append(status)
    for w in _words(q):
        where.append(
            "(v.number ILIKE %s OR v.lot ILIKE %s OR v.name ILIKE %s"
            " OR EXISTS (SELECT 1 FROM project_aliases a WHERE a.unit_id = v.id AND a.alias ILIKE %s))"
        )
        pat = _like(w)
        params.extend([pat, pat, pat, pat])
    sql = UNIT_SELECT + " WHERE " + " AND ".join(where) + f" ORDER BY v.project_number, v.lot LIMIT {LIST_LIMIT}"
    with conn.cursor(row_factory=dict_row) as cur:
        rows = cur.execute(sql, params).fetchall()
    return [unit_out(r) for r in rows]


# ---------------------------------------------------------------- writes: helpers


def _audit(conn: Any, who: Who, action: str, subject_type: str, subject_id: Any, detail: Any) -> None:
    audit.append(
        conn,
        action=action,
        user_id=who.user_id,
        session_id=who.session_id,
        role_snapshot=who.role_snapshot,
        subject_type=subject_type,
        subject_id=subject_id,
        detail=detail,
        it_reason=who.it_reason,
        ip=who.ip,
    )


def _publish(conn: Any, project_id: int, unit_id: int | None = 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 _lock_project(conn: Any, project_id: int) -> dict[str, Any]:
    with conn.cursor(row_factory=dict_row) as cur:
        r = cur.execute(
            "SELECT * FROM projects WHERE id = %s AND removed_at IS NULL FOR UPDATE", (project_id,)
        ).fetchone()
    if r is None:
        raise ApiError(404, "not_found", "No such project.")
    return dict(r)


def _lock_unit(conn: Any, unit_id: int) -> tuple[dict[str, Any], dict[str, Any]]:
    """Lock the unit's project, then the unit (always in that order). Returns (unit, project)."""
    row = conn.execute("SELECT project_id FROM units WHERE id = %s AND removed_at IS NULL", (unit_id,)).fetchone()
    if row is None:
        raise ApiError(404, "not_found", "No such unit.")
    try:
        project = _lock_project(conn, int(row[0]))
    except ApiError:
        raise ApiError(404, "not_found", "No such unit.") from None
    with conn.cursor(row_factory=dict_row) as cur:
        u = cur.execute("SELECT * FROM units WHERE id = %s AND removed_at IS NULL FOR UPDATE", (unit_id,)).fetchone()
    if u is None:
        raise ApiError(404, "not_found", "No such unit.")
    return dict(u), project


def _cmp(v: Any) -> Any:
    if v is None:
        return None
    if isinstance(v, date):
        return v.isoformat()
    text = str(v).strip()
    return text or None


def _norm_field(key: str, value: Any, *, unit: bool) -> Any:
    if key == "due":
        return parse_due(value)
    if key == "status":
        return parse_status(value)
    if key == "name":
        return clean_opt(value, LIMITS["unit_name"] if unit else LIMITS["project_name"])
    if key == "customer":
        return clean_opt(value, LIMITS["customer"])
    if key == "po":
        return clean_opt(value, LIMITS["po"])
    if key == "notes":
        return clean_opt(value, LIMITS["notes"])
    if key == "order_pin":
        pin = clean_opt(value, LIMITS["order_pin"] + 20)
        if pin is not None and not RX_ORDER_ID.match(pin):
            raise ApiError(422, "invalid_order", "Give a VISUAL customer order ID, or nothing to clear the pin.")
        return pin
    if key == "order_pin_reason":
        return clean_opt(value, LIMITS["reason"])
    raise ApiError(422, "invalid_field", f"{key} cannot be changed here.")


def field_cas(
    row: Mapping[str, Any], values: Mapping[str, Any], base: Mapping[str, Any], fields: Iterable[str], *, unit: bool
) -> dict[str, Any]:
    """The field-level CAS: returns {key: new value} for the keys that change; 409 field_conflict on a stale base.

    A key whose base differs from the current value conflicts, unless the new value already equals the current one.
    Keys without a base entry are written without a check.
    """
    allowed = set(fields)
    bad = [k for k in values if k not in allowed]
    if bad:
        raise ApiError(422, "invalid_field", f"{', '.join(sorted(bad))} cannot be changed here.")
    new = {k: _norm_field(k, v, unit=unit) for k, v in values.items()}
    conflicts = {
        k: _iso(row[k]) for k in new if k in base and _cmp(base[k]) != _cmp(row[k]) and _cmp(new[k]) != _cmp(row[k])
    }
    if conflicts:
        raise ApiError(
            409,
            "field_conflict",
            "Someone else changed this since you opened it. Check the new values.",
            extra={"current": conflicts},
        )
    return {k: v for k, v in new.items() if _cmp(v) != _cmp(row[k])}


def _change_detail(row: Mapping[str, Any], changes: Mapping[str, Any]) -> dict[str, Any]:
    out: dict[str, Any] = {}
    for k, v in changes.items():
        out[k] = {"changed": True} if k == "notes" else {"from": _iso(row[k]), "to": _iso(v)}
    return out


def _columns(row: Mapping[str, Any], changes: Mapping[str, Any], who: Who) -> dict[str, Any]:
    """The columns a field edit writes: a hand customer or po marks its source, a pin records who and when."""
    cols = dict(changes)
    for k in ("customer", "po"):
        if k in cols:
            cols[f"{k}_src"] = "hand" if cols[k] else None
    if "order_pin" in cols:
        if cols["order_pin"] is None:
            cols.update(order_pin_by=None, order_pin_at=None, order_pin_reason=None)
        else:
            cols["order_pin_by"] = who.user_id
            cols["order_pin_at"] = datetime.now(UTC)
    return cols


def _check_pin(
    conn: Any, row: Mapping[str, Any], changes: Mapping[str, Any], *, project_number: str, bases: list[str]
) -> None:
    """A new pin needs a reason and, when VISUAL's orders are cached for this project, must name one of them."""
    if "order_pin" not in changes and "order_pin_reason" in changes and row.get("order_pin") is None:
        raise ApiError(422, "invalid_field", "There is no order pin to give a reason for.")
    pin = changes.get("order_pin")
    if pin is None:
        return
    if not changes.get("order_pin_reason"):
        raise ApiError(400, "reason_required", "Say why this order is pinned.")
    known = [o["cust_order_id"] for o in cached_orders(conn, project_number, bases)]
    if known and pin.upper() not in {k.upper() for k in known}:
        raise ApiError(
            422,
            "unknown_order",
            f"Order {pin} is not one of the VISUAL orders for {project_number}.",
            extra={"orders": known},
        )


def _project_bases(conn: Any, project_id: int) -> list[str]:
    return [
        r[0]
        for r in conn.execute(
            "SELECT DISTINCT wo.wo_base FROM packet_work_orders wo JOIN packets k ON k.id = wo.packet_id"
            " JOIN units u ON u.id = k.unit_id WHERE u.project_id = %s ORDER BY 1",
            (project_id,),
        )
    ]


def _unit_bases(conn: Any, unit_id: int) -> list[str]:
    return [
        r[0]
        for r in conn.execute(
            "SELECT DISTINCT wo.wo_base FROM packet_work_orders wo JOIN packets k ON k.id = wo.packet_id"
            " WHERE k.unit_id = %s ORDER BY 1",
            (unit_id,),
        )
    ]


def cached_orders(conn: Any, project_number: str, bases: Iterable[str]) -> list[dict[str, Any]]:
    """Every cached VISUAL order of a project: the ones linked to the project number, then the ones linked to its
    work order bases. Each once, with `via` (project | base) and the link's lines and top jobs."""
    with conn.cursor(row_factory=dict_row) as cur:
        rows = cur.execute(
            "SELECT o.*, l.kind AS via, l.key, l.rank, l.lines, l.top_jobs FROM visual_order_links l"
            " JOIN visual_orders o ON o.cust_order_id = l.cust_order_id"
            " WHERE (l.kind = 'project' AND upper(l.key) = upper(%s)) OR (l.kind = 'base' AND l.key = ANY(%s))"
            " ORDER BY CASE l.kind WHEN 'project' THEN 0 ELSE 1 END, l.rank, l.key, o.cust_order_id",
            (project_number, list(bases)),
        ).fetchall()
    out: list[dict[str, Any]] = []
    seen: set[str] = set()
    for r in rows:
        if r["cust_order_id"] in seen:
            continue
        seen.add(r["cust_order_id"])
        out.append(
            {
                "cust_order_id": r["cust_order_id"],
                "status": r["status"],
                "customer_id": r["customer_id"],
                "customer_name": r["customer_name"],
                "customer_po_ref": r["customer_po_ref"],
                "order_date": r["order_date"],
                "desired_ship_date": r["desired_ship_date"],
                "promise_date": r["promise_date"],
                "via": r["via"],
                "key": r["key"],
                "lines": list(r["lines"] or []),
                "top_jobs": list(r["top_jobs"] or []),
            }
        )
    return out


# ---------------------------------------------------------------- writes: projects


def create_project(conn: Any, who: Who, data: Mapping[str, Any]) -> dict[str, Any]:
    number = normalize_number(data.get("number"))
    name = clean_opt(data.get("name"), LIMITS["project_name"])
    customer = clean_opt(data.get("customer"), LIMITS["customer"])
    po = clean_opt(data.get("po"), LIMITS["po"])
    notes = clean_opt(data.get("notes"), LIMITS["notes"])
    due = parse_due(data.get("due"))
    status = parse_status(data.get("status") or "open")
    try:
        with conn.transaction():
            conn.execute("SELECT pg_advisory_xact_lock(7312)")  # one number allocation at a time
            assert_free(conn, number)
            pid = conn.execute(
                "INSERT INTO projects (number, name, customer, customer_src, po, po_src, due, status, notes, source,"
                " created_by) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, 'hand', %s) RETURNING id",
                (
                    number,
                    name,
                    customer,
                    "hand" if customer else None,
                    po,
                    "hand" if po else None,
                    due,
                    status,
                    notes,
                    who.user_id,
                ),
            ).fetchone()[0]
            _audit(
                conn,
                who,
                "project.created",
                "project",
                pid,
                {"number": number, "name": name, "customer": customer, "po": po, "due": _iso(due), "status": status},
            )
            _publish(conn, pid, action="project.created")
    except pgerr.UniqueViolation as e:
        raise _unique_taken(e) from None
    return _project_or_404(conn, pid)


def update_project(
    conn: Any, who: Who, project_id: int, values: Mapping[str, Any], base: Mapping[str, Any]
) -> dict[str, Any]:
    with conn.transaction():
        row = _lock_project(conn, project_id)
        changes = field_cas(row, values, base, PROJECT_FIELDS, unit=False)
        if changes:
            _check_pin(conn, row, changes, project_number=row["number"], bases=_project_bases(conn, project_id))
            cols = _columns(row, changes, who)
            sets = ", ".join(f"{k} = %s" for k in cols)
            conn.execute(
                f"UPDATE projects SET {sets}, version = version + 1, updated_at = now() WHERE id = %s",  # noqa: S608
                [*cols.values(), project_id],
            )
            _audit(
                conn,
                who,
                "project.updated",
                "project",
                project_id,
                {"number": row["number"], "fields": _change_detail(row, changes)},
            )
            _publish(conn, project_id, action="project.updated")
    return _project_or_404(conn, project_id)


def renumber_project(conn: Any, who: Who, project_id: int, number: Any, reason: Any) -> dict[str, Any]:
    new = normalize_number(number)
    why = clean_reason(reason)
    try:
        with conn.transaction():
            conn.execute("SELECT pg_advisory_xact_lock(7312)")
            row = _lock_project(conn, project_id)
            old = row["number"]
            if new == old.upper():
                raise ApiError(400, "same_number", f"The project is already {old}.")
            units = conn.execute(
                "SELECT id, lot FROM units WHERE project_id = %s AND removed_at IS NULL ORDER BY lot FOR UPDATE",
                (project_id,),
            ).fetchall()
            assert_free(conn, new, own_project=project_id)
            for uid, lot in units:
                for o in owners(conn, new + lot):
                    own_live = o["kind"] == "unit" and not o["alias"] and o["project_id"] == project_id
                    own_old = o["kind"] == "unit" and o["alias"] and o["id"] == uid
                    if not (own_live or own_old):
                        raise _taken(o)
            old_numbers = {old.upper(): None, **{(old + lot).upper(): uid for uid, lot in units}}
            new_numbers = [new, *[new + lot for _, lot in units]]
            if set(old_numbers) & set(new_numbers):
                raise ApiError(409, "number_taken", "The new numbers overlap the old ones. Pick another number.")
            conn.execute(
                "DELETE FROM project_aliases WHERE alias = ANY(%s) AND project_id = %s", (new_numbers, project_id)
            )
            for alias, uid in old_numbers.items():
                conn.execute(
                    "INSERT INTO project_aliases (alias, project_id, unit_id, kind) VALUES (%s, %s, %s, 'renumber')"
                    " ON CONFLICT (alias) DO UPDATE SET project_id = EXCLUDED.project_id, unit_id = EXCLUDED.unit_id,"
                    " kind = EXCLUDED.kind, created_at = now()",
                    (alias, project_id, uid),
                )
            conn.execute(
                "UPDATE projects SET number = %s, version = version + 1, updated_at = now() WHERE id = %s",
                (new, project_id),
            )
            conn.execute(
                "INSERT INTO project_renumbers (by, project_id, unit_id, old_number, new_number, reason)"
                " VALUES (%s, %s, NULL, %s, %s, %s)",
                (who.user_id, project_id, old, new, why),
            )
            for uid, lot in units:
                conn.execute(
                    "INSERT INTO project_renumbers (by, project_id, unit_id, old_number, new_number, reason)"
                    " VALUES (%s, %s, %s, %s, %s, %s)",
                    (who.user_id, project_id, uid, old + lot, new + lot, why),
                )
            _audit(
                conn,
                who,
                "project.renumbered",
                "project",
                project_id,
                {"from": old, "to": new, "reason": why, "units": len(units)},
            )
            _publish(conn, project_id, action="project.renumbered")
    except pgerr.UniqueViolation as e:
        raise _unique_taken(e) from None
    return _project_or_404(conn, project_id)


def _live_packets(conn: Any, unit_ids: list[int]) -> int:
    if not unit_ids:
        return 0
    return int(
        conn.execute(
            "SELECT count(*) FROM packets WHERE unit_id = ANY(%s) AND removed_at IS NULL", (unit_ids,)
        ).fetchone()[0]
    )


def remove_project(conn: Any, who: Who, project_id: int, reason: Any) -> dict[str, Any]:
    """Soft delete with its units. Refused (409 has_packets) while any of its units has a live packet."""
    why = clean_reason(reason)
    with conn.transaction():
        row = _lock_project(conn, project_id)
        ids = [
            int(r[0])
            for r in conn.execute(
                "SELECT id FROM units WHERE project_id = %s AND removed_at IS NULL ORDER BY id FOR UPDATE",
                (project_id,),
            ).fetchall()
        ]
        n = _live_packets(conn, ids)
        if n:
            raise ApiError(
                409,
                "has_packets",
                f"{row['number']} still has {n} packet{'' if n == 1 else 's'}. Move or remove them first.",
                extra={"packets": n},
            )
        conn.execute(
            "UPDATE units SET removed_at = now(), removed_by = %s, removed_reason = %s, version = version + 1,"
            " updated_at = now() WHERE id = ANY(%s)",
            (who.user_id, why, ids),
        )
        conn.execute(
            "UPDATE projects SET removed_at = now(), removed_by = %s, removed_reason = %s, version = version + 1,"
            " updated_at = now() WHERE id = %s",
            (who.user_id, why, project_id),
        )
        _audit(
            conn,
            who,
            "project.removed",
            "project",
            project_id,
            {"number": row["number"], "reason": why, "units": ids},
        )
        _publish(conn, project_id, action="project.removed")
    out = get_project(conn, project_id, removed=True)
    assert out is not None
    return out


# ---------------------------------------------------------------- writes: units


def lots_from_request(
    project_number: str,
    *,
    lot: Any = None,
    lots: list[Any] | None = None,
    count: int | None = None,
    start: Any = None,
    suggest: str | None = None,
) -> list[str]:
    """The unit sheet's one-or-several mode: one lot, a list of lots, or `count` lots counting up from `start`."""
    if lots:
        out = [lot_input(project_number, x) for x in lots]
    elif count is not None:
        if not 1 <= count <= MAX_UNITS_AT_ONCE:
            raise ApiError(422, "invalid_count", f"How many: 1 to {MAX_UNITS_AT_ONCE}.")
        first = lot_input(project_number, start) if start not in (None, "") else suggest
        if first is None or not _DIGITS.match(first):
            raise ApiError(422, "invalid_lot", "The starting lot has to be a number, e.g. 001.")
        b = int(first)
        if b + count - 1 > 999:
            raise ApiError(422, "invalid_lot", "Lots go up to 999. Start lower or add fewer.")
        out = [f"{b + i:03d}" for i in range(count)]
    elif lot not in (None, ""):
        out = [lot_input(project_number, lot)]
    else:
        if suggest is None:
            raise ApiError(422, "invalid_lot", "No free lot left in this project. Give one.")
        out = [suggest]
    if len(out) > MAX_UNITS_AT_ONCE:
        raise ApiError(422, "invalid_count", f"How many: 1 to {MAX_UNITS_AT_ONCE}.")
    dup = sorted({x for x in out if out.count(x) > 1})
    if dup:
        raise ApiError(409, "number_taken", f"Lot {dup[0]} is in the list twice.", extra={"lots": dup})
    return out


def create_units(conn: Any, who: Who, project_id: int, data: Mapping[str, Any]) -> list[dict[str, Any]]:
    name = clean_opt(data.get("name"), LIMITS["unit_name"])
    notes = clean_opt(data.get("notes"), LIMITS["notes"])
    due = parse_due(data.get("due"))
    status = parse_status(data.get("status") or "open")
    try:
        with conn.transaction():
            conn.execute("SELECT pg_advisory_xact_lock(7312)")
            project = _lock_project(conn, project_id)
            if project["status"] == "closed":
                raise ApiError(409, "project_closed", f"Project {project['number']} is closed. Reopen it to add units.")
            pn = project["number"]
            wants_suggest = not data.get("lots") and data.get("lot") in (None, "") and data.get("start") in (None, "")
            lots = lots_from_request(
                pn,
                lot=data.get("lot"),
                lots=data.get("lots"),
                count=data.get("count"),
                start=data.get("start"),
                suggest=next_lot(conn, project_id, pn) if wants_suggest else None,
            )
            for lot in lots:
                assert_free(conn, pn + lot)
            ids: list[int] = []
            for lot in lots:
                uid = conn.execute(
                    "INSERT INTO units (project_id, lot, name, due, status, notes, created_by)"
                    " VALUES (%s, %s, %s, %s, %s, %s, %s) RETURNING id",
                    (project_id, lot, name, due, status, notes, who.user_id),
                ).fetchone()[0]
                ids.append(int(uid))
                _audit(
                    conn,
                    who,
                    "unit.created",
                    "unit",
                    uid,
                    {"number": pn + lot, "project_id": project_id, "lot": lot, "name": name, "due": _iso(due)},
                )
            _publish(conn, project_id, action="unit.created")
    except pgerr.UniqueViolation as e:
        raise _unique_taken(e) from None
    with conn.cursor(row_factory=dict_row) as cur:
        rows = cur.execute(UNIT_SELECT + " WHERE v.id = ANY(%s) ORDER BY v.lot", (ids,)).fetchall()
    return [unit_out(r) for r in rows]


def update_unit(
    conn: Any, who: Who, unit_id: int, values: Mapping[str, Any], base: Mapping[str, Any]
) -> dict[str, Any]:
    with conn.transaction():
        row, project = _lock_unit(conn, unit_id)
        changes = field_cas(row, values, base, UNIT_FIELDS, unit=True)
        if changes:
            _check_pin(conn, row, changes, project_number=project["number"], bases=_unit_bases(conn, unit_id))
            cols = _columns(row, changes, who)
            sets = ", ".join(f"{k} = %s" for k in cols)
            conn.execute(
                f"UPDATE units SET {sets}, version = version + 1, updated_at = now() WHERE id = %s",  # noqa: S608
                [*cols.values(), unit_id],
            )
            _audit(
                conn,
                who,
                "unit.updated",
                "unit",
                unit_id,
                {"number": project["number"] + row["lot"], "fields": _change_detail(row, changes)},
            )
            _publish(conn, project["id"], unit_id, action="unit.updated")
    return _unit_or_404(conn, unit_id)


def renumber_unit(conn: Any, who: Who, unit_id: int, lot: Any, reason: Any) -> dict[str, Any]:
    why = clean_reason(reason)
    try:
        with conn.transaction():
            conn.execute("SELECT pg_advisory_xact_lock(7312)")
            row, project = _lock_unit(conn, unit_id)
            pn = project["number"]
            new_lot = lot_input(pn, lot)
            if new_lot == row["lot"]:
                raise ApiError(400, "same_number", f"The unit is already {pn + new_lot}.")
            old, new = pn + row["lot"], pn + new_lot
            assert_free(conn, new, own_unit=unit_id)
            conn.execute("DELETE FROM project_aliases WHERE alias = %s AND unit_id = %s", (new.upper(), unit_id))
            conn.execute(
                "INSERT INTO project_aliases (alias, project_id, unit_id, kind) VALUES (%s, %s, %s, 'unit_renumber')"
                " ON CONFLICT (alias) DO UPDATE SET project_id = EXCLUDED.project_id, unit_id = EXCLUDED.unit_id,"
                " kind = EXCLUDED.kind, created_at = now()",
                (old.upper(), project["id"], unit_id),
            )
            conn.execute(
                "UPDATE units SET lot = %s, version = version + 1, updated_at = now() WHERE id = %s", (new_lot, unit_id)
            )
            conn.execute(
                "INSERT INTO project_renumbers (by, project_id, unit_id, old_number, new_number, reason)"
                " VALUES (%s, %s, %s, %s, %s, %s)",
                (who.user_id, project["id"], unit_id, old, new, why),
            )
            _audit(conn, who, "unit.renumbered", "unit", unit_id, {"from": old, "to": new, "reason": why})
            _publish(conn, project["id"], unit_id, action="unit.renumbered")
    except pgerr.UniqueViolation as e:
        raise _unique_taken(e) from None
    return _unit_or_404(conn, unit_id)


def remove_unit(conn: Any, who: Who, unit_id: int, reason: Any) -> dict[str, Any]:
    why = clean_reason(reason)
    with conn.transaction():
        row, project = _lock_unit(conn, unit_id)
        n = _live_packets(conn, [unit_id])
        number = project["number"] + row["lot"]
        if n:
            raise ApiError(
                409,
                "has_packets",
                f"{number} still has {n} packet{'' if n == 1 else 's'}. Move or remove them first.",
                extra={"packets": n},
            )
        conn.execute(
            "UPDATE units SET removed_at = now(), removed_by = %s, removed_reason = %s, version = version + 1,"
            " updated_at = now() WHERE id = %s",
            (who.user_id, why, unit_id),
        )
        _audit(conn, who, "unit.removed", "unit", unit_id, {"number": number, "reason": why})
        _publish(conn, project["id"], unit_id, action="unit.removed")
    out = get_unit(conn, unit_id, removed=True)
    assert out is not None
    return out