pbwraith/gitea-saas-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Billing ledger core (clients/ledger/holds, idempotent, fail-closed) (#1) #10

Merged
pbwraith merged 1 commits from issue-1 into main 2026-08-25 12:23:01 +02:00
Conversation 2 Commits 1 Files Changed 6 +1039
6 changed files with 1039 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

534
gitea_saas/billing_core.py Normal file
View File

@ -0,0 +1,534 @@
"""Billing ledger core for gitea-saas-core.
Pure Python + SQLite (stdlib only). No HTTP, no Stripe, no Gitea. This is the
shared billing core everything else builds on.
Invariants (fail-closed)
-----------------------
* ``balance = sum(credits) + sum(refunds) - sum(debits) - sum(active holds)``
and may never go negative: any mutation that would push it below zero is
rejected with a structured error and writes NO rows.
* Every mutating operation is idempotent via ``idempotency_key`` (UNIQUE in
the ledger). Replaying a key that already exists is a no-op that returns
the original entry -- never an error.
* Holds expire (default 30 minutes) and are auto-released on the next
read/query, so a stale hold can never lock balance forever.
Public API contract
-------------------
Every operation returns ``(ok, details)`` -- a bool and a dict. Failures are
structured and never raise: ``(False, {"error": {"code": ..., "message":
...}})``. Success payloads carry ``ledger_id``/``hold_id`` and the client's
``balance`` after the mutation.
Ledger audit convention for holds: the ``hold``/``release`` ledger rows use
``ref_type='hold', ref_id=str(hold_id)`` (a system pointer) so idempotent
replays resolve the exact hold row. The caller-facing semantic ref supplied
to ``reserve_hold`` lives on the holds row itself.
"""
from __future__ import annotations
import json
import sqlite3
import threading
from datetime import datetime, timedelta, timezone
from pathlib import Path
from .db import SCHEMA_PATH, connect, create_tables
# Structured error codes (public contract).
UNKNOWN_CLIENT = "UNKNOWN_CLIENT"
INVALID_AMOUNT = "INVALID_AMOUNT"
INVALID_KEY = "INVALID_KEY"
INVALID_CURRENCY = "INVALID_CURRENCY"
INVALID_NAME = "INVALID_NAME"
INVALID_HOLD_TTL = "INVALID_HOLD_TTL"
INSUFFICIENT_BALANCE = "INSUFFICIENT_BALANCE"
DUPLICATE_CLIENT = "DUPLICATE_CLIENT"
HOLD_NOT_FOUND = "HOLD_NOT_FOUND"
HOLD_ALREADY_RELEASED = "HOLD_ALREADY_RELEASED"
IDEMPOTENCY_CONFLICT = "IDEMPOTENCY_CONFLICT"
INTERNAL = "INTERNAL"
KINDS = frozenset({"credit", "debit", "hold", "release", "refund"})
LEDGER_CREDIT_KINDS = ("credit", "refund") # money-in kinds counted in balance
DEFAULT_HOLD_MINUTES = 30
DEFAULT_CURRENCY = "USD"
def _ok(details: dict) -> tuple[bool, dict]:
return True, details
def _err(code: str, message: str) -> tuple[bool, dict]:
return False, {"error": {"code": code, "message": message}}
def _coerce_ts(value) -> str:
"""Accept an ISO string or a (possibly naive) datetime -> ISO string."""
if isinstance(value, datetime):
if value.tzinfo is None:
value = value.replace(tzinfo=timezone.utc)
return value.isoformat()
return str(value)
def _load_meta(raw: str | None) -> dict:
try:
parsed = json.loads(raw or "{}")
return parsed if isinstance(parsed, dict) else {}
except (ValueError, TypeError):
return {}
class BillingCore:
"""Thread-safe billing ledger backed by a single SQLite file.
``db_path`` is a ``pathlib.Path`` (tests use temp-dir DBs). All access is
serialized by one re-entrant lock; WAL keeps readers non-blocking.
"""
def __init__(self, db_path: Path, schema_path: Path | None = None,
now_fn=None):
"""Open (or create) the ledger DB.
``db_path`` is a ``pathlib.Path`` (tests use temp-dir DBs).
``now_fn`` is a test seam: a zero-arg callable returning the current
UTC ``datetime``; defaults to the real clock. All access is
serialized by one re-entrant lock; WAL keeps readers non-blocking.
"""
self._path = Path(db_path)
self._schema_path = Path(schema_path) if schema_path is not None else SCHEMA_PATH
self._now_fn = now_fn or (lambda: datetime.now(timezone.utc))
self._lock = threading.RLock()
self._conn = connect(self._path)
self.create_tables()
# ------------------------------------------------------------------
# Time helpers (injectable clock)
# ------------------------------------------------------------------
def _ts(self) -> str:
"""Current time as a sortable ISO-8601 UTC string."""
return self._now_fn().isoformat()
def _ts_plus_minutes(self, minutes: float) -> str:
"""ISO-8601 UTC timestamp ``minutes`` from now (negative => the past)."""
return (self._now_fn() + timedelta(minutes=minutes)).isoformat()
# ------------------------------------------------------------------
# Schema / low-level helpers
# ------------------------------------------------------------------
def create_tables(self) -> None:
"""Create the schema idempotently (safe to call any number of times)."""
with self._lock:
create_tables(self._conn, schema_path=self._schema_path)
def _client_exists(self, client_id: int) -> bool:
row = self._conn.execute("SELECT id FROM clients WHERE id = ?", (client_id,)).fetchone()
return row is not None
def _active_hold_sum(self, client_id: int) -> int:
row = self._conn.execute(
"SELECT COALESCE(SUM(amount_minor), 0) FROM holds"
" WHERE client_id = ? AND released_at IS NULL AND expires_at > ?",
(client_id, self._ts()),
).fetchone()
return int(row[0])
def _ledger_sum(self, client_id: int, kinds: tuple[str, ...]) -> int:
placeholders = ",".join("?" * len(kinds))
row = self._conn.execute(
f"SELECT COALESCE(SUM(amount_minor), 0) FROM ledger"
f" WHERE client_id = ? AND kind IN ({placeholders})",
(client_id, *kinds),
).fetchone()
return int(row[0])
def _balance(self, client_id: int) -> int:
"""Available balance: credits + refunds - debits - active holds."""
money_in = self._ledger_sum(client_id, LEDGER_CREDIT_KINDS)
debits = self._ledger_sum(client_id, ("debit",))
holds = self._active_hold_sum(client_id)
return money_in - debits - holds
def _expire_stale_holds(self, client_id: int | None = None) -> list[dict]:
"""Release expired holds, writing deterministic 'release' ledger rows.
Idempotent: only ``released_at IS NULL`` holds are touched, and the
auto-release ledger key is derived from the hold id, so re-running is
a no-op.
"""
now = self._ts()
if client_id is None:
rows = self._conn.execute(
"SELECT * FROM holds WHERE released_at IS NULL AND expires_at <= ?",
(now,),
).fetchall()
else:
rows = self._conn.execute(
"SELECT * FROM holds WHERE client_id = ? AND released_at IS NULL AND expires_at <= ?",
(client_id, now),
).fetchall()
expired = []
for h in rows:
hold_id = h["id"]
cur = self._conn.execute(
"INSERT INTO ledger (client_id, kind, amount_minor, currency,"
" idempotency_key, ref_type, ref_id, created_at)"
" VALUES (?, 'release', ?, ?, ?, 'hold', ?, ?)",
(h["client_id"], h["amount_minor"], DEFAULT_CURRENCY,
f"auto-release:{hold_id}", str(hold_id), now),
)
self._conn.execute("UPDATE holds SET released_at = ? WHERE id = ?", (now, hold_id))
expired.append({"hold_id": hold_id, "amount_minor": h["amount_minor"],
"ledger_id": cur.lastrowid})
return expired
# ------------------------------------------------------------------
# Transaction wrapper (fail-closed)
# ------------------------------------------------------------------
def _mutate(self, fn) -> tuple[bool, dict]:
"""Run ``fn`` inside BEGIN IMMEDIATE with structured error isolation.
Writes are atomic: on any SQLite failure the whole transaction is
rolled back and a structured INTERNAL error is returned -- no partial
rows, and no exceptions leak to callers.
"""
with self._lock:
try:
self._conn.execute("BEGIN IMMEDIATE")
except sqlite3.Error as e:
return _err(INTERNAL, f"could not start transaction: {e}")
try:
result = fn()
except sqlite3.Error as e:
try:
self._conn.rollback()
except sqlite3.Error:
pass
return _err(INTERNAL, f"database error: {e}")
try:
self._conn.commit()
except sqlite3.Error as e:
try:
self._conn.rollback()
except sqlite3.Error:
pass
return _err(INTERNAL, f"could not commit transaction: {e}")
return result
@staticmethod
def _validate_amount(amount_minor) -> str | None:
"""Return an error code if the amount is invalid, else None."""
if not isinstance(amount_minor, int) or isinstance(amount_minor, bool) or amount_minor <= 0:
return INVALID_AMOUNT
return None
@staticmethod
def _validate_key(idempotency_key) -> str | None:
if not isinstance(idempotency_key, str) or not idempotency_key.strip():
return INVALID_KEY
return None
@staticmethod
def _validate_currency(currency) -> str | None:
if not isinstance(currency, str) or not currency.strip():
return INVALID_CURRENCY
return None
# ------------------------------------------------------------------
# Clients
# ------------------------------------------------------------------
def create_client(self, name: str, tier: str = "free", status: str = "active",
meta: dict | None = None) -> tuple[bool, dict]:
"""Create a client. Duplicate names are rejected (UNIQUE)."""
name = (name or "").strip()
if not name:
return _err(INVALID_NAME, "client name must be a non-empty string")
meta = meta if isinstance(meta, dict) else {}
meta_json = json.dumps(meta, sort_keys=True)
def fn() -> tuple[bool, dict]:
now = self._ts()
try:
cur = self._conn.execute(
"INSERT INTO clients (name, tier, status, meta_json, created_at)"
" VALUES (?, ?, ?, ?, ?)",
(name, tier, status, meta_json, now),
)
except sqlite3.IntegrityError:
return _err(DUPLICATE_CLIENT, f"client name already exists: {name}")
return _ok({"client_id": cur.lastrowid, "name": name, "tier": tier,
"status": status, "meta": meta, "created_at": now})
return self._mutate(fn)
def get_client(self, client_id: int, include_balance: bool = True) -> tuple[bool, dict]:
"""Fetch a client. Stale holds auto-release before the balance is read."""
def fn() -> tuple[bool, dict]:
row = self._conn.execute("SELECT * FROM clients WHERE id = ?", (client_id,)).fetchone()
if row is None:
return _err(UNKNOWN_CLIENT, f"no client with id {client_id}")
client = dict(row)
client["meta"] = _load_meta(client.pop("meta_json"))
self._expire_stale_holds(client_id)
details: dict = {"client": client}
if include_balance:
details["balance"] = self._balance(client_id)
return _ok(details)
return self._mutate(fn)
def get_balance(self, client_id: int) -> tuple[bool, dict]:
"""Available balance: credits + refunds - debits - active holds."""
def fn() -> tuple[bool, dict]:
if not self._client_exists(client_id):
return _err(UNKNOWN_CLIENT, f"no client with id {client_id}")
self._expire_stale_holds(client_id)
return _ok({"client_id": client_id, "balance": self._balance(client_id)})
return self._mutate(fn)
# ------------------------------------------------------------------
# Ledger mutations (credit / debit / refund)
# ------------------------------------------------------------------
def credit(self, client_id: int, amount_minor: int, idempotency_key: str,
currency: str = DEFAULT_CURRENCY, ref_type: str | None = None,
ref_id: str | None = None) -> tuple[bool, dict]:
"""Add money. Replaying the same ``idempotency_key`` is a no-op."""
return self._apply_ledger(client_id, "credit", amount_minor, idempotency_key,
currency, ref_type, ref_id)
def debit(self, client_id: int, amount_minor: int, idempotency_key: str,
currency: str = DEFAULT_CURRENCY, ref_type: str | None = None,
ref_id: str | None = None) -> tuple[bool, dict]:
"""Remove money. Rejected with INSUFFICIENT_BALANCE (no rows written)
if it would push the balance below zero."""
return self._apply_ledger(client_id, "debit", amount_minor, idempotency_key,
currency, ref_type, ref_id)
def refund(self, client_id: int, amount_minor: int, idempotency_key: str,
currency: str = DEFAULT_CURRENCY, ref_type: str | None = None,
ref_id: str | None = None) -> tuple[bool, dict]:
"""Add money back (audited as 'refund'). Preserves idempotency."""
return self._apply_ledger(client_id, "refund", amount_minor, idempotency_key,
currency, ref_type, ref_id)
def _apply_ledger(self, client_id: int, kind: str, amount_minor: int,
idempotency_key: str, currency: str, ref_type: str | None,
ref_id: str | None) -> tuple[bool, dict]:
if kind not in KINDS or kind in ("hold", "release"):
return _err(INTERNAL, f"invalid ledger kind: {kind}")
if (code := self._validate_amount(amount_minor)) is not None:
return _err(code, "amount_minor must be a positive integer (minor units, e.g. cents)")
if (code := self._validate_key(idempotency_key)) is not None:
return _err(code, "idempotency_key must be a non-empty string")
if (code := self._validate_currency(currency)) is not None:
return _err(code, "currency must be a non-empty string (default USD)")
currency = currency.strip().upper()
def fn() -> tuple[bool, dict]:
if not self._client_exists(client_id):
return _err(UNKNOWN_CLIENT, f"no client with id {client_id}")
self._expire_stale_holds(client_id)
existing = self._conn.execute(
"SELECT * FROM ledger WHERE idempotency_key = ?", (idempotency_key,)
).fetchone()
if existing is not None:
return self._replay(existing, client_id)
balance = self._balance(client_id)
if kind == "debit" and amount_minor > balance:
return _err(INSUFFICIENT_BALANCE,
f"debit {amount_minor} exceeds available balance {balance}")
cur = self._conn.execute(
"INSERT INTO ledger (client_id, kind, amount_minor, currency,"
" idempotency_key, ref_type, ref_id, created_at)"
" VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(client_id, kind, amount_minor, currency, idempotency_key,
ref_type, ref_id, self._ts()),
)
return _ok({"ledger_id": cur.lastrowid, "kind": kind, "client_id": client_id,
"amount_minor": amount_minor, "currency": currency,
"ref_type": ref_type, "ref_id": ref_id,
"balance": self._balance(client_id)})
return self._mutate(fn)
# ------------------------------------------------------------------
# Holds
# ------------------------------------------------------------------
def reserve_hold(self, client_id: int, amount_minor: int, idempotency_key: str,
ref_type: str | None = None, ref_id: str | None = None,
expires_in_minutes: float = DEFAULT_HOLD_MINUTES,
currency: str = DEFAULT_CURRENCY) -> tuple[bool, dict]:
"""Reserve balance with a hold. Replaying the same key is a no-op.
``expires_in_minutes`` defaults to 30; <= 0 means already expired
(deterministic expiry for tests / backdated holds).
"""
if (code := self._validate_amount(amount_minor)) is not None:
return _err(code, "amount_minor must be a positive integer (minor units, e.g. cents)")
if (code := self._validate_key(idempotency_key)) is not None:
return _err(code, "idempotency_key must be a non-empty string")
if (code := self._validate_currency(currency)) is not None:
return _err(code, "currency must be a non-empty string (default USD)")
if not isinstance(expires_in_minutes, (int, float)) or isinstance(expires_in_minutes, bool):
return _err(INVALID_HOLD_TTL, "expires_in_minutes must be a number")
currency = currency.strip().upper()
def fn() -> tuple[bool, dict]:
if not self._client_exists(client_id):
return _err(UNKNOWN_CLIENT, f"no client with id {client_id}")
self._expire_stale_holds(client_id)
existing = self._conn.execute(
"SELECT * FROM ledger WHERE idempotency_key = ?", (idempotency_key,)
).fetchone()
if existing is not None:
return self._replay(existing, client_id)
balance = self._balance(client_id)
if amount_minor > balance:
return _err(INSUFFICIENT_BALANCE,
f"hold {amount_minor} exceeds available balance {balance}")
expires_at = self._ts_plus_minutes(float(expires_in_minutes))
cur = self._conn.execute(
"INSERT INTO holds (client_id, amount_minor, ref_type, ref_id, expires_at)"
" VALUES (?, ?, ?, ?, ?)",
(client_id, amount_minor, ref_type, ref_id, expires_at),
)
hold_id = cur.lastrowid
# System pointer so idempotent replays resolve the exact hold row.
lcur = self._conn.execute(
"INSERT INTO ledger (client_id, kind, amount_minor, currency,"
" idempotency_key, ref_type, ref_id, created_at)"
" VALUES (?, 'hold', ?, ?, ?, 'hold', ?, ?)",
(client_id, amount_minor, currency, idempotency_key, str(hold_id), self._ts()),
)
return _ok({"hold_id": hold_id, "client_id": client_id,
"amount_minor": amount_minor, "currency": currency,
"ref_type": ref_type, "ref_id": ref_id,
"expires_at": expires_at, "ledger_id": lcur.lastrowid,
"balance": self._balance(client_id)})
return self._mutate(fn)
def release_hold(self, hold_id: int, idempotency_key: str) -> tuple[bool, dict]:
"""Release a hold, returning its amount to the balance.
Replaying the same ``idempotency_key`` returns the original release
(no-op). Releasing an already-released hold with a *different* key is
a structured HOLD_ALREADY_RELEASED error.
"""
if (code := self._validate_key(idempotency_key)) is not None:
return _err(code, "idempotency_key must be a non-empty string")
def fn() -> tuple[bool, dict]:
hold = self._conn.execute("SELECT * FROM holds WHERE id = ?", (hold_id,)).fetchone()
if hold is None:
return _err(HOLD_NOT_FOUND, f"no hold with id {hold_id}")
client_id = hold["client_id"]
existing = self._conn.execute(
"SELECT * FROM ledger WHERE idempotency_key = ?", (idempotency_key,)
).fetchone()
if existing is not None:
return self._replay(existing, client_id)
if hold["released_at"] is not None:
return _err(HOLD_ALREADY_RELEASED,
f"hold {hold_id} already released at {hold['released_at']}")
# The hold may be stale: expire it first, then re-check state.
self._expire_stale_holds(client_id)
hold = self._conn.execute("SELECT * FROM holds WHERE id = ?", (hold_id,)).fetchone()
if hold["released_at"] is not None:
return _err(HOLD_ALREADY_RELEASED,
f"hold {hold_id} auto-released on expiry at {hold['released_at']}")
now = self._ts()
self._conn.execute("UPDATE holds SET released_at = ? WHERE id = ?", (now, hold_id))
lcur = self._conn.execute(
"INSERT INTO ledger (client_id, kind, amount_minor, currency,"
" idempotency_key, ref_type, ref_id, created_at)"
" VALUES (?, 'release', ?, ?, ?, 'hold', ?, ?)",
(client_id, hold["amount_minor"], DEFAULT_CURRENCY,
idempotency_key, str(hold_id), now),
)
return _ok({"hold_id": hold_id, "client_id": client_id,
"amount_minor": hold["amount_minor"],
"released_at": now, "ledger_id": lcur.lastrowid,
"balance": self._balance(client_id)})
return self._mutate(fn)
def expire_stale_holds(self, client_id: int | None = None) -> tuple[bool, dict]:
"""Explicitly release every expired hold (optionally for one client)."""
def fn() -> tuple[bool, dict]:
if client_id is not None and not self._client_exists(client_id):
return _err(UNKNOWN_CLIENT, f"no client with id {client_id}")
expired = self._expire_stale_holds(client_id)
return _ok({"expired": expired, "count": len(expired)})
return self._mutate(fn)
# ------------------------------------------------------------------
# Queries
# ------------------------------------------------------------------
def list_ledger(self, client_id: int, from_=None, to=None) -> tuple[bool, dict]:
"""Ledger entries for a client, newest-compatible ordering.
``from_``/``to`` bound ``created_at`` (ISO string or datetime).
Stale holds auto-release before the read.
"""
def fn() -> tuple[bool, dict]:
if not self._client_exists(client_id):
return _err(UNKNOWN_CLIENT, f"no client with id {client_id}")
self._expire_stale_holds(client_id)
sql = "SELECT * FROM ledger WHERE client_id = ?"
params: list = [client_id]
if from_ is not None:
sql += " AND created_at >= ?"
params.append(_coerce_ts(from_))
if to is not None:
sql += " AND created_at <= ?"
params.append(_coerce_ts(to))
sql += " ORDER BY created_at, id"
rows = self._conn.execute(sql, params).fetchall()
entries = [dict(r) for r in rows]
return _ok({"entries": entries, "count": len(entries)})
return self._mutate(fn)
# ------------------------------------------------------------------
# Idempotent replay
# ------------------------------------------------------------------
def _replay(self, row, client_id: int) -> tuple[bool, dict]:
"""Return the original result for a replayed idempotency key."""
if row["client_id"] != client_id:
return _err(IDEMPOTENCY_CONFLICT,
f"idempotency_key already used for client {row['client_id']}")
original = {"ledger_id": row["id"], "kind": row["kind"], "client_id": row["client_id"],
"amount_minor": row["amount_minor"], "currency": row["currency"],
"ref_type": row["ref_type"], "ref_id": row["ref_id"],
"created_at": row["created_at"]}
details: dict = {"replayed": True, "original": original,
"client_id": client_id, "balance": self._balance(client_id)}
if row["kind"] in ("hold", "release") and row["ref_id"] is not None:
try:
hold = self._conn.execute(
"SELECT * FROM holds WHERE id = ?", (int(row["ref_id"]),)
).fetchone()
except (ValueError, TypeError):
hold = None
if hold is not None:
details["hold_id"] = hold["id"]
if row["kind"] == "release":
details["released_at"] = hold["released_at"]
return _ok(details)

37
gitea_saas/db.py Normal file
View File

@ -0,0 +1,37 @@
"""SQLite connection + schema bootstrap for gitea-saas-core.
Pure stdlib (sqlite3, pathlib). No HTTP, no Stripe, no Gitea.
"""
from __future__ import annotations
import sqlite3
from pathlib import Path
SCHEMA_PATH = Path(__file__).resolve().parent / "schema.sql"
def connect(db_path: Path) -> sqlite3.Connection:
"""Open a SQLite connection tuned for the billing core.
- WAL journal: readers never block writers (writes are additionally
serialized by BillingCore's single lock).
- Foreign keys enforced.
- ``check_same_thread=False``: BillingCore serializes all access with
one lock, so a single long-lived connection is safe across threads.
"""
path = Path(db_path)
path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(path), check_same_thread=False)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
conn.execute("PRAGMA busy_timeout=5000")
return conn
def create_tables(conn: sqlite3.Connection, schema_path: Path | None = None) -> None:
"""Apply schema.sql idempotently (safe to call any number of times)."""
path = Path(schema_path) if schema_path is not None else SCHEMA_PATH
with open(path, encoding="utf-8") as f:
conn.executescript(f.read())

49
gitea_saas/schema.sql Normal file
View File

@ -0,0 +1,49 @@
-- gitea-saas-core billing schema (SQLite).
--
-- Idempotent by construction: every object is created with IF NOT EXISTS,
-- so executing this file any number of times is safe.
--
-- Conventions:
-- * All amounts are INTEGER minor units (cents). No floats, ever.
-- * Currency defaults to USD.
-- * Timestamps are ISO-8601 UTC text written by the application layer;
-- the DEFAULT expressions are fallbacks for manual/one-off inserts only.
-- * ledger.kind is constrained to {credit, debit, hold, release, refund}.
-- * ledger.idempotency_key is globally UNIQUE: every mutation is replay-safe.
CREATE TABLE IF NOT EXISTS clients (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
tier TEXT NOT NULL DEFAULT 'free',
status TEXT NOT NULL DEFAULT 'active',
meta_json TEXT NOT NULL DEFAULT '{}',
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);
CREATE TABLE IF NOT EXISTS ledger (
id INTEGER PRIMARY KEY AUTOINCREMENT,
client_id INTEGER NOT NULL REFERENCES clients(id),
kind TEXT NOT NULL CHECK (kind IN ('credit', 'debit', 'hold', 'release', 'refund')),
amount_minor INTEGER NOT NULL CHECK (amount_minor >= 0),
currency TEXT NOT NULL DEFAULT 'USD',
idempotency_key TEXT NOT NULL UNIQUE,
ref_type TEXT,
ref_id TEXT,
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);
CREATE TABLE IF NOT EXISTS holds (
id INTEGER PRIMARY KEY AUTOINCREMENT,
client_id INTEGER NOT NULL REFERENCES clients(id),
amount_minor INTEGER NOT NULL CHECK (amount_minor >= 0),
ref_type TEXT,
ref_id TEXT,
expires_at TEXT NOT NULL,
released_at TEXT
);
-- Balance / listing lookups.
CREATE INDEX IF NOT EXISTS idx_ledger_client_created ON ledger (client_id, created_at, id);
CREATE INDEX IF NOT EXISTS idx_ledger_client_kind ON ledger (client_id, kind);
CREATE INDEX IF NOT EXISTS idx_holds_client ON holds (client_id);
CREATE INDEX IF NOT EXISTS idx_holds_stale ON holds (client_id) WHERE released_at IS NULL;

8
pyproject.toml Normal file
View File

@ -0,0 +1,8 @@
[project]
name = "gitea-saas-core"
version = "0.1.0"
description = "Shared billing core (clients/ledger/holds, idempotent, fail-closed) for gitea-saas"
requires-python = ">=3.10"
[tool.pytest.ini_options]
testpaths = ["tests"]

3
requirements.txt Normal file
View File

@ -0,0 +1,3 @@
# Runtime: pure stdlib (sqlite3, pathlib, threading, json). No third-party deps.
# Test/dev dependency:
pytest>=8.0

408
tests/test_billing_core.py Normal file
View File

@ -0,0 +1,408 @@
"""Hermetic tests for the billing ledger core.
Zero reliance on network / Stripe / Gitea: every test uses a temp-dir SQLite
file and pure stdlib. Run with: python -m pytest tests/test_billing_core.py -q
"""
import sys
import threading
from datetime import datetime, timedelta, timezone
from pathlib import Path
import pytest
# The package is a PEP 420 namespace package; make the project root importable
# regardless of how pytest was invoked.
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from gitea_saas.billing_core import ( # noqa: E402
BillingCore,
DUPLICATE_CLIENT,
HOLD_ALREADY_RELEASED,
HOLD_NOT_FOUND,
IDEMPOTENCY_CONFLICT,
INSUFFICIENT_BALANCE,
INVALID_AMOUNT,
INVALID_KEY,
UNKNOWN_CLIENT,
)
class FakeClock:
"""Deterministic clock for hold-expiry tests (injected via now_fn)."""
def __init__(self, start: datetime | None = None):
self.now = start or datetime(2026, 1, 1, tzinfo=timezone.utc)
def __call__(self) -> datetime:
return self.now
def advance(self, minutes: float) -> None:
self.now += timedelta(minutes=minutes)
@pytest.fixture
def core(tmp_path):
return BillingCore(tmp_path / "billing.db")
def make_client(core, name="alice"):
ok, details = core.create_client(name)
assert ok, details
return details["client_id"]
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
def test_create_tables_idempotent(core, tmp_path):
# Constructor already created them; calling again must be safe and
# non-destructive.
core.create_tables()
core.create_tables()
cid = make_client(core)
ok, details = core.get_client(cid)
assert ok and details["client"]["id"] == cid
# ---------------------------------------------------------------------------
# Credit / debit
# ---------------------------------------------------------------------------
def test_credit_balance_up_debit_balance_down(core):
cid = make_client(core)
ok, d = core.credit(cid, 500, "c1")
assert ok and d["kind"] == "credit" and d["amount_minor"] == 500
assert d["balance"] == 500
ok, d = core.credit(cid, 250, "c2")
assert ok and d["balance"] == 750
ok, d = core.debit(cid, 300, "d1")
assert ok and d["kind"] == "debit" and d["balance"] == 450
ok, details = core.get_client(cid, include_balance=True)
assert ok and details["balance"] == 450
# Debit down to exactly zero is allowed (never *negative*).
ok, d = core.debit(cid, 450, "d2")
assert ok and d["balance"] == 0
def test_debit_below_zero_rejected_and_no_row_written(core):
cid = make_client(core)
core.credit(cid, 100, "c1")
ok, details = core.debit(cid, 200, "d1")
assert ok is False
assert details["error"]["code"] == INSUFFICIENT_BALANCE
# No debit row was written; balance unchanged.
ok, ledger = core.list_ledger(cid)
assert ok and ledger["count"] == 1
assert ledger["entries"][0]["kind"] == "credit"
ok, bal = core.get_balance(cid)
assert ok and bal["balance"] == 100
def test_duplicate_idempotency_key_credit_applied_once(core):
cid = make_client(core)
ok, first = core.credit(cid, 100, "k1")
assert ok and first["balance"] == 100
ok, second = core.credit(cid, 100, "k1")
assert ok and second["replayed"] is True
assert second["original"]["ledger_id"] == first["ledger_id"]
assert second["original"]["amount_minor"] == 100
assert second["balance"] == 100 # applied once
ok, ledger = core.list_ledger(cid)
assert ok and ledger["count"] == 1
def test_duplicate_idempotency_key_debit_applied_once(core):
cid = make_client(core)
core.credit(cid, 200, "c1")
ok, d1 = core.debit(cid, 50, "k1")
assert ok and d1["balance"] == 150
ok, d2 = core.debit(cid, 50, "k1")
assert ok and d2["replayed"] is True and d2["balance"] == 150
ok, ledger = core.list_ledger(cid)
assert ok and ledger["count"] == 2 # one credit, one debit
def test_idempotency_key_conflict_across_clients_is_an_error(core):
cid1, cid2 = make_client(core, "a"), make_client(core, "b")
ok, _ = core.credit(cid1, 100, "shared-key")
assert ok
ok, details = core.credit(cid2, 100, "shared-key")
assert ok is False and details["error"]["code"] == IDEMPOTENCY_CONFLICT
# ---------------------------------------------------------------------------
# Holds
# ---------------------------------------------------------------------------
def test_hold_reserves_release_returns(core):
cid = make_client(core)
core.credit(cid, 1000, "c1")
ok, d = core.reserve_hold(cid, 300, "h1")
assert ok and d["hold_id"] and d["balance"] == 700
ok, details = core.get_client(cid)
assert ok and details["balance"] == 700
ok, d = core.release_hold(d["hold_id"], "r1")
assert ok and d["amount_minor"] == 300 and d["balance"] == 1000
ok, ledger = core.list_ledger(cid)
assert ok and ledger["count"] == 3 # credit + hold + release
def test_hold_cannot_exceed_available_balance(core):
cid = make_client(core)
core.credit(cid, 1000, "c1")
ok, _ = core.reserve_hold(cid, 700, "h1")
assert ok
ok, details = core.reserve_hold(cid, 400, "h2")
assert ok is False and details["error"]["code"] == INSUFFICIENT_BALANCE
ok, bal = core.get_balance(cid)
assert ok and bal["balance"] == 300
def test_hold_expiry_auto_releases_on_next_query(tmp_path):
clock = FakeClock()
core = BillingCore(tmp_path / "billing.db", now_fn=clock)
cid = make_client(core)
core.credit(cid, 1000, "c1")
ok, d = core.reserve_hold(cid, 200, "h1") # default 30 min
assert ok and d["balance"] == 800 # reserved
# 31 minutes later the hold has expired; the next read auto-releases it.
clock.advance(minutes=31)
ok, details = core.get_client(cid)
assert ok and details["balance"] == 1000
ok, ledger = core.list_ledger(cid)
assert ledger["count"] == 3 # credit + hold + auto-release
kinds = {e["kind"] for e in ledger["entries"]}
assert kinds == {"credit", "hold", "release"}
assert any(e["idempotency_key"].startswith("auto-release:") for e in ledger["entries"])
def test_expire_stale_holds_releases_only_expired(tmp_path):
clock = FakeClock()
core = BillingCore(tmp_path / "billing.db", now_fn=clock)
cid = make_client(core)
core.credit(cid, 1000, "c1")
ok, hold_live = core.reserve_hold(cid, 200, "h-live", expires_in_minutes=60)
assert ok
clock.advance(minutes=30)
ok, hold_stale = core.reserve_hold(cid, 400, "h-stale", expires_in_minutes=10)
assert ok
ok, bal = core.get_balance(cid)
assert ok and bal["balance"] == 400 # both holds active
clock.advance(minutes=11) # h-stale expired (t+41 > t+40), h-live not (t+41 < t+90)
ok, d = core.expire_stale_holds(cid)
assert ok and d["count"] == 1 and d["expired"][0]["hold_id"] == hold_stale["hold_id"]
ok, bal = core.get_balance(cid)
assert ok and bal["balance"] == 800 # only the live hold still reserves
# Re-running is a no-op.
ok, d = core.expire_stale_holds(cid)
assert ok and d["count"] == 0
def test_hold_expired_at_reservation_never_reserves(tmp_path):
# A hold born already-expired (negative TTL / backdated) does not lock
# balance and is cleaned up on the next query.
clock = FakeClock()
core = BillingCore(tmp_path / "billing.db", now_fn=clock)
cid = make_client(core)
core.credit(cid, 1000, "c1")
ok, d = core.reserve_hold(cid, 200, "h1", expires_in_minutes=-1)
assert ok and d["balance"] == 1000 # already expired: no reservation
ok, bal = core.get_balance(cid)
assert ok and bal["balance"] == 1000
ok, ledger = core.list_ledger(cid)
assert ledger["count"] == 3 # credit + hold + auto-release
def test_reserve_hold_replay_is_noop(core):
cid = make_client(core)
core.credit(cid, 1000, "c1")
ok, first = core.reserve_hold(cid, 300, "h1")
assert ok
ok, second = core.reserve_hold(cid, 300, "h1")
assert ok and second["replayed"] is True
assert second["hold_id"] == first["hold_id"]
ok, bal = core.get_balance(cid)
assert ok and bal["balance"] == 700
ok, ledger = core.list_ledger(cid)
assert ledger["count"] == 2 # credit + one hold row
def test_release_hold_replay_and_double_release(core):
cid = make_client(core)
core.credit(cid, 1000, "c1")
ok, hold = core.reserve_hold(cid, 300, "h1")
ok, rel = core.release_hold(hold["hold_id"], "r1")
assert ok and rel["balance"] == 1000
# Same key: idempotent replay, not an error.
ok, rel2 = core.release_hold(hold["hold_id"], "r1")
assert ok and rel2["replayed"] is True and rel2["balance"] == 1000
# Different key on an already-released hold: structured error.
ok, details = core.release_hold(hold["hold_id"], "r2")
assert ok is False and details["error"]["code"] == HOLD_ALREADY_RELEASED
def test_release_unknown_hold(core):
ok, details = core.release_hold(99999, "r1")
assert ok is False and details["error"]["code"] == HOLD_NOT_FOUND
# ---------------------------------------------------------------------------
# Refunds
# ---------------------------------------------------------------------------
def test_refund_adds_back_and_preserves_idempotency(core):
cid = make_client(core)
core.credit(cid, 100, "c1")
core.debit(cid, 100, "d1")
ok, bal = core.get_balance(cid)
assert ok and bal["balance"] == 0
ok, d = core.refund(cid, 100, "rf1")
assert ok and d["kind"] == "refund" and d["balance"] == 100
ok, d2 = core.refund(cid, 100, "rf1")
assert ok and d2["replayed"] is True and d2["balance"] == 100
ok, ledger = core.list_ledger(cid)
assert ok and ledger["count"] == 3
# ---------------------------------------------------------------------------
# Structured errors / validation (never raise, always tuples)
# ---------------------------------------------------------------------------
def test_unknown_client_structured_errors(core):
for op in (
lambda: core.get_client(999),
lambda: core.get_balance(999),
lambda: core.credit(999, 100, "x"),
lambda: core.debit(999, 100, "x"),
lambda: core.refund(999, 100, "x"),
lambda: core.reserve_hold(999, 100, "x"),
lambda: core.list_ledger(999),
lambda: core.expire_stale_holds(999),
):
ok, details = op()
assert ok is False
assert details["error"]["code"] == UNKNOWN_CLIENT
def test_invalid_amounts_rejected(core):
cid = make_client(core)
for bad in (0, -5, True, "10", 1.5, None):
ok, details = core.credit(cid, bad, f"k-{bad}")
assert ok is False and details["error"]["code"] == INVALID_AMOUNT
def test_missing_idempotency_key_rejected(core):
cid = make_client(core)
for bad in ("", " ", None, 42):
ok, details = core.credit(cid, 100, bad)
assert ok is False and details["error"]["code"] == INVALID_KEY
def test_duplicate_client_name_rejected(core):
make_client(core, "bob")
ok, details = core.create_client("bob")
assert ok is False and details["error"]["code"] == DUPLICATE_CLIENT
def test_empty_client_name_rejected(core):
ok, details = core.create_client(" ")
assert ok is False
def test_all_operations_return_tuples(core):
# Smoke check across the whole public surface: (bool, dict) always.
cid = make_client(core)
ops = [
core.get_client(cid),
core.get_balance(cid),
core.credit(cid, 50, "c1"),
core.debit(cid, 10, "d1"),
core.refund(cid, 5, "rf1"),
core.reserve_hold(cid, 5, "h1"),
core.list_ledger(cid),
core.expire_stale_holds(cid),
]
ok, hold = core.reserve_hold(cid, 5, "h2")
ops.append(core.release_hold(hold["hold_id"], "r1"))
ops.append(core.get_client(424242))
for result in ops:
assert isinstance(result, tuple) and len(result) == 2
assert isinstance(result[0], bool)
assert isinstance(result[1], dict)
# ---------------------------------------------------------------------------
# Queries / misc
# ---------------------------------------------------------------------------
def test_list_ledger_filters(core):
cid = make_client(core)
core.credit(cid, 100, "c1")
core.credit(cid, 50, "c2")
ok, ledger = core.list_ledger(cid)
assert ok and ledger["count"] == 2
ok, ledger = core.list_ledger(cid, to="2000-01-01T00:00:00+00:00")
assert ok and ledger["count"] == 0
ok, ledger = core.list_ledger(cid, from_="3000-01-01T00:00:00+00:00")
assert ok and ledger["count"] == 0
def test_get_client_include_balance_false(core):
cid = make_client(core)
core.credit(cid, 100, "c1")
ok, details = core.get_client(cid, include_balance=False)
assert ok and "balance" not in details
assert details["client"]["name"] == "alice"
def test_create_client_meta_roundtrip(core):
cid = make_client(core)
ok, carol = core.create_client("carol", tier="enterprise", status="active",
meta={"plan": "pro", "seats": 3})
assert ok
ok, details = core.get_client(cid)
assert ok and details["client"]["tier"] == "free" # default tier
ok, details = core.get_client(carol["client_id"])
assert ok and details["client"]["name"] == "carol"
assert details["client"]["meta"] == {"plan": "pro", "seats": 3}
def test_ledger_entries_expose_all_columns(core):
cid = make_client(core)
core.credit(cid, 100, "c1", ref_type="issue", ref_id="42")
ok, ledger = core.list_ledger(cid)
entry = ledger["entries"][0]
assert entry["ref_type"] == "issue" and entry["ref_id"] == "42"
assert entry["currency"] == "USD"
assert entry["amount_minor"] == 100
assert entry["client_id"] == cid
# ---------------------------------------------------------------------------
# Thread safety
# ---------------------------------------------------------------------------
def test_concurrent_credits_are_serialized(core):
cid = make_client(core)
n_threads, per_thread = 8, 25
errors = []
def worker(t):
for i in range(per_thread):
ok, details = core.credit(cid, 1, f"t{t}-i{i}")
if not ok:
errors.append(details)
threads = [threading.Thread(target=worker, args=(t,)) for t in range(n_threads)]
for t in threads:
t.start()
for t in threads:
t.join()
assert errors == []
ok, bal = core.get_balance(cid)
assert ok and bal["balance"] == n_threads * per_thread
ok, ledger = core.list_ledger(cid)
assert ok and ledger["count"] == n_threads * per_thread