diff --git a/gitea_saas/billing_core.py b/gitea_saas/billing_core.py new file mode 100644 index 0000000..c64acf2 --- /dev/null +++ b/gitea_saas/billing_core.py @@ -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) diff --git a/gitea_saas/db.py b/gitea_saas/db.py new file mode 100644 index 0000000..84ecd9c --- /dev/null +++ b/gitea_saas/db.py @@ -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()) diff --git a/gitea_saas/schema.sql b/gitea_saas/schema.sql new file mode 100644 index 0000000..54a5f99 --- /dev/null +++ b/gitea_saas/schema.sql @@ -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; diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..ecd4de4 --- /dev/null +++ b/pyproject.toml @@ -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"] diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..3f0f5f0 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +# Runtime: pure stdlib (sqlite3, pathlib, threading, json). No third-party deps. +# Test/dev dependency: +pytest>=8.0 diff --git a/tests/test_billing_core.py b/tests/test_billing_core.py new file mode 100644 index 0000000..f51fa0b --- /dev/null +++ b/tests/test_billing_core.py @@ -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