dodil/erp-brick-payments
DODIL ERP brick: payments
src/brick_payments/application/use_cases.py
"""The use cases - what a person (or their agent) can do with money received, refunded and chased.
Each is declared ONCE and served as an API route and as an MCP tool. Two things are decided here and nowhere
else: WHO MAY (a PERMISSION on the declaration, never a role's name) and WHICH ROWS. A row the caller may not
see is a 404, never a 403.
The rules that shape every use case (module.yaml in_code_not_rows):
- outstanding(invoice) = total - SUM(allocations), derived on read; nothing stores "paid", nothing increments a balance
- 'today' is the date in business_timezone: an allocation, an undo, a credit limit and an override are never dated
before today; a receipt, a return and a refund payment are never dated after it
- a correction is a reversing row or a new row - no committed pay_ row is edited (only a suggestion's, a refund's and a
receipt's status move forward)
- an AI suggestion becomes an allocation only through a person (decide_match), except an EXACT match in the opt-in
auto-with-review, which is recorded as auto and listed for review
- no read-your-writes inside an open transaction: each use case reads committed rows first, then writes, and works
the figures it reports out from what it wrote instead of querying them back
"""
import uuid
from contextlib import contextmanager
from datetime import date, datetime, timedelta
from decimal import Decimal
from typing import Annotated, Literal, Optional, Union
from zoneinfo import ZoneInfo
from erp_platform import work
from erp_platform.bricks import Brick, Refused
from erp_platform.types import BigId, Limit, Name, text
from pydantic import Field
from sqlalchemy.exc import IntegrityError
from brick_payments.application import fx
from brick_payments.application.inputs import (
Code,
Currency,
Day,
LineIn,
Money,
Offset,
Rate,
Reason,
Ref,
ReminderLevelIn,
Seq,
day,
)
from brick_payments.domain import allocation as alloc
from brick_payments.domain import collection as coll
from brick_payments.domain import credit, matching
from brick_payments.domain import refund as refund_module
from brick_payments.domain import settlement as stl
from brick_payments.domain.errors import RuleBroken
from brick_payments.domain.money import ZERO, money
from brick_payments.domain.readiness import fmt, met, refuse, unmet
from brick_payments.domain.receipt import Receipt, received_not_future, return_dated_ok
from brick_payments.domain.refund import Refund
from brick_payments.domain.settlement import Allocation
from brick_payments.domain.values import (
MODES,
SETTING_DEFS,
SourceKind,
check_setting,
needs_controls,
)
from brick_payments.infrastructure import ai_steps, seams
# Every decision is a PERMISSION (module.yaml permissions.catalog); no role's name is ever compared.
READ, WRITE, IMPORT = "pay:receipt:read", "pay:receipt:write", "pay:receipt:import"
ALLOC = "pay:allocation:write"
REFUND_REQUEST, REFUND_APPROVE = "pay:refund:request", "pay:refund:approve"
COLLECT, EVENT, REMINDER, STATEMENT = "pay:collection:write", "pay:collection_event:write", "pay:reminder:send", "pay:statement:write"
CREDIT_WRITE = "pay:credit:write"
CREDIT_READ = "pay:credit:read" # sales reads a customer's balance and hold through credit_status ONLY (an addition to the spec's catalog)
SETTINGS_WRITE, CONTROLS = "pay:settings:write", "pay:settings:controls"
MIGRATION_SIGN = "pay:migration:sign"
# machine: held only by a service client; no wildcard reaches them
INGEST, MATCH, TICK = "pay:receipt:ingest", "pay:receipt:match", "pay:collection:tick"
READ_UNMASKED = "pay:bank_account:read_unmasked"
FROM_SOURCE, MIGRATION_IMPORT = "pay:allocation:from_source", "pay:migration:import"
STATEMENT_INGEST, LINE_SUGGEST, DD_COLLECT = "pay:statement:ingest", "pay:statement_line:suggest", "pay:dd:collect"
PAY = Brick("payments", version="0.2.2", ids_as_strings=True,
machine_permissions=(INGEST, MATCH, TICK, FROM_SOURCE, MIGRATION_IMPORT, STATEMENT_INGEST, LINE_SUGGEST, DD_COLLECT, READ_UNMASKED,
"pay:deposit_batch:intake", "pay:card_settlement:intake", "pay:payout:intake"), instructions=(
"Payments & collections: record what customers pay (typed in, or imported from the bank statement as camt.053 or "
"CSV), match it to the invoices it settles, refund what is owed back, and chase what is late. What is "
"outstanding on an invoice is DERIVED - its total less the allocations - so ask settlement or open_items, never "
"keep a balance. A receipt is a bank fact dated today or earlier; an allocation, an undo, a credit limit and a "
"hold are dated when they are recorded; a correction is a reversing row, never an edit. The AI only suggests a "
"match; a person accepts, changes or rejects it (decide_match). Money going OUT (a refund) needs a second "
"person's approval. credit_status and credit_ok say whether a customer is on hold and why. A foreign-currency receipt keeps its rate: allocating it posts the "
"realised FX difference. A bank statement must add up and follow the previous one; reconcile_statement proves the ledger "
"balance equals the closing balance before it locks. A debit is matched to purchasing's supplier payment (supplier_payment), "
"a receipt can be allocated to a down-payment request, and a contractor's CIS deduction is recorded with tax first."))
# The process view (BRICKS.md): the workflows as BPMN, GET /process/bpmn.
PAY.declare_process("process.yaml")
@contextmanager
def _repository_default():
from erp_platform import db # imported late: it needs BUCKET
with _on_engine(db.engine) as repo:
yield repo
def _lost_a_unique_race(exc: IntegrityError) -> bool:
text = f"{getattr(exc, 'orig', '')} {exc}"
return "23505" in text or "duplicate key" in text or "UNIQUE constraint" in text
@contextmanager
def _on_engine(engine):
"""One transaction. On DataK3 a write conflict on a claim row can surface at COMMIT (verified live 2026-10-24: two allocations of one
receipt, the loser's COMMIT raised 23505), after the use case has returned - so it is caught here, around the commit, and answered as
the same retry-safe 409 a conflict seen mid-use-case gets, never a 500."""
from brick_payments.infrastructure.tables import Repository
try:
with engine.begin() as conn:
yield Repository(conn)
except IntegrityError as exc:
if not _lost_a_unique_race(exc):
raise
raise Refused(409, "Someone else wrote the same record a moment ago - load it again and retry") from None
repository = _repository_default # tests and a composition swap in their own
@contextmanager
def _decide():
"""A broken business rule is a 409 with the rule's own words, through both doors."""
try:
yield
except RuleBroken as broken:
raise Refused(409, str(broken)) from None
except seams.Unavailable as down:
raise Refused(down.http, down.reason) from None
except IntegrityError:
raise Refused(409, "Someone else wrote the same record a moment ago - load it again and retry") from None
def _who(caller: dict) -> str:
return (caller.get("email") or caller.get("sub") or "").strip().lower()
def _ids(caller: dict) -> tuple:
"""Every identity the caller has (email and sub, lower-case): the same person under either is the same person."""
return tuple(dict.fromkeys(x.strip().lower() for x in (caller.get("email"), caller.get("sub")) if x and x.strip()))
def _now() -> datetime:
return work.now()
def _today(repo) -> date:
"""Business today: the date NOW (work.now()) in business_timezone - read from ONE place (SettingsRepository.timezone) - and
never below the last business date already stamped on a row, whatever a change of zone would say."""
day = _now().astimezone(ZoneInfo(repo.settings.timezone())).date()
floor = repo.settings.used_through()
return max(day, floor) if floor else day
def _date(v: Optional[str], what: str) -> Optional[date]:
try:
return day(v)
except ValueError:
raise Refused(422, f"{what} is a date like 2026-11-01") from None
def _s(v) -> str:
"""A Decimal as the exact string both doors speak."""
return f"{Decimal(v):.2f}"
def _cur(v: str) -> str:
"""An ISO 4217 alphabetic code (GBP, EUR, JPY), upper-cased."""
v = (v or "").strip().upper()
if len(v) != 3 or not v.isalpha() or not v.isascii():
raise Refused(422, f"{v or 'a blank currency'!r} is not a currency code: use three letters such as GBP")
return v
def _receipt(repo, payment_id: str) -> Receipt:
r = repo.receipts.get(payment_id)
if r is None:
raise Refused(404, f"no receipt {payment_id}")
return r
def _group(rows: list) -> dict:
out: dict = {}
for r in rows:
out.setdefault(r.invoice_id, []).append(r)
return out
def _credit_notes(bp_id: Optional[int], caller: Optional[dict]) -> list:
"""Credit notes are optional for credit: ar not installed means none; an ar that is down is an error."""
try:
return seams.credit_notes(bp_id=bp_id, caller=caller)
except seams.Unavailable as down:
if down.http == 409:
return []
raise
def _invoice_view(d: seams.InvoiceDoc, rows: list, today: date, settings, as_at: Optional[date] = None) -> dict:
out = stl.outstanding(d.total, rows, as_at)
days = stl.age_days(d.due_date, d.issued_on, as_at or today, settings.age_from)
overdue = bool(d.due_date and d.due_date < (as_at or today) and out > ZERO)
return {"invoice_id": d.invoice_id, "number": d.number, "bp_id": d.bp_id, "currency": d.currency,
"total": _s(d.total), "outstanding": _s(out), "state": stl.state_of(d.total, out, rows, as_at),
"issued_on": d.issued_on.isoformat() if d.issued_on else None,
"due_date": d.due_date.isoformat() if d.due_date else None,
"age_days": days if overdue else 0, "bucket": stl.bucket_label(days, settings.aging_buckets) if overdue else None}
def _remainder(r: Receipt, rows: list, as_at: Optional[date] = None) -> Decimal:
"""What a receipt still holds, given ALL the customer's allocation rows. A reversed receipt holds nothing."""
if as_at is None:
if not r.live:
return ZERO
elif not r.live_at(as_at):
return ZERO
return stl.source_remainder(r.amount, SourceKind.RECEIPT.value, r.payment_id, rows, as_at)
def _credits(repo, bp_id: int, caller: Optional[dict], as_at: Optional[date] = None, with_notes: bool = True) -> list:
"""The customer's unallocated credit as documents: [{kind, id, number, date, currency, remainder}]."""
rows = repo.allocations.for_bp(bp_id)
out = []
for r in repo.receipts.live_of(bp_id):
rem = _remainder(r, rows, as_at)
if rem > ZERO:
out.append({"kind": "receipt", "id": r.payment_id, "number": r.bank_reference or r.payment_id,
"date": r.received_on, "currency": r.currency, "remainder": rem})
if with_notes:
for n in _credit_notes(bp_id, caller):
if as_at is not None and n.issued_on and n.issued_on > as_at:
continue
rem = stl.source_remainder(n.total, SourceKind.CREDIT_NOTE.value, n.credit_note_id, rows, as_at)
if rem > ZERO:
out.append({"kind": "credit_note", "id": n.credit_note_id, "number": n.number, "date": n.issued_on,
"currency": n.currency, "remainder": rem})
return out
def _credit_available(repo, bp_id: int, currency: str, caller: Optional[dict]) -> Decimal:
"""Unallocated credit in a currency, less refunds already asked for (they hold it), never summed across currencies."""
total = sum((c["remainder"] for c in _credits(repo, bp_id, caller) if c["currency"] == currency), ZERO)
return total - repo.refunds.reserved(bp_id, currency)
def _open_items(repo, bp_id: int, as_at: date, caller: Optional[dict]) -> dict:
"""Everything unsettled for a customer at the end of `as_at`: invoices (positive), credit notes and receipts
(negative), with a closing balance PER CURRENCY."""
settings = repo.settings.current()
docs = [d for d in seams.invoices(bp_id=bp_id, caller=caller) if d.issued_on is None or d.issued_on <= as_at]
by = _group(repo.allocations.for_invoices([d.invoice_id for d in docs]))
today = _today(repo)
invoices = []
for d in docs:
v = _invoice_view(d, by.get(d.invoice_id, []), today, settings, as_at)
if Decimal(v["outstanding"]) != ZERO:
invoices.append(v)
credits = [{"kind": c["kind"], "id": c["id"], "number": c["number"], "date": c["date"].isoformat() if c["date"] else None,
"currency": c["currency"], "amount": _s(-c["remainder"])} for c in _credits(repo, bp_id, caller, as_at)]
closing: dict = {}
for v in invoices:
closing[v["currency"]] = closing.get(v["currency"], ZERO) + Decimal(v["outstanding"])
for c in credits:
closing[c["currency"]] = closing.get(c["currency"], ZERO) + Decimal(c["amount"])
return {"bp_id": bp_id, "as_at": as_at.isoformat(), "invoices": invoices, "credits": credits,
"closing_balance": {k: _s(v) for k, v in sorted(closing.items())}}
# ═══ reads other bricks and screens use: settlement, open items ═════════════════════════════════════════
@PAY.use_case(kind="query", route="GET /settlement", permission=READ)
def settlement(caller: dict, invoice_ids: text(8000, min_length=1), as_at: Optional[Day] = None) -> dict:
"""Outstanding, state and age per invoice - derived on read from the invoice total and the allocations, as at a
date (today by default). Invoicing, sales and crm ask THIS, never pay_ tables. invoice_ids is comma-separated."""
ids = [i.strip() for i in invoice_ids.split(",") if i.strip()][:200]
with repository() as repo, _decide():
on, today, settings = _date(as_at, "as_at"), _today(repo), repo.settings.current()
docs = seams.invoices(ids=ids, caller=caller)
by = _group(repo.allocations.for_invoices(ids))
found = {d.invoice_id for d in docs}
return {"as_at": (on or today).isoformat(),
"invoices": [_invoice_view(d, by.get(d.invoice_id, []), today, settings, on) for d in docs],
"unknown": [i for i in ids if i not in found]}
@PAY.use_case(kind="query", route="GET /customers/{bp_id}/open-items", permission=READ)
def open_items(caller: dict, bp_id: BigId, as_at: Optional[Day] = None) -> dict:
"""A customer's open items as at a date (today by default): the unsettled invoices with outstanding, state and
age, their unallocated receipts and credit notes, and the closing balance per currency."""
with repository() as repo, _decide():
return _open_items(repo, bp_id, _date(as_at, "as_at") or _today(repo), caller)
# ═══ receipts ════════════════════════════════════════════════════════════════════════════════════════════
def _alloc_view(a: Allocation) -> dict:
return {"source_kind": a.source_kind, "source_id": a.source_id, "invoice_id": a.invoice_id, "seq": a.seq,
"amount": _s(a.amount), "currency": a.currency, "allocated_on": a.allocated_on.isoformat(),
"allocated_by": a.allocated_by, "reason": a.reason, "reverses_seq": a.reverses_seq,
"suggestion_id": a.suggestion_id, "decided_how": a.decided_how, "origin": a.origin}
def _receipt_view(repo, r: Receipt, rows: Optional[list] = None, suggestion=...) -> dict:
rows = rows if rows is not None else (repo.allocations.for_bp(r.bp_id) if r.bp_id is not None else [])
mine = [a for a in rows if a.source_kind == SourceKind.RECEIPT.value and a.source_id == r.payment_id]
ref = stl.credit_ref(SourceKind.RECEIPT.value, r.payment_id)
refunded = sum((a.amount for a in rows if a.source_kind == SourceKind.REFUND.value and a.invoice_id == ref), ZERO)
allocated = sum((a.drawn for a in mine), ZERO)
out = r.as_dict()
out.update({"allocated": _s(allocated), "refunded": _s(refunded), "unallocated": _s(_remainder(r, rows)),
"allocation_state": ("reversed" if not r.live else "unallocated" if allocated + refunded == ZERO
else "fully_allocated" if r.amount - allocated - refunded == ZERO else "partly_allocated"),
"allocations": [_alloc_view(a) for a in mine],
"suggestion": repo.suggestions.open_of(r.payment_id) if suggestion is ... else suggestion})
return out
def _post_receipt(repo, r: Receipt, *, reverse: bool = False, actor: str = "") -> dict:
"""Money in posts to the ledger (PAY-42): bank / receivable, journal_id derived from the payment_id so re-posting
changes nothing. No ledger installed, or no ledger account on the bank account, is a reply - never an error."""
bank = repo.config.bank(r.bank_account_id) if r.bank_account_id else None
if not bank or not bank.get("gl_account_id"):
return {"posted": False, "reason": f"bank account {r.bank_account_id or '(none)'} has no ledger account (gl_account_id), so nothing was posted"}
amt = _s(r.amount)
bank_line, ar_line = seams.account_line(bank["gl_account_id"]), seams.receivable_line(r.bp_id)
debit, credit_ = (ar_line, bank_line) if reverse else (bank_line, ar_line)
lines = [{**debit, "debit": amt}, {**credit_, "credit": amt}]
return seams.post_journal(r.payment_id, "receipt_reversal" if reverse else "receipt",
(r.returned_on if reverse else r.received_on) or r.received_on, r.currency,
f"{'Returned' if reverse else 'Receipt'} {r.bank_reference or r.payment_id}", lines, actor,
bank.get("company_code") or "", r.fx_rate)
def _open_for_match(repo, r: Receipt, caller: Optional[dict]) -> list:
docs = [d for d in seams.invoices(bp_id=r.bp_id, caller=caller) if d.currency == r.currency]
by = _group(repo.allocations.for_invoices([d.invoice_id for d in docs]))
out = []
for d in docs:
o = stl.outstanding(d.total, by.get(d.invoice_id, []))
if o > ZERO:
out.append(matching.Open(d.invoice_id, d.number, o, d.due_date))
return out
def _allocate_core(repo, caller: dict, *, source_kind: str, source_id: str, source_currency: str, source_bp: Optional[int],
source_remainder: Decimal, lines: list, decided_how: str = "manual", suggestion_id: str = "",
by: Optional[str] = None, what: str = "receipt") -> dict:
"""Allocate `lines` from one source, today, refusing in the spec's words. One unit of work. Returns (the answer, the rows
written): a caller that reports the receipt afterwards uses the rows in hand - there is no read-your-writes inside an open
transaction, so it must not query them back."""
if source_bp is None:
raise RuleBroken("This receipt is not on a customer yet. Name the customer first (assign-customer)")
today, now = _today(repo), _now()
ids = [ln.invoice_id for ln in lines]
docs = {d.invoice_id: d for d in seams.invoices(ids=ids, caller=caller)}
for ln in lines:
d = docs.get(ln.invoice_id)
if d is None:
raise Refused(404, f"no settleable invoice {ln.invoice_id}")
if d.bp_id != source_bp:
raise RuleBroken(f"Invoice {d.number} belongs to customer {d.bp_id}, the {what} to customer {source_bp}. "
"Allocate only to the same customer's invoices")
by_inv = _group(repo.allocations.for_invoices(ids))
plan_lines = [alloc.Line(ln.invoice_id, ln.amount, stl.outstanding(docs[ln.invoice_id].total, by_inv.get(ln.invoice_id, [])),
docs[ln.invoice_id].currency, source_bp) for ln in lines]
next_seq = repo.allocations.next_seqs(source_kind, source_id, [ln.invoice_id for ln in lines])
rows = alloc.plan(source_kind=source_kind, source_id=source_id, source_currency=source_currency,
source_remainder=source_remainder, lines=plan_lines, next_seq=next_seq, today=today,
by=by or _who(caller), decided_how=decided_how, suggestion_id=suggestion_id, what=what)
repo.allocations.claim(invoices=ids, sources=[(source_kind, source_id)], at=now) # two writers of this money collide here
repo.allocations.add(rows, now)
fx_posted = []
if source_kind == SourceKind.RECEIPT.value and (src := repo.receipts.get(source_id)) is not None:
fx_posted = fx.post_allocations(repo, src, rows, docs, _who(caller), now) # PAY-33: the realised difference, same transaction
tolerance = repo.settings.current().match_tolerance
after, proposals = [], []
for row, pl in zip(rows, plan_lines):
left = pl.outstanding - row.amount
after.append({"invoice_id": row.invoice_id, "number": docs[row.invoice_id].number, "outstanding": _s(left)})
if tolerance > ZERO and ZERO < left <= tolerance and source_kind == SourceKind.RECEIPT.value:
got = seams.propose_writeoff(row.invoice_id, left, row.currency, f"difference within tolerance {fmt(tolerance)}",
f"receipt:{source_id}", _who(caller))
proposals.append({"invoice_id": row.invoice_id, "amount": _s(left), **got})
total = sum((r.amount for r in rows), ZERO)
return {"allocations": [_alloc_view(r) for r in rows], "allocated": _s(total),
"source_remainder": _s(source_remainder - total), "invoices": after, "write_off_proposals": proposals,
**({"fx": fx_posted} if fx_posted else {})}, rows
def _suggest(repo, r: Receipt, caller: dict, mode: str) -> dict:
"""The match suggestion for one receipt (PAY-32): a ROW with its evidence, confidence and model. Nothing is
allocated unless the mode is auto-with-review AND the match is exact."""
if mode == "off":
return {"suggested": False, "reason": "ai_suggest_match is off"}
if not r.live or r.bp_id is None:
return {"suggested": False, "reason": "the receipt is reversed or not on a customer yet"}
rows = repo.allocations.for_bp(r.bp_id)
rem = _remainder(r, rows)
cur = repo.suggestions.open_of(r.payment_id)
if rem <= ZERO:
return {"suggested": False, "reason": "the receipt has no unallocated money"}
try:
open_inv = _open_for_match(repo, r, caller)
except seams.Unavailable as down:
return {"suggested": False, "reason": down.reason}
found = matching.suggest(amount=r.amount, remainder=rem, reference=f"{r.bank_reference} {r.remittance_text}",
open_invoices=open_inv)
now = _now()
if found is None:
if cur:
repo.suggestions.move(cur["suggestion_id"], status="superseded")
return {"suggested": False, "reason": "nothing in the reference or the amounts names an open invoice"}
lines = [{"invoice_id": ln["invoice_id"], "amount": _s(ln["amount"])} for ln in found.lines]
if cur and cur["proposed_lines"] == lines:
return {"suggested": True, "unchanged": True, "suggestion": cur}
if cur:
repo.suggestions.move(cur["suggestion_id"], status="superseded")
sid = f"{r.payment_id}#{len(repo.suggestions.of_receipt(r.payment_id)) + 1}"
view = {"suggestion_id": sid, "payment_id": r.payment_id, "proposed_lines": lines, "evidence": found.evidence,
"confidence": str(found.confidence), "model": found.model, "exact": found.exact, "status": "open"}
auto, auto_rows = None, []
if mode == "auto-with-review" and found.exact and rem == r.amount:
auto, auto_rows = _allocate_core(repo, {"email": "system"}, source_kind=SourceKind.RECEIPT.value, source_id=r.payment_id,
source_currency=r.currency, source_bp=r.bp_id, source_remainder=rem,
lines=[LineIn(invoice_id=ln["invoice_id"], amount=Decimal(ln["amount"])) for ln in lines],
decided_how="auto", suggestion_id=sid, by="system:ai_suggest_match")
view["status"] = "accepted"
repo.suggestions.add(view, now)
if auto is not None:
repo.suggestions.move(sid, decided_by="auto", decided_at=now)
return {"suggested": True, "suggestion": {**view, "decided_by": "auto" if auto else ""}, "auto_allocated": auto,
"_rows": auto_rows if auto is not None else []}
@PAY.use_case(kind="command", route="POST /receipts", permission=WRITE)
def record_receipt(caller: dict, bp_id: BigId, amount: Money, currency: Currency, received_on: Day,
method: Optional[text(40)] = "bank_transfer", bank_account_id: Optional[Code] = None,
bank_reference: Optional[Ref] = None, remittance_text: Optional[text(500)] = None,
fx_rate: Optional[Rate] = None) -> dict:
"""Record money a customer paid, with its date, amount, currency and bank reference: it lands on the customer
as unallocated cash, and the AI (when on) suggests which invoices it pays. A received_on after today (in
business_timezone) is refused; the same bank reference on the same bank account twice is refused as a
duplicate; a wrong amount is corrected by reversing the receipt and recording the right one, never by editing."""
with repository() as repo, _decide():
today, now, settings = _today(repo), _now(), repo.settings.current()
on = _date(received_on, "received_on")
known = seams.check_customer(bp_id, caller)
if known.status != "known":
raise Refused(known.http, known.note)
if bank_account_id and repo.config.bank(bank_account_id) is None:
raise Refused(422, f"there is no bank account {bank_account_id} - set it up in the settings first")
account, ref = bank_account_id or "", (bank_reference or "").strip()
if ref: # a transfer typed in with no account and the same one on a bank account are one transfer
holder = (repo.receipts.ref_holder(account, ref) or repo.receipts.ref_holder("", ref)) if account else repo.receipts.ref_holder_anywhere(ref)
if holder:
raise RuleBroken(f"Bank reference {ref} is already on receipt {holder}. Open receipt {holder}")
r = Receipt.record(payment_id=str(uuid.uuid4()), bp_id=bp_id, currency=_cur(currency), amount=amount, received_on=on,
today=today, tz=settings.business_timezone, by=_who(caller), at=now,
method=method or "", bank_account_id=account, bank_reference=ref,
remittance_text=remittance_text or "", fx_rate=fx_rate)
repo.receipts.add(r)
if ref:
repo.receipts.claim_ref(account, ref, r.payment_id, now)
suggestion = _suggest(repo, r, caller, settings.ai_suggest_match)
posted = _post_receipt(repo, r, actor=_who(caller))
made = suggestion.pop("_rows", [])
sug = suggestion.get("suggestion") if suggestion.get("suggested") and not made else None
view = _receipt_view(repo, r, made, sug)
return {**view, "suggestion_result": suggestion, "ledger": posted}
@PAY.use_case(kind="command", route="POST /receipts/import", permission=IMPORT)
def import_statement(caller: dict, format: Literal["camt053", "mt940", "csv"], content: text(5_000_000, min_length=1),
bank_account_id: Code, currency: Optional[Currency] = None) -> dict:
"""Import the bank statement (ISO 20022 camt.053, SWIFT MT940, or CSV with a header). EVERY line, credit or debit, is kept as a
statement line (reconciled in PAY-21) keyed bank:<account>:<entry reference>; a customer's credit is also a receipt with the
same key, dated by the bank's booking date. A line already imported is recognised and skipped, so importing the same file twice
makes no duplicates; lines booked before the migration cut-over (statement_import_from) are skipped as already migrated; a
credit from one of our own bank accounts, or one a reconciliation rule names (interest, a supplier's refund), is not made a
customer receipt. Credits that name no customer land on the suspense customer, for the cashier to clear."""
return statements.import_and_reconcile(caller, format, content, bank_account_id, currency)
@PAY.use_case(kind="command", route="POST /receipts/ingest-statement", permission=STATEMENT_INGEST, tool=False)
def ingest_statement(caller: dict, format: Literal["camt053", "mt940", "csv"], content: text(5_000_000, min_length=1),
bank_account_id: Code, currency: Optional[Currency] = None) -> dict:
"""The system imports a statement file a bank feed or a scheduled job delivered (workflow import_statement, by the
system): exactly the cashier's import, held by a service client with pay:statement:ingest (v4; it was pay:receipt:ingest)."""
return statements.import_and_reconcile(caller, format, content, bank_account_id, currency)
@PAY.use_case(kind="command", route="POST /statements/import", permission=IMPORT)
def import_bank_statement(caller: dict, format: Literal["camt053", "mt940", "csv"], content: text(5_000_000, min_length=1),
bank_account_id: Code, currency: Optional[Currency] = None) -> dict:
"""The bank_statement workflow's import (PAY-21): the same import as import_statement under the name the statement process uses -
a statement for a known bank account, a line per entry, a re-import adds nothing."""
return statements.import_and_reconcile(caller, format, content, bank_account_id, currency)
@PAY.use_case(kind="command", route="POST /receipts/ingest", permission=INGEST, idempotent=True, tool=False)
def ingest_provider_payment(caller: dict, provider: Code, payment_ref: Ref, bp_id: BigId, amount: Money, currency: Currency,
received_on: Day, invoice_id: Optional[Code] = None) -> dict:
"""A card or direct-debit provider confirms a payment (PAY-43): it becomes a receipt keyed
provider:<provider>:<payment ref> and, when it names an invoice, is allocated to it. A duplicate notification adds
nothing. The provider calls THIS; it never writes pay_ tables."""
with repository() as repo, _decide():
pid = f"provider:{provider}:{payment_ref}"
have = repo.receipts.get(pid)
if have is not None:
return {**_receipt_view(repo, have), "duplicate": True}
today, now, settings = _today(repo), _now(), repo.settings.current()
r = Receipt.record(payment_id=pid, bp_id=bp_id, currency=_cur(currency), amount=amount,
received_on=_date(received_on, "received_on"), today=today, tz=settings.business_timezone,
by=f"provider:{provider}", at=now, method=provider, bank_reference=payment_ref,
origin=f"provider:{provider}")
repo.receipts.add(r)
if not repo.receipts.ref_holder("", payment_ref):
repo.receipts.claim_ref("", payment_ref, pid, now)
done, made = None, []
if invoice_id:
d = seams.invoice(invoice_id)
if d is None:
raise Refused(404, f"no settleable invoice {invoice_id}")
out = stl.outstanding(d.total, repo.allocations.for_invoices([invoice_id]))
take = min(r.amount, out)
if take > ZERO:
done, made = _allocate_core(repo, {"email": f"provider:{provider}"}, source_kind=SourceKind.RECEIPT.value,
source_id=pid, source_currency=r.currency, source_bp=bp_id, source_remainder=r.amount,
lines=[LineIn(invoice_id=invoice_id, amount=take)], by=f"provider:{provider}")
return {**_receipt_view(repo, r, made, None), "duplicate": False, "allocation": done,
"ledger": _post_receipt(repo, r, actor=f"provider:{provider}")}
@PAY.use_case(kind="query", route="GET /receipts", permission=READ)
def list_receipts(caller: dict, bp_id: Optional[BigId] = None, status: Optional[Literal["received", "reversed"]] = None,
limit: Limit = 100, offset: Offset = 0) -> dict:
"""Receipts, newest first, with how much of each is still unallocated."""
with repository() as repo:
found = repo.receipts.search(bp_id=bp_id, status=status, limit=limit, offset=offset)
rows = repo.allocations.for_bps({r.bp_id for r in found if r.bp_id is not None})
open_ones = repo.suggestions.open_of_many(r.payment_id for r in found)
return {"receipts": [_receipt_view(repo, r, rows.get(r.bp_id, []), open_ones[r.payment_id]) for r in found]}
@PAY.use_case(kind="query", route="GET /receipts/auto-matches", permission=READ)
def auto_matches(caller: dict, on: Optional[Day] = None) -> dict:
"""In auto-with-review mode: the exact matches the system allocated on a day (today by default), for a person
to confirm (confirm_auto_match) or undo (unallocate). Each names the suggestion it came from."""
with repository() as repo:
day = _date(on, "on") or _today(repo)
out = []
found = repo.allocations.auto_on(day)
sugs = repo.suggestions.get_many(a.suggestion_id for a in found if a.suggestion_id)
for a in found:
sug = sugs.get(a.suggestion_id) if a.suggestion_id else None
out.append({**_alloc_view(a), "confidence": sug["confidence"] if sug else None,
"reviewed_by": sug["reviewed_by"] if sug else "", "to_review": not (sug and sug["reviewed_by"])})
return {"day": day.isoformat(), "auto_matches": out}
@PAY.use_case(kind="query", route="GET /receipts/{payment_id}", permission=READ)
def get_receipt(caller: dict, payment_id: Ref) -> dict:
"""One receipt with its allocations, what is unallocated, its open match suggestion and its readiness (one
{gate, met, reason, fix} per gate of its next transitions)."""
with repository() as repo:
r = _receipt(repo, payment_id)
return {**_receipt_view(repo, r), "readiness": _receipt_gates(repo, r, caller)}
def _receipt_gates(repo, r: Receipt, caller: dict) -> list:
"""The readiness of a receipt's next transitions, answered so no screen re-derives a rule."""
today, settings = _today(repo), repo.settings.current()
gates = [received_not_future(r.received_on, today, settings.business_timezone)]
holder = repo.receipts.ref_holder(r.bank_account_id, r.bank_reference) if r.bank_reference else None
gates.append(met("reference_unused") if holder in (None, r.payment_id) else
unmet("reference_unused", f"Bank reference {r.bank_reference} is already on receipt {holder}", f"Open receipt {holder}"))
rows = repo.allocations.for_bp(r.bp_id) if r.bp_id is not None else []
rem = _remainder(r, rows)
open_inv = []
try:
open_inv = _open_for_match(repo, r, caller) if r.bp_id is not None and r.live else []
down = ""
except seams.Unavailable as e:
down = e.reason
smallest = min((o.outstanding for o in open_inv), default=ZERO)
if not r.live:
gates.append(unmet("within_remainder", f"Receipt {r.payment_id} is reversed and holds nothing", "Record the payment again if the bank sends it"))
elif rem <= ZERO:
gates.append(unmet("within_remainder", f"Allocating {fmt(ZERO)}; the receipt has {fmt(rem)} left and the invoice {fmt(smallest)} outstanding",
"Allocate at most the smaller figure - undo an allocation first, or record another receipt"))
elif down:
gates.append(unmet("within_remainder", f"Invoicing could not be asked what is outstanding: {down}", "Try again"))
elif not open_inv:
gates.append(unmet("within_remainder", f"Allocating {fmt(rem)}; the receipt has {fmt(rem)} left and no {r.currency} invoice is outstanding",
"Allocate at most the smaller figure - there is nothing to allocate to yet; it stays as credit"))
else:
gates.append(met("within_remainder"))
gates.append(met("same_currency") if open_inv or not r.live else
unmet("same_currency", f"The receipt is {r.currency}, and the customer has no open {r.currency} invoice",
"Allocate a receipt in the invoice's currency"))
gates.append(met("within_tolerance") if settings.match_tolerance > ZERO else
unmet("within_tolerance", f"The difference is above the tolerance {fmt(settings.match_tolerance)}",
"Leave it outstanding, or ask the owner (pay:settings:controls)"))
sug = repo.suggestions.open_of(r.payment_id)
gates.append(met("suggestion_current") if sug else
unmet("suggestion_current", "There is no open suggestion for this receipt",
"Ask for a fresh suggestion, or allocate by hand"))
gates.append(return_dated_ok(today, r.received_on, today) if r.live else
unmet("return_dated_ok", f"The receipt was already returned on {r.returned_on}", "Nothing more to do"))
return [g.as_dict() for g in gates]
@PAY.use_case(kind="command", route="POST /receipts/{payment_id}/suggest-match", permission=MATCH, tool=False)
def suggest_match(caller: dict, payment_id: Ref) -> dict:
"""The system suggests which invoices this receipt pays (a suggestion row with its evidence, confidence and model;
nothing is allocated unless ai_suggest_match is auto-with-review and the match is exact). Running it again
supersedes an open suggestion and never a decided one."""
with repository() as repo, _decide():
r = _receipt(repo, payment_id)
got = _suggest(repo, r, caller, repo.settings.current().ai_suggest_match)
got.pop("_rows", None)
return got
@PAY.use_case(kind="command", route="POST /receipts/{payment_id}/decide-match", permission=ALLOC)
def decide_match(caller: dict, payment_id: Ref, decision: Literal["accept", "change", "reject"],
suggestion_id: Optional[Ref] = None, lines: Optional[list[LineIn]] = None) -> dict:
"""Accept, change or reject the AI's suggested match for a receipt. Accept allocates what it proposed; change
allocates the lines you give (the row keeps both what was proposed and what you chose); reject allocates nothing.
Every allocation made this way cites the suggestion and the person who decided."""
with repository() as repo, _decide():
r = _receipt(repo, payment_id)
sug = repo.suggestions.get(suggestion_id) if suggestion_id else repo.suggestions.open_of(payment_id)
if sug is None or sug["payment_id"] != payment_id:
raise Refused(404, f"no suggestion for receipt {payment_id}" + (f" named {suggestion_id}" if suggestion_id else ""))
return _decide_core(repo, caller, r, sug, decision, lines)
def _decide_core(repo, caller: dict, r: Receipt, sug: dict, decision: str, lines: Optional[list]) -> dict:
"""The decision on one suggestion (PAY-14); a remittance's suggestion (PAY-17) is decided the same way and records
decided_how remittance. A decided suggestion is final: a late second decision is refused."""
payment_id = r.payment_id
if sug["status"] != "open":
raise RuleBroken(f"Suggestion {sug['suggestion_id']} is already {sug['status']}. Ask for a fresh suggestion, or allocate by hand")
now, who = _now(), _who(caller)
from_remittance = sug["model"] == "remittance"
if decision == "reject":
repo.suggestions.move(sug["suggestion_id"], status="rejected", decided_by=who, decided_at=now)
if from_remittance:
_remittance_state(repo, payment_id, "rejected")
return {"decision": "rejected", "suggestion_id": sug["suggestion_id"], "receipt": _receipt_view(repo, r, None, None)}
proposed = [LineIn(invoice_id=ln["invoice_id"], amount=Decimal(ln["amount"])) for ln in sug["proposed_lines"]]
chosen = proposed
if decision == "change":
if not lines:
raise RuleBroken("Changing a suggestion needs the lines you choose instead. Give at least one invoice and amount")
chosen = lines
if not r.live or r.bp_id is None:
raise RuleBroken(f"Receipt {payment_id} is reversed or not on a customer. Nothing to allocate")
rows = repo.allocations.for_bp(r.bp_id)
if decision == "accept": # gate suggestion_current: the amounts it proposed still hold
docs = {d.invoice_id: d for d in seams.invoices(ids=[ln.invoice_id for ln in proposed], caller=caller)}
by = _group(repo.allocations.for_invoices(list(docs)))
for ln in proposed:
d = docs.get(ln.invoice_id)
if d is None or stl.outstanding(d.total, by.get(ln.invoice_id, [])) < ln.amount or _remainder(r, rows) < sum(
(p.amount for p in proposed), ZERO):
refuse(unmet("suggestion_current", f"Since the suggestion, {ln.invoice_id} was settled or the receipt allocated elsewhere",
"Ask for a fresh suggestion, or allocate by hand"))
how = "remittance" if from_remittance and decision == "accept" else "accepted" if decision == "accept" else "changed"
done, _ = _allocate_core(repo, caller, source_kind=SourceKind.RECEIPT.value, source_id=r.payment_id,
source_currency=r.currency, source_bp=r.bp_id, source_remainder=_remainder(r, rows), lines=chosen,
decided_how=how, suggestion_id=sug["suggestion_id"])
repo.suggestions.move(sug["suggestion_id"], status="accepted" if decision == "accept" else "changed", decided_by=who,
decided_at=now, **({"decided_lines": [{"invoice_id": c.invoice_id, "amount": _s(c.amount)} for c in chosen]}
if decision == "change" else {}))
if from_remittance:
_remittance_state(repo, payment_id, "applied")
return {"decision": "accepted" if decision == "accept" else "changed", "suggestion_id": sug["suggestion_id"], **done}
def _remittance_state(repo, payment_id: str, status: str) -> None:
"""The remittance lines of a receipt move forward with the decision on their suggestion (imported -> applied | rejected)."""
for ln in repo.remittances.of_payment(payment_id):
if ln["status"] == "imported":
repo.remittances.move(ln["remittance_id"], ln["line_no"], status=status)
@PAY.use_case(kind="command", route="POST /receipts/{payment_id}/confirm-auto-match", permission=ALLOC, idempotent=True)
def confirm_auto_match(caller: dict, payment_id: Ref) -> dict:
"""A person confirms a match the system made in auto-with-review mode. To undo it instead, use unallocate."""
with repository() as repo, _decide():
_receipt(repo, payment_id)
autos = [s for s in repo.suggestions.of_receipt(payment_id) if s["status"] == "accepted" and s["decided_by"] == "auto"]
if not autos:
raise Refused(404, f"no automatic match to review on receipt {payment_id}")
for s in autos:
repo.suggestions.move(s["suggestion_id"], reviewed_by=_who(caller), reviewed_at=_now())
return {"payment_id": payment_id, "reviewed": [s["suggestion_id"] for s in autos], "reviewed_by": _who(caller)}
@PAY.use_case(kind="command", route="POST /receipts/{payment_id}/allocate", permission=ALLOC)
def allocate(caller: dict, payment_id: Ref, lines: list[LineIn]) -> dict:
"""Allocate a receipt across invoices: one receipt can settle several. Each allocation is dated today. Allocating
more than the receipt has left, or more than an invoice has outstanding, is refused naming both figures; the
receipt and the invoice must be in the same currency and the same customer's. A shortfall within the
tolerance is proposed to invoicing as a write-off."""
with repository() as repo, _decide():
r = _receipt(repo, payment_id)
if not r.live:
raise RuleBroken(f"Receipt {payment_id} was returned on {r.returned_on}. A returned payment settles nothing")
rows = repo.allocations.for_bp(r.bp_id) if r.bp_id is not None else []
done, _ = _allocate_core(repo, caller, source_kind=SourceKind.RECEIPT.value, source_id=r.payment_id,
source_currency=r.currency, source_bp=r.bp_id, source_remainder=_remainder(r, rows), lines=lines)
if r.bp_id is not None: # a hand allocation closes the open suggestion it overtakes
cur = repo.suggestions.open_of(payment_id)
if cur:
repo.suggestions.move(cur["suggestion_id"], status="superseded")
return {"payment_id": payment_id, **done}
@PAY.use_case(kind="command", route="POST /credit-notes/{credit_note_id}/allocate", permission=ALLOC)
def allocate_credit_note(caller: dict, credit_note_id: Ref, lines: list[LineIn]) -> dict:
"""Apply a customer's unallocated credit note to their invoices (same currency, dated today). A credit is never
applied across currencies."""
with repository() as repo, _decide():
notes = {n.credit_note_id: n for n in _credit_notes(None, caller)}
n = notes.get(credit_note_id)
if n is None:
raise Refused(404, f"no credit note {credit_note_id}")
rows = repo.allocations.for_bp(n.bp_id)
rem = stl.source_remainder(n.total, SourceKind.CREDIT_NOTE.value, n.credit_note_id, rows)
done, _ = _allocate_core(repo, caller, source_kind=SourceKind.CREDIT_NOTE.value, source_id=n.credit_note_id,
source_currency=n.currency, source_bp=n.bp_id, source_remainder=rem, lines=lines, what="credit note")
return {"credit_note_id": credit_note_id, **done}
@PAY.use_case(kind="command", route="POST /receipts/{payment_id}/settle-difference", permission=ALLOC)
def settle_difference(caller: dict, payment_id: Ref, invoice_id: Code) -> dict:
"""Propose a small difference left on an invoice this receipt paid (a bank charge, a rounding) to invoicing as a
write-off. Only within match_tolerance (0.00 = off); above it the difference stays outstanding."""
with repository() as repo, _decide():
_receipt(repo, payment_id)
d = seams.invoice(invoice_id, caller)
if d is None:
raise Refused(404, f"no settleable invoice {invoice_id}")
rows = repo.allocations.for_invoices([invoice_id])
if not any(a.source_kind == "receipt" and a.source_id == payment_id and a.amount > ZERO for a in rows):
raise RuleBroken(f"Receipt {payment_id} has not paid invoice {d.number}. Allocate it first")
diff, tol = stl.outstanding(d.total, rows), repo.settings.current().match_tolerance
if diff <= ZERO:
raise RuleBroken(f"Invoice {d.number} has nothing outstanding. Nothing to write off")
refuse(met("within_tolerance") if diff <= tol else
unmet("within_tolerance", f"The difference {fmt(diff)} is above the tolerance {fmt(tol)}",
"Leave it outstanding, or ask the owner (pay:settings:controls)"))
got = seams.propose_writeoff(invoice_id, diff, d.currency, f"difference within tolerance {fmt(tol)}", f"receipt:{payment_id}", _who(caller))
return {"invoice_id": invoice_id, "difference": _s(diff), "tolerance": _s(tol), **got}
@PAY.use_case(kind="command", route="POST /receipts/{payment_id}/unallocate", permission=ALLOC)
def unallocate(caller: dict, payment_id: Ref, invoice_id: Code, seq: Seq, reason: Reason) -> dict:
"""Undo one allocation of a receipt with a reason: a reversing row (negative, same source, dated today, naming the
row it reverses) is added and both are kept; the invoice's outstanding rises again from today. Nothing is deleted
or edited, so a statement sent for an earlier date still re-computes to what it said."""
with repository() as repo, _decide():
_receipt(repo, payment_id)
pair = repo.allocations.pair(SourceKind.RECEIPT.value, payment_id, invoice_id)
original = next((a for a in pair if a.seq == seq), None)
if original is None:
raise Refused(404, f"receipt {payment_id} has no allocation {seq} to invoice {invoice_id}")
row = alloc.reverse_row(original, pair, next_seq=repo.allocations.next_seq(SourceKind.RECEIPT.value, payment_id, invoice_id),
today=_today(repo), by=_who(caller), reason=reason)
repo.allocations.claim(invoices=[invoice_id], sources=[(SourceKind.RECEIPT.value, payment_id)], at=_now()) # collides with an allocate or a reverse
repo.allocations.add([row], _now())
rcpt = repo.receipts.get(payment_id)
undone = fx.post_allocations(repo, rcpt, [row], {}, _who(caller), _now()) if rcpt is not None else [] # PAY-33: fx_reversal
return {"payment_id": payment_id, "reversal": _alloc_view(row), "reverses": _alloc_view(original), **({"fx": undone} if undone else {})}
@PAY.use_case(kind="command", route="POST /receipts/{payment_id}/reverse", permission=WRITE, idempotent=True)
def reverse_receipt(caller: dict, payment_id: Ref, bank_return_ref: Ref, returned_on: Optional[Day] = None) -> dict:
"""Reverse a returned payment (a bounced direct debit, a recalled transfer): the receipt is reversed on the bank's
return date (today or earlier, not before it was received), its allocations are reversed with it by reversing
rows, so the invoices it paid are owed again, and the reversal posts in the ledger. A receipt part of which was
already refunded cannot be reversed."""
with repository() as repo, _decide():
r = _receipt(repo, payment_id)
if not r.live:
return {**_receipt_view(repo, r), "unchanged": True}
return reverse_core(repo, caller, r, bank_return_ref, _date(returned_on, "returned_on"))
def reverse_core(repo, caller: dict, r: Receipt, bank_return_ref: str, returned_on: Optional[date], reason_code: str = "") -> dict:
"""PAY-07, shared with the direct-debit returns (PAY-20). It takes the SAME source claim an allocation of this receipt takes, so a
reversal and an allocation racing for one receipt collide (409) instead of both landing."""
payment_id = r.payment_id
today, now = _today(repo), _now()
rows = repo.allocations.for_bp(r.bp_id) if r.bp_id is not None else []
ref = stl.credit_ref("receipt", payment_id)
if any(a.source_kind == "refund" and a.invoice_id == ref for a in rows):
raise RuleBroken(f"Part of receipt {payment_id} was already refunded. The refund went out; reverse the receipt "
"only once the money is back")
r.reverse(returned_on=returned_on or today, bank_return_ref=bank_return_ref, today=today, reason_code=reason_code)
made = []
touched = sorted({a.invoice_id for a in rows if a.source_kind == "receipt" and a.source_id == payment_id})
for inv in touched:
pair = [a for a in rows if (a.source_kind, a.source_id, a.invoice_id) == ("receipt", payment_id, inv)]
seq = repo.allocations.next_seq("receipt", payment_id, inv)
for a in pair:
if a.amount > ZERO and not any(x.reverses_seq == a.seq for x in pair):
made.append(alloc.reverse_row(a, pair, next_seq=seq, today=today, by=_who(caller),
reason=f"payment returned: {bank_return_ref}"))
seq += 1
repo.allocations.claim(invoices=touched if made else [], sources=[("receipt", payment_id)], at=now)
repo.allocations.add(made, now)
fx.post_allocations(repo, r, made, {}, _who(caller), now) # PAY-33: the returned payment's FX differences come back
repo.receipts.save(r)
cur = repo.suggestions.open_of(payment_id)
if cur:
repo.suggestions.move(cur["suggestion_id"], status="superseded")
posted = _post_receipt(repo, r, reverse=True, actor=_who(caller))
view = r.as_dict()
return {**view, "allocation_reversals": [_alloc_view(a) for a in made], "unallocated": _s(ZERO), "ledger": posted}
@PAY.use_case(kind="command", route="POST /receipts/{payment_id}/assign-customer", permission=WRITE, idempotent=True)
def assign_receipt_customer(caller: dict, payment_id: Ref, bp_id: BigId) -> dict:
"""Clear money that landed on no customer, or on the suspense customer (an imported credit, a migrated bank line
nobody named): put it on the customer who paid. Only a receipt with nothing allocated can move."""
with repository() as repo, _decide():
r = _receipt(repo, payment_id)
suspense = repo.settings.current().suspense_bp_id
if r.bp_id == bp_id:
return {**_receipt_view(repo, r), "unchanged": True}
if r.bp_id is not None and r.bp_id != suspense:
raise RuleBroken(f"Receipt {payment_id} is already on customer {r.bp_id}. Only money on no customer, or on the "
"suspense customer, can be moved: reverse the receipt and record it again")
if any(a.source_kind == "receipt" and a.source_id == payment_id for a in (repo.allocations.for_bp(r.bp_id) if r.bp_id else [])):
raise RuleBroken(f"Receipt {payment_id} already has allocations. Undo them first")
known = seams.check_customer(bp_id, caller)
if known.status != "known":
raise Refused(known.http, known.note)
r.bp_id = bp_id
repo.receipts.clear_suspense(payment_id, bp_id, _who(caller))
return _receipt_view(repo, r)
@PAY.use_case(kind="command", route="POST /allocations/from-source", permission=FROM_SOURCE, idempotent=True, tool=False)
def allocate_from_source(caller: dict, source_kind: Literal["credit_note", "writeoff"], source_id: Ref, invoice_id: Code,
amount: Money, seq: Seq = 1) -> dict:
"""Invoicing allocates an approved credit note or write-off against an invoice (PAY-41), so every reduction of a
balance is in this one table. Calling twice with the same (source, invoice, seq) adds nothing."""
with repository() as repo, _decide():
key = (source_kind, source_id, invoice_id, seq)
have = repo.allocations.get(*key)
if have is not None:
return {"created": False, "allocation": _alloc_view(have)}
d = seams.invoice(invoice_id)
if d is None:
raise Refused(404, f"no settleable invoice {invoice_id}")
rows = repo.allocations.for_invoices([invoice_id])
out = stl.outstanding(d.total, rows)
if source_kind == "credit_note":
note = next((n for n in _credit_notes(d.bp_id, None) if n.credit_note_id == source_id), None)
if note is None:
raise Refused(404, f"no credit note {source_id}")
if note.currency != d.currency:
refuse(alloc.same_currency(note.currency, d.currency))
rem = stl.source_remainder(note.total, "credit_note", source_id, repo.allocations.for_bp(d.bp_id))
else:
rem = out
row = alloc.plan(source_kind=source_kind, source_id=source_id, source_currency=d.currency, source_remainder=rem,
lines=[alloc.Line(invoice_id, amount, out, d.currency, d.bp_id)], next_seq={invoice_id: seq},
today=_today(repo), by=_who(caller) or "invoicing", what="credit note" if source_kind == "credit_note" else "write-off")[0]
repo.allocations.claim(invoices=[invoice_id], sources=[(source_kind, source_id)], at=_now())
repo.allocations.add([row], _now())
return {"created": True, "allocation": _alloc_view(row), "outstanding": _s(out - row.amount)}
# ═══ refunds ═════════════════════════════════════════════════════════════════════════════════════════════
def _refund_view(r: Refund) -> dict:
return {"refund_id": r.refund_id, "bp_id": r.bp_id, "currency": r.currency, "amount": _s(r.amount), "status": r.status,
"requested_by": r.requested_by, "requested_at": r.requested_at.isoformat() if r.requested_at else None,
"approved_by": r.approved_by, "approved_at": r.approved_at.isoformat() if r.approved_at else None,
"reason": r.reason, "bank_ref": r.bank_ref, "paid_on": r.paid_on.isoformat() if r.paid_on else None, "origin": r.origin}
def _refund(repo, refund_id: str) -> Refund:
r = repo.refunds.get(refund_id)
if r is None:
raise Refused(404, f"no refund {refund_id}")
return r
@PAY.use_case(kind="command", route="POST /refunds", permission=REFUND_REQUEST)
def request_refund(caller: dict, bp_id: BigId, amount: Money, currency: Currency, reason: Optional[Reason] = None) -> dict:
"""Ask to pay a customer's unallocated credit (an overpayment, a credit note) back to them. Refunding more than
their credit in that currency is refused; credit already asked for in another open refund is not counted twice.
Nothing moves until an approver, who is not you, approves it."""
with repository() as repo, _decide():
cur = _cur(currency)
r = Refund.request(refund_id=str(uuid.uuid4()), bp_id=bp_id, currency=cur, amount=amount,
credit=_credit_available(repo, bp_id, cur, caller), by=_who(caller), at=_now(), reason=reason or "",
sub=(caller.get("sub") or ""))
repo.refunds.add(r)
return _refund_view(r)
@PAY.use_case(kind="command", route="POST /refunds/{refund_id}/approve", permission=REFUND_APPROVE, idempotent=True)
def approve_refund(caller: dict, refund_id: Ref) -> dict:
"""Approve a requested refund. Money going out always needs a second person: the requester cannot approve their
own refund, and the credit must still cover it."""
with repository() as repo, _decide():
r = _refund(repo, refund_id)
if r.status == "approved" and r.approved_by == _who(caller):
return _refund_view(r)
credit_now = _credit_available(repo, r.bp_id, r.currency, caller) + (r.amount if r.status == "requested" else ZERO)
r.approve(by=_who(caller), credit=credit_now, at=_now(), on=_today(repo), identities=_ids(caller))
repo.refunds.save(r)
return _refund_view(r)
@PAY.use_case(kind="command", route="POST /refunds/{refund_id}/reject", permission=REFUND_APPROVE, idempotent=True)
def reject_refund(caller: dict, refund_id: Ref, reason: Reason) -> dict:
"""Reject a requested refund, with the reason. The credit stays on the customer."""
with repository() as repo, _decide():
r = _refund(repo, refund_id)
if r.status == "rejected":
return _refund_view(r)
r.reject(by=_who(caller), reason=reason)
repo.refunds.save(r)
return _refund_view(r)
@PAY.use_case(kind="command", route="POST /refunds/{refund_id}/pay", permission=REFUND_REQUEST, idempotent=True)
def pay_refund(caller: dict, refund_id: Ref, bank_ref: Ref, paid_on: Optional[Day] = None,
bank_account_id: Optional[Code] = None) -> dict:
"""Record an approved refund paid, with the bank's reference: it is allocated against the customer's credit (the
oldest credit first), so their credit falls by the amount, and posts in the ledger on the bank's payment date
(today or earlier, not before the approval)."""
with repository() as repo, _decide():
r = _refund(repo, refund_id)
if r.status == "paid":
return _refund_view(r)
today, now = _today(repo), _now()
on = _date(paid_on, "paid_on") or today
r.pay(paid_on=on, bank_ref=bank_ref, today=today)
need, rows, drawn = r.amount, [], []
for c in sorted((c for c in _credits(repo, r.bp_id, caller) if c["currency"] == r.currency), key=lambda c: (c["date"] or today, c["id"])):
take = min(need, c["remainder"])
if take > ZERO:
ref = stl.credit_ref(c["kind"], c["id"])
rows.append(Allocation("refund", r.refund_id, ref, repo.allocations.next_seq("refund", r.refund_id, ref), take,
r.currency, today, _who(caller), "refund paid", None, "", "manual", "", r.bp_id))
need -= take
drawn.append((c["kind"], c["id"]))
if need > ZERO:
refuse(unmet("credit_covers", f"Refund {fmt(r.amount)}; the customer's unallocated credit in {r.currency} is {fmt(r.amount - need)}",
f"Refund at most {fmt(r.amount - need)}"))
repo.allocations.claim(invoices=[], sources=drawn, at=now)
repo.allocations.add(rows, now)
repo.refunds.save(r)
bank = repo.config.bank(bank_account_id) if bank_account_id else None
if bank and bank.get("gl_account_id"):
amt = _s(r.amount)
posted = seams.post_journal(r.refund_id, "refund", on, r.currency, f"Refund {r.bank_ref}",
[{**seams.receivable_line(r.bp_id), "debit": amt},
{**seams.account_line(bank["gl_account_id"]), "credit": amt}], _who(caller),
bank.get("company_code") or "")
else:
posted = {"posted": False, "reason": "no bank account with a ledger account was named, so nothing was posted"}
return {**_refund_view(r), "allocations": [_alloc_view(a) for a in rows], "ledger": posted}
@PAY.use_case(kind="query", route="GET /refunds", permission=READ)
def list_refunds(caller: dict, bp_id: Optional[BigId] = None, status: Optional[Literal["requested", "approved", "rejected", "paid"]] = None,
limit: Limit = 100) -> dict:
"""Refunds, newest first."""
with repository() as repo:
return {"refunds": [_refund_view(r) for r in repo.refunds.search(bp_id=bp_id, status=status, limit=limit)]}
@PAY.use_case(kind="query", route="GET /refunds/{refund_id}", permission=READ)
def get_refund(caller: dict, refund_id: Ref) -> dict:
"""One refund with its readiness: one {gate, met, reason, fix} per gate of its next transitions."""
with repository() as repo, _decide():
r = _refund(repo, refund_id)
who = _who(caller)
credit_now = _credit_available(repo, r.bp_id, r.currency, caller) + (r.amount if r.status in ("requested", "approved") else ZERO)
gates = [refund_module.credit_covers(r.amount, credit_now, r.currency),
refund_module.second_person(r.requested_by, who, r.requested_sub, _ids(caller)),
refund_module.paid_dated_ok(_today(repo), r.approved_on, _today(repo))]
return {**_refund_view(r), "readiness": [g.as_dict() for g in gates]}
# ═══ collections: the overdue list, promises, disputes, reminders, statements ═══════════════════════════
def _naive_now() -> datetime:
return _now().astimezone(ZoneInfo("UTC")).replace(tzinfo=None)
def _active_promise(events: list, today: date) -> Optional[date]:
dates = [e.promise_date for e in events if e.kind == "promise" and e.promise_date and e.promise_date >= today]
return max(dates) if dates else None
@PAY.use_case(kind="query", route="GET /collections/overdue", permission=READ)
def overdue_list(caller: dict, bp_id: Optional[BigId] = None, bucket: Optional[text(10)] = None, limit: Limit = 200) -> dict:
"""What is overdue and by how long: the customer's unpaid invoices past their due date, in the configured aging
buckets, aged from the basis the business chose, with customer, outstanding, age, last contact, any promise or
dispute, the state (overdue, reminded, promised, disputed) and - when late_payment_risk is on - the risk."""
with repository() as repo, _decide():
today, settings = _today(repo), repo.settings.current()
docs = seams.invoices(bp_id=bp_id, caller=caller)
by = _group(repo.allocations.for_invoices([d.invoice_id for d in docs]))
events = repo.collections.all_events() if bp_id is None else {bp_id: repo.collections.events_of(bp_id)}
history: dict = {}
for d in docs:
if stl.outstanding(d.total, by.get(d.invoice_id, [])) <= ZERO and d.due_date:
last = max((a.allocated_on for a in by.get(d.invoice_id, []) if a.amount > ZERO), default=None)
h = history.setdefault(d.bp_id, [0, 0])
h[0] += 1
h[1] += bool(last and last > d.due_date)
items, totals, labels = [], {}, stl.bucket_labels(settings.aging_buckets)
for d in docs:
rows = by.get(d.invoice_id, [])
out = stl.outstanding(d.total, rows)
if out <= ZERO or d.due_date is None or d.due_date >= today:
continue
v = _invoice_view(d, rows, today, settings)
if bucket and v["bucket"] != bucket:
continue
ev = events.get(d.bp_id, [])
last = max((e.occurred_on for e in ev if e.occurred_on and (e.invoice_id in ("", d.invoice_id))), default=None)
promise = _active_promise(ev, today)
v.update({"state": coll.state_of(outstanding=out, due_date=d.due_date, today=today, events=ev, invoice_id=d.invoice_id),
"last_contact": last.isoformat() if last else None, "promise_until": promise.isoformat() if promise else None,
"level_reached": max((e.level for e in ev if e.kind == "reminder" and e.invoice_id == d.invoice_id), default=0)})
if settings.late_payment_risk != "off":
paid, late = history.get(d.bp_id, [0, 0])
v["late_payment_risk"] = ai_steps.risk(v["age_days"], late, paid)
items.append(v)
k = (v["bucket"], v["currency"])
totals[k] = totals.get(k, ZERO) + out
items.sort(key=lambda v: (-v["age_days"], -Decimal(v["outstanding"]), v["invoice_id"]))
return {"as_at": today.isoformat(), "age_from": settings.age_from, "invoices": items[:limit],
"buckets": [{"bucket": b, "currency": c, "outstanding": _s(t)}
for (b, c), t in sorted(totals.items(), key=lambda kv: (labels.index(kv[0][0]), kv[0][1]))]}
@PAY.use_case(kind="command", route="POST /collections/promises", permission=EVENT)
def record_promise(caller: dict, bp_id: BigId, promise_date: Day, note: Optional[Reason] = None) -> dict:
"""Record what a customer promised: one promise to pay on a date, recorded once against the customer, shown on
each of their overdue invoices, and suppressing reminders until that date. A promise dated today or earlier is
refused."""
with repository() as repo, _decide():
today, now = _today(repo), _now()
on = _date(promise_date, "promise_date")
refuse(coll.promise_in_future(on, today))
level = repo.collections.next_level(bp_id, "promise")
repo.collections.add({"bp_id": bp_id, "kind": "promise", "level": level, "ref_id": on.isoformat(), "invoice_id": "",
"occurred_on": today, "promise_date": on, "reason": "", "outcome": "", "note": note or "",
"draft_text": "", "status": "", "origin": "", "recorded_by": _who(caller)}, now)
return {"bp_id": bp_id, "promise_date": on.isoformat(), "recorded_on": today.isoformat(), "reminders_suppressed_until": on.isoformat()}
def open_dispute(repo, bp_id: int, invoice_id: str, reason: str, who: str) -> Optional[int]:
"""An open dispute on an invoice with the customer's reason (reminders for it are suppressed); None when one is already open."""
if coll.open_disputes(repo.collections.events_of(bp_id), invoice_id):
return None
level = repo.collections.next_level(bp_id, "dispute")
repo.collections.add({"bp_id": bp_id, "kind": "dispute", "level": level, "ref_id": invoice_id, "invoice_id": invoice_id,
"occurred_on": _today(repo), "promise_date": None, "reason": reason, "outcome": "", "note": "",
"draft_text": "", "status": "open", "origin": "", "recorded_by": who}, _now())
return level
@PAY.use_case(kind="command", route="POST /collections/disputes", permission=EVENT)
def record_dispute(caller: dict, bp_id: BigId, invoice_id: Code, reason: Reason) -> dict:
"""Record that a customer disputes an invoice, with the reason: reminders for it are suppressed until the dispute
is resolved."""
with repository() as repo, _decide():
if not reason.strip():
raise RuleBroken("A dispute needs the customer's reason. Write what they said")
if coll.open_disputes(repo.collections.events_of(bp_id), invoice_id):
raise RuleBroken(f"Invoice {invoice_id} already has an open dispute. Resolve it first")
level = repo.collections.next_level(bp_id, "dispute")
repo.collections.add({"bp_id": bp_id, "kind": "dispute", "level": level, "ref_id": invoice_id, "invoice_id": invoice_id,
"occurred_on": _today(repo), "promise_date": None, "reason": reason, "outcome": "", "note": "",
"draft_text": "", "status": "open", "origin": "", "recorded_by": _who(caller)}, _now())
return {"bp_id": bp_id, "invoice_id": invoice_id, "dispute": level, "reminders_suppressed": True}
@PAY.use_case(kind="command", route="POST /collections/disputes/resolve", permission=EVENT, idempotent=True)
def resolve_dispute(caller: dict, bp_id: BigId, invoice_id: Code, outcome: Reason) -> dict:
"""Resolve a customer's dispute of an invoice with its outcome (a credit note in invoicing, or reminders resume)."""
with repository() as repo, _decide():
events = repo.collections.events_of(bp_id)
open_ = coll.open_disputes(events, invoice_id)
if not open_:
if any(e.kind == "resolution" and e.invoice_id == invoice_id for e in events):
return {"bp_id": bp_id, "invoice_id": invoice_id, "resolved": True, "unchanged": True}
raise Refused(404, f"no open dispute on invoice {invoice_id} for customer {bp_id}")
if not outcome.strip():
raise RuleBroken("Resolving a dispute needs its outcome. Say what was agreed")
lvl = open_[-1].level
repo.collections.add({"bp_id": bp_id, "kind": "resolution", "level": lvl, "ref_id": invoice_id, "invoice_id": invoice_id,
"occurred_on": _today(repo), "promise_date": None, "reason": "", "outcome": outcome, "note": "",
"draft_text": "", "status": "resolved", "origin": "", "recorded_by": _who(caller)}, _now())
return {"bp_id": bp_id, "invoice_id": invoice_id, "resolved": True, "outcome": outcome}
@PAY.use_case(kind="query", route="GET /collections/reminders", permission=READ)
def reminder_drafts(caller: dict, bp_id: Optional[BigId] = None) -> dict:
"""The reminder drafts waiting for a collector to send (mode suggest), with the text the level's template made."""
with repository() as repo:
rows = [r for r in repo.collections.drafts() if bp_id is None or r["bp_id"] == bp_id]
return {"drafts": [{"bp_id": r["bp_id"], "invoice_id": r["invoice_id"], "level": r["level"], "draft_text": r["draft_text"],
"created_on": r["occurred_on"].isoformat() if r["occurred_on"] else None} for r in rows]}
@PAY.use_case(kind="command", route="POST /collections/reminders/send", permission=REMINDER, idempotent=True)
def send_reminder(caller: dict, bp_id: BigId, invoice_id: Code, level: Annotated[int, Field(ge=1, le=20)]) -> dict:
"""Send (approve) a drafted reminder: the draft for this invoice and level becomes sent. A promise or an open
dispute suppresses it. Sending it again changes nothing."""
with repository() as repo, _decide():
row = repo.collections.get(bp_id, "reminder", level, invoice_id)
if row is None:
raise Refused(404, f"no reminder draft at level {level} for invoice {invoice_id}")
if row["status"] == "sent":
return {"bp_id": bp_id, "invoice_id": invoice_id, "level": level, "status": "sent", "unchanged": True}
refuse(coll.not_suppressed(repo.collections.events_of(bp_id), invoice_id, _today(repo)))
repo.collections.move(bp_id, "reminder", level, invoice_id, status="sent", occurred_on=_today(repo), recorded_by=_who(caller))
return {"bp_id": bp_id, "invoice_id": invoice_id, "level": level, "status": "sent", "draft_text": row["draft_text"], "sent_by": _who(caller)}
@PAY.use_case(kind="command", route="POST /collections/tick", permission=TICK, idempotent=True, tool=False)
def collections_tick(caller: dict) -> dict:
"""The daily collections pass: for each overdue invoice, the next reminder level that is due is drafted for the
collector (reminder_mode suggest) or sent (auto), once per (customer, invoice, level) - a second pass adds
nothing. A promise or open dispute suppresses it; with reminders off nothing happens. Also reports promises that
lapsed unpaid and invoices settled since."""
with repository() as repo, _decide():
today, now, settings = _today(repo), _now(), repo.settings.current()
out = {"day": today.isoformat(), "overdue": 0, "drafted": 0, "sent": 0, "suppressed": 0, "lapsed_promises": 0,
"settled_since": 0, "mode": settings.reminder_mode}
docs = seams.invoices()
by = _group(repo.allocations.for_invoices([d.invoice_id for d in docs]))
events = repo.collections.all_events()
levels = repo.config.levels()
lapsed = set()
for d in docs:
out_ = stl.outstanding(d.total, by.get(d.invoice_id, []))
ev = events.get(d.bp_id, [])
if out_ <= ZERO:
out["settled_since"] += any(e.kind == "reminder" and e.invoice_id == d.invoice_id for e in ev)
continue
if d.due_date is None or d.due_date >= today:
continue
out["overdue"] += 1
if any(e.kind == "promise" and e.promise_date and e.promise_date < today for e in ev) and not _active_promise(ev, today):
lapsed.add(d.bp_id)
if settings.reminder_mode == "off":
continue
if coll.suppression(ev, d.invoice_id, today):
out["suppressed"] += 1
continue
lv = coll.next_level(levels, ev, d.invoice_id, d.due_date, today)
if lv is None:
continue
auto = settings.reminder_mode == "auto" and lv.mode == "auto"
text_ = ai_steps.draft_reminder(lv.template, customer=str(d.bp_id), invoice_number=d.number, outstanding=out_,
currency=d.currency, due_date=d.due_date, level=lv.level)
repo.collections.add({"bp_id": d.bp_id, "kind": "reminder", "level": lv.level, "ref_id": d.invoice_id,
"invoice_id": d.invoice_id, "occurred_on": today, "promise_date": None, "reason": "", "outcome": "",
"note": lv.channel, "draft_text": text_, "status": "sent" if auto else "draft", "origin": "",
"recorded_by": "system"}, now)
ev.append(coll.Event("reminder", lv.level, d.invoice_id, d.invoice_id, today, None, "sent" if auto else "draft"))
out["sent" if auto else "drafted"] += 1
out["lapsed_promises"] = len(lapsed)
return out
@PAY.use_case(kind="command", route="POST /statements", permission=STATEMENT)
def produce_statement(caller: dict, bp_id: BigId, as_at: Day) -> dict:
"""Produce an open-item statement for a customer as at a date: every document unsettled at the end of that day
(documents and allocations dated on or before it), with the closing balance per currency, and record that it was
sent. Produced again later for the same date it shows the same balance, whatever was allocated since."""
with repository() as repo, _decide():
today, now = _today(repo), _now()
on = _date(as_at, "as_at")
if on > today:
raise RuleBroken(f"A statement is as at a day that has ended; {on} is after today ({today}). Use today or an earlier date")
got = _open_items(repo, bp_id, on, caller)
sent = []
for cur, bal in got["closing_balance"].items():
have = repo.statements.get(bp_id, on, cur)
if have is None:
repo.statements.add({"bp_id": bp_id, "as_at": on, "currency": cur, "closing_balance": Decimal(bal),
"sent_at": _naive_now(), "sent_by": _who(caller)}, now)
sent.append({"currency": cur, "closing_balance": bal, "recorded": True, "matches_recorded": True})
else:
same = Decimal(str(have["closing_balance"])) == Decimal(bal)
sent.append({"currency": cur, "closing_balance": bal, "recorded": False, "recorded_balance": _s(have["closing_balance"]),
"matches_recorded": same})
return {**got, "sent": sent}
@PAY.use_case(kind="query", route="GET /customers/{bp_id}/statements", permission=READ)
def list_statements(caller: dict, bp_id: BigId) -> dict:
"""The statements sent to a customer: the date each is as at, its closing balance and when it was sent."""
with repository() as repo:
return {"bp_id": bp_id, "statements": [{"as_at": s["as_at"].isoformat(), "currency": s["currency"],
"closing_balance": _s(s["closing_balance"]), "sent_by": s["sent_by"],
"sent_at": s["sent_at"].isoformat() if s["sent_at"] else None}
for s in repo.statements.of(bp_id)]}
# ═══ credit: limits, holds, overrides ════════════════════════════════════════════════════════════════════
def _credit_status(repo, bp_id: int, extra_overrides: tuple = ()) -> dict:
"""Derived: the limit in force today, the net balance, the oldest overdue invoice and the people's overrides.
Invoices are read as payments' own service identity - sales sees this narrow projection, never the documents."""
today, settings = _today(repo), repo.settings.current()
docs = seams.invoices(bp_id=bp_id)
by = _group(repo.allocations.for_invoices([d.invoice_id for d in docs]))
receivable, oldest = {}, 0
for d in docs:
o = stl.outstanding(d.total, by.get(d.invoice_id, []))
if o > ZERO:
receivable[d.currency] = receivable.get(d.currency, ZERO) + o
if d.due_date and d.due_date < today:
oldest = max(oldest, (today - d.due_date).days)
held_credit: dict = {}
for c in _credits(repo, bp_id, None):
held_credit[c["currency"]] = held_credit.get(c["currency"], ZERO) + c["remainder"]
net = {cur: receivable.get(cur, ZERO) - held_credit.get(cur, ZERO) for cur in set(receivable) | set(held_credit)}
limit = credit.limit_on(repo.credit.limits(bp_id), today)
hold = credit.derive(today=today, now=_naive_now(), limit=limit, balance_in_limit_currency=net.get(limit.currency, ZERO) if limit else ZERO,
oldest_overdue_days=oldest, days_rule=settings.credit_hold_days_overdue,
over_limit_rule=settings.credit_hold_over_limit,
overrides=[*repo.credit.overrides(bp_id), *extra_overrides])
gate = credit.credit_ok(hold, settings.credit_hold_mode)
return {"bp_id": bp_id, "as_at": today.isoformat(), "hold": hold.held, "reason": hold.reason, "hold_source": hold.source,
"hold_mode": settings.credit_hold_mode, "balance": {k: _s(v) for k, v in sorted(net.items())},
"receivable": {k: _s(v) for k, v in sorted(receivable.items())},
"unallocated_credit": {k: _s(v) for k, v in sorted(held_credit.items())}, "oldest_overdue_days": oldest,
"credit_limit": ({"amount": _s(limit.amount), "currency": limit.currency, "valid_from": limit.valid_from.isoformat()}
if limit else None),
"readiness": [gate.as_dict()]}
@PAY.use_case(kind="query", route="GET /customers/{bp_id}/credit-status", permission=CREDIT_READ)
def credit_status(caller: dict, bp_id: BigId) -> dict:
"""Is this customer on credit hold, and why: the hold (derived from the credit limit, the oldest overdue invoice
and the owner's overrides), the hold mode (warn or block), the net balance and the limit - and the hold as a
readiness gate {gate credit_ok, met, reason, fix} that sales-order shows on confirm. A narrow route: balance, hold
and reason only, never receipts or documents."""
with repository() as repo, _decide():
return _credit_status(repo, bp_id)
@PAY.use_case(kind="query", route="GET /customers/{bp_id}/credit-ok", permission=CREDIT_READ)
def credit_ok(caller: dict, bp_id: BigId) -> dict:
"""The readiness gate sales-order asks before confirming an order: {gate: credit_ok, met, reason, fix}, with the
hold mode (warn | block) it should act on."""
with repository() as repo, _decide():
st = _credit_status(repo, bp_id)
return {**st["readiness"][0], "hold_mode": st["hold_mode"], "bp_id": bp_id}
@PAY.use_case(kind="command", route="POST /customers/{bp_id}/credit-limit", permission=CREDIT_WRITE)
def set_credit_limit(caller: dict, bp_id: BigId, amount: Money, currency: Currency, valid_from: Optional[Day] = None,
reason: Optional[Reason] = None) -> dict:
"""Set a customer's credit limit from a date. A new limit takes effect from tomorrow at the earliest - it is a new
dated row, never an edit - so what the limit WAS on any past date stays answerable. Only an early release of a
hold (set_credit_override) is same-day."""
with repository() as repo, _decide():
today, now = _today(repo), _now()
tomorrow = today + timedelta(days=1)
start = _date(valid_from, "valid_from") or tomorrow
if start < tomorrow:
raise RuleBroken(f"A new credit limit takes effect from tomorrow ({tomorrow}) at the earliest; {start} would change what "
f"the limit was on a day already past. Use {tomorrow} or later")
known = seams.check_customer(bp_id, caller)
if known.status != "known":
raise Refused(known.http, known.note)
if repo.credit.has(bp_id, start):
raise RuleBroken(f"Customer {bp_id} already has a limit from {start}, and a limit is never edited. Choose a later date")
repo.credit.add_limit({"bp_id": bp_id, "valid_from": start, "credit_limit_amount": money(amount), "currency": _cur(currency),
"reason": reason or "", "set_by": _who(caller), "set_at": _naive_now(), "origin": ""}, now)
return {"bp_id": bp_id, "valid_from": start.isoformat(), "credit_limit": _s(amount), "currency": _cur(currency),
"limit_today": (lambda lim: {"amount": _s(lim.amount), "currency": lim.currency} if lim else None)(
credit.limit_on(repo.credit.limits(bp_id), today))}
@PAY.use_case(kind="command", route="POST /customers/{bp_id}/credit-override", permission=CREDIT_WRITE)
def set_credit_override(caller: dict, bp_id: BigId, kind: Literal["hold", "release"], ends_on: Day, reason: Reason) -> dict:
"""Put a customer on hold by hand, or release a hold early, with a reason and an end date after today. It takes
effect from the moment it is recorded, never earlier. Lifting one early is a new row, not an edit."""
with repository() as repo, _decide():
today, now = _today(repo), _now()
end = _date(ends_on, "ends_on")
if end <= today:
raise RuleBroken(f"The end date {end} is today or earlier. Choose a date after today ({today})")
if not reason.strip():
raise RuleBroken("A hold or release needs a reason. Say why")
starts = _naive_now()
if repo.credit.has_override(bp_id, kind, starts):
raise RuleBroken("That was just recorded. Check the customer's credit status")
repo.credit.add_override({"bp_id": bp_id, "kind": kind, "starts_at": starts, "ends_on": end, "reason": reason,
"set_by": _who(caller), "origin": ""}, now)
return {"bp_id": bp_id, "kind": kind, "effective_from": starts.isoformat(), "ends_on": end.isoformat(), "reason": reason,
"credit_status": _credit_status_safe(repo, bp_id, (credit.Override(kind, starts, end, reason, _who(caller)),))}
def _credit_status_safe(repo, bp_id: int, extra: tuple = ()) -> Optional[dict]:
"""The status after a change, with the new row in hand (it is not readable back inside this transaction)."""
try:
return _credit_status(repo, bp_id, extra)
except seams.Unavailable as down:
return {"unavailable": down.reason}
@PAY.use_case(kind="query", route="GET /customers/{bp_id}/credit", permission=CREDIT_WRITE)
def credit_history(caller: dict, bp_id: BigId) -> dict:
"""A customer's credit limits (every dated row) and holds/releases (every override), for the owner's Credit screen."""
with repository() as repo:
return {"bp_id": bp_id,
"limits": [{"valid_from": r["valid_from"].isoformat(), "amount": _s(r["credit_limit_amount"]), "currency": r["currency"],
"reason": r["reason"], "set_by": r["set_by"], "origin": r["origin"]} for r in repo.credit.limit_rows(bp_id)],
"overrides": [{"kind": o.kind, "starts_at": o.starts_at.isoformat(), "ends_on": o.ends_on.isoformat(),
"reason": o.reason, "set_by": o.set_by} for o in repo.credit.overrides(bp_id)]}
# ═══ configuration: settings, controls, bank accounts, reminder levels ═══════════════════════════════════
@PAY.use_case(kind="query", route="GET /settings", permission=READ)
def list_pay_settings(caller: dict) -> dict:
"""The settings the business edits (aging buckets and basis, the tolerance, the credit-hold rule and mode, the time
zone, the import cut-over, reminder mode, each AI step's mode), with the default, and whether changing it needs
pay:settings:controls."""
with repository() as repo:
return {"settings": repo.settings.all()}
@PAY.use_case(kind="query", route="GET /settings-history", permission=READ)
def settings_history(caller: dict, setting_key: Optional[Code] = None) -> dict:
"""Who changed a setting, from what, to what and when - the record behind the controls on the cashier."""
with repository() as repo:
return {"history": repo.settings.history(setting_key)}
def _set(repo, caller: dict, key: str, value) -> dict:
if key not in SETTING_DEFS:
raise Refused(404, f"no setting {key}")
why = check_setting(key, value)
if why:
raise Refused(422, f"{key} is {why}")
if key == "match_tolerance" and value is not None:
value = f"{Decimal(str(value)):.2f}"
now = repo.settings.get(key)
if (now["value"] if now["value"] is not None else now["default"]) == (value if value is not None else now["default"]):
return {"setting_key": key, "value": value, "unchanged": True}
return repo.settings.set(key, value, _who(caller), _now())
@PAY.use_case(kind="command", route="POST /settings/{setting_key}", permission=SETTINGS_WRITE, idempotent=True)
def set_pay_setting(caller: dict, setting_key: Code, value: Optional[Union[bool, int, Name]] = None) -> dict:
"""Change one day-to-day setting (aging buckets and basis, reminder mode, an AI step's mode off or suggest, the
suspense customer). null puts the default back. The CONTROLS on the people who record money - match_tolerance,
credit_hold_*, business_timezone, statement_import_from, and turning on auto-with-review - are refused here (403);
they are set with set_pay_control by whoever holds pay:settings:controls."""
if setting_key in SETTING_DEFS and needs_controls(setting_key, value):
raise Refused(403, f"missing permission: {CONTROLS} ({setting_key} is a control on the people who record money: use set_pay_control)")
with repository() as repo, _decide():
return _set(repo, caller, setting_key, value)
@PAY.use_case(kind="command", route="POST /controls/{setting_key}", permission=CONTROLS, idempotent=True)
def set_pay_control(caller: dict, setting_key: Code, value: Optional[Union[bool, int, Name]] = None) -> dict:
"""Change a control on the people who record money: match_tolerance, credit_hold_days_overdue, credit_hold_over_limit,
credit_hold_mode, business_timezone, statement_import_from, or switch ai_suggest_match to auto-with-review (exact
matches only). Applies to allocations made from now on and none already made; who changed it, from what, to what
and when is recorded. Only pay:settings:controls - never held by the cashier it controls."""
if setting_key not in SETTING_DEFS:
raise Refused(404, f"no setting {setting_key}")
if not needs_controls(setting_key, value) and setting_key != "ai_suggest_match":
raise Refused(422, f"{setting_key} is not a control: use set_pay_setting")
with repository() as repo, _decide():
return _set(repo, caller, setting_key, value)
@PAY.use_case(kind="query", route="GET /bank-accounts", permission=READ)
def list_bank_accounts(caller: dict) -> dict:
"""The accounts customers pay into, with the ledger account each posts to."""
with repository() as repo:
return {"bank_accounts": [{k: b[k] for k in ("bank_account_id", "name", "currency", "iban_or_sort_code_masked", "gl_account_id")}
for b in repo.config.banks()]}
@PAY.use_case(kind="query", route="GET /reminder-levels", permission=READ)
def list_reminder_levels(caller: dict) -> dict:
"""The follow-up levels: days after due, channel, wording and mode per level."""
with repository() as repo:
return {"levels": [{"level": lv.level, "days_after_due": lv.days_after_due, "channel": lv.channel, "template": lv.template,
"mode": lv.mode} for lv in repo.config.levels()]}
@PAY.use_case(kind="command", route="POST /reminder-levels", permission=SETTINGS_WRITE, idempotent=True)
def put_reminder_level(caller: dict, level: ReminderLevelIn) -> dict:
"""Set one follow-up level. A change applies from the next daily pass; reminders already sent are unchanged. auto
is allowed only for level 1 with the business's own approved template; later levels stay suggest."""
with repository() as repo, _decide():
if level.mode not in MODES["reminder_mode"]:
raise Refused(422, "mode is off, suggest or auto")
if level.mode == "auto" and (level.level != 1 or not level.template.strip()):
raise RuleBroken("Automatic sending is only for level 1 with the business's own approved template. Later levels stay suggest")
repo.config.put_level("default", coll.Level(level.level, level.days_after_due, level.channel, level.template, level.mode))
return {"level": level.level, "days_after_due": level.days_after_due, "mode": level.mode}
# the Odoo migration slice declares its use cases on PAY too
from brick_payments.application import migration # noqa: E402,F401
# ═══ the Odoo id map other bricks read (shared.yaml conventions.odoo_migration.id_map) ═══════════════
@PAY.use_case(kind="query", route="GET /migration-map/{db}/{model}", permission=MIGRATION_IMPORT, tool=False)
def pay_migration_map(caller: dict, db: Name, model: Name, ids: Optional[text(8000)] = None) -> list:
"""What Odoo records became here: (odoo db, model, ids) -> our natural key, as this brick's Odoo migration wrote it in
pay_odoo_map (an account.payment becomes a payment_id, an account.journal a bank account, a reconciliation an allocation).
Later bricks resolve Odoo ids through this map, never by matching names (shared.yaml conventions.odoo_migration). `ids` is
comma-separated (at most 500); an id with no mapping is left out, never a 404."""
from brick_payments.domain import odoo_map
want = [i.strip() for i in (ids or "").split(",") if i.strip()][:500]
with repository() as repo:
got = repo.migration.targets([odoo_map.key(db, model, i) for i in want])
return [{"odoo_model": model, "odoo_id": i, "ours": got[odoo_map.key(db, model, i)]} for i in want if odoo_map.key(db, model, i) in got]
# The v4 use cases live beside this module and register on the same brick.
from brick_payments.application import ( # noqa: E402,F401
accounts,
bankrec,
cash,
deposits,
directdebit,
expected,
forecast,
payouts,
remittances,
sources,
statements,
transfers,
)