dodil/erp-brick-manufacturing

DODIL ERP brick: manufacturing

src/brick_manufacturing/application/dispatch.py

"""Side effects, after the local commit, from pending rows.

A use case writes its movement and a PENDING outbox row in ONE transaction; this module sends the row to the neighbour
afterwards and records the answer in a second one. So a rolled-back movement sends nothing, a sent one is never lost, and a
retry is the neighbour's replay on the same reference (move_ref, adjustment_ref, journal_id) - never a second effect.

  sent      the neighbour took it (201, or 200 for a reference it knew): final
  rejected  the neighbour refused it (a 4xx): final, and the movement it carried is void (inventory has no such move)
  pending   it could not be asked (not installed, down, 5xx) - or gl/inventory said the period is closed (409): retried
"""
import json
from decimal import Decimal, InvalidOperation
from typing import Callable

from brick_manufacturing.domain.values import MAX_MONEY
from brick_manufacturing.infrastructure import seams

WAITS = ("gl.journal", "inventory.adjustment")      # a 409 here is a closed period: wait, never reject


def _refs(target: str, body: dict) -> list:
    if target == "inventory.issue":
        return [x["move_ref"] for x in body["lines"]]
    if target == "inventory.receipt":
        return [body["move_ref"]]
    return []


def _answers(target: str, body: dict, data: dict) -> list:
    """The movement updates an accepted call carries: inventory's move id, its value, and the lot it chose."""
    if target == "inventory.issue":
        got = {m["move_ref"]: m for m in data.get("moves", [])}
        return [{"move_ref": r, "status": "sent", "inventory_move_id": str(got[r].get("move_id", "")) if r in got else "",
                 "value": got[r].get("value") if r in got else None} for r in _refs(target, body)]
    if target == "inventory.receipt":
        return [{"move_ref": body["move_ref"], "status": "sent", "inventory_move_id": str(data.get("move_id", "")),
                 "value": data.get("value")}]
    return []


def _out_of_range(answers: list) -> str:
    """inventory's value is NUMERIC(18,2) by its contract; an answer our column cannot hold is not stored (a 500 at the UPDATE)."""
    for a in answers:
        try:
            if a.get("value") is not None and abs(Decimal(str(a["value"]))) > MAX_MONEY:
                return f"{a['move_ref']}: value {a['value']} is above what a ledger amount can hold"
        except InvalidOperation:
            return f"{a['move_ref']}: value {a['value']!r} is not an amount"
    return ""


def send(repository: Callable, order_id: str = None) -> dict:
    """Send every pending outbox row (of one order, or all): {sent, rejected: [{ref, reason, gate}], pending}."""
    with repository() as repo:
        rows = repo.outbox.pending(order_id)
    marks, moves, rejected = [], [], []
    for row in rows:
        seam, path = seams.TARGETS[row["target"]]
        body = json.loads(row["payload"])
        attempts = (row["attempts"] or 0) + 1
        if not seam.installed:
            marks.append({**row, "status": "pending", "attempts": attempts, "last_error": f"{seam.name} is not installed"})
            continue
        r = seam.post(path, body, actor="manufacturing")
        answers = _answers(row["target"], body, r.data) if r.ok else []
        bad = _out_of_range(answers)
        if r.ok and not bad:
            marks.append({**row, "status": "sent", "attempts": attempts, "last_error": ""})
            moves += answers
        elif bad:
            marks.append({**row, "status": "rejected", "attempts": attempts, "last_error": bad[:500]})
            moves += [{"move_ref": ref, "status": "rejected", "reject_reason": bad[:500]} for ref in _refs(row["target"], body)]
            rejected.append({"ref": row["ref"], "reason": bad, "gate": "bounds"})
        elif r.status and 400 <= r.status < 500 and not (r.status == 409 and row["target"] in WAITS):
            try:
                problem = json.loads(r.reason)
            except ValueError:
                problem = {}
            problem = problem if isinstance(problem, dict) else {}
            reason = str(problem.get("reason") or problem.get("detail") or r.reason or f"{seam.name} answered {r.status}")
            marks.append({**row, "status": "rejected", "attempts": attempts, "last_error": reason[:500]})
            moves += [{"move_ref": ref, "status": "rejected", "reject_reason": reason[:500]} for ref in _refs(row["target"], body)]
            rejected.append({"ref": row["ref"], "reason": reason, "gate": problem.get("gate", "")})
        else:
            marks.append({**row, "status": "pending", "attempts": attempts,
                          "last_error": (r.reason or f"{seam.name} answered {r.status}")[:500]})
    if marks:
        voided = [m["ref"] for m in marks if m["status"] == "rejected" and m["target"] == "gl.journal" and ":activity:" in m["ref"]]
        with repository() as repo:
            repo.outbox.mark(marks)
            repo.movements.answer(moves)
            repo.floor.void_by_ref(voided)             # an activity journal gl refused: the confirmation it carried is void
    return {"sent": sum(1 for m in marks if m["status"] == "sent"), "rejected": rejected,
            "pending": sum(1 for m in marks if m["status"] == "pending")}