285 lines
11 KiB
Python
285 lines
11 KiB
Python
"""Off-chain USDT billing ledger (ADR-0015, US-031).
|
|
|
|
Tracks client API-key balances, node pending balances, and the protocol cut.
|
|
All mutations are expressed as append-only billing events with unique ids so
|
|
the ledger converges when events are gossiped across the tracker hive: every
|
|
event is a commutative balance delta, and each tracker applies an event at
|
|
most once (dedupe by event id).
|
|
|
|
No Solana calls live here — on-chain deposit/settlement adapters (US-032/033)
|
|
consume this ledger through `snapshot()` and `settle_node_payout()`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
import uuid
|
|
|
|
DEFAULT_PRICE_PER_1K_TOKENS = 0.02 # USDT
|
|
DEFAULT_STARTING_CREDIT = 1.0 # USDT of Caller Credit for a new API key
|
|
NODE_REVENUE_SHARE = 0.90 # nodes 90% / protocol cut 10% (ADR-0015)
|
|
|
|
|
|
class BillingLedger:
|
|
"""Thread-safe USDT ledger with SQLite persistence and event replication."""
|
|
|
|
SAVE_INTERVAL = 10.0
|
|
|
|
def __init__(
|
|
self,
|
|
db_path: str | None = None,
|
|
*,
|
|
prices: dict[str, float] | None = None,
|
|
default_price_per_1k: float = DEFAULT_PRICE_PER_1K_TOKENS,
|
|
starting_credit: float = DEFAULT_STARTING_CREDIT,
|
|
node_share: float = NODE_REVENUE_SHARE,
|
|
) -> None:
|
|
self._db_path = db_path
|
|
self._prices = dict(prices) if prices else {}
|
|
self._default_price_per_1k = default_price_per_1k
|
|
self._starting_credit = starting_credit
|
|
self._node_share = node_share
|
|
self._lock = threading.Lock()
|
|
self._client_balances: dict[str, float] = {}
|
|
self._node_pending: dict[str, float] = {}
|
|
self._protocol_cut: float = 0.0
|
|
self._seen_event_ids: set[str] = set()
|
|
self._event_log: list[dict] = [] # full log, order of local application
|
|
self._dirty = False
|
|
if self._db_path:
|
|
self._init_db()
|
|
self._load_from_db()
|
|
|
|
# ---- pricing ----
|
|
|
|
def price_for(self, model: str) -> float:
|
|
return self._prices.get(model, self._default_price_per_1k)
|
|
|
|
def set_price(self, model: str, price_per_1k: float) -> None:
|
|
with self._lock:
|
|
self._prices[model] = price_per_1k
|
|
|
|
# ---- local operations (create + apply + log an event) ----
|
|
|
|
def ensure_client(self, api_key: str) -> float:
|
|
"""Return the client's balance, granting Caller Credit on first touch."""
|
|
with self._lock:
|
|
if api_key not in self._client_balances:
|
|
self._apply_locked({
|
|
"id": f"credit-{uuid.uuid4().hex}",
|
|
"type": "credit",
|
|
"api_key": api_key,
|
|
"amount": self._starting_credit,
|
|
"ts": time.time(),
|
|
"note": "caller-credit",
|
|
})
|
|
return self._client_balances[api_key]
|
|
|
|
def has_funds(self, api_key: str) -> bool:
|
|
return self.ensure_client(api_key) > 0.0
|
|
|
|
def credit_client(self, api_key: str, amount: float, *, note: str = "deposit") -> float:
|
|
if amount <= 0:
|
|
raise ValueError("credit amount must be positive")
|
|
with self._lock:
|
|
self._apply_locked({
|
|
"id": f"credit-{uuid.uuid4().hex}",
|
|
"type": "credit",
|
|
"api_key": api_key,
|
|
"amount": amount,
|
|
"ts": time.time(),
|
|
"note": note,
|
|
})
|
|
return self._client_balances[api_key]
|
|
|
|
def charge_request(
|
|
self,
|
|
api_key: str,
|
|
model: str,
|
|
total_tokens: int,
|
|
node_work: list[tuple[str | None, int]],
|
|
) -> dict:
|
|
"""Debit the client and split the fee 90/10.
|
|
|
|
``node_work`` is ``[(wallet_address | None, work_units), ...]`` for the
|
|
nodes that served the request. Work units of nodes without a wallet
|
|
accrue to the protocol cut — there is nowhere to pay them out.
|
|
The client balance may dip below zero on the final request; the next
|
|
request is then rejected by ``has_funds`` (post-pay drift, standard
|
|
metered-billing behavior).
|
|
"""
|
|
cost = self.price_for(model) * max(0, total_tokens) / 1000.0
|
|
total_work = sum(max(0, w) for _, w in node_work)
|
|
node_pool = cost * self._node_share
|
|
shares: dict[str, float] = {}
|
|
if total_work > 0:
|
|
for wallet, work in node_work:
|
|
if wallet and work > 0:
|
|
shares[wallet] = shares.get(wallet, 0.0) + node_pool * work / total_work
|
|
protocol_amount = cost - sum(shares.values())
|
|
with self._lock:
|
|
if api_key not in self._client_balances:
|
|
self._apply_locked({
|
|
"id": f"credit-{uuid.uuid4().hex}",
|
|
"type": "credit",
|
|
"api_key": api_key,
|
|
"amount": self._starting_credit,
|
|
"ts": time.time(),
|
|
"note": "caller-credit",
|
|
})
|
|
event = {
|
|
"id": f"charge-{uuid.uuid4().hex}",
|
|
"type": "charge",
|
|
"api_key": api_key,
|
|
"model": model,
|
|
"total_tokens": total_tokens,
|
|
"cost": cost,
|
|
"shares": shares,
|
|
"protocol_amount": protocol_amount,
|
|
"ts": time.time(),
|
|
}
|
|
self._apply_locked(event)
|
|
return event
|
|
|
|
def settle_node_payout(self, wallet: str, amount: float, *, reference: str = "") -> dict:
|
|
"""Deduct a paid-out amount from a node's pending balance (US-033 hook)."""
|
|
if amount <= 0:
|
|
raise ValueError("payout amount must be positive")
|
|
with self._lock:
|
|
event = {
|
|
"id": f"payout-{uuid.uuid4().hex}",
|
|
"type": "payout",
|
|
"wallet": wallet,
|
|
"amount": amount,
|
|
"reference": reference,
|
|
"ts": time.time(),
|
|
}
|
|
self._apply_locked(event)
|
|
return event
|
|
|
|
def forfeit_pending(self, wallet: str, *, reason: str = "fraud") -> dict:
|
|
"""Forfeit a node's entire pending balance to the protocol cut (US-034 hook)."""
|
|
with self._lock:
|
|
event = {
|
|
"id": f"forfeit-{uuid.uuid4().hex}",
|
|
"type": "forfeit",
|
|
"wallet": wallet,
|
|
"amount": self._node_pending.get(wallet, 0.0),
|
|
"reason": reason,
|
|
"ts": time.time(),
|
|
}
|
|
self._apply_locked(event)
|
|
return event
|
|
|
|
# ---- replication ----
|
|
|
|
def events_since(self, index: int) -> tuple[list[dict], int]:
|
|
"""Return events after ``index`` in the local log and the new cursor."""
|
|
with self._lock:
|
|
return list(self._event_log[index:]), len(self._event_log)
|
|
|
|
def apply_events(self, events: list[dict]) -> int:
|
|
"""Apply peer events not yet seen locally. Returns how many applied."""
|
|
applied = 0
|
|
with self._lock:
|
|
for event in events:
|
|
event_id = event.get("id")
|
|
if not event_id or event_id in self._seen_event_ids:
|
|
continue
|
|
self._apply_locked(event)
|
|
applied += 1
|
|
return applied
|
|
|
|
def _apply_locked(self, event: dict) -> None:
|
|
etype = event.get("type")
|
|
if etype == "credit":
|
|
key = event["api_key"]
|
|
self._client_balances[key] = self._client_balances.get(key, 0.0) + float(event["amount"])
|
|
elif etype == "charge":
|
|
key = event["api_key"]
|
|
self._client_balances[key] = self._client_balances.get(key, 0.0) - float(event["cost"])
|
|
for wallet, amount in (event.get("shares") or {}).items():
|
|
self._node_pending[wallet] = self._node_pending.get(wallet, 0.0) + float(amount)
|
|
self._protocol_cut += float(event.get("protocol_amount", 0.0))
|
|
elif etype == "payout":
|
|
wallet = event["wallet"]
|
|
self._node_pending[wallet] = self._node_pending.get(wallet, 0.0) - float(event["amount"])
|
|
elif etype == "forfeit":
|
|
wallet = event["wallet"]
|
|
amount = float(event.get("amount", 0.0))
|
|
self._node_pending[wallet] = self._node_pending.get(wallet, 0.0) - amount
|
|
self._protocol_cut += amount
|
|
else:
|
|
return
|
|
self._seen_event_ids.add(event["id"])
|
|
self._event_log.append(event)
|
|
self._dirty = True
|
|
|
|
# ---- views ----
|
|
|
|
def get_client_balance(self, api_key: str) -> float:
|
|
with self._lock:
|
|
return self._client_balances.get(api_key, 0.0)
|
|
|
|
def get_node_pending(self, wallet: str) -> float:
|
|
with self._lock:
|
|
return self._node_pending.get(wallet, 0.0)
|
|
|
|
def snapshot(self) -> dict:
|
|
with self._lock:
|
|
return {
|
|
"clients": dict(self._client_balances),
|
|
"node_pending": dict(self._node_pending),
|
|
"protocol_cut": self._protocol_cut,
|
|
"prices": dict(self._prices),
|
|
"default_price_per_1k_tokens": self._default_price_per_1k,
|
|
"node_share": self._node_share,
|
|
"event_count": len(self._event_log),
|
|
}
|
|
|
|
# ---- persistence ----
|
|
|
|
def _init_db(self) -> None:
|
|
con = sqlite3.connect(self._db_path) # type: ignore[arg-type]
|
|
con.execute(
|
|
"CREATE TABLE IF NOT EXISTS billing_events "
|
|
"(event_id TEXT PRIMARY KEY, payload TEXT NOT NULL, ts REAL NOT NULL)"
|
|
)
|
|
con.commit()
|
|
con.close()
|
|
|
|
def _load_from_db(self) -> None:
|
|
con = sqlite3.connect(self._db_path) # type: ignore[arg-type]
|
|
rows = con.execute(
|
|
"SELECT payload FROM billing_events ORDER BY ts, event_id"
|
|
).fetchall()
|
|
con.close()
|
|
with self._lock:
|
|
for (payload,) in rows:
|
|
try:
|
|
event = json.loads(payload)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if event.get("id") not in self._seen_event_ids:
|
|
self._apply_locked(event)
|
|
self._dirty = False
|
|
|
|
def save_to_db(self) -> None:
|
|
if not self._db_path:
|
|
return
|
|
with self._lock:
|
|
if not self._dirty:
|
|
return
|
|
events = list(self._event_log)
|
|
self._dirty = False
|
|
con = sqlite3.connect(self._db_path) # type: ignore[arg-type]
|
|
con.executemany(
|
|
"INSERT OR IGNORE INTO billing_events (event_id, payload, ts) VALUES (?, ?, ?)",
|
|
[(e["id"], json.dumps(e), float(e.get("ts", 0.0))) for e in events],
|
|
)
|
|
con.commit()
|
|
con.close()
|