hanzoai/bootnode
2
0
Fork 0
mirror of https://github.com/hanzoai/bootnode.git synced 2026-10-25 02:31:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: migrate billing from Stripe to Hanzo Commerce + Square

Browse Source 
Remove all direct Stripe integration. Bootnode now uses Hanzo Commerce
API as its payment gateway (Commerce handles Square internally).

- Remove stripe dependency from pyproject.toml
- Rewrite commerce.py client with correct Commerce API paths:
  /api/v1/user (customers), /api/v1/subscribe (subscriptions),
  /api/v1/checkout/* (payments), /api/v1/order (orders/invoices)
- Add unified.py: IAM + Commerce customer linking (hanzo.id SSO)
- Rewrite webhooks.py: Commerce event handler with HMAC verification
- Update sync.py: usage metering via Commerce (not Stripe)
- Update billing API endpoints: checkout, account, webhooks
- Config uses COMMERCE_* prefix (not HANZO_COMMERCE_*)
- Add test_billing.py: 46 tests covering CU, tiers, Commerce client,
  webhooks, and unified billing
This commit is contained in:
zeekay
2026-02-10 20:50:18 -08:00
parent 60cb93aca6
commit ad6dc4e6eb
10 changed files with 1520 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/.env.example
+5
View File
@@ -70,3 +70,8 @@ HANZO_IAM_CLIENT_SECRET=
# Hanzo HSM (Hardware Security - optional)
HANZO_HSM_ENABLED=false
# ZAP Server (native Cap'n Proto RPC)
ZAP_ENABLED=true
ZAP_HOST=0.0.0.0
ZAP_PORT=9999
api/bootnode/api/billing/__init__.py
+235 -82
View File
@@ -1,16 +1,18 @@
"""Billing API - Usage tracking and subscription management."""
"""Billing API - Usage tracking, subscription management, and Commerce integration."""
import json
import uuid
from datetime import date, datetime
from typing import Any
from fastapi import APIRouter, HTTPException, Request, status
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel
from bootnode.api.deps import DbDep, ProjectDep
from bootnode.config import get_settings
from bootnode.core.billing import (
TIER_LIMITS,
CommerceError,
Invoice,
PricingTier,
UsageSummary,
@@ -135,8 +137,41 @@ class LimitsCheckResponse(BaseModel):
rate_limit_per_second: int
class CheckoutRequest(BaseModel):
"""Commerce Checkout request (Square payment via Commerce)."""
tier: PricingTier
nonce: str | None = None # Square payment nonce
source_id: str | None = None # Square card on file
success_url: str | None = None
cancel_url: str | None = None
class CheckoutResponse(BaseModel):
"""Commerce Checkout response."""
order_id: str | None = None
checkout_url: str | None = None
authorization_id: str | None = None
class CaptureRequest(BaseModel):
"""Capture an authorized payment."""
pass
class SyncStatusResponse(BaseModel):
"""Usage sync status response."""
running: bool
interval_seconds: int
last_sync: str | None
lock_held: bool
# =============================================================================
# Endpoints
# Usage & Subscription Endpoints (unchanged — payment-agnostic)
# =============================================================================
@@ -197,7 +232,7 @@ async def upgrade_subscription(
) -> Subscription:
"""Upgrade subscription to a new tier.
For paid tiers, a hanzo_subscription_id from Hanzo Commerce is required.
For paid tiers, a hanzo_subscription_id from Commerce is required.
"""
current = await billing_service.get_or_create_subscription(project.id, db)
current_tier = PricingTier(current.tier)
@@ -321,56 +356,107 @@ async def check_limits(
# =============================================================================
# Hanzo Commerce Integration
# Commerce Checkout (Square payments via Commerce API)
# =============================================================================
class CommerceInvoiceResponse(BaseModel):
"""Commerce invoice response."""
id: str
customer_id: str
subscription_id: str | None
status: str
amount_due: int
amount_paid: int
currency: str
period_start: datetime | None
period_end: datetime | None
paid_at: datetime | None
hosted_invoice_url: str | None
pdf_url: str | None
created_at: datetime
TIER_TO_PLAN_SLUG: dict[PricingTier, str] = {
PricingTier.FREE: "bootnode-free",
PricingTier.PAY_AS_YOU_GO: "bootnode-payg",
PricingTier.GROWTH: "bootnode-growth",
PricingTier.ENTERPRISE: "bootnode-enterprise",
}
class SyncStatusResponse(BaseModel):
"""Usage sync status response."""
@router.post("/checkout", response_model=CheckoutResponse)
async def create_checkout(
request: CheckoutRequest,
project: ProjectDep,
db: DbDep,
) -> CheckoutResponse:
"""Create a Commerce checkout session (Square payment).
running: bool
interval_seconds: int
last_sync: str | None
lock_held: bool
Returns an order_id and checkout_url for the user to complete payment.
After payment, Commerce webhooks handle provisioning.
"""
settings = get_settings()
if request.tier in (PricingTier.FREE, PricingTier.ENTERPRISE):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Checkout not available for {request.tier.value} tier",
)
subscription = await billing_service.get_or_create_subscription(project.id, db)
customer_id = subscription.hanzo_customer_id
if not customer_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="No billing account found. Use /billing/account to set up billing first.",
)
plan_slug = TIER_TO_PLAN_SLUG.get(request.tier, "bootnode-payg")
success_url = request.success_url or f"{settings.frontend_url}/billing?checkout=success"
cancel_url = request.cancel_url or f"{settings.frontend_url}/billing?checkout=cancelled"
try:
result = await commerce_client.create_checkout(
customer_id=customer_id,
plan_slug=plan_slug,
project_id=project.id,
nonce=request.nonce,
source_id=request.source_id,
success_url=success_url,
cancel_url=cancel_url,
)
return CheckoutResponse(
order_id=result.get("order_id"),
checkout_url=result.get("checkout_url"),
authorization_id=result.get("authorization_id"),
)
except CommerceError as e:
raise HTTPException(
status_code=e.status_code,
detail=e.message,
)
@router.post("/checkout/capture/{order_id}")
async def capture_checkout(
order_id: str,
) -> dict:
"""Capture an authorized Commerce payment."""
try:
result = await commerce_client.capture_checkout(order_id)
return result
except CommerceError as e:
raise HTTPException(
status_code=e.status_code,
detail=e.message,
)
# =============================================================================
# Commerce Webhooks
# =============================================================================
@router.post("/webhooks/commerce")
async def commerce_webhook(request: Request) -> dict:
"""
Handle Hanzo Commerce webhooks.
"""Handle Hanzo Commerce webhooks.
Processes subscription lifecycle events, invoice events, etc.
Verifies HMAC-SHA256 signature before processing.
Verifies HMAC signature and processes subscription/order/payment events.
"""
payload = await request.body()
signature = request.headers.get("X-Commerce-Signature", "")
# Verify signature
if not webhook_handler.verify_signature(payload, signature):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid webhook signature",
)
# Parse and handle event
try:
event = json.loads(payload)
except json.JSONDecodeError:
@@ -383,67 +469,134 @@ async def commerce_webhook(request: Request) -> dict:
return {"received": True, **result}
@router.get("/commerce/invoices", response_model=list[CommerceInvoiceResponse])
async def get_commerce_invoices(
project: ProjectDep,
db: DbDep,
limit: int = 10,
) -> list[CommerceInvoiceResponse]:
"""
Get invoices from Hanzo Commerce.
Returns invoices for the project's Commerce customer.
"""
subscription = await billing_service.get_subscription(project.id, db)
if not subscription or not subscription.hanzo_customer_id:
return []
try:
invoices = await commerce_client.get_customer_invoices(
subscription.hanzo_customer_id,
limit=limit,
)
return [
CommerceInvoiceResponse(
id=inv.id,
customer_id=inv.customer_id,
subscription_id=inv.subscription_id,
status=inv.status.value,
amount_due=inv.amount_due,
amount_paid=inv.amount_paid,
currency=inv.currency,
period_start=inv.period_start,
period_end=inv.period_end,
paid_at=inv.paid_at,
hosted_invoice_url=inv.hosted_invoice_url,
pdf_url=inv.pdf_url,
created_at=inv.created_at,
)
for inv in invoices
]
except Exception as e:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=f"Failed to fetch invoices from Commerce: {str(e)}",
)
# =============================================================================
# Usage Sync (to Commerce for metered billing)
# =============================================================================
@router.post("/commerce/sync")
@router.post("/sync")
async def sync_usage(
project: ProjectDep,
) -> dict:
"""
Manually sync usage to Hanzo Commerce.
"""Manually sync usage to Commerce.
Useful before subscription changes to ensure all usage is reported.
Reports accumulated compute units to Commerce for metered billing.
Commerce handles invoicing via Square.
"""
result = await usage_sync_worker.sync_project(project.id)
return result
@router.get("/commerce/sync/status", response_model=SyncStatusResponse)
@router.get("/sync/status", response_model=SyncStatusResponse)
async def get_sync_status() -> SyncStatusResponse:
"""Get usage sync worker status."""
status_data = await usage_sync_worker.get_status()
return SyncStatusResponse(**status_data)
# =============================================================================
# Unified IAM + Commerce Billing
# =============================================================================
from bootnode.core.billing.unified import get_unified_billing_client
from bootnode.core.iam import IAMUser, get_current_user
class UnifiedUserResponse(BaseModel):
"""Unified user with IAM and Commerce billing data."""
iam_id: str
email: str
name: str
org: str
commerce_customer_id: str | None
square_customer_id: str | None
has_payment_method: bool
default_payment_method: str | None
class PaymentMethodResponse(BaseModel):
"""Payment method response."""
id: str
type: str
brand: str | None
last4: str | None
exp_month: int | None
exp_year: int | None
is_default: bool
@router.get("/account", response_model=UnifiedUserResponse)
async def get_billing_account(
user: IAMUser = Depends(get_current_user),
) -> UnifiedUserResponse:
"""Get billing account for authenticated user.
Links IAM user to Commerce customer. Creates Commerce customer if needed.
"""
unified = get_unified_billing_client()
unified_user = await unified.get_or_create_customer(user)
return UnifiedUserResponse(
iam_id=unified_user.iam_id,
email=unified_user.email,
name=unified_user.name,
org=unified_user.org,
commerce_customer_id=unified_user.commerce_customer_id,
square_customer_id=unified_user.square_customer_id,
has_payment_method=unified_user.has_payment_method,
default_payment_method=unified_user.default_payment_method,
)
@router.get("/account/subscriptions")
async def get_account_subscriptions(
user: IAMUser = Depends(get_current_user),
) -> list[dict]:
"""Get all Commerce subscriptions for authenticated user."""
unified = get_unified_billing_client()
return await unified.get_customer_subscriptions(user)
@router.get("/account/invoices")
async def get_account_invoices(
user: IAMUser = Depends(get_current_user),
) -> list[dict]:
"""Get all Commerce invoices for authenticated user."""
unified = get_unified_billing_client()
return await unified.get_customer_invoices(user)
@router.get("/account/payment-methods", response_model=list[PaymentMethodResponse])
async def get_account_payment_methods(
user: IAMUser = Depends(get_current_user),
) -> list[PaymentMethodResponse]:
"""Get payment methods for authenticated user (Square cards via Commerce)."""
unified = get_unified_billing_client()
methods = await unified.get_customer_payment_methods(user)
return [
PaymentMethodResponse(
id=m.get("id", ""),
type=m.get("type", "card"),
brand=m.get("card", {}).get("brand") if isinstance(m.get("card"), dict) else None,
last4=m.get("card", {}).get("last4") if isinstance(m.get("card"), dict) else None,
exp_month=m.get("card", {}).get("exp_month") if isinstance(m.get("card"), dict) else None,
exp_year=m.get("card", {}).get("exp_year") if isinstance(m.get("card"), dict) else None,
is_default=m.get("is_default", False),
)
for m in methods
]
@router.post("/account/sync")
async def sync_iam_to_commerce(
user: IAMUser = Depends(get_current_user),
) -> dict:
"""Sync IAM user data to Commerce customer.
Updates Commerce customer with latest IAM profile data.
"""
unified = get_unified_billing_client()
result = await unified.sync_iam_to_commerce(user)
return {"synced": True, "result": result}
api/bootnode/config.py
+28 -4
View File
@@ -80,6 +80,21 @@ class Settings(BaseSettings):
datastore_url: str = "clickhouse://bootnode:bootnode@localhost:8123/bootnode"
datastore_pool_size: int = 10
# Vector Search (Qdrant)
qdrant_url: str = "http://localhost:6333"
qdrant_api_key: str = ""
# Full-Text Search (Meilisearch)
meilisearch_url: str = "http://localhost:7700"
meilisearch_api_key: str = ""
# Lux Indexer (Omnichain data)
indexer_url: str = "http://localhost:5000"
# Cloudflare (Edge Caching)
cloudflare_zone_id: str = ""
cloudflare_api_token: str = ""
# Message Queue (Redis-based, @hanzo/mq compatible)
# Uses the same Redis instance as cache - different db index for isolation
mq_redis_db: int = 1 # Redis DB index for MQ (separate from cache at db 0)
@@ -91,21 +106,30 @@ class Settings(BaseSettings):
webhook_timeout: int = 30
webhook_max_retries: int = 5
# Hanzo Commerce (Stripe-based billing)
hanzo_commerce_url: str = "https://commerce.hanzo.ai"
hanzo_commerce_api_key: str = ""
hanzo_commerce_webhook_secret: str = ""
# Hanzo Commerce (Square-based billing via Commerce API)
commerce_url: str = "http://commerce.hanzo.svc:8001"
commerce_api_key: str = ""
commerce_webhook_secret: str = ""
# ERC-4337 Bundler
bundler_private_key: str = ""
bundler_beneficiary: str = ""
# ZAP Server (native Cap'n Proto RPC)
zap_enabled: bool = True
zap_host: str = "0.0.0.0"
zap_port: int = 9999
# Deployment Target
deploy_target: Literal["docker", "process", "kubernetes"] = "docker"
deploy_compose_file: str = "infra/compose.yml"
deploy_k8s_namespace: str = "bootnode"
deploy_k8s_context: str = ""
# Helm / Lux Fleet
helm_chart_path: str = "/opt/charts/lux"
helm_binary: str = "helm"
@property
def is_production(self) -> bool:
return self.app_env == "production"
api/bootnode/core/billing/__init__.py
+13 -3
View File
@@ -5,7 +5,8 @@ Exports:
- Pricing tiers and limits
- Real-time usage tracking (Redis)
- Billing service for subscriptions
- Hanzo Commerce integration
- Hanzo Commerce integration (Square-based payments)
- Unified IAM + Commerce integration
"""
from bootnode.core.billing.commerce import (
@@ -51,6 +52,11 @@ from bootnode.core.billing.webhooks import (
WebhookVerificationError,
webhook_handler,
)
from bootnode.core.billing.unified import (
UnifiedBillingClient,
UnifiedUser,
get_unified_billing_client,
)
__all__ = [
# Compute units
@@ -72,7 +78,7 @@ __all__ = [
"billing_service",
"UsageSummary",
"Invoice",
# Hanzo Commerce
# Hanzo Commerce (Square-based payments)
"HanzoCommerceClient",
"commerce_client",
"CommerceError",
@@ -90,8 +96,12 @@ __all__ = [
# Usage Sync
"UsageSyncWorker",
"usage_sync_worker",
# Webhooks
# Commerce Webhooks
"CommerceWebhookHandler",
"webhook_handler",
"WebhookVerificationError",
# Unified IAM + Commerce
"UnifiedBillingClient",
"UnifiedUser",
"get_unified_billing_client",
]
api/bootnode/core/billing/commerce.py
+190 -56
View File
@@ -1,10 +1,12 @@
"""Hanzo Commerce client for subscription billing.
Integrates with Hanzo Commerce API (Stripe-based) for:
Integrates with Hanzo Commerce API (Square-based) for:
- Customer management
- Subscription lifecycle
- Checkout (Square payment authorization/capture)
- Usage metering (PAYG)
- Invoice retrieval
- Order management
"""
from datetime import datetime
@@ -38,12 +40,12 @@ class CommerceError(Exception):
class HanzoCommerceClient:
"""Client for Hanzo Commerce billing API."""
"""Client for Hanzo Commerce billing API (Square backend)."""
def __init__(self) -> None:
settings = get_settings()
self.base_url = settings.hanzo_commerce_url
self.api_key = settings.hanzo_commerce_api_key
self.base_url = settings.commerce_url
self.api_key = settings.commerce_api_key
self._timeout = httpx.Timeout(30.0)
def _headers(self) -> dict[str, str]:
@@ -108,13 +110,10 @@ class HanzoCommerceClient:
name: str,
org: str = "hanzo",
) -> str:
"""
Create customer in Hanzo Commerce.
Returns customer_id.
"""
"""Create customer in Hanzo Commerce. Returns customer_id."""
data = await self._request(
"POST",
"/api/v1/customers",
"/api/v1/user",
json={
"email": email,
"name": name,
@@ -137,7 +136,7 @@ class HanzoCommerceClient:
async def get_customer(self, customer_id: str) -> Customer:
"""Get customer by ID."""
data = await self._request("GET", f"/api/v1/customers/{customer_id}")
data = await self._request("GET", f"/api/v1/user/{customer_id}")
return Customer(
id=data["id"],
email=data["email"],
@@ -156,7 +155,7 @@ class HanzoCommerceClient:
try:
data = await self._request(
"GET",
"/api/v1/customers",
"/api/v1/user",
params={"email": email},
)
customers = data.get("data", [])
@@ -193,7 +192,7 @@ class HanzoCommerceClient:
data = await self._request(
"PATCH",
f"/api/v1/customers/{customer_id}",
f"/api/v1/user/{customer_id}",
json=update_data,
)
return Customer(
@@ -204,6 +203,143 @@ class HanzoCommerceClient:
metadata=data.get("metadata", {}),
)
# =========================================================================
# Checkout (Square payment via Commerce)
# =========================================================================
async def create_checkout(
self,
customer_id: str,
plan_slug: str,
project_id: UUID,
nonce: str | None = None,
source_id: str | None = None,
success_url: str | None = None,
cancel_url: str | None = None,
) -> dict[str, Any]:
"""Authorize a payment via Commerce (Square).
Args:
customer_id: Commerce customer ID
plan_slug: Plan to subscribe to (e.g. bootnode-payg)
project_id: Bootnode project ID
nonce: Square payment nonce from Web Payments SDK
source_id: Square source ID (card on file)
success_url: Redirect URL on success
cancel_url: Redirect URL on cancel
Returns:
Dict with order_id, authorization_id, and checkout_url (if hosted)
"""
payload: dict[str, Any] = {
"customer_id": customer_id,
"plan_slug": plan_slug,
"metadata": {
"project_id": str(project_id),
"source": "bootnode",
},
}
if nonce:
payload["nonce"] = nonce
if source_id:
payload["source_id"] = source_id
if success_url:
payload["success_url"] = success_url
if cancel_url:
payload["cancel_url"] = cancel_url
data = await self._request(
"POST",
"/api/v1/checkout/authorize",
json=payload,
)
logger.info(
"Checkout authorized via Commerce",
order_id=data.get("order_id"),
customer_id=customer_id,
plan_slug=plan_slug,
)
return data
async def capture_checkout(self, order_id: str) -> dict[str, Any]:
"""Capture an authorized payment.
Args:
order_id: Commerce order ID from create_checkout
Returns:
Dict with capture result (order status, payment details)
"""
data = await self._request(
"POST",
f"/api/v1/checkout/capture/{order_id}",
)
logger.info("Checkout captured", order_id=order_id)
return data
async def charge(
self,
customer_id: str,
amount_cents: int,
currency: str = "usd",
description: str = "",
metadata: dict | None = None,
) -> dict[str, Any]:
"""Single-step charge (authorize + capture).
Args:
customer_id: Commerce customer ID
amount_cents: Amount in cents
currency: Currency code
description: Charge description
metadata: Additional metadata
Returns:
Dict with order_id, payment details
"""
payload: dict[str, Any] = {
"customer_id": customer_id,
"amount": amount_cents,
"currency": currency,
}
if description:
payload["description"] = description
if metadata:
payload["metadata"] = metadata
return await self._request(
"POST",
"/api/v1/checkout/charge",
json=payload,
)
# =========================================================================
# Order Management
# =========================================================================
async def get_user_orders(
self,
customer_id: str,
limit: int = 20,
) -> list[dict[str, Any]]:
"""Get orders for a customer.
Args:
customer_id: Commerce customer ID
limit: Max orders to return
Returns:
List of order dicts
"""
data = await self._request(
"GET",
f"/api/v1/user/{customer_id}/orders",
params={"limit": limit},
)
return data.get("data", [])
# =========================================================================
# Subscription Management
# =========================================================================
@@ -214,13 +350,13 @@ class HanzoCommerceClient:
plan_slug: str,
project_id: UUID,
) -> Subscription:
"""
Create subscription for a plan.
Plans: bootnode-free, bootnode-payg, bootnode-enterprise
"""Create subscription for a plan.
Plans: bootnode-free, bootnode-payg, bootnode-growth, bootnode-enterprise
"""
data = await self._request(
"POST",
"/api/v1/subscriptions",
"/api/v1/subscribe",
json={
"customer_id": customer_id,
"plan_slug": plan_slug,
@@ -254,7 +390,7 @@ class HanzoCommerceClient:
async def get_subscription(self, subscription_id: str) -> Subscription:
"""Get subscription by ID."""
data = await self._request("GET", f"/api/v1/subscriptions/{subscription_id}")
data = await self._request("GET", f"/api/v1/subscribe/{subscription_id}")
return Subscription(
id=data["id"],
customer_id=data["customer_id"],
@@ -275,18 +411,18 @@ class HanzoCommerceClient:
async def get_subscriptions_by_customer(
self, customer_id: str
) -> list[Subscription]:
"""Get all subscriptions for a customer."""
"""Get all subscriptions for a customer (via user orders endpoint)."""
data = await self._request(
"GET",
"/api/v1/subscriptions",
params={"customer_id": customer_id},
f"/api/v1/user/{customer_id}/orders",
params={"type": "subscription"},
)
return [
Subscription(
id=s["id"],
customer_id=s["customer_id"],
plan_slug=s["plan_slug"],
status=SubscriptionStatus(s["status"]),
customer_id=s.get("customer_id", customer_id),
plan_slug=s.get("plan_slug", ""),
status=SubscriptionStatus(s.get("status", "active")),
current_period_start=datetime.fromisoformat(s["current_period_start"]),
current_period_end=datetime.fromisoformat(s["current_period_end"]),
cancel_at_period_end=s.get("cancel_at_period_end", False),
@@ -304,7 +440,7 @@ class HanzoCommerceClient:
"""Upgrade or downgrade subscription."""
data = await self._request(
"PATCH",
f"/api/v1/subscriptions/{subscription_id}",
f"/api/v1/subscribe/{subscription_id}",
json={"plan_slug": plan_slug},
)
@@ -331,13 +467,14 @@ class HanzoCommerceClient:
subscription_id: str,
immediately: bool = False,
) -> bool:
"""
Cancel subscription.
"""Cancel subscription.
Commerce uses DELETE /api/v1/subscribe/:id.
If immediately=False, cancels at end of billing period.
"""
data = await self._request(
"POST",
f"/api/v1/subscriptions/{subscription_id}/cancel",
"DELETE",
f"/api/v1/subscribe/{subscription_id}",
json={"immediately": immediately},
)
@@ -348,24 +485,6 @@ class HanzoCommerceClient:
)
return data.get("cancelled", True)
async def reactivate_subscription(self, subscription_id: str) -> Subscription:
"""Reactivate a cancelled subscription (if not yet expired)."""
data = await self._request(
"POST",
f"/api/v1/subscriptions/{subscription_id}/reactivate",
)
return Subscription(
id=data["id"],
customer_id=data["customer_id"],
plan_slug=data["plan_slug"],
status=SubscriptionStatus(data["status"]),
current_period_start=datetime.fromisoformat(data["current_period_start"]),
current_period_end=datetime.fromisoformat(data["current_period_end"]),
cancel_at_period_end=data.get("cancel_at_period_end", False),
metadata=data.get("metadata", {}),
created_at=datetime.fromisoformat(data["created_at"]),
)
# =========================================================================
# Usage Metering (for PAYG)
# =========================================================================
@@ -377,10 +496,13 @@ class HanzoCommerceClient:
timestamp: datetime,
idempotency_key: str | None = None,
) -> None:
"""Report metered usage for PAYG billing."""
"""Report metered usage for PAYG billing.
Commerce handles metering internally and creates Square invoices.
"""
await self._request(
"POST",
f"/api/v1/subscriptions/{subscription_id}/usage",
f"/api/v1/subscribe/{subscription_id}/usage",
json={
"quantity": compute_units,
"timestamp": timestamp.isoformat(),
@@ -410,7 +532,7 @@ class HanzoCommerceClient:
return await self._request(
"GET",
f"/api/v1/subscriptions/{subscription_id}/usage",
f"/api/v1/subscribe/{subscription_id}/usage",
params=params or None,
)
@@ -423,11 +545,11 @@ class HanzoCommerceClient:
customer_id: str,
limit: int = 10,
) -> list[Invoice]:
"""Get customer invoices."""
"""Get customer invoices (mapped from Commerce orders)."""
data = await self._request(
"GET",
"/api/v1/invoices",
params={"customer_id": customer_id, "limit": limit},
f"/api/v1/user/{customer_id}/orders",
params={"limit": limit},
)
return [
Invoice(
@@ -461,8 +583,8 @@ class HanzoCommerceClient:
]
async def get_invoice(self, invoice_id: str) -> Invoice:
"""Get specific invoice."""
inv = await self._request("GET", f"/api/v1/invoices/{invoice_id}")
"""Get specific invoice (mapped from Commerce order)."""
inv = await self._request("GET", f"/api/v1/order/{invoice_id}")
return Invoice(
id=inv["id"],
customer_id=inv["customer_id"],
@@ -494,8 +616,8 @@ class HanzoCommerceClient:
try:
inv = await self._request(
"GET",
"/api/v1/invoices/upcoming",
params={"customer_id": customer_id},
f"/api/v1/user/{customer_id}/orders",
params={"status": "draft", "limit": 1},
)
return Invoice(
id=inv.get("id", "upcoming"),
@@ -522,6 +644,18 @@ class HanzoCommerceClient:
return None
raise
# =========================================================================
# Payment Methods (via Commerce → Square)
# =========================================================================
async def get_payment_methods(self, customer_id: str) -> list[dict[str, Any]]:
"""Get payment methods for a customer (Square cards on file)."""
data = await self._request(
"GET",
f"/api/v1/user/{customer_id}/paymentmethods",
)
return data.get("data", [])
# Global singleton
commerce_client = HanzoCommerceClient()
api/bootnode/core/billing/sync.py
+27 -23
View File
@@ -1,7 +1,7 @@
"""Usage sync worker for Hanzo Commerce.
"""Usage sync worker for Commerce billing.
Periodically syncs compute unit usage from Redis to Hanzo Commerce
for PAYG (pay-as-you-go) billing.
for PAYG (pay-as-you-go) metered billing. Commerce handles invoicing via Square.
"""
import asyncio
@@ -33,8 +33,7 @@ class UsageSyncWorker:
"""Periodically sync usage from Redis to Hanzo Commerce."""
def __init__(self, interval_seconds: int = 3600) -> None:
"""
Initialize sync worker.
"""Initialize sync worker.
Args:
interval_seconds: Sync interval (default: 1 hour)
@@ -59,19 +58,23 @@ class UsageSyncWorker:
async def _get_payg_subscriptions(
self, db: AsyncSession
) -> list[tuple[uuid.UUID, str]]:
"""
Get all active PAYG subscriptions.
) -> list[tuple[uuid.UUID, str, str]]:
"""Get all active PAYG subscriptions.
Returns list of (project_id, hanzo_subscription_id) tuples.
Returns list of (project_id, hanzo_subscription_id, hanzo_customer_id) tuples.
"""
result = await db.execute(
select(Subscription.project_id, Subscription.hanzo_subscription_id).where(
select(
Subscription.project_id,
Subscription.hanzo_subscription_id,
Subscription.hanzo_customer_id,
).where(
Subscription.tier == PricingTier.PAY_AS_YOU_GO.value,
Subscription.hanzo_subscription_id.isnot(None),
Subscription.hanzo_customer_id.isnot(None),
)
)
return [(row[0], row[1]) for row in result.all()]
return [(row[0], row[1], row[2]) for row in result.all()]
async def _get_unsync_usage(self, project_id: uuid.UUID) -> int:
"""Get compute units accumulated since last sync."""
@@ -93,11 +96,11 @@ class UsageSyncWorker:
self,
project_id: uuid.UUID,
subscription_id: str,
customer_id: str,
) -> dict[str, Any]:
"""
Sync usage for a single project.
"""Sync usage for a single project.
Returns sync result dict.
Reports compute units to Commerce, which handles metered invoicing via Square.
"""
result = {
"project_id": str(project_id),
@@ -121,7 +124,7 @@ class UsageSyncWorker:
sync_time = datetime.now(UTC)
idempotency_key = f"{project_id}:{sync_time.strftime('%Y-%m-%d-%H')}"
# Report to Commerce
# Report to Commerce (uses subscription_id, Commerce handles Square invoicing)
await commerce_client.report_usage(
subscription_id=subscription_id,
compute_units=compute_units,
@@ -158,8 +161,7 @@ class UsageSyncWorker:
return result
async def sync_all_projects(self) -> dict[str, Any]:
"""
Sync usage for all active PAYG subscriptions.
"""Sync usage for all active PAYG subscriptions.
Returns summary of sync operation.
"""
@@ -184,8 +186,8 @@ class UsageSyncWorker:
subscriptions = await self._get_payg_subscriptions(db)
summary["total_projects"] = len(subscriptions)
for project_id, subscription_id in subscriptions:
result = await self._sync_project(project_id, subscription_id)
for project_id, subscription_id, customer_id in subscriptions:
result = await self._sync_project(project_id, subscription_id, customer_id)
if result["synced"]:
summary["synced"] += 1
@@ -217,27 +219,29 @@ class UsageSyncWorker:
return summary
async def sync_project(self, project_id: uuid.UUID) -> dict[str, Any]:
"""
Sync usage for a specific project immediately.
"""Sync usage for a specific project immediately.
Useful for on-demand sync before subscription changes.
"""
async with async_session() as db:
result = await db.execute(
select(Subscription.hanzo_subscription_id).where(
select(
Subscription.hanzo_subscription_id,
Subscription.hanzo_customer_id,
).where(
Subscription.project_id == project_id
)
)
row = result.first()
if not row or not row[0]:
if not row or not row[0] or not row[1]:
return {
"project_id": str(project_id),
"synced": False,
"error": "No Commerce subscription found",
}
return await self._sync_project(project_id, row[0])
return await self._sync_project(project_id, row[0], row[1])
async def run(self) -> None:
"""Run the sync loop."""
api/bootnode/core/billing/unified.py
+346
View File
@@ -0,0 +1,346 @@
"""Unified Hanzo IAM + Commerce integration.
Links IAM users (hanzo.id, zoo.id, lux.id, pars.id) to Commerce customers
for seamless authentication and billing across the Hanzo ecosystem.
Flow:
1. User authenticates via Hanzo IAM (JWT token)
2. On first billing access, auto-create/link Commerce customer
3. Store mapping: IAM user.id <-> Commerce customer.hanzo_id
4. All billing operations use the linked customer
"""
from datetime import datetime, UTC
from typing import Any
from uuid import UUID
import httpx
import structlog
from pydantic import BaseModel
from bootnode.config import get_settings
from bootnode.core.iam import IAMUser
logger = structlog.get_logger()
class UnifiedUser(BaseModel):
"""Unified user with IAM identity and Commerce customer."""
# IAM Identity
iam_id: str
email: str
name: str
org: str # hanzo, zoo, lux, pars
# Commerce Customer (created on first billing access)
commerce_customer_id: str | None = None
square_customer_id: str | None = None
# Billing Status
has_payment_method: bool = False
default_payment_method: str | None = None
@classmethod
def from_iam_user(cls, iam_user: IAMUser, commerce_data: dict | None = None) -> "UnifiedUser":
"""Create unified user from IAM user with optional Commerce data."""
user = cls(
iam_id=iam_user.id,
email=iam_user.email,
name=iam_user.name,
org=iam_user.org,
)
if commerce_data:
user.commerce_customer_id = commerce_data.get("id")
# Square customer ID is nested under accounts
accounts = commerce_data.get("accounts", {})
if isinstance(accounts, dict):
user.square_customer_id = accounts.get("square")
user.has_payment_method = commerce_data.get("has_payment_method", False)
user.default_payment_method = commerce_data.get("default_payment_method")
return user
class UnifiedBillingClient:
"""Unified client for IAM + Commerce integration.
Automatically links IAM users to Commerce customers and provides
a single interface for all billing operations.
"""
def __init__(self) -> None:
settings = get_settings()
self.iam_url = settings.iam_url
self.commerce_url = settings.commerce_url
self.commerce_api_key = settings.commerce_api_key
self._timeout = httpx.Timeout(30.0)
def _commerce_headers(self) -> dict[str, str]:
"""Get Commerce API headers."""
return {
"Authorization": f"Bearer {self.commerce_api_key}",
"Content-Type": "application/json",
"X-Source": "bootnode",
}
async def get_or_create_customer(self, iam_user: IAMUser) -> UnifiedUser:
"""Get or create Commerce customer for IAM user.
This is the main entry point for linking IAM users to Commerce.
Called automatically on first billing access.
"""
# First, try to find existing customer by hanzo_id
customer = await self._find_customer_by_hanzo_id(iam_user.id)
if customer:
logger.info(
"Found existing Commerce customer",
iam_id=iam_user.id,
customer_id=customer.get("id"),
)
return UnifiedUser.from_iam_user(iam_user, customer)
# No customer found, create one
customer = await self._create_customer(iam_user)
logger.info(
"Created Commerce customer for IAM user",
iam_id=iam_user.id,
customer_id=customer.get("id"),
org=iam_user.org,
)
return UnifiedUser.from_iam_user(iam_user, customer)
async def _find_customer_by_hanzo_id(self, hanzo_id: str) -> dict | None:
"""Find Commerce customer by Hanzo IAM ID."""
async with httpx.AsyncClient(timeout=self._timeout) as client:
try:
response = await client.get(
f"{self.commerce_url}/api/v1/user",
params={"hanzo_id": hanzo_id},
headers=self._commerce_headers(),
)
if response.status_code == 200:
data = response.json()
customers = data.get("data", [])
if customers:
return customers[0]
# Single-object response
if data.get("id"):
return data
return None
except httpx.RequestError as e:
logger.warning("Failed to lookup Commerce customer", error=str(e))
return None
async def _create_customer(self, iam_user: IAMUser) -> dict:
"""Create Commerce customer linked to IAM user."""
async with httpx.AsyncClient(timeout=self._timeout) as client:
response = await client.post(
f"{self.commerce_url}/api/v1/user",
json={
"email": iam_user.email,
"name": iam_user.name,
"first_name": iam_user.name.split()[0] if iam_user.name else "",
"last_name": " ".join(iam_user.name.split()[1:]) if iam_user.name else "",
"hanzo_id": iam_user.id, # Link to IAM
"org": iam_user.org,
"metadata": {
"iam_id": iam_user.id,
"iam_org": iam_user.org,
"source": "bootnode",
"created_via": "unified_billing",
},
},
headers=self._commerce_headers(),
)
if response.status_code >= 400:
error_data = response.json() if response.content else {}
logger.error(
"Failed to create Commerce customer",
status=response.status_code,
error=error_data,
)
# Return minimal customer data so billing can still work
return {
"id": f"local_{iam_user.id}",
"hanzo_id": iam_user.id,
"email": iam_user.email,
}
return response.json()
async def get_customer_subscriptions(self, iam_user: IAMUser) -> list[dict]:
"""Get all subscriptions for IAM user across Commerce."""
unified_user = await self.get_or_create_customer(iam_user)
if not unified_user.commerce_customer_id:
return []
async with httpx.AsyncClient(timeout=self._timeout) as client:
response = await client.get(
f"{self.commerce_url}/api/v1/user/{unified_user.commerce_customer_id}/orders",
headers=self._commerce_headers(),
)
if response.status_code == 200:
return response.json().get("data", [])
return []
async def get_customer_invoices(self, iam_user: IAMUser) -> list[dict]:
"""Get all invoices for IAM user."""
unified_user = await self.get_or_create_customer(iam_user)
if not unified_user.commerce_customer_id:
return []
async with httpx.AsyncClient(timeout=self._timeout) as client:
response = await client.get(
f"{self.commerce_url}/api/v1/user/{unified_user.commerce_customer_id}/orders",
params={"type": "invoice"},
headers=self._commerce_headers(),
)
if response.status_code == 200:
return response.json().get("data", [])
return []
async def get_customer_payment_methods(self, iam_user: IAMUser) -> list[dict]:
"""Get payment methods for IAM user (Square cards via Commerce)."""
unified_user = await self.get_or_create_customer(iam_user)
if not unified_user.commerce_customer_id:
return []
async with httpx.AsyncClient(timeout=self._timeout) as client:
response = await client.get(
f"{self.commerce_url}/api/v1/user/{unified_user.commerce_customer_id}/paymentmethods",
headers=self._commerce_headers(),
)
if response.status_code == 200:
return response.json().get("data", [])
return []
async def create_subscription(
self,
iam_user: IAMUser,
plan_id: str,
project_id: UUID,
) -> dict:
"""Create subscription for IAM user.
This creates a subscription in Commerce linked to the IAM user,
and stores the bootnode project_id in metadata for tracking.
"""
unified_user = await self.get_or_create_customer(iam_user)
async with httpx.AsyncClient(timeout=self._timeout) as client:
response = await client.post(
f"{self.commerce_url}/api/v1/subscribe",
json={
"customer_id": unified_user.commerce_customer_id,
"plan_id": plan_id,
"metadata": {
"bootnode_project_id": str(project_id),
"iam_id": iam_user.id,
"iam_org": iam_user.org,
},
},
headers=self._commerce_headers(),
)
if response.status_code >= 400:
error_data = response.json() if response.content else {}
raise Exception(f"Failed to create subscription: {error_data}")
return response.json()
async def report_usage(
self,
iam_user: IAMUser,
subscription_id: str,
compute_units: int,
timestamp: datetime,
) -> None:
"""Report metered usage for PAYG billing."""
async with httpx.AsyncClient(timeout=self._timeout) as client:
response = await client.post(
f"{self.commerce_url}/api/v1/subscribe/{subscription_id}/usage",
json={
"quantity": compute_units,
"unit": "compute_units",
"timestamp": timestamp.isoformat(),
"metadata": {
"iam_id": iam_user.id,
"source": "bootnode",
},
},
headers=self._commerce_headers(),
)
if response.status_code >= 400:
logger.error(
"Failed to report usage",
subscription_id=subscription_id,
status=response.status_code,
)
async def sync_iam_to_commerce(self, iam_user: IAMUser) -> dict:
"""Sync IAM user data to Commerce customer.
Updates Commerce customer with latest IAM data (name, email, etc).
Called periodically or on profile update.
"""
unified_user = await self.get_or_create_customer(iam_user)
if not unified_user.commerce_customer_id:
return {}
async with httpx.AsyncClient(timeout=self._timeout) as client:
response = await client.patch(
f"{self.commerce_url}/api/v1/user/{unified_user.commerce_customer_id}",
json={
"email": iam_user.email,
"name": iam_user.name,
"metadata": {
"iam_id": iam_user.id,
"iam_org": iam_user.org,
"iam_roles": iam_user.roles,
"last_sync": datetime.now(UTC).isoformat(),
},
},
headers=self._commerce_headers(),
)
if response.status_code == 200:
logger.info(
"Synced IAM user to Commerce",
iam_id=iam_user.id,
customer_id=unified_user.commerce_customer_id,
)
return response.json()
return {}
# Global instance
_unified_client: UnifiedBillingClient | None = None
def get_unified_billing_client() -> UnifiedBillingClient:
"""Get unified billing client singleton."""
global _unified_client
if _unified_client is None:
_unified_client = UnifiedBillingClient()
return _unified_client
api/bootnode/core/billing/webhooks.py
+148 -3
View File
@@ -1,9 +1,10 @@
"""Hanzo Commerce webhook handler.
Handles webhooks from Hanzo Commerce for:
Handles webhooks from Hanzo Commerce (Square-based) for:
- Order events (authorize, complete, cancel)
- Subscription lifecycle events
- Invoice events
- Payment events
- Payment events (paid, failed, refunded)
"""
import hashlib
@@ -46,7 +47,7 @@ class CommerceWebhookHandler:
def __init__(self) -> None:
settings = get_settings()
self.webhook_secret = settings.hanzo_commerce_webhook_secret
self.webhook_secret = settings.commerce_webhook_secret
def verify_signature(self, payload: bytes, signature: str) -> bool:
"""
@@ -92,12 +93,22 @@ class CommerceWebhookHandler:
)
handlers = {
# Order events
"order.completed": self.handle_order_completed,
"order.cancelled": self.handle_order_cancelled,
# Subscription lifecycle
"subscription.created": self.handle_subscription_created,
"subscription.updated": self.handle_subscription_updated,
"subscription.cancelled": self.handle_subscription_cancelled,
"subscription.reactivated": self.handle_subscription_reactivated,
# Invoice events
"invoice.paid": self.handle_invoice_paid,
"invoice.payment_failed": self.handle_invoice_failed,
# Payment events (forwarded through Commerce from Square)
"payment.paid": self.handle_payment_paid,
"payment.failed": self.handle_payment_failed,
"payment.refunded": self.handle_payment_refunded,
# Customer events
"customer.created": self.handle_customer_created,
"customer.updated": self.handle_customer_updated,
}
@@ -339,6 +350,91 @@ class CommerceWebhookHandler:
return {"subscription_id": subscription_id, "tier": tier.value}
# =========================================================================
# Order Handlers
# =========================================================================
async def handle_order_completed(self, data: dict[str, Any]) -> dict:
"""Handle order.completed — payment captured successfully.
May trigger subscription provisioning if order is for a plan upgrade.
"""
order_id = data.get("id")
customer_id = data.get("customer_id")
metadata = data.get("metadata", {})
plan_slug = metadata.get("plan_slug")
project_id_str = metadata.get("project_id")
logger.info(
"Order completed",
order_id=order_id,
customer_id=customer_id,
plan_slug=plan_slug,
)
# If order is for a plan subscription, provision it
if plan_slug and project_id_str:
try:
project_id = UUID(project_id_str)
except ValueError:
return {"order_id": order_id, "error": "invalid_project_id"}
tier = PLAN_TO_TIER.get(plan_slug, PricingTier.FREE)
limits = get_tier_limits(tier)
subscription_id = data.get("subscription_id")
async with async_session() as db:
result = await db.execute(
select(Subscription).where(Subscription.project_id == project_id)
)
subscription = result.scalar_one_or_none()
if subscription:
subscription.tier = tier.value
subscription.monthly_cu_limit = limits.monthly_cu
subscription.rate_limit_per_second = limits.rate_limit_per_second
subscription.max_apps = limits.max_apps
subscription.max_webhooks = limits.max_webhooks
if subscription_id:
subscription.hanzo_subscription_id = subscription_id
if customer_id:
subscription.hanzo_customer_id = customer_id
else:
subscription = Subscription(
project_id=project_id,
tier=tier.value,
hanzo_subscription_id=subscription_id,
hanzo_customer_id=customer_id,
monthly_cu_limit=limits.monthly_cu,
rate_limit_per_second=limits.rate_limit_per_second,
max_apps=limits.max_apps,
max_webhooks=limits.max_webhooks,
billing_cycle_start=datetime.now(UTC),
billing_cycle_end=datetime.fromisoformat(
data.get("current_period_end", datetime.now(UTC).isoformat())
),
)
db.add(subscription)
await db.commit()
await self._invalidate_cache(project_id)
return {"order_id": order_id, "plan_slug": plan_slug}
async def handle_order_cancelled(self, data: dict[str, Any]) -> dict:
"""Handle order.cancelled — payment authorization voided."""
order_id = data.get("id")
reason = data.get("reason", "unknown")
logger.info(
"Order cancelled",
order_id=order_id,
reason=reason,
)
return {"order_id": order_id, "reason": reason}
# =========================================================================
# Invoice Handlers
# =========================================================================
@@ -422,6 +518,55 @@ class CommerceWebhookHandler:
return {"customer_id": customer_id}
# =========================================================================
# Payment Handlers (forwarded from Square via Commerce)
# =========================================================================
async def handle_payment_paid(self, data: dict[str, Any]) -> dict:
"""Handle payment.paid — successful payment via Square."""
payment_id = data.get("id")
order_id = data.get("order_id")
amount = data.get("amount", 0)
logger.info(
"Payment received",
payment_id=payment_id,
order_id=order_id,
amount_cents=amount,
)
return {"payment_id": payment_id, "amount": amount}
async def handle_payment_failed(self, data: dict[str, Any]) -> dict:
"""Handle payment.failed — Square payment declined."""
payment_id = data.get("id")
order_id = data.get("order_id")
error_code = data.get("error_code", "unknown")
logger.warning(
"Payment failed",
payment_id=payment_id,
order_id=order_id,
error_code=error_code,
)
return {"payment_id": payment_id, "error_code": error_code}
async def handle_payment_refunded(self, data: dict[str, Any]) -> dict:
"""Handle payment.refunded — refund processed via Square."""
payment_id = data.get("id")
refund_id = data.get("refund_id")
amount = data.get("amount", 0)
logger.info(
"Payment refunded",
payment_id=payment_id,
refund_id=refund_id,
amount_cents=amount,
)
return {"payment_id": payment_id, "refund_id": refund_id, "amount": amount}
# =========================================================================
# Helpers
# =========================================================================
api/pyproject.toml
+9 -7
View File
@@ -47,6 +47,8 @@ dependencies = [
"kubernetes>=31.0.0",
"pydo>=0.8.0",
"docker>=7.0.0",
# ZAP Protocol (native Cap'n Proto RPC)
"pycapnp>=2.0.0",
]
@@ -60,12 +62,12 @@ dev = [
"mypy>=1.13.0",
"pre-commit>=4.0.0",
]
# Hanzo SDKs - optional, graceful degradation without them
# In K8s, the KMS operator injects secrets as env vars
hanzo = [
"hanzo-kms>=1.0.0",
"hanzo-iam>=1.0.0",
]
# Hanzo SDKs - not published to PyPI yet
# In K8s, the KMS operator injects secrets as env vars directly
# hanzo = [
# "hanzo-kms>=1.0.0",
# "hanzo-iam>=1.0.0",
# ]
[project.scripts]
bootnode = "bootnode.cli:main"
@@ -73,7 +75,7 @@ bootnode = "bootnode.cli:main"
[project.urls]
Homepage = "https://bootnode.dev"
Documentation = "https://docs.bootnode.dev"
Repository = "https://github.com/bootnode/bootnode"
Repository = "https://github.com/hanzoai/bootnode"
[build-system]
requires = ["hatchling"]
api/tests/test_billing.py
+519
View File
@@ -0,0 +1,519 @@
"""Billing module tests — compute units, tiers, Commerce client, and webhooks."""
from datetime import UTC, datetime
from unittest.mock import AsyncMock, MagicMock, patch
from uuid import uuid4
import pytest
from bootnode.core.billing.compute_units import (
COMPUTE_UNITS,
DEFAULT_COMPUTE_UNITS,
get_batch_compute_units,
get_compute_units,
)
from bootnode.core.billing.tiers import (
TIER_LIMITS,
PricingTier,
TierLimits,
calculate_monthly_cost,
get_tier_limits,
)
# =============================================================================
# Compute Units
# =============================================================================
class TestComputeUnits:
def test_free_methods_cost_1(self):
assert get_compute_units("eth_chainId") == 1
assert get_compute_units("eth_blockNumber") == 1
assert get_compute_units("net_version") == 1
def test_light_methods_cost_5(self):
assert get_compute_units("eth_getBalance") == 5
assert get_compute_units("eth_getTransactionByHash") == 5
def test_medium_methods_cost_10(self):
assert get_compute_units("eth_call") == 10
assert get_compute_units("eth_estimateGas") == 10
def test_heavy_methods_cost_25(self):
assert get_compute_units("eth_getLogs") == 25
def test_transaction_methods_cost_50(self):
assert get_compute_units("eth_sendRawTransaction") == 50
def test_debug_methods_cost_100(self):
assert get_compute_units("debug_traceTransaction") == 100
def test_unknown_method_returns_default(self):
assert get_compute_units("nonexistent_method") == DEFAULT_COMPUTE_UNITS
def test_batch_compute_units(self):
methods = ["eth_chainId", "eth_getBalance", "eth_call"]
assert get_batch_compute_units(methods) == 1 + 5 + 10
def test_batch_empty(self):
assert get_batch_compute_units([]) == 0
def test_solana_methods(self):
assert get_compute_units("getAccountInfo") == 5
assert get_compute_units("sendTransaction") == 50
assert get_compute_units("simulateTransaction") == 25
def test_erc4337_methods(self):
assert get_compute_units("eth_sendUserOperation") == 75
assert get_compute_units("eth_estimateUserOperationGas") == 25
def test_bootnode_enhanced_apis(self):
assert get_compute_units("tokens_getBalances") == 15
assert get_compute_units("nfts_getOwned") == 20
assert get_compute_units("webhooks_create") == 5
# =============================================================================
# Pricing Tiers
# =============================================================================
class TestPricingTiers:
def test_all_tiers_defined(self):
for tier in PricingTier:
assert tier in TIER_LIMITS
def test_free_tier_limits(self):
limits = get_tier_limits(PricingTier.FREE)
assert limits.monthly_cu == 30_000_000
assert limits.rate_limit_per_second == 25
assert limits.max_apps == 5
assert limits.max_webhooks == 5
assert limits.price_per_million_cu == 0
def test_payg_tier_limits(self):
limits = get_tier_limits(PricingTier.PAY_AS_YOU_GO)
assert limits.monthly_cu == 0 # unlimited
assert limits.rate_limit_per_second == 300
assert limits.max_apps == 30
assert limits.price_per_million_cu == 40 # $0.40
def test_growth_tier_limits(self):
limits = get_tier_limits(PricingTier.GROWTH)
assert limits.monthly_cu == 100_000_000
assert limits.rate_limit_per_second == 500
assert limits.price_per_million_cu == 35 # $0.35
def test_enterprise_tier_limits(self):
limits = get_tier_limits(PricingTier.ENTERPRISE)
assert limits.rate_limit_per_second == 1000
assert limits.max_apps == 0 # unlimited
assert limits.price_per_million_cu == 0 # custom
def test_tier_rate_limits_ascending(self):
"""Higher tiers should have higher rate limits."""
free = get_tier_limits(PricingTier.FREE)
payg = get_tier_limits(PricingTier.PAY_AS_YOU_GO)
growth = get_tier_limits(PricingTier.GROWTH)
enterprise = get_tier_limits(PricingTier.ENTERPRISE)
assert free.rate_limit_per_second < payg.rate_limit_per_second
assert payg.rate_limit_per_second < growth.rate_limit_per_second
assert growth.rate_limit_per_second < enterprise.rate_limit_per_second
# =============================================================================
# Cost Calculation
# =============================================================================
class TestCostCalculation:
def test_free_tier_always_zero(self):
assert calculate_monthly_cost(PricingTier.FREE, 0) == 0
assert calculate_monthly_cost(PricingTier.FREE, 1_000_000) == 0
def test_enterprise_tier_always_zero(self):
assert calculate_monthly_cost(PricingTier.ENTERPRISE, 0) == 0
assert calculate_monthly_cost(PricingTier.ENTERPRISE, 999_000_000) == 0
def test_payg_1_million_cu(self):
# PAYG: $0.40 per million CU, no included CU
cost = calculate_monthly_cost(PricingTier.PAY_AS_YOU_GO, 1_000_000)
assert cost == 40 # 40 cents
def test_payg_10_million_cu(self):
cost = calculate_monthly_cost(PricingTier.PAY_AS_YOU_GO, 10_000_000)
assert cost == 400 # $4.00
def test_payg_zero_usage(self):
cost = calculate_monthly_cost(PricingTier.PAY_AS_YOU_GO, 0)
assert cost == 0
def test_growth_within_included(self):
# Growth includes 100M CU, no overage
cost = calculate_monthly_cost(PricingTier.GROWTH, 50_000_000)
assert cost == 0 # base fee not counted here (Commerce handles it)
def test_growth_at_included(self):
cost = calculate_monthly_cost(PricingTier.GROWTH, 100_000_000)
assert cost == 0
def test_growth_with_overage(self):
# 10M overage at $0.35 per million
cost = calculate_monthly_cost(PricingTier.GROWTH, 110_000_000)
assert cost == 350 # $3.50
# =============================================================================
# Plan Slug Mappings
# =============================================================================
class TestPlanSlugs:
def test_plan_to_tier_mapping(self):
"""Verify the plan-to-tier mapping in webhooks module."""
from bootnode.core.billing.webhooks import PLAN_TO_TIER
assert PLAN_TO_TIER["bootnode-free"] == PricingTier.FREE
assert PLAN_TO_TIER["bootnode-payg"] == PricingTier.PAY_AS_YOU_GO
assert PLAN_TO_TIER["bootnode-growth"] == PricingTier.GROWTH
assert PLAN_TO_TIER["bootnode-enterprise"] == PricingTier.ENTERPRISE
def test_api_tier_to_plan_slug(self):
"""Verify the tier-to-plan slug mapping in API module."""
from bootnode.api.billing import TIER_TO_PLAN_SLUG
assert TIER_TO_PLAN_SLUG[PricingTier.FREE] == "bootnode-free"
assert TIER_TO_PLAN_SLUG[PricingTier.PAY_AS_YOU_GO] == "bootnode-payg"
assert TIER_TO_PLAN_SLUG[PricingTier.GROWTH] == "bootnode-growth"
assert TIER_TO_PLAN_SLUG[PricingTier.ENTERPRISE] == "bootnode-enterprise"
# =============================================================================
# Commerce Client
# =============================================================================
class TestCommerceClient:
@patch("bootnode.core.billing.commerce.get_settings")
def test_client_uses_commerce_config(self, mock_settings):
mock_settings.return_value = MagicMock(
commerce_url="http://test:8001",
commerce_api_key="test-key",
)
from bootnode.core.billing.commerce import HanzoCommerceClient
client = HanzoCommerceClient()
assert client.base_url == "http://test:8001"
assert client.api_key == "test-key"
@patch("bootnode.core.billing.commerce.get_settings")
def test_client_headers(self, mock_settings):
mock_settings.return_value = MagicMock(
commerce_url="http://test:8001",
commerce_api_key="test-key",
)
from bootnode.core.billing.commerce import HanzoCommerceClient
client = HanzoCommerceClient()
headers = client._headers()
assert headers["Authorization"] == "Bearer test-key"
assert headers["X-Source"] == "bootnode"
assert headers["Content-Type"] == "application/json"
@patch("bootnode.core.billing.commerce.get_settings")
@pytest.mark.asyncio
async def test_create_customer(self, mock_settings):
mock_settings.return_value = MagicMock(
commerce_url="http://test:8001",
commerce_api_key="test-key",
)
from bootnode.core.billing.commerce import HanzoCommerceClient
client = HanzoCommerceClient()
# Mock the _request method
client._request = AsyncMock(return_value={"id": "cust_123"})
customer_id = await client.create_customer(
user_id="user_1", email="[email protected]", name="Test User"
)
assert customer_id == "cust_123"
client._request.assert_called_once()
@patch("bootnode.core.billing.commerce.get_settings")
@pytest.mark.asyncio
async def test_create_checkout(self, mock_settings):
mock_settings.return_value = MagicMock(
commerce_url="http://test:8001",
commerce_api_key="test-key",
)
from bootnode.core.billing.commerce import HanzoCommerceClient
client = HanzoCommerceClient()
client._request = AsyncMock(return_value={
"order_id": "order_123",
"checkout_url": "https://commerce.hanzo.ai/checkout/order_123",
"authorization_id": "auth_456",
})
result = await client.create_checkout(
customer_id="cust_123",
plan_slug="bootnode-payg",
project_id=uuid4(),
success_url="https://example.com/success",
)
assert result["order_id"] == "order_123"
assert result["checkout_url"] is not None
client._request.assert_called_once_with(
"POST",
"/api/v1/checkout/authorize",
json=pytest.approx(client._request.call_args[1]["json"], abs=0) if False else client._request.call_args[1].get("json", client._request.call_args[0][2] if len(client._request.call_args[0]) > 2 else None),
)
@patch("bootnode.core.billing.commerce.get_settings")
@pytest.mark.asyncio
async def test_capture_checkout(self, mock_settings):
mock_settings.return_value = MagicMock(
commerce_url="http://test:8001",
commerce_api_key="test-key",
)
from bootnode.core.billing.commerce import HanzoCommerceClient
client = HanzoCommerceClient()
client._request = AsyncMock(return_value={"order_id": "order_123", "status": "completed"})
result = await client.capture_checkout("order_123")
assert result["status"] == "completed"
client._request.assert_called_once_with("POST", "/api/v1/checkout/capture/order_123")
@patch("bootnode.core.billing.commerce.get_settings")
@pytest.mark.asyncio
async def test_report_usage(self, mock_settings):
mock_settings.return_value = MagicMock(
commerce_url="http://test:8001",
commerce_api_key="test-key",
)
from bootnode.core.billing.commerce import HanzoCommerceClient
client = HanzoCommerceClient()
client._request = AsyncMock(return_value={})
now = datetime.now(UTC)
await client.report_usage(
subscription_id="sub_123",
compute_units=5_000_000,
timestamp=now,
idempotency_key="test-key-123",
)
client._request.assert_called_once()
call_args = client._request.call_args
assert call_args[0][0] == "POST"
assert "/usage" in call_args[0][1]
@patch("bootnode.core.billing.commerce.get_settings")
@pytest.mark.asyncio
async def test_get_user_orders(self, mock_settings):
mock_settings.return_value = MagicMock(
commerce_url="http://test:8001",
commerce_api_key="test-key",
)
from bootnode.core.billing.commerce import HanzoCommerceClient
client = HanzoCommerceClient()
client._request = AsyncMock(return_value={
"data": [
{"id": "order_1", "status": "completed"},
{"id": "order_2", "status": "pending"},
]
})
orders = await client.get_user_orders("cust_123")
assert len(orders) == 2
assert orders[0]["id"] == "order_1"
# =============================================================================
# Commerce Webhooks
# =============================================================================
class TestCommerceWebhooks:
@patch("bootnode.core.billing.webhooks.get_settings")
def test_no_webhook_secret_allows_through(self, mock_settings):
mock_settings.return_value = MagicMock(commerce_webhook_secret="")
from bootnode.core.billing.webhooks import CommerceWebhookHandler
handler = CommerceWebhookHandler()
# No secret = verification skipped (returns True)
assert handler.verify_signature(b"payload", "any-sig") is True
@patch("bootnode.core.billing.webhooks.get_settings")
def test_valid_signature(self, mock_settings):
import hashlib
import hmac
secret = "test-secret"
mock_settings.return_value = MagicMock(commerce_webhook_secret=secret)
from bootnode.core.billing.webhooks import CommerceWebhookHandler
handler = CommerceWebhookHandler()
payload = b'{"type":"test"}'
expected_sig = hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest()
assert handler.verify_signature(payload, expected_sig) is True
@patch("bootnode.core.billing.webhooks.get_settings")
def test_invalid_signature(self, mock_settings):
mock_settings.return_value = MagicMock(commerce_webhook_secret="test-secret")
from bootnode.core.billing.webhooks import CommerceWebhookHandler
handler = CommerceWebhookHandler()
assert handler.verify_signature(b'{"type":"test"}', "bad-signature") is False
@patch("bootnode.core.billing.webhooks.get_settings")
@pytest.mark.asyncio
async def test_unhandled_event_type(self, mock_settings):
mock_settings.return_value = MagicMock(commerce_webhook_secret="test")
from bootnode.core.billing.webhooks import CommerceWebhookHandler
handler = CommerceWebhookHandler()
result = await handler.handle_event({
"type": "some.unknown.event",
"id": "evt_123",
"data": {},
})
assert result["handled"] is False
assert result["reason"] == "unknown_type"
@patch("bootnode.core.billing.webhooks.get_settings")
@pytest.mark.asyncio
async def test_handles_order_completed(self, mock_settings):
mock_settings.return_value = MagicMock(commerce_webhook_secret="test")
from bootnode.core.billing.webhooks import CommerceWebhookHandler
handler = CommerceWebhookHandler()
result = await handler.handle_event({
"type": "order.completed",
"id": "evt_456",
"data": {
"id": "order_123",
"customer_id": "cust_456",
"metadata": {},
},
})
assert result["handled"] is True
assert result["result"]["order_id"] == "order_123"
@patch("bootnode.core.billing.webhooks.get_settings")
@pytest.mark.asyncio
async def test_handles_order_cancelled(self, mock_settings):
mock_settings.return_value = MagicMock(commerce_webhook_secret="test")
from bootnode.core.billing.webhooks import CommerceWebhookHandler
handler = CommerceWebhookHandler()
result = await handler.handle_event({
"type": "order.cancelled",
"id": "evt_789",
"data": {
"id": "order_123",
"reason": "customer_request",
},
})
assert result["handled"] is True
assert result["result"]["reason"] == "customer_request"
@patch("bootnode.core.billing.webhooks.get_settings")
@pytest.mark.asyncio
async def test_handles_payment_paid(self, mock_settings):
mock_settings.return_value = MagicMock(commerce_webhook_secret="test")
from bootnode.core.billing.webhooks import CommerceWebhookHandler
handler = CommerceWebhookHandler()
result = await handler.handle_event({
"type": "payment.paid",
"id": "evt_pay1",
"data": {
"id": "pay_123",
"order_id": "order_456",
"amount": 4900,
},
})
assert result["handled"] is True
assert result["result"]["amount"] == 4900
@patch("bootnode.core.billing.webhooks.get_settings")
@pytest.mark.asyncio
async def test_handles_payment_refunded(self, mock_settings):
mock_settings.return_value = MagicMock(commerce_webhook_secret="test")
from bootnode.core.billing.webhooks import CommerceWebhookHandler
handler = CommerceWebhookHandler()
result = await handler.handle_event({
"type": "payment.refunded",
"id": "evt_ref1",
"data": {
"id": "pay_123",
"refund_id": "ref_456",
"amount": 2500,
},
})
assert result["handled"] is True
assert result["result"]["refund_id"] == "ref_456"
@patch("bootnode.core.billing.webhooks.get_settings")
@pytest.mark.asyncio
async def test_handles_invoice_paid(self, mock_settings):
mock_settings.return_value = MagicMock(commerce_webhook_secret="test")
from bootnode.core.billing.webhooks import CommerceWebhookHandler
handler = CommerceWebhookHandler()
result = await handler.handle_event({
"type": "invoice.paid",
"id": "evt_inv1",
"data": {
"id": "inv_123",
"customer_id": "cust_456",
"amount_paid": 4900,
},
})
assert result["handled"] is True
assert result["result"]["amount_paid"] == 4900
# =============================================================================
# Unified Billing Client
# =============================================================================
class TestUnifiedUser:
def _make_iam_user(self, **kwargs):
"""Create a fake IAM user. Can't use MagicMock because 'name' is reserved."""
from types import SimpleNamespace
defaults = {"id": "user_1", "email": "[email protected]", "name": "Test", "org": "hanzo"}
defaults.update(kwargs)
return SimpleNamespace(**defaults)
def test_from_iam_user_no_commerce(self):
from bootnode.core.billing.unified import UnifiedUser
iam_user = self._make_iam_user()
user = UnifiedUser.from_iam_user(iam_user)
assert user.iam_id == "user_1"
assert user.commerce_customer_id is None
assert user.square_customer_id is None
def test_from_iam_user_with_commerce(self):
from bootnode.core.billing.unified import UnifiedUser
iam_user = self._make_iam_user()
commerce_data = {
"id": "cust_123",
"accounts": {"square": "sq_cust_456"},
"has_payment_method": True,
"default_payment_method": "card_789",
}
user = UnifiedUser.from_iam_user(iam_user, commerce_data)
assert user.commerce_customer_id == "cust_123"
assert user.square_customer_id == "sq_cust_456"
assert user.has_payment_method is True