T012: ledger API routes per contract

ledger/src/m2_ledger/api.py

@@ -0,0 +1,148 @@
+"""m2-ledger FastAPI app: wires models.py (tx log + balances) to auth.py
+(X-API-Key service/admin classes) per contracts/ledger-api.md (frozen v1).
+"""
+
+import json
+import os
+from datetime import datetime, timezone
+from typing import Optional
+
+from fastapi import Depends, FastAPI
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+
+from m2_ledger import models
+from m2_ledger.auth import require_admin, require_service
+
+app = FastAPI(title="m2-ledger")
+
+
+def _db_path() -> str:
+    return os.environ.get("LEDGER_DB_PATH", "./ledger.db")
+
+
+def _platform_pct_default() -> float:
+    return float(os.environ.get("LEDGER_PLATFORM_PCT", "0.10"))
+
+
+def _starter_grant_default() -> int:
+    return int(os.environ.get("LEDGER_STARTER_GRANT", "100"))
+
+
+def get_conn():
+    conn = models.init_db(_db_path())
+    try:
+        yield conn
+    finally:
+        conn.close()
+
+
+class InstallRequest(BaseModel):
+    buyer: str
+    seller: str
+    amount: int
+    platform_pct: Optional[float] = None
+    ref: str
+
+
+class RefundRequest(BaseModel):
+    ref: str
+
+
+class StarterGrantRequest(BaseModel):
+    operator_id: str
+    amount: Optional[int] = None
+
+
+@app.get("/health")
+def health():
+    return {"status": "healthy"}
+
+
+@app.get("/balance/{operator_id}")
+def get_balance(operator_id: str, conn=Depends(get_conn), _key: str = Depends(require_service)):
+    bal = models.balance(conn, operator_id)
+    return {
+        "operator_id": operator_id,
+        "balance": bal,
+        "as_of": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
+    }
+
+
+@app.post("/tx/install")
+def install(
+    body: InstallRequest,
+    conn=Depends(get_conn),
+    _key: str = Depends(require_service),
+):
+    pct = body.platform_pct if body.platform_pct is not None else _platform_pct_default()
+    try:
+        result = models.record_install(conn, body.buyer, body.seller, body.amount, pct, body.ref)
+    except models.InsufficientFunds:
+        return JSONResponse(
+            status_code=402,
+            content={"error": "insufficient_funds", "balance": models.balance(conn, body.buyer)},
+        )
+    except models.DuplicateRef:
+        return JSONResponse(status_code=409, content={"error": "duplicate_ref"})
+    return result
+
+
+@app.post("/tx/refund")
+def refund(
+    body: RefundRequest,
+    conn=Depends(get_conn),
+    _key: str = Depends(require_service),
+):
+    try:
+        result = models.record_refund(conn, body.ref)
+    except models.UnknownRef:
+        return JSONResponse(status_code=404, content={"error": "unknown_ref"})
+    except models.AlreadyRefunded:
+        return JSONResponse(status_code=409, content={"error": "already_refunded"})
+    return result
+
+
+@app.get("/tx")
+def list_tx(
+    operator_id: Optional[str] = None,
+    reason: Optional[str] = None,
+    since: Optional[str] = None,
+    conn=Depends(get_conn),
+    _key: str = Depends(require_service),
+):
+    return {"transactions": models.transactions(conn, operator_id=operator_id, reason=reason, since=since)}
+
+
+@app.get("/license/{operator_id}/{listing_id}")
+def get_license(
+    operator_id: str,
+    listing_id: str,
+    conn=Depends(get_conn),
+    _key: str = Depends(require_service),
+):
+    grant = models.license_for(conn, operator_id, listing_id)
+    if grant is None:
+        return JSONResponse(status_code=404, content={"error": "not_found"})
+    return {"grant": grant}
+
+
+@app.post("/grant/starter")
+def grant_starter(
+    body: StarterGrantRequest,
+    conn=Depends(get_conn),
+    _key: str = Depends(require_admin),
+):
+    amount = body.amount if body.amount is not None else _starter_grant_default()
+    return models.record_grant(conn, body.operator_id, amount)
+
+
+@app.post("/snapshot")
+def snapshot(conn=Depends(get_conn), _key: str = Depends(require_admin)):
+    balances = models.all_balances(conn)
+    date = datetime.now(timezone.utc).strftime("%Y-%m-%d")
+    path = f"audit/{date}.json"
+    content = json.dumps(balances, sort_keys=True, indent=2)
+    # TODO(T034): commit `content` to `path` in m2/market-registry via Forgejo API
+    # (REGISTRY_REPO_URL + FORGEJO_TOKEN); this endpoint only produces the JSON.
+    return {"path": path, "content": content}