T010: ledger models — append-only tx, derived balances

ledger/src/m2_ledger/models.py

@@ -0,0 +1,295 @@
+"""m2-ledger SQLite models: append-only tx log, derived balances.
+
+Per contracts/ledger-api.md and data-model.md (Transaction, LicenseGrant, Wallet):
+balances are never stored authoritatively — every read folds the tx log. The tx
+table has no update/delete path exposed anywhere in this module.
+"""
+
+import sqlite3
+import uuid
+from datetime import datetime, timezone
+
+
+class InsufficientFunds(Exception):
+    pass
+
+
+class DuplicateRef(Exception):
+    pass
+
+
+class UnknownRef(Exception):
+    pass
+
+
+class AlreadyRefunded(Exception):
+    pass
+
+
+DEFAULT_PLATFORM_PCT = 0.10
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
+
+
+def init_db(path: str) -> sqlite3.Connection:
+    conn = sqlite3.connect(path)
+    conn.row_factory = sqlite3.Row
+    conn.execute("PRAGMA journal_mode=WAL")
+    conn.execute(
+        """
+        CREATE TABLE IF NOT EXISTS tx (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            ts TEXT NOT NULL,
+            from_id TEXT NOT NULL,
+            to_id TEXT NOT NULL,
+            amount INTEGER NOT NULL CHECK (amount > 0),
+            reason TEXT NOT NULL CHECK (reason IN
+                ('install', 'payout', 'grant', 'route', 'earn', 'refund')),
+            ref TEXT
+        )
+        """
+    )
+    conn.execute(
+        """
+        CREATE TABLE IF NOT EXISTS grant_ (
+            grant_id TEXT PRIMARY KEY,
+            operator_id TEXT NOT NULL,
+            listing_id TEXT NOT NULL,
+            solution_version_major INTEGER NOT NULL,
+            tx_id INTEGER NOT NULL REFERENCES tx(id),
+            ts TEXT NOT NULL
+        )
+        """
+    )
+    conn.execute("CREATE INDEX IF NOT EXISTS idx_tx_ref ON tx(ref, reason)")
+    conn.execute("CREATE INDEX IF NOT EXISTS idx_tx_from ON tx(from_id)")
+    conn.execute("CREATE INDEX IF NOT EXISTS idx_tx_to ON tx(to_id)")
+    conn.commit()
+    return conn
+
+
+def balance(conn: sqlite3.Connection, operator_id: str) -> int:
+    row = conn.execute(
+        """
+        SELECT
+            (SELECT COALESCE(SUM(amount), 0) FROM tx WHERE to_id = ?) -
+            (SELECT COALESCE(SUM(amount), 0) FROM tx WHERE from_id = ?) AS bal
+        """,
+        (operator_id, operator_id),
+    ).fetchone()
+    return row["bal"]
+
+
+def _insert_tx(conn, ts, from_id, to_id, amount, reason, ref) -> int:
+    cur = conn.execute(
+        "INSERT INTO tx (ts, from_id, to_id, amount, reason, ref) VALUES (?, ?, ?, ?, ?, ?)",
+        (ts, from_id, to_id, amount, reason, ref),
+    )
+    return cur.lastrowid
+
+
+def record_install(
+    conn: sqlite3.Connection,
+    buyer: str,
+    seller: str,
+    amount: int,
+    platform_pct: float,
+    ref: str,
+) -> dict:
+    """One SQLite transaction: debit buyer, credit seller net + platform cut,
+    then a license grant referencing the paying tx (contracts/ledger-api.md
+    POST /tx/install; SC-003/SC-004).
+    """
+    if platform_pct is None:
+        platform_pct = DEFAULT_PLATFORM_PCT
+
+    with conn:
+        dup = conn.execute(
+            "SELECT 1 FROM tx WHERE ref = ? AND reason = 'install' LIMIT 1", (ref,)
+        ).fetchone()
+        if dup is not None:
+            raise DuplicateRef(f"ref already settled: {ref}")
+
+        if balance(conn, buyer) < amount:
+            raise InsufficientFunds(
+                f"buyer {buyer} balance below {amount} for ref {ref}"
+            )
+
+        ts = _now()
+        cut = int(amount * platform_pct)
+        seller_amount = amount - cut
+
+        net_tx_id = _insert_tx(conn, ts, buyer, seller, seller_amount, "install", ref)
+        cut_tx_id = _insert_tx(conn, ts, buyer, "platform", cut, "install", ref)
+
+        grant_id = f"gr_{uuid.uuid4().hex}"
+        conn.execute(
+            """
+            INSERT INTO grant_ (grant_id, operator_id, listing_id,
+                                 solution_version_major, tx_id, ts)
+            VALUES (?, ?, ?, ?, ?, ?)
+            """,
+            (grant_id, buyer, ref, 1, net_tx_id, ts),
+        )
+
+        grant_row = conn.execute(
+            "SELECT * FROM grant_ WHERE grant_id = ?", (grant_id,)
+        ).fetchone()
+
+    return {"tx_ids": [net_tx_id, cut_tx_id], "grant": dict(grant_row)}
+
+
+def record_refund(conn: sqlite3.Connection, ref: str) -> dict:
+    """Compensating rows reversing every install row for `ref` (FR-005 edge case)."""
+    with conn:
+        original_rows = conn.execute(
+            "SELECT * FROM tx WHERE ref = ? AND reason = 'install'", (ref,)
+        ).fetchall()
+        if not original_rows:
+            raise UnknownRef(f"no install tx found for ref: {ref}")
+
+        already = conn.execute(
+            "SELECT 1 FROM tx WHERE ref = ? AND reason = 'refund' LIMIT 1", (ref,)
+        ).fetchone()
+        if already is not None:
+            raise AlreadyRefunded(f"ref already refunded: {ref}")
+
+        ts = _now()
+        tx_ids = [
+            _insert_tx(conn, ts, row["to_id"], row["from_id"], row["amount"], "refund", ref)
+            for row in original_rows
+        ]
+
+    return {"tx_ids": tx_ids}
+
+
+def record_grant(conn: sqlite3.Connection, operator_id: str, amount: int) -> dict:
+    """Mint -> operator, reason 'grant' (e.g. POST /grant/starter)."""
+    with conn:
+        ts = _now()
+        tx_id = _insert_tx(conn, ts, "mint", operator_id, amount, "grant", operator_id)
+        new_balance = balance(conn, operator_id)
+
+    return {"tx_id": tx_id, "balance": new_balance}
+
+
+def transactions(
+    conn: sqlite3.Connection,
+    operator_id: str | None = None,
+    reason: str | None = None,
+    since: str | None = None,
+) -> list[dict]:
+    clauses = []
+    params: list = []
+    if operator_id is not None:
+        clauses.append("(from_id = ? OR to_id = ?)")
+        params.extend([operator_id, operator_id])
+    if reason is not None:
+        clauses.append("reason = ?")
+        params.append(reason)
+    if since is not None:
+        clauses.append("ts >= ?")
+        params.append(since)
+
+    query = "SELECT * FROM tx"
+    if clauses:
+        query += " WHERE " + " AND ".join(clauses)
+    query += " ORDER BY id ASC"
+
+    rows = conn.execute(query, params).fetchall()
+    return [dict(row) for row in rows]
+
+
+def license_for(conn: sqlite3.Connection, operator_id: str, listing_id: str) -> dict | None:
+    row = conn.execute(
+        """
+        SELECT * FROM grant_ WHERE operator_id = ? AND listing_id = ?
+        ORDER BY ts DESC LIMIT 1
+        """,
+        (operator_id, listing_id),
+    ).fetchone()
+    return dict(row) if row is not None else None
+
+
+def all_balances(conn: sqlite3.Connection) -> dict:
+    ids = conn.execute(
+        """
+        SELECT DISTINCT id FROM (
+            SELECT from_id AS id FROM tx
+            UNION
+            SELECT to_id AS id FROM tx
+        ) WHERE id != 'mint'
+        """
+    ).fetchall()
+    return {row["id"]: balance(conn, row["id"]) for row in ids}
+
+
+if __name__ == "__main__":
+    import os
+    import tempfile
+
+    tmp_dir = tempfile.mkdtemp(prefix="m2_ledger_smoke_")
+    db_path = os.path.join(tmp_dir, "ledger.sqlite3")
+    conn = init_db(db_path)
+
+    print("== grant starter credits to buyer op_alice ==")
+    print(record_grant(conn, "op_alice", 100))
+    assert balance(conn, "op_alice") == 100
+
+    print("== install: op_alice buys lst_pdf_export from op_bob for 40cr ==")
+    result = record_install(conn, "op_alice", "op_bob", 40, 0.10, "lst_pdf_export-install-1")
+    print(result)
+    assert balance(conn, "op_alice") == 60
+    assert balance(conn, "op_bob") == 36
+    assert balance(conn, "platform") == 4
+
+    print("== duplicate ref is rejected ==")
+    try:
+        record_install(conn, "op_alice", "op_bob", 40, 0.10, "lst_pdf_export-install-1")
+        raise SystemExit("expected DuplicateRef")
+    except DuplicateRef as exc:
+        print("DuplicateRef:", exc)
+
+    print("== insufficient funds is rejected ==")
+    try:
+        record_install(conn, "op_alice", "op_bob", 10_000, 0.10, "lst_pdf_export-install-2")
+        raise SystemExit("expected InsufficientFunds")
+    except InsufficientFunds as exc:
+        print("InsufficientFunds:", exc)
+
+    print("== license_for reflects the grant ==")
+    lic = license_for(conn, "op_alice", "lst_pdf_export-install-1")
+    print(lic)
+    assert lic is not None and lic["tx_id"] == result["tx_ids"][0]
+
+    print("== refund reverses the install ==")
+    refund = record_refund(conn, "lst_pdf_export-install-1")
+    print(refund)
+    assert balance(conn, "op_alice") == 100
+    assert balance(conn, "op_bob") == 0
+    assert balance(conn, "platform") == 0
+
+    print("== refunding twice is rejected ==")
+    try:
+        record_refund(conn, "lst_pdf_export-install-1")
+        raise SystemExit("expected AlreadyRefunded")
+    except AlreadyRefunded as exc:
+        print("AlreadyRefunded:", exc)
+
+    print("== unknown ref refund is rejected ==")
+    try:
+        record_refund(conn, "no-such-ref")
+        raise SystemExit("expected UnknownRef")
+    except UnknownRef as exc:
+        print("UnknownRef:", exc)
+
+    print("== transactions() filters ==")
+    print(transactions(conn, operator_id="op_alice"))
+
+    print("== all_balances() for snapshot ==")
+    print(all_balances(conn))
+
+    conn.close()
+    print("\nSMOKE OK")