feat(schema): dispense outcome on settlements, dispense_reports, cash-out hold mirror (ADR-005)
m016: an append-only `dispense_reports` table (lamassu-server's cash_out_actions shape — one row per report the machine sent, so a retry, a late report and a remediation report stay distinct); dispense_confirmed / dispense_error / dispense_error_code / dispense_raw_code / dispense_error_class / dispense_reported_at / dispensed_fiat_cents on dca_settlements; cash_out_held_since / _reason / _code on dca_machines beside counts_uncertain_since. Settlement lifecycle gains awaiting_dispense (cash_out at insert — paid, waiting for the machine's report), partial_pending (some notes out, value short; held whole until the operator records the resolution) and cash_owed (nothing out; legs never run). dispense_unreported is derived by the worklist, not stored. resume_cash_out joins CASSETTE_OP_TYPES as a machine-wide op: position 0, no position on the wire, no bay fields. It rides the operator channel the machine already consumes; the machine honours it only if stamped after the hold began. A recount releases the hold too. crud: get_settlement_by_txid (the join the machine's extra.txid already provides), apply_dispense_outcome (copies the report onto the settlement and finally writes bills_json / cassettes_json with what actually came out), the dispense_reports accessors incl. adopting a report that arrived before its payment, set_machine_cash_out_hold, and the three new worklist buckets.
This commit is contained in:
parent
d09c51f277
commit
b8a5e6352a
3 changed files with 511 additions and 18 deletions
250
crud.py
250
crud.py
|
|
@ -27,6 +27,8 @@ from .models import (
|
|||
DcaLpPreferences,
|
||||
DcaPayment,
|
||||
DcaSettlement,
|
||||
DispenseReport,
|
||||
DispenseReportIn,
|
||||
Machine,
|
||||
PublishCassettesPayload,
|
||||
SuperConfig,
|
||||
|
|
@ -257,6 +259,32 @@ async def set_machine_unpaired(machine_id: str) -> Machine | None:
|
|||
return await get_machine(machine_id)
|
||||
|
||||
|
||||
async def set_machine_cash_out_hold(
|
||||
machine_id: str,
|
||||
since: datetime | None,
|
||||
reason: str | None,
|
||||
code: str | None,
|
||||
) -> None:
|
||||
"""Mirror the machine's cash-out hold (ADR-005 §5) onto its registry row.
|
||||
|
||||
Written on every state event, including when it is None: the machine
|
||||
clearing the hold — after an operator recount or resume_cash_out — is as
|
||||
important as it setting one. `updated_at` is left alone for the same
|
||||
reason as counts_uncertain_since: this is the machine reporting about
|
||||
itself on a heartbeat, not an operator editing the machine.
|
||||
"""
|
||||
await db.execute(
|
||||
"""
|
||||
UPDATE spirekeeper.dca_machines
|
||||
SET cash_out_held_since = :since,
|
||||
cash_out_held_reason = :reason,
|
||||
cash_out_held_code = :code
|
||||
WHERE id = :id
|
||||
""",
|
||||
{"id": machine_id, "since": since, "reason": reason, "code": code},
|
||||
)
|
||||
|
||||
|
||||
async def set_machine_counts_uncertain(machine_id: str, since: datetime | None) -> None:
|
||||
"""Record (or clear) the machine's own "I can't vouch for these counts"
|
||||
marker, straight from its state document.
|
||||
|
|
@ -711,6 +739,171 @@ async def create_settlement_idempotent(
|
|||
return await get_settlement(settlement_id)
|
||||
|
||||
|
||||
async def get_settlement_by_txid(
|
||||
machine_id: str, bitspire_txid: str
|
||||
) -> DcaSettlement | None:
|
||||
"""The settlement a machine report refers to. `bitspire_txid` comes from
|
||||
the invoice's extra.txid, stamped by the machine at create_invoice time,
|
||||
so it is the natural join for a report that names the same txid."""
|
||||
return await db.fetchone(
|
||||
"SELECT * FROM spirekeeper.dca_settlements "
|
||||
"WHERE machine_id = :mid AND bitspire_txid = :txid",
|
||||
{"mid": machine_id, "txid": bitspire_txid},
|
||||
DcaSettlement,
|
||||
)
|
||||
|
||||
|
||||
async def apply_dispense_outcome(
|
||||
settlement_id: str,
|
||||
report: DispenseReportIn,
|
||||
new_status: str,
|
||||
reported_at: datetime,
|
||||
) -> DcaSettlement | None:
|
||||
"""Copy the machine's report onto the settlement and move it (ADR-005 §1).
|
||||
|
||||
Fills the never-before-written bills_json / cassettes_json with what
|
||||
actually came out, not what was provisioned. `error_message` is left to
|
||||
the distribution path; the dispense error lives in its own columns.
|
||||
"""
|
||||
import json as _json
|
||||
|
||||
await db.execute(
|
||||
"""
|
||||
UPDATE spirekeeper.dca_settlements
|
||||
SET status = :status,
|
||||
dispense_confirmed = :confirmed,
|
||||
dispense_error = :error,
|
||||
dispense_error_code = :error_code,
|
||||
dispense_raw_code = :raw_code,
|
||||
dispense_error_class = :error_class,
|
||||
dispense_reported_at = :reported_at,
|
||||
dispensed_fiat_cents = :dispensed_fiat_cents,
|
||||
bills_json = :bills_json,
|
||||
cassettes_json = :cassettes_json,
|
||||
processing_claim = NULL
|
||||
WHERE id = :id
|
||||
""",
|
||||
{
|
||||
"id": settlement_id,
|
||||
"status": new_status,
|
||||
"confirmed": report.dispense_confirmed,
|
||||
"error": report.error,
|
||||
"error_code": report.error_code,
|
||||
"raw_code": report.raw_code,
|
||||
"error_class": report.error_class,
|
||||
"reported_at": reported_at,
|
||||
"dispensed_fiat_cents": report.dispensed_fiat_cents,
|
||||
"bills_json": _json.dumps([b.dict() for b in report.bills]),
|
||||
"cassettes_json": _json.dumps([c.dict() for c in report.cassettes]),
|
||||
},
|
||||
)
|
||||
return await get_settlement(settlement_id)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dispense reports (ADR-005 §2) — append-only
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def get_dispense_report(
|
||||
machine_id: str, txid: str, reported_at: int
|
||||
) -> DispenseReport | None:
|
||||
"""A report is identified by (machine, txid, at): the machine resends the
|
||||
same report until acked, and a byte-identical resend must not grow the
|
||||
log. A remediation report for the same txid carries a later `at`."""
|
||||
return await db.fetchone(
|
||||
"SELECT * FROM spirekeeper.dispense_reports "
|
||||
"WHERE machine_id = :mid AND txid = :txid AND reported_at = :at",
|
||||
{"mid": machine_id, "txid": txid, "at": datetime.fromtimestamp(reported_at)},
|
||||
DispenseReport,
|
||||
)
|
||||
|
||||
|
||||
async def insert_dispense_report(
|
||||
machine_id: str, settlement_id: str | None, report: DispenseReportIn
|
||||
) -> DispenseReport:
|
||||
import json as _json
|
||||
|
||||
report_id = urlsafe_short_hash()
|
||||
await db.execute(
|
||||
"""
|
||||
INSERT INTO spirekeeper.dispense_reports
|
||||
(id, machine_id, settlement_id, txid, payment_hash, dispense_confirmed,
|
||||
error, error_code, raw_code, error_class, fiat_cents, currency,
|
||||
bills_json, cassettes_json, counts_uncertain, remediates_txid,
|
||||
reported_at, received_at)
|
||||
VALUES (:id, :machine_id, :settlement_id, :txid, :payment_hash,
|
||||
:confirmed, :error, :error_code, :raw_code, :error_class,
|
||||
:fiat_cents, :currency, :bills_json, :cassettes_json,
|
||||
:counts_uncertain, :remediates_txid, :reported_at, :received_at)
|
||||
""",
|
||||
{
|
||||
"id": report_id,
|
||||
"machine_id": machine_id,
|
||||
"settlement_id": settlement_id,
|
||||
"txid": report.txid,
|
||||
"payment_hash": report.payment_hash,
|
||||
"confirmed": report.dispense_confirmed,
|
||||
"error": report.error,
|
||||
"error_code": report.error_code,
|
||||
"raw_code": report.raw_code,
|
||||
"error_class": report.error_class,
|
||||
"fiat_cents": report.fiat_cents,
|
||||
"currency": report.currency,
|
||||
"bills_json": _json.dumps([b.dict() for b in report.bills]),
|
||||
"cassettes_json": _json.dumps([c.dict() for c in report.cassettes]),
|
||||
"counts_uncertain": report.counts_uncertain,
|
||||
"remediates_txid": report.remediates_txid,
|
||||
"reported_at": datetime.fromtimestamp(report.at),
|
||||
"received_at": datetime.now(),
|
||||
},
|
||||
)
|
||||
row = await db.fetchone(
|
||||
"SELECT * FROM spirekeeper.dispense_reports WHERE id = :id",
|
||||
{"id": report_id},
|
||||
DispenseReport,
|
||||
)
|
||||
assert row is not None, "Newly inserted dispense report couldn't be retrieved"
|
||||
return row
|
||||
|
||||
|
||||
async def link_dispense_reports_to_settlement(
|
||||
machine_id: str, txid: str, settlement_id: str
|
||||
) -> int:
|
||||
"""A report can arrive before its payment lands (hold invoices settle
|
||||
after the dispense; the invoice listener can lag). When the settlement is
|
||||
finally inserted, adopt the orphan rows."""
|
||||
result = await db.execute(
|
||||
"UPDATE spirekeeper.dispense_reports SET settlement_id = :sid "
|
||||
"WHERE machine_id = :mid AND txid = :txid AND settlement_id IS NULL",
|
||||
{"sid": settlement_id, "mid": machine_id, "txid": txid},
|
||||
)
|
||||
return getattr(result, "rowcount", 0) or 0
|
||||
|
||||
|
||||
async def get_latest_unlinked_dispense_report(
|
||||
machine_id: str, txid: str
|
||||
) -> DispenseReport | None:
|
||||
return await db.fetchone(
|
||||
"SELECT * FROM spirekeeper.dispense_reports "
|
||||
"WHERE machine_id = :mid AND txid = :txid AND settlement_id IS NULL "
|
||||
"ORDER BY reported_at DESC LIMIT 1",
|
||||
{"mid": machine_id, "txid": txid},
|
||||
DispenseReport,
|
||||
)
|
||||
|
||||
|
||||
async def get_dispense_reports_for_settlement(
|
||||
settlement_id: str,
|
||||
) -> list[DispenseReport]:
|
||||
return await db.fetchall(
|
||||
"SELECT * FROM spirekeeper.dispense_reports WHERE settlement_id = :sid "
|
||||
"ORDER BY reported_at ASC",
|
||||
{"sid": settlement_id},
|
||||
DispenseReport,
|
||||
)
|
||||
|
||||
|
||||
async def get_settlement(settlement_id: str) -> DcaSettlement | None:
|
||||
return await db.fetchone(
|
||||
"SELECT * FROM spirekeeper.dca_settlements WHERE id = :id",
|
||||
|
|
@ -752,7 +945,14 @@ async def get_stuck_settlements_for_operator(
|
|||
) -> dict:
|
||||
"""Operator worklist of settlements that didn't process cleanly.
|
||||
|
||||
Returns a dict with four keyed lists:
|
||||
Returns a dict with seven keyed lists. The first three are ADR-005 §6 —
|
||||
the only ones whose meaning is "a customer is owed money":
|
||||
- 'cash_owed': the machine reported nothing dispensed; legs never ran.
|
||||
- 'partial_pending': some notes out, value short; held whole until the
|
||||
operator records the resolution.
|
||||
- 'dispense_unreported': awaiting_dispense older than the threshold —
|
||||
the machine never reported (crashed, offline, or an old build).
|
||||
Then the original four:
|
||||
- 'rejected': any status='rejected' (Nostr attribution cross-check
|
||||
failed — signer didn't match the machine identity). Distinct
|
||||
from 'errored' because retry is wrong: the row was misrouted,
|
||||
|
|
@ -817,7 +1017,48 @@ async def get_stuck_settlements_for_operator(
|
|||
{"uid": operator_user_id, "threshold": threshold_at},
|
||||
DcaSettlement,
|
||||
)
|
||||
# ADR-005 §6 — the owed-cash buckets. cash_owed / partial_pending are
|
||||
# stored statuses; dispense_unreported is derived: a cash-out that landed
|
||||
# and never heard from its machine within the threshold.
|
||||
cash_owed = await db.fetchall(
|
||||
"""
|
||||
SELECT s.*
|
||||
FROM spirekeeper.dca_settlements s
|
||||
JOIN spirekeeper.dca_machines m ON m.id = s.machine_id
|
||||
WHERE m.operator_user_id = :uid AND s.status = 'cash_owed'
|
||||
ORDER BY s.created_at DESC
|
||||
""",
|
||||
{"uid": operator_user_id},
|
||||
DcaSettlement,
|
||||
)
|
||||
partial_pending = await db.fetchall(
|
||||
"""
|
||||
SELECT s.*
|
||||
FROM spirekeeper.dca_settlements s
|
||||
JOIN spirekeeper.dca_machines m ON m.id = s.machine_id
|
||||
WHERE m.operator_user_id = :uid AND s.status = 'partial_pending'
|
||||
ORDER BY s.created_at DESC
|
||||
""",
|
||||
{"uid": operator_user_id},
|
||||
DcaSettlement,
|
||||
)
|
||||
dispense_unreported = await db.fetchall(
|
||||
"""
|
||||
SELECT s.*
|
||||
FROM spirekeeper.dca_settlements s
|
||||
JOIN spirekeeper.dca_machines m ON m.id = s.machine_id
|
||||
WHERE m.operator_user_id = :uid
|
||||
AND s.status = 'awaiting_dispense'
|
||||
AND s.created_at < :threshold
|
||||
ORDER BY s.created_at ASC
|
||||
""",
|
||||
{"uid": operator_user_id, "threshold": threshold_at},
|
||||
DcaSettlement,
|
||||
)
|
||||
return {
|
||||
"cash_owed": cash_owed,
|
||||
"partial_pending": partial_pending,
|
||||
"dispense_unreported": dispense_unreported,
|
||||
"rejected": rejected,
|
||||
"errored": errored,
|
||||
"stuck_pending": stuck_pending,
|
||||
|
|
@ -870,9 +1111,10 @@ async def mark_settlement_status(
|
|||
status: str,
|
||||
error_message: str | None = None,
|
||||
) -> DcaSettlement | None:
|
||||
"""Status: 'pending' | 'processing' | 'processed' | 'partial' |
|
||||
'refunded' | 'errored'. Clears processing_claim on terminal states so a
|
||||
fresh claim attempt won't see a stale token."""
|
||||
"""Status: 'awaiting_dispense' | 'pending' | 'processing' | 'processed' |
|
||||
'partial_pending' | 'cash_owed' | 'partial' | 'refunded' | 'errored'.
|
||||
Clears processing_claim on terminal states so a fresh claim attempt won't
|
||||
see a stale token."""
|
||||
await db.execute(
|
||||
"""
|
||||
UPDATE spirekeeper.dca_settlements
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue