154 lines
5.7 KiB
Python
154 lines
5.7 KiB
Python
"""Persistence abstraction for Invoice objects.
|
|
|
|
The SDK does not prescribe a database. Consumers implement
|
|
:class:`InvoiceStore` against whatever backend suits them — SQLAlchemy,
|
|
asyncpg, SQLite, in-memory dict.
|
|
|
|
All methods are async so the same Protocol works for both
|
|
asyncpg/asyncio-sqlalchemy backends and synchronous backends wrapped
|
|
with ``asyncio.to_thread``.
|
|
|
|
Also ships :class:`InMemoryStore` — a reference implementation used by
|
|
the test suite and useful for local development.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import copy
|
|
from dataclasses import dataclass, field
|
|
from typing import Optional, Protocol, runtime_checkable
|
|
|
|
from cardano_checkout.invoice import Invoice, InvoiceStatus
|
|
|
|
|
|
@runtime_checkable
|
|
class InvoiceStore(Protocol):
|
|
"""Persistence backend for invoices.
|
|
|
|
Consumers implement the six methods below. The SDK's monitor + scheduler
|
|
modules operate entirely through this interface, never touching a specific
|
|
ORM or driver.
|
|
"""
|
|
|
|
async def create(self, invoice: Invoice) -> None:
|
|
"""Insert a new invoice. Should raise if an invoice with the same id exists."""
|
|
...
|
|
|
|
async def get(self, invoice_id: str) -> Optional[Invoice]:
|
|
"""Fetch one invoice by id. Returns None if not found."""
|
|
...
|
|
|
|
async def list_by_status(
|
|
self, status: InvoiceStatus, limit: int = 100
|
|
) -> list[Invoice]:
|
|
"""List invoices in a given state, newest-first. Used by the monitor poll loop."""
|
|
...
|
|
|
|
async def update(self, invoice: Invoice) -> None:
|
|
"""Persist the current state of an invoice.
|
|
|
|
Implementations should compare-and-set on ``invoice.id`` — if the row
|
|
doesn't exist the call should raise. Does NOT create; see :meth:`create`.
|
|
"""
|
|
...
|
|
|
|
async def next_derivation_index(self, merchant_id: str) -> int:
|
|
"""Return the next unused receive-address index for a merchant.
|
|
|
|
Should be transactionally safe against concurrent invoice creation;
|
|
consumers typically implement this via ``SELECT COALESCE(MAX(index), -1) + 1 ... FOR UPDATE``
|
|
or an atomic counter row.
|
|
"""
|
|
...
|
|
|
|
async def record_tx(
|
|
self, invoice_id: str, tx_hash: str, lovelace_delta: int
|
|
) -> None:
|
|
"""Record an observed inbound UTxO against an invoice.
|
|
|
|
Must be idempotent on (invoice_id, tx_hash) — monitor loops will
|
|
re-observe the same UTxO until the invoice transitions to a terminal
|
|
state.
|
|
"""
|
|
...
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Reference implementation: InMemoryStore
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class InMemoryStore:
|
|
"""In-memory :class:`InvoiceStore` — intended for tests and local dev.
|
|
|
|
Uses an ``asyncio.Lock`` around mutating operations so concurrent callers
|
|
see a consistent view. Objects are deep-copied on read so callers can't
|
|
mutate the stored state by accident.
|
|
"""
|
|
|
|
_invoices: dict[str, Invoice] = field(default_factory=dict)
|
|
_tx_log: dict[tuple[str, str], int] = field(default_factory=dict)
|
|
_index_counters: dict[str, int] = field(default_factory=dict)
|
|
_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
|
|
|
|
async def create(self, invoice: Invoice) -> None:
|
|
async with self._lock:
|
|
if invoice.id in self._invoices:
|
|
raise ValueError(f"Invoice {invoice.id!r} already exists")
|
|
self._invoices[invoice.id] = copy.deepcopy(invoice)
|
|
# Bump the per-merchant index cursor so next_derivation_index doesn't
|
|
# hand out the same slot again.
|
|
cur = self._index_counters.get(invoice.merchant_id, -1)
|
|
if invoice.derivation_index > cur:
|
|
self._index_counters[invoice.merchant_id] = invoice.derivation_index
|
|
|
|
async def get(self, invoice_id: str) -> Optional[Invoice]:
|
|
async with self._lock:
|
|
stored = self._invoices.get(invoice_id)
|
|
return copy.deepcopy(stored) if stored else None
|
|
|
|
async def list_by_status(
|
|
self, status: InvoiceStatus, limit: int = 100
|
|
) -> list[Invoice]:
|
|
async with self._lock:
|
|
matching = [
|
|
copy.deepcopy(inv)
|
|
for inv in self._invoices.values()
|
|
if inv.status == status
|
|
]
|
|
# Newest first per contract.
|
|
matching.sort(key=lambda inv: inv.created_at, reverse=True)
|
|
return matching[:limit]
|
|
|
|
async def update(self, invoice: Invoice) -> None:
|
|
async with self._lock:
|
|
if invoice.id not in self._invoices:
|
|
raise KeyError(f"Invoice {invoice.id!r} not found")
|
|
self._invoices[invoice.id] = copy.deepcopy(invoice)
|
|
|
|
async def next_derivation_index(self, merchant_id: str) -> int:
|
|
async with self._lock:
|
|
nxt = self._index_counters.get(merchant_id, -1) + 1
|
|
self._index_counters[merchant_id] = nxt
|
|
return nxt
|
|
|
|
async def record_tx(
|
|
self, invoice_id: str, tx_hash: str, lovelace_delta: int
|
|
) -> None:
|
|
async with self._lock:
|
|
# Idempotent: (invoice_id, tx_hash) overwrites the delta rather than
|
|
# adding it a second time — see InvoiceStore.record_tx contract.
|
|
self._tx_log[(invoice_id, tx_hash)] = lovelace_delta
|
|
inv = self._invoices.get(invoice_id)
|
|
if inv and tx_hash not in inv.tx_hashes:
|
|
inv.tx_hashes.append(tx_hash)
|
|
|
|
# --- test helpers (not part of the Protocol) ---
|
|
|
|
def _all(self) -> list[Invoice]:
|
|
return [copy.deepcopy(inv) for inv in self._invoices.values()]
|
|
|
|
def _tx_records(self) -> dict[tuple[str, str], int]:
|
|
return dict(self._tx_log)
|