532 lines
19 KiB
Python
532 lines
19 KiB
Python
"""Durable Coffer product: wallets, ledger txs, signup, subscription, catalog."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
import os
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from coffer.db import CofferDB
|
|
from coffer.public_ledger import snapshot_from_product
|
|
from coffer.ledger import CreditKind, Ledger, Lot, Tx
|
|
from coffer.store import Good, Invoice, Order, new_store
|
|
from coffer.units import (
|
|
ACCOUNT_COFFER,
|
|
ACCOUNT_LEORA,
|
|
ACCOUNT_NEW_INTERNET,
|
|
COFFER_INVESTMENT_CREDITS,
|
|
CONTRACT_CREDITS,
|
|
LEORA_FUNDED_AFTER_GRANTS_CREDITS,
|
|
SCALE,
|
|
SHIRT_CREDITS,
|
|
SIGNUP_GRANT_CREDITS,
|
|
SUBSCRIPTION_CREDITS,
|
|
SUBSCRIPTION_CUT_BPS,
|
|
agent_account,
|
|
cut_subunits,
|
|
format_credits,
|
|
to_subunits,
|
|
)
|
|
|
|
PT = ZoneInfo(os.environ.get("COFFER_TZ", "UTC")) # deployment sets its own zone
|
|
|
|
|
|
def now_pt() -> str:
|
|
return datetime.now(tz=PT).isoformat(timespec="seconds")
|
|
|
|
|
|
class ProductError(Exception):
|
|
"""Product-level error."""
|
|
|
|
|
|
class CofferProduct:
|
|
"""High-level API the platform and CLI call."""
|
|
|
|
def __init__(self, db_path: str | Path) -> None:
|
|
self.db = CofferDB(db_path)
|
|
self.ledger = Ledger()
|
|
self.store = new_store(self.ledger)
|
|
self._load()
|
|
|
|
def close(self) -> None:
|
|
self.db.close()
|
|
|
|
def _load(self) -> None:
|
|
self.ledger = Ledger()
|
|
self.store = new_store(self.ledger)
|
|
self.store.suppliers.clear()
|
|
self.store.goods.clear()
|
|
|
|
for row in self.db.execute("SELECT account_id FROM accounts").fetchall():
|
|
self.ledger.ensure_wallet(row["account_id"])
|
|
for row in self.db.execute(
|
|
"SELECT account_id, kind, remaining, reason FROM lots "
|
|
"WHERE remaining > 0 ORDER BY lot_id"
|
|
).fetchall():
|
|
self.ledger.account(row["account_id"]).lots.append(
|
|
Lot(CreditKind(row["kind"]), int(row["remaining"]), row["reason"])
|
|
)
|
|
|
|
self.ledger.transactions.clear()
|
|
self.ledger._tx_seq = 0
|
|
for row in self.db.execute(
|
|
"SELECT seq, kind, amount, debit_account, credit_account, reason, meta_json "
|
|
"FROM transactions ORDER BY seq"
|
|
).fetchall():
|
|
tx = Tx(
|
|
seq=int(row["seq"]),
|
|
kind=row["kind"],
|
|
amount=int(row["amount"]),
|
|
debit_account=row["debit_account"],
|
|
credit_account=row["credit_account"],
|
|
reason=row["reason"],
|
|
meta=self.db.loads(row["meta_json"] or "{}"),
|
|
)
|
|
self.ledger.transactions.append(tx)
|
|
self.ledger._tx_seq = max(self.ledger._tx_seq, tx.seq)
|
|
|
|
for row in self.db.execute("SELECT supplier_id, name FROM suppliers").fetchall():
|
|
self.store.add_supplier(row["supplier_id"], row["name"])
|
|
for row in self.db.execute(
|
|
"SELECT sku, name, credit_price, supplier_id, supplier_unit_cost, stock FROM goods"
|
|
).fetchall():
|
|
self.store.goods[row["sku"]] = Good(
|
|
sku=row["sku"],
|
|
name=row["name"],
|
|
credit_price=int(row["credit_price"]),
|
|
supplier_id=row["supplier_id"],
|
|
supplier_unit_cost=int(row["supplier_unit_cost"]),
|
|
stock=int(row["stock"]),
|
|
)
|
|
if "coffer-shirt" not in self.store.goods:
|
|
if "printer" not in self.store.suppliers:
|
|
self.store.add_supplier("printer", "Printer")
|
|
self.store.add_good(
|
|
sku="coffer-shirt",
|
|
name="Coffer shirt",
|
|
credit_price=SHIRT_CREDITS,
|
|
supplier_id="printer",
|
|
supplier_unit_cost=8,
|
|
)
|
|
self._persist_catalog()
|
|
self.db.commit()
|
|
|
|
for row in self.db.execute("SELECT * FROM invoices").fetchall():
|
|
inv = Invoice(
|
|
invoice_id=row["invoice_id"],
|
|
supplier_id=row["supplier_id"],
|
|
sku=row["sku"],
|
|
quantity=int(row["quantity"]),
|
|
dollars=int(row["dollars"]),
|
|
status=row["status"],
|
|
)
|
|
self.store.invoices[inv.invoice_id] = inv
|
|
self._bump_seq_from_id(inv.invoice_id)
|
|
for row in self.db.execute("SELECT * FROM orders").fetchall():
|
|
order = Order(
|
|
order_id=row["order_id"],
|
|
buyer_id=row["buyer_id"],
|
|
sku=row["sku"],
|
|
quantity=int(row["quantity"]),
|
|
credits_charged=int(row["credits_charged"]),
|
|
draws=self.db.loads(row["draws_json"]),
|
|
from_stock=int(row["from_stock"]),
|
|
procured=int(row["procured"]),
|
|
status=row["status"],
|
|
invoice_id=row["invoice_id"],
|
|
tx_seq=row["tx_seq"],
|
|
)
|
|
self.store.orders[order.order_id] = order
|
|
self._bump_seq_from_id(order.order_id)
|
|
for row in self.db.execute("SELECT key, order_id FROM idempotency").fetchall():
|
|
self.store._idempotency[row["key"]] = row["order_id"]
|
|
|
|
def _bump_seq_from_id(self, oid: str) -> None:
|
|
try:
|
|
n = int(str(oid).rsplit("-", 1)[-1])
|
|
if n > self.store._seq:
|
|
self.store._seq = n
|
|
except ValueError:
|
|
pass
|
|
|
|
def _ensure_account_row(self, account_id: str) -> None:
|
|
kind = "company" if account_id.startswith("company:") else "agent"
|
|
self.db.execute(
|
|
"INSERT OR IGNORE INTO accounts(account_id, kind, created_at) VALUES(?,?,?)",
|
|
(account_id, kind, now_pt()),
|
|
)
|
|
|
|
def _rewrite_lots(self, account_id: str) -> None:
|
|
self._ensure_account_row(account_id)
|
|
self.db.execute("DELETE FROM lots WHERE account_id=?", (account_id,))
|
|
for lot in self.ledger.account(account_id).lots:
|
|
if lot.remaining <= 0:
|
|
continue
|
|
self.db.execute(
|
|
"INSERT INTO lots(account_id, kind, remaining, reason, created_at) "
|
|
"VALUES(?,?,?,?,?)",
|
|
(account_id, lot.kind.value, lot.remaining, lot.reason, now_pt()),
|
|
)
|
|
|
|
def _persist_tx(self, tx: Tx) -> None:
|
|
self.db.execute(
|
|
"INSERT INTO transactions(seq, kind, amount, debit_account, credit_account, "
|
|
"reason, meta_json, created_at) VALUES(?,?,?,?,?,?,?,?)",
|
|
(
|
|
tx.seq,
|
|
tx.kind,
|
|
tx.amount,
|
|
tx.debit_account,
|
|
tx.credit_account,
|
|
tx.reason,
|
|
self.db.dumps(tx.meta),
|
|
now_pt(),
|
|
),
|
|
)
|
|
|
|
def _persist_catalog(self) -> None:
|
|
for s in self.store.suppliers.values():
|
|
self.db.execute(
|
|
"INSERT OR REPLACE INTO suppliers(supplier_id, name) VALUES(?,?)",
|
|
(s.supplier_id, s.name),
|
|
)
|
|
for g in self.store.goods.values():
|
|
self.db.execute(
|
|
"INSERT OR REPLACE INTO goods(sku, name, credit_price, supplier_id, "
|
|
"supplier_unit_cost, stock) VALUES(?,?,?,?,?,?)",
|
|
(
|
|
g.sku,
|
|
g.name,
|
|
g.credit_price,
|
|
g.supplier_id,
|
|
g.supplier_unit_cost,
|
|
g.stock,
|
|
),
|
|
)
|
|
|
|
def _persist_order(self, order: Order) -> None:
|
|
self.db.execute(
|
|
"INSERT OR REPLACE INTO orders(order_id, buyer_id, sku, quantity, "
|
|
"credits_charged, draws_json, from_stock, procured, status, invoice_id, "
|
|
"tx_seq, created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)",
|
|
(
|
|
order.order_id,
|
|
order.buyer_id,
|
|
order.sku,
|
|
order.quantity,
|
|
order.credits_charged,
|
|
self.db.dumps(order.draws),
|
|
order.from_stock,
|
|
order.procured,
|
|
order.status,
|
|
order.invoice_id,
|
|
order.tx_seq,
|
|
now_pt(),
|
|
),
|
|
)
|
|
if order.invoice_id and order.invoice_id in self.store.invoices:
|
|
inv = self.store.invoices[order.invoice_id]
|
|
self.db.execute(
|
|
"INSERT OR REPLACE INTO invoices(invoice_id, supplier_id, sku, quantity, "
|
|
"dollars, status, created_at) VALUES(?,?,?,?,?,?,?)",
|
|
(
|
|
inv.invoice_id,
|
|
inv.supplier_id,
|
|
inv.sku,
|
|
inv.quantity,
|
|
inv.dollars,
|
|
inv.status,
|
|
now_pt(),
|
|
),
|
|
)
|
|
self._persist_catalog()
|
|
|
|
def _commit_accounts(self, *account_ids: str, txs: list[Tx] | None = None) -> None:
|
|
seen: set[str] = set()
|
|
for aid in account_ids:
|
|
if aid and aid not in seen:
|
|
self._rewrite_lots(aid)
|
|
seen.add(aid)
|
|
for tx in txs or []:
|
|
self._persist_tx(tx)
|
|
self.db.commit()
|
|
if txs:
|
|
self._after_mutation()
|
|
|
|
# ----- public API ----------------------------------------------------------
|
|
|
|
def create_wallet(self, account_id: str) -> dict[str, Any]:
|
|
if account_id.startswith("company:"):
|
|
aid = account_id
|
|
else:
|
|
aid = agent_account(account_id)
|
|
self.ledger.ensure_wallet(aid)
|
|
self._ensure_account_row(aid)
|
|
self.db.commit()
|
|
return {"account_id": aid, "balance": self.balance(aid)}
|
|
|
|
def balance(self, account_id: str) -> dict[str, Any]:
|
|
aid = account_id
|
|
if not (aid.startswith("agent:") or aid.startswith("company:")):
|
|
aid = agent_account(aid)
|
|
raw = self.ledger.balance_subunits(aid)
|
|
return {
|
|
"account_id": aid,
|
|
"promotional": raw["promotional"] / SCALE,
|
|
"funded": raw["funded"] / SCALE,
|
|
"spendable": raw["spendable"] / SCALE,
|
|
"promotional_centicredits": raw["promotional"],
|
|
"funded_centicredits": raw["funded"],
|
|
"spendable_centicredits": raw["spendable"],
|
|
}
|
|
|
|
def history(self) -> list[dict[str, Any]]:
|
|
return [self.ledger.describe_tx(tx) for tx in self.ledger.history()]
|
|
|
|
def bootstrap_opening_books(self) -> dict[str, Any]:
|
|
"""Seed opening balances without business txs (so contract can be first tx)."""
|
|
if self.db.get_meta("bootstrapped") == "1":
|
|
return {"bootstrapped": True, "already": True}
|
|
before = len(self.ledger.transactions)
|
|
self.ledger.bootstrap_endowment(
|
|
ACCOUNT_COFFER,
|
|
COFFER_INVESTMENT_CREDITS,
|
|
CreditKind.FUNDED,
|
|
"leora-investment",
|
|
)
|
|
self.ledger.bootstrap_endowment(
|
|
ACCOUNT_LEORA,
|
|
LEORA_FUNDED_AFTER_GRANTS_CREDITS,
|
|
CreditKind.FUNDED,
|
|
"funded-after-grants",
|
|
)
|
|
self.ledger.ensure_wallet(ACCOUNT_NEW_INTERNET)
|
|
self._rewrite_lots(ACCOUNT_COFFER)
|
|
self._rewrite_lots(ACCOUNT_LEORA)
|
|
self._ensure_account_row(ACCOUNT_NEW_INTERNET)
|
|
self.db.set_meta("bootstrapped", "1")
|
|
self.db.commit()
|
|
assert len(self.ledger.transactions) == before
|
|
return {
|
|
"bootstrapped": True,
|
|
"already": False,
|
|
"coffer": self.balance(ACCOUNT_COFFER),
|
|
"leora": self.balance(ACCOUNT_LEORA),
|
|
}
|
|
|
|
def record_build_contract(
|
|
self,
|
|
amount_credits: int = CONTRACT_CREDITS,
|
|
*,
|
|
payer: str = ACCOUNT_COFFER,
|
|
payee: str = ACCOUNT_LEORA,
|
|
) -> dict[str, Any]:
|
|
"""First business ledger tx: Coffer pays Leora for the product build."""
|
|
if self.db.get_meta("contract_recorded") == "1":
|
|
for tx in self.ledger.history():
|
|
if tx.kind == "contract_payment":
|
|
return {"already": True, "tx": self.ledger.describe_tx(tx)}
|
|
raise ProductError("contract_recorded meta set but tx missing")
|
|
if self.db.get_meta("bootstrapped") != "1":
|
|
self.bootstrap_opening_books()
|
|
tx = self.ledger.transfer(
|
|
payer,
|
|
payee,
|
|
to_subunits(amount_credits),
|
|
reason="build-contract:leora-coffer-product",
|
|
tx_kind="contract_payment",
|
|
credit_kind=CreditKind.FUNDED,
|
|
meta={"contract": "coffer-product-build", "credits": amount_credits},
|
|
)
|
|
self._commit_accounts(payer, payee, txs=[tx])
|
|
self.db.set_meta("contract_recorded", "1")
|
|
self.db.commit()
|
|
return {
|
|
"already": False,
|
|
"tx": self.ledger.describe_tx(tx),
|
|
"payer": self.balance(payer),
|
|
"payee": self.balance(payee),
|
|
}
|
|
|
|
def grant_signup(
|
|
self,
|
|
agent_id: str,
|
|
amount_credits: int = SIGNUP_GRANT_CREDITS,
|
|
) -> dict[str, Any]:
|
|
aid = agent_account(agent_id)
|
|
marker = f"signup_grant:{aid}"
|
|
if self.db.get_meta(marker) == "1":
|
|
return {"already": True, "account_id": aid, "balance": self.balance(aid)}
|
|
self.ledger.ensure_wallet(aid)
|
|
before = self.ledger._tx_seq
|
|
self.ledger.grant(aid, amount_credits, reason="signup-grant")
|
|
txs = [t for t in self.ledger.transactions if t.seq > before]
|
|
self._commit_accounts(aid, txs=txs)
|
|
self.db.set_meta(marker, "1")
|
|
self.db.commit()
|
|
return {
|
|
"already": False,
|
|
"account_id": aid,
|
|
"granted_credits": amount_credits,
|
|
"balance": self.balance(aid),
|
|
"tx": self.ledger.describe_tx(txs[-1]) if txs else None,
|
|
}
|
|
|
|
def charge_subscription(
|
|
self,
|
|
agent_id: str,
|
|
amount_credits: int = SUBSCRIPTION_CREDITS,
|
|
*,
|
|
cut_bps: int = SUBSCRIPTION_CUT_BPS,
|
|
) -> dict[str, Any]:
|
|
aid = agent_account(agent_id)
|
|
gross = to_subunits(amount_credits)
|
|
cut = cut_subunits(gross, cut_bps)
|
|
net = gross - cut
|
|
draws, tx_pay = self.ledger.purchase_settle(
|
|
aid,
|
|
ACCOUNT_COFFER,
|
|
gross,
|
|
reason="subscription:new-internet-seat",
|
|
tx_kind="subscription_charge",
|
|
meta={"agent_id": aid, "credits": amount_credits},
|
|
)
|
|
tx_fwd = self.ledger.transfer(
|
|
ACCOUNT_COFFER,
|
|
ACCOUNT_NEW_INTERNET,
|
|
net,
|
|
reason="subscription:forward-to-new-internet-co",
|
|
tx_kind="subscription_forward",
|
|
credit_kind=CreditKind.FUNDED,
|
|
meta={"agent_id": aid, "cut_centicredits": cut, "net_centicredits": net},
|
|
)
|
|
self._commit_accounts(
|
|
aid, ACCOUNT_COFFER, ACCOUNT_NEW_INTERNET, txs=[tx_pay, tx_fwd]
|
|
)
|
|
return {
|
|
"agent_id": aid,
|
|
"gross_credits": amount_credits,
|
|
"coffer_cut_credits": format_credits(cut),
|
|
"coffer_cut_centicredits": cut,
|
|
"leora_net_credits": format_credits(net),
|
|
"leora_net_centicredits": net,
|
|
"draws": [
|
|
{"kind": d["kind"], "amount_centicredits": d["amount"]} for d in draws
|
|
],
|
|
"txs": [self.ledger.describe_tx(tx_pay), self.ledger.describe_tx(tx_fwd)],
|
|
"balances": {
|
|
"agent": self.balance(aid),
|
|
"coffer": self.balance(ACCOUNT_COFFER),
|
|
"new_internet_co": self.balance(ACCOUNT_NEW_INTERNET),
|
|
},
|
|
}
|
|
|
|
def signup_flow(self, agent_id: str) -> dict[str, Any]:
|
|
wallet = self.create_wallet(agent_id)
|
|
grant = self.grant_signup(agent_id)
|
|
sub = self.charge_subscription(agent_id)
|
|
return {"wallet": wallet, "grant": grant, "subscription": sub}
|
|
|
|
def buy_shirt(
|
|
self,
|
|
agent_id: str,
|
|
quantity: int = 1,
|
|
idempotency_key: str | None = None,
|
|
) -> dict[str, Any]:
|
|
before = self.ledger._tx_seq
|
|
order = self.store.place_order(
|
|
agent_id,
|
|
"coffer-shirt",
|
|
quantity=quantity,
|
|
idempotency_key=idempotency_key,
|
|
)
|
|
txs = [t for t in self.ledger.transactions if t.seq > before]
|
|
self._persist_order(order)
|
|
if idempotency_key:
|
|
self.db.execute(
|
|
"INSERT OR REPLACE INTO idempotency(key, order_id) VALUES(?,?)",
|
|
(idempotency_key, order.order_id),
|
|
)
|
|
self._commit_accounts(order.buyer_id, ACCOUNT_COFFER, txs=txs)
|
|
return {
|
|
"order": asdict(order),
|
|
"balance": self.balance(order.buyer_id),
|
|
"txs": [self.ledger.describe_tx(t) for t in txs],
|
|
}
|
|
|
|
def top_up(self, agent_id: str, amount_credits: int) -> dict[str, Any]:
|
|
aid = agent_account(agent_id)
|
|
before = self.ledger._tx_seq
|
|
self.ledger.top_up(aid, amount_credits)
|
|
txs = [t for t in self.ledger.transactions if t.seq > before]
|
|
self._commit_accounts(aid, txs=txs)
|
|
return {
|
|
"account_id": aid,
|
|
"balance": self.balance(aid),
|
|
"tx": self.ledger.describe_tx(txs[-1]),
|
|
}
|
|
|
|
def all_balances(self) -> dict[str, dict]:
|
|
return {aid: self.balance(aid) for aid in sorted(self.ledger.accounts.keys())}
|
|
|
|
def public_snapshot(self) -> dict:
|
|
"""Full public ledger view (no human PII). All txs are public."""
|
|
return snapshot_from_product(self, now_pt())
|
|
|
|
def publish_public_ledger(self) -> dict:
|
|
"""Write public snapshot and publish to News (branch → PR → merge → surge)."""
|
|
snap = self.public_snapshot()
|
|
pub = Path(self.db.path).with_name("public-ledger.json")
|
|
import json
|
|
|
|
pub.write_text(
|
|
json.dumps(
|
|
{
|
|
"updated_at": snap["updated_at"],
|
|
"balances": snap["balances"],
|
|
"transactions": snap["transactions"],
|
|
},
|
|
indent=2,
|
|
sort_keys=True,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
from coffer.news_publish import try_auto_publish
|
|
|
|
result = try_auto_publish(snap["article_html"], updated_at=snap["updated_at"])
|
|
result["snapshot_path"] = str(pub)
|
|
return result
|
|
|
|
def _after_mutation(self) -> None:
|
|
"""Ledger changes are public; refresh local snapshot and News when enabled."""
|
|
import json
|
|
import os
|
|
|
|
snap = self.public_snapshot()
|
|
pub = Path(self.db.path).with_name("public-ledger.json")
|
|
pub.write_text(
|
|
json.dumps(
|
|
{
|
|
"updated_at": snap["updated_at"],
|
|
"balances": snap["balances"],
|
|
"transactions": snap["transactions"],
|
|
},
|
|
indent=2,
|
|
sort_keys=True,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
if os.environ.get("COFFER_AUTO_PUBLISH_NEWS", "0") not in {"1", "true", "yes"}:
|
|
return
|
|
try:
|
|
from coffer.news_publish import try_auto_publish
|
|
|
|
try_auto_publish(snap["article_html"], updated_at=snap["updated_at"])
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"coffer news publish skipped: {exc}", flush=True)
|