Compare commits

..

17 Commits

Author SHA1 Message Date
Jon Staab 7134915665 Refactor infra 2026-05-15 14:30:26 -07:00
Jon Staab cfa52d739f Clean up relay validation 2026-05-15 13:15:57 -07:00
Jon Staab 6abe62b569 remove invoice auto collection on nwc_url update 2026-05-15 12:54:40 -07:00
Jon Staab cd7b84439e define defaults on the model, simplify create relay payload 2026-05-15 11:25:25 -07:00
Jon Staab 1c3e0d619a Refactor error handling 2026-05-15 11:07:27 -07:00
Jon Staab 5590b14074 Refactor api into different route files 2026-05-15 09:28:12 -07:00
Jon Staab 26f05e8b8f Add env struct 2026-05-14 15:33:28 -07:00
Jon Staab 066c91a4d1 Refactor bitcoin exchange rate fetching and wallet 2026-05-14 12:47:32 -07:00
userAdityaa 3ed021214a feat(infra): pass Blossom S3 config to Zooid with schema key prefix (#69)
Reviewed-on: coracle/caravel#69
Co-authored-by: userAdityaa <aditya.chaudhary1558@gmail.com>
Co-committed-by: userAdityaa <aditya.chaudhary1558@gmail.com>
2026-05-13 15:47:08 +00:00
Jon Staab c0aff5f7cf Refactor billing module 2026-05-12 16:32:05 -07:00
Jon Staab c9c1dd2c4c Group subscription items by price 2026-05-12 15:53:17 -07:00
Jon Staab 679a56edc3 Add docker publish workflow 2026-05-12 14:48:50 -07:00
userAdityaa e7efd9d08b fix: stripe portal dead-end with callback return flow (#67)
Co-authored-by: userAdityaa <aditya.chaudhary1558@gmail.com>
Co-committed-by: userAdityaa <aditya.chaudhary1558@gmail.com>
2026-05-11 20:48:55 +00:00
userAdityaa 0151762362 chore: improve billing customer name using Nostr kind 0 with pubkey fallback (#66)
Co-authored-by: userAdityaa <aditya.chaudhary1558@gmail.com>
Co-committed-by: userAdityaa <aditya.chaudhary1558@gmail.com>
2026-05-08 22:52:13 +00:00
userAdityaa a79c43e17e feat: open payment modal immediately on relay plan upgrade (#64)
Co-authored-by: userAdityaa <aditya.chaudhary1558@gmail.com>
Co-committed-by: userAdityaa <aditya.chaudhary1558@gmail.com>
2026-05-07 18:35:24 +00:00
Jon Staab dbe25c372f Conflate id and schema 2026-05-05 17:47:13 -07:00
userAdityaa 80a86452d0 chore: encrypt tenant NWC URL at rest and stop secret exposure in tenant APIs (#58)
Co-authored-by: userAdityaa <aditya.chaudhary1558@gmail.com>
Co-committed-by: userAdityaa <aditya.chaudhary1558@gmail.com>
2026-05-05 20:42:12 +00:00
45 changed files with 2492 additions and 2574 deletions
+59
View File
@@ -0,0 +1,59 @@
name: Docker
on:
push:
branches: [master]
env:
REGISTRY: gitea.coracle.social
jobs:
build-and-push-image:
runs-on: ubuntu-latest
permissions:
contents: read
packages: write
strategy:
matrix:
include:
- component: frontend
image: coracle/caravel-frontend
context: frontend
- component: backend
image: coracle/caravel-backend
context: backend
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Log in to the Container registry
uses: docker/login-action@v3
with:
registry: ${{ env.REGISTRY }}
username: hodlbod
password: ${{ secrets.PACKAGE_TOKEN }}
- name: Extract metadata (tags, labels) for Docker
id: meta
uses: docker/metadata-action@v5
with:
images: ${{ env.REGISTRY }}/${{ matrix.image }}
tags: |
type=raw,value=latest,enable=${{ github.ref == 'refs/heads/master' }}
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
with:
driver: docker-container
- name: Build and push Docker image
id: push
uses: docker/build-push-action@v5
with:
context: ${{ matrix.context }}
push: true
platforms: linux/amd64,linux/arm64
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
+1
View File
@@ -1,4 +1,5 @@
ref
todo.md
node_modules
target
data
+1 -1
View File
@@ -26,7 +26,7 @@ docker run -it \
-v ./config:/app/config \
-v ./media:/app/media \
-v ./data:/app/data \
ghcr.io/coracle-social/zooid
gitea.coracle.social/coracle/zooid
```
### 2. Configure the backend
+17 -12
View File
@@ -1,10 +1,8 @@
# Server
HOST=127.0.0.1
PORT=2892
ALLOW_ORIGINS= # Optional comma-separated allowed CORS origins; empty = permissive
# Auth
ADMINS= # Comma-separated hex pubkeys with admin access
SERVER_HOST=127.0.0.1
SERVER_PORT=2892
SERVER_ALLOW_ORIGINS= # Optional comma-separated allowed CORS origins; empty = permissive
SERVER_ADMIN_PUBKEYS= # Comma-separated hex pubkeys with admin access
# Database
DATABASE_URL=sqlite://data/caravel.db
@@ -14,20 +12,27 @@ ROBOT_SECRET= # Nostr private key (hex)
ROBOT_NAME=
ROBOT_DESCRIPTION=
ROBOT_PICTURE=
ROBOT_OUTBOX_RELAYS=relay.damus.io,relay.primal.net,nos.lol
ROBOT_INDEXER_RELAYS=purplepag.es,relay.damus.io,indexer.coracle.social
ROBOT_MESSAGING_RELAYS=auth.nostr1.com,relay.keychat.io,relay.ditto.pub
ROBOT_WALLET= # Nostr Wallet Connect URL for generating Lightning invoices
ROBOT_OUTBOX_RELAYS=wss://relay.damus.io,wss://relay.primal.net,wss://nos.lol
ROBOT_INDEXER_RELAYS=wss://purplepag.es,wss://relay.damus.io,wss://indexer.coracle.social
ROBOT_MESSAGING_RELAYS=wss://auth.nostr1.com,wss://relay.keychat.io,wss://relay.ditto.pub
# Zooid
ZOOID_API_URL=http://127.0.0.1:3334
ZOOID_API_SECRET=
RELAY_DOMAIN=spaces.coracle.social
LIVEKIT_URL=
LIVEKIT_API_KEY=
LIVEKIT_API_SECRET=
# Blossom S3 (optional; when region, bucket, access key, and secret are all set, relays with blossom enabled sync with adapter s3 and key_prefix = relay schema)
BLOSSOM_S3_ENDPOINT=
BLOSSOM_S3_REGION=
BLOSSOM_S3_BUCKET=
BLOSSOM_S3_ACCESS_KEY=
BLOSSOM_S3_SECRET_KEY=
# Billing
NWC_URL= # Nostr Wallet Connect URL for generating Lightning invoices
ENCRYPTION_SECRET= # Nostr secret key (hex or nsec) used to encrypt tenant NWC URLs at rest
STRIPE_SECRET_KEY= # Required Stripe API secret key (sk_...)
STRIPE_WEBHOOK_SECRET=whsec_test_00000000000000000000000000 # Webhook signing secret (use real value in production)
STRIPE_PRICE_BASIC= # Stripe price ID (price_...) for the Basic plan; required for paid plans
STRIPE_PRICE_GROWTH= # Stripe price ID (price_...) for the Growth plan; required for paid plans
+13
View File
@@ -210,6 +210,7 @@ dependencies = [
"nostr-sdk",
"nwc",
"rand 0.8.5",
"regex",
"reqwest",
"serde",
"serde_json",
@@ -1894,6 +1895,18 @@ dependencies = [
"bitflags",
]
[[package]]
name = "regex"
version = "1.12.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e10754a14b9137dd7b1e3e5b0493cc9171fdd105e0ab477f51b72e7f3ac0e276"
dependencies = [
"aho-corasick",
"memchr",
"regex-automata",
"regex-syntax",
]
[[package]]
name = "regex-automata"
version = "0.4.14"
+1
View File
@@ -24,6 +24,7 @@ hmac = "0.12"
sha2 = "0.10"
dotenvy = "0.15.7"
base64 = "0.22"
regex = "1"
[dev-dependencies]
tower = { version = "0.5", features = ["util"] }
+7 -1
View File
@@ -38,15 +38,21 @@ Environment variables:
| `ADMINS` | Comma-separated admin pubkeys (hex) | _optional_ |
| `ALLOW_ORIGINS` | Comma-separated CORS origins. If empty, CORS is permissive. | _optional_ |
| `ZOOID_API_URL` | Zooid API base URL used by infra worker | _required for infra sync_ |
| `ZOOID_API_SECRET` | Nostr secret key used for authentication of requests to the zooid API | _required_ |
| `RELAY_DOMAIN` | Base domain appended to relay subdomains | empty |
| `LIVEKIT_URL` | LiveKit URL sent to zooid when relay livekit is enabled | _optional_ |
| `LIVEKIT_API_KEY` | LiveKit API key sent to zooid | _optional_ |
| `LIVEKIT_API_SECRET` | LiveKit API secret sent to zooid | _optional_ |
| `BLOSSOM_S3_ENDPOINT` | S3-compatible endpoint URL for Blossom; omit for AWS S3 | _optional_ |
| `BLOSSOM_S3_REGION` | S3 region; with bucket, access key, and secret enables S3 for Blossom | _optional_ |
| `BLOSSOM_S3_BUCKET` | S3 bucket name | _optional_ |
| `BLOSSOM_S3_ACCESS_KEY` | S3 access key ID | _optional_ |
| `BLOSSOM_S3_SECRET_KEY` | S3 secret access key | _optional_ |
| `NWC_URL` | Platform NWC URL used to generate BOLT11 invoices | _required for invoice generation_ |
| `ENCRYPTION_SECRET` | Nostr secret key (hex or nsec) used to encrypt tenant NWC URLs at rest | _required_ |
| `STRIPE_SECRET_KEY` | Stripe API secret key used for billing API operations | _required_ |
| `STRIPE_WEBHOOK_SECRET` | Stripe webhook signing secret used to verify `Stripe-Signature` headers | _required_ |
| `STRIPE_PRICE_BASIC` | Stripe price ID (`price_...`) backing the Basic plan | _required for paid plans_ |
| `STRIPE_PRICE_GROWTH` | Stripe price ID (`price_...`) backing the Growth plan | _required for paid plans_ |
| `ROBOT_SECRET` | Robot Nostr secret key | _required_ |
| `ROBOT_NAME` | Robot display name (kind `0`) | _optional_ |
| `ROBOT_DESCRIPTION` | Robot description (kind `0`) | _optional_ |
+56 -41
View File
@@ -2,45 +2,47 @@
Billing encapsulates logic related to synchronizing state with Stripe, processing payments via NWC, and managing subscription lifecycle.
It owns the domain logic only: Stripe REST calls go through `Stripe` (see `spec/stripe.md`), NWC wallet operations through `Wallet` (see `spec/wallet.md`), and fiat → msats conversion through `bitcoin` (see `spec/bitcoin.md`).
Members:
- `nwc_url: String` - a nostr wallet connect URL used to **create** bolt11 invoices (i.e. receive payments), from `NWC_URL`
- `stripe_secret_key: String` - Stripe API key used for billing API operations, from `STRIPE_SECRET_KEY`
- `stripe_webhook_secret: String` - secret for verifying Stripe webhook signatures, from `STRIPE_WEBHOOK_SECRET`
- `stripe: Stripe` - Stripe REST wrapper (see `spec/stripe.md`)
- `wallet: Wallet` - the system NWC wallet, used to issue and look up bolt11 invoices (see `spec/wallet.md`)
- `query: Query`
- `command: Command`
- `robot: Robot`
## `pub fn new(query: Query, command: Command, robot: Robot) -> Self`
- Reads environment and populates members
- Panics if `STRIPE_SECRET_KEY` is missing/empty
- Panics if `STRIPE_WEBHOOK_SECRET` is missing/empty
- Builds `stripe` via `Stripe::from_env()` and `wallet` from `NWC_URL`
- Panics if `STRIPE_SECRET_KEY`, `STRIPE_WEBHOOK_SECRET`, or `NWC_URL` is missing or malformed
## `pub fn start(&self)`
- Subscribes to `command.notify.subscribe()`
- On `create_relay`, `update_relay`, `activate_relay`, `deactivate_relay`, `fail_relay_sync`, and `complete_relay_sync`, call `self.sync_relay_subscription`.
- On `create_relay`, `update_relay`, `activate_relay`, `deactivate_relay`, `fail_relay_sync`, and `complete_relay_sync`: resolve the relay named by the activity (skip if it no longer exists) and reconcile its tenant via `sync_tenant_subscription`.
- The startup/lagged reconcile loop calls `sync_tenant_subscription` for every tenant.
## `pub fn sync_relay_subscription(&self, activity: &Activity)`
## `fn sync_tenant_subscription(&self, tenant_pubkey: &str)`
Manages the Stripe subscription and subscription items for a relay's tenant. Only paid (non-free) relays interact with Stripe. Free-only tenants have no subscription. Must be idempotent.
Reconciles a tenant's single Stripe subscription with the set of relays that should be billed. Only paid (non-free) relays interact with Stripe. Free-only tenants have no subscription. Must be idempotent.
Stripe forbids two subscription items on the same subscription from sharing a price, so billing is modeled as **one subscription item per plan (price), with `quantity` equal to the number of the tenant's `active` relays on that plan**. Every such relay's `stripe_subscription_item_id` points at the shared item for its plan; relays that aren't billed (free, inactive, delinquent) have it cleared.
Stripe uses **pay-in-advance** by default: when a subscription is first created, Stripe immediately generates an open invoice for the current period. The `invoice.created` webhook fires shortly after and `handle_invoice_created` attempts payment.
- Fetch the relay and tenant associated with the `activity`
- **If relay plan is `free`**: if the relay has a `stripe_subscription_item_id`, delete it via the Stripe API and call `command.delete_relay_subscription_item`. Then run downgrade proration validation by previewing the upcoming invoice and logging proration lines/amounts. Then check cleanup (below). Return early.
- **If relay is `inactive` or `delinquent`**: if the relay has a `stripe_subscription_item_id`, delete it via the Stripe API and call `command.delete_relay_subscription_item`. Then check cleanup (below). Return early.
- **If relay is `active` and on a paid plan**:
- **Ensure subscription exists**: If the tenant has no `stripe_subscription_id`, create a Stripe subscription for the customer with `collection_method: "charge_automatically"` and the relay's price as the first item. Save the subscription ID via `command.set_tenant_subscription` and the item ID via `command.set_relay_subscription_item`. Return early.
- **Sync the subscription item**: If the tenant already has a subscription, create or update the relay's Stripe subscription item to the plan's `stripe_price_id` via the Stripe API, then call `command.set_relay_subscription_item`.
- **Downgrade validation**: when changing an existing subscription item, detect if the new plan amount is lower than the current one. If yes, preview the upcoming Stripe invoice and log proration line/amount details to validate expected credit/proration behavior.
- **Clean up empty subscription**: After any item deletion, check if the tenant has any remaining active paid relays. If none and the tenant has a `stripe_subscription_id`, cancel the Stripe subscription immediately and call `command.clear_tenant_subscription`.
- Fetch the tenant and its relays. Build the desired state: for each `active` relay on a paid plan with a non-empty `stripe_price_id`, count relays per price.
- **Resolve the live subscription**: if the tenant has a `stripe_subscription_id`, fetch it. If Stripe no longer knows about it, or its status is `canceled`/`incomplete_expired`, call `command.clear_tenant_subscription` and treat the tenant as having no subscription.
- **No relays to bill** (desired state empty): if the tenant still has a `stripe_subscription_id`, cancel the Stripe subscription and call `command.clear_tenant_subscription`. Clear `stripe_subscription_item_id` on every relay that has one. Return.
- **No subscription yet**: create a Stripe subscription for the customer with `collection_method: "charge_automatically"` and one item per `(price, quantity)`. Save the subscription ID via `command.set_tenant_subscription`.
- **Existing subscription**: fetch its current items. For each desired `(price, quantity)`: update the matching item's quantity if it differs, otherwise create the item. Delete any item whose price no longer appears in the desired state.
- **Point relays at items**: for each relay, set `stripe_subscription_item_id` (via `command.set_relay_subscription_item`) to the shared item for its plan, or clear it (via `command.delete_relay_subscription_item`) if the relay is not billed.
- **Downgrade validation**: if any quantity decreased or any item was removed, preview the upcoming Stripe invoice and log proration line/amount details to validate expected credit/proration behavior.
## `pub fn handle_webhook(&self, payload: &str, signature: &str) -> Result<()>`
- Verify the webhook signature using `self.stripe_webhook_secret`
- Parse the event and dispatch by type:
- Verify and parse the event via `self.stripe.construct_event(payload, signature)` (checks the `Stripe-Signature` HMAC and timestamp tolerance — see `spec/stripe.md`)
- Dispatch by type:
- `invoice.created` -> `self.handle_invoice_created`
- `invoice.paid` -> `self.handle_invoice_paid`
- `invoice.payment_failed` -> `self.handle_invoice_payment_failed`
@@ -50,56 +52,69 @@ Stripe uses **pay-in-advance** by default: when a subscription is first created,
- `payment_method.attached` -> `self.handle_payment_method_attached`
- Unknown event types are ignored (return Ok)
## `pub async fn stripe_create_customer(&self, tenant_pubkey: &str) -> Result<String>`
- Resolves a display name via `robot.fetch_nostr_name(tenant_pubkey)`, falling back to the first 8 chars of the pubkey
- Creates the Stripe customer via `stripe.create_customer(display_name, tenant_pubkey)` and returns its id
## `pub async fn stripe_list_invoices(&self, customer_id: &str) -> Result<Value>`
- Fetches invoices from Stripe API for the given customer
- Returns the `data` array from the Stripe response
- Delegates to `stripe.list_invoices` — returns the `data` array of the customer's invoices
## `pub async fn stripe_get_invoice(&self, invoice_id: &str) -> Result<Value>`
## `pub async fn stripe_create_portal_session(&self, customer_id: &str, return_url: Option<&str>) -> Result<String>`
- Fetches a single invoice from Stripe API by ID
- Returns the full Stripe invoice object
- Delegates to `stripe.create_portal_session` — returns the Customer Portal session URL
## `pub async fn create_bolt11(&self, amount_due_cents: i64) -> Result<String>`
## `pub async fn get_invoice_with_tenant(&self, invoice_id: &str) -> Result<(Value, Tenant), InvoiceLookupError>`
- Creates a bolt11 Lightning invoice for the given amount using the system NWC wallet (`self.nwc_url`)
- Fetches the invoice via `stripe.get_invoice` (a Stripe 4xx surfaces as `InvoiceLookupError::StripeClient`)
- Looks up the tenant by the invoice's `customer` field; errors if the invoice has no customer or no tenant matches
## `pub async fn reconcile_manual_lightning_invoice(&self, invoice_id: &str, invoice: &Value) -> Result<Value, InvoiceLookupError>`
If `invoice.status == "open"` and a manual-Lightning bolt11 was previously issued for it (`query.get_invoice_manual_lightning_bolt11`), check whether that bolt11 has settled (`self.wallet.is_settled(...)`). If it has, mark the Stripe invoice paid out of band (`stripe.pay_invoice_out_of_band`) and return the refreshed invoice. On any lookup/settlement failure, log and return the invoice unchanged.
## `pub async fn get_or_create_manual_lightning_bolt11(&self, invoice_id: &str, tenant_pubkey: &str, amount_due_minor: i64, currency: &str) -> Result<String>`
- Returns the existing bolt11 if one is already recorded for the invoice
- Otherwise creates one via `create_bolt11`, records it with `command.insert_manual_lightning_invoice_payment`, and returns it (re-reading the stored row if the insert lost a race)
## `pub async fn create_bolt11(&self, amount_due_minor: i64, currency: &str) -> Result<String>`
- Converts the fiat amount to msats via `bitcoin::fiat_to_msats` (fetches the live BTC spot price — see `spec/bitcoin.md`)
- Issues a bolt11 invoice for that amount on the system NWC wallet (`self.wallet.make_invoice(...)`)
- Returns the bolt11 invoice string
## `pub async fn stripe_create_portal_session(&self, customer_id: &str) -> Result<String>`
- Creates a Stripe Customer Portal session for the given customer
- Returns the portal session URL
## `pub async fn pay_outstanding_nwc_invoices(&self, tenant: &Tenant) -> Result<()>`
Called when a tenant first sets their NWC URL (via `PUT /tenants/:pubkey`). Attempts to pay any currently open invoices for the tenant using their NWC wallet, so that invoices created before NWC was configured are not left unpaid.
- If `tenant.nwc_url` is empty, return early.
- List all Stripe invoices for `tenant.stripe_customer_id` via `stripe_list_invoices`.
- List all Stripe invoices for `tenant.stripe_customer_id` via `stripe.list_invoices`.
- For each invoice with `status == "open"` and `amount_due > 0`:
- Attempt NWC payment via `nwc_pay_invoice`.
- On success: call `stripe_pay_invoice_out_of_band` and `command.clear_tenant_nwc_error`.
- On success: mark the Stripe invoice paid out of band (`stripe.pay_invoice_out_of_band`) and call `command.clear_tenant_nwc_error`.
- On failure: call `command.set_tenant_nwc_error` and log the error; continue to the next invoice.
## `pub async fn pay_outstanding_card_invoices(&self, tenant: &Tenant) -> Result<()>`
Attempts Stripe-side collection for open invoices when the tenant has a card on file.
- If tenant has no card payment method, return early.
- List all Stripe invoices for `tenant.stripe_customer_id`.
- If tenant has no card payment method (`stripe.has_payment_method`), return early.
- List all Stripe invoices for `tenant.stripe_customer_id` via `stripe.list_invoices`.
- For each invoice with `status == "open"` and `amount_due > 0`:
- Call Stripe `POST /v1/invoices/:id/pay` to retry collection using the card on file.
- Call `stripe.pay_invoice` to retry collection using the card on file.
- Log and continue on failures.
## `fn handle_invoice_created(&self, invoice: &Invoice)`
Attempts to pay a new subscription invoice. Because Stripe defaults to pay-in-advance, this webhook fires immediately when a subscription is created (i.e. when a paid relay is added or a plan is upgraded). Payment priority:
1. **NWC auto-pay**: If the tenant has a `nwc_url`:
- Create a bolt11 Lightning invoice for the invoice amount using `self.nwc_url` (the receiving/system wallet)
- Pay the bolt11 invoice using the tenant's `nwc_url` (the spending/tenant wallet)
- If payment succeeds: call Stripe `POST /v1/invoices/:id/pay` with `paid_out_of_band: true`. Clear `nwc_error` via `command.clear_tenant_nwc_error`.
- If payment fails: set `nwc_error` on tenant via `command.set_tenant_nwc_error`. Fall through to next option.
1. **NWC auto-pay**: If the tenant has a `nwc_url`, run `nwc_pay_invoice` (decrypting the tenant's stored `nwc_url` first):
- The system wallet (`self.wallet`) issues a bolt11 invoice for the fiat amount; the tenant's wallet (`Wallet::from_url` of the decrypted URL) pays it. A `pending` row in `invoice_nwc_payment` guards against double-charging across retries.
- If payment succeeds: mark the Stripe invoice paid out of band (`stripe.pay_invoice_out_of_band`) and clear `nwc_error` via `command.clear_tenant_nwc_error`. Done.
- If it fails before any charge could have gone out: set `nwc_error` on the tenant via `command.set_tenant_nwc_error`, and fall through to the next option (carrying a short summary of the error into the eventual DM).
- If it fails after a charge may have gone out (needs reconciliation): set `nwc_error` and return the error without falling through — a human must reconcile before any retry.
2. **Card on file**: If the tenant has a payment method on the Stripe customer, do nothing here — Stripe will charge automatically for this invoice attempt.
3. **Manual payment**: If neither NWC nor card is available, send a DM via `robot.send_dm` notifying the tenant that payment is due with a link to the application for manual Lightning payment.
+11
View File
@@ -0,0 +1,11 @@
# `bitcoin` — fiat ↔ Bitcoin conversion
Free async helpers for pricing fiat amounts in Lightning units against a live BTC spot price. The NWC wallet lives in `spec/wallet.md`; billing orchestration lives in `spec/billing.md`.
## `pub async fn fiat_to_msats(amount_fiat_minor: i64, currency: &str) -> Result<u64>`
Converts a Stripe-style minor-unit fiat amount to millisatoshis using the live BTC spot price for `currency` and Stripe's per-currency decimal exponent (most currencies 2; `JPY`/`KRW`/… 0; `BHD`/`KWD`/… 3).
## `pub async fn get_bitcoin_price(currency: &str) -> Result<f64>`
Returns the current BTC spot price in `currency`, fetched from Coinbase's public spot-price endpoint.
+2
View File
@@ -5,6 +5,7 @@ Infra is a service which listens for activity and synchronizes relay updates to
Members:
- `api_url: String` - the URL of the zooid instance to be managed, from `ZOOID_API_URL`
- `blossom_s3: Option<BlossomS3Sync>` - shared Blossom S3 settings from `BLOSSOM_S3_*` when region, bucket, access key, and secret are all non-empty after trim
- `query: Query`
- `command: Command`
@@ -36,3 +37,4 @@ Members:
- Otherwise, sends `PATCH /relay/:id` to update it.
- Includes `secret` only for relay creation (`POST`) so updates do not rotate relay identity.
- Passes relay configuration in the body including host, schema, inactive flag, info, policy, groups, management, blossom, livekit, push, and roles.
- When `blossom_s3` is configured and the relay has blossom enabled, the blossom section includes `adapter: "s3"`, S3 fields from the environment, and `s3.key_prefix` set to the relay's `schema`. Otherwise blossom omits S3 (zooid defaults to local storage) or sends `{ "enabled": false }` when blossom is disabled.
+2 -2
View File
@@ -63,9 +63,9 @@ Tenants are customers of the service, identified by a nostr `pubkey`. Public met
A relay is a nostr relay owned by a `tenant` and hosted by the attached zooid instance. Relay subdomains MUST be unique.
- `id` - a random ID identifying the relay
- `id` - calculated based on `subdomain` + 8 random hex chars
- `tenant` - the tenant's pubkey
- `schema` - the relay's db schema (read_only, calculated based on `subdomain` + `id`)
- `schema` - the relay's db schema (read only, same as `id`)
- `subdomain` - the relay's subdomain
- `plan` - the relay's plan
- `stripe_subscription_item_id` (nullable) - the Stripe subscription item id. Only set for relays on paid plans.
-4
View File
@@ -39,10 +39,6 @@ Members:
- Returns the tenant matching the given `stripe_customer_id`
## `pub fn has_active_paid_relays(&self, tenant_id: &str) -> Result<bool>`
- Returns true if the tenant has any relays where `status = 'active'` and `plan != 'free'`
## `pub fn list_activity_for_relay(&self, relay_id: &str) -> Result<Vec<Activity>>`
- Returns all activity where `resource_type = 'relay'` and `resource_id = relay_id`
+113
View File
@@ -0,0 +1,113 @@
# `pub struct Stripe`
A thin async wrapper around the subset of the Stripe REST API this service uses. It knows nothing about relays, tenants, or the database — it just speaks HTTP to Stripe and returns `serde_json::Value` (or small typed results). The domain logic that drives it lives in `spec/billing.md`.
Members:
- `secret_key: String` - Stripe API key, used as the bearer token and as the HMAC key for idempotency keys
- `webhook_secret: String` - secret for verifying Stripe webhook signatures
- `http: reqwest::Client`
All requests authenticate with `secret_key` via `Authorization: Bearer`. Write requests that must not be retried twice (customer/subscription/item creation, invoice payment) send a deterministic `Idempotency-Key` derived by HMAC-ing the operation name and its arguments with `secret_key`. Reconcile-to-desired-state writes (e.g. setting an item quantity) intentionally omit the idempotency key, since re-applying the same target is a no-op.
On any 4xx/5xx response, the wrapper reads the body and folds Stripe's JSON error payload (`error.message` / `error.type` / `error.code` / `error.param`) into the returned error so callers get an actionable message instead of a bare status line.
## `pub fn from_env() -> Self`
Reads `STRIPE_SECRET_KEY` and `STRIPE_WEBHOOK_SECRET` (both required) and constructs the client. Panics if either is missing or blank. This is what `Billing::new` calls.
## `pub fn new(secret_key: String, webhook_secret: String) -> Self`
Constructs the client with a fresh `reqwest::Client` from explicit keys (does not touch the environment).
## `pub async fn create_customer(&self, name: &str, tenant_pubkey: &str) -> Result<String>`
- `POST /v1/customers` with `name` and `metadata[tenant_pubkey]`
- Idempotent on `tenant_pubkey`
- Returns the new customer id; errors if it isn't a `cus_…` id
## `pub async fn get_subscription(&self, subscription_id: &str) -> Result<Option<Value>>`
- `GET /v1/subscriptions/:id`
- Returns `None` if Stripe responds `404` (so callers can recover from a stale subscription id), otherwise the subscription object
## `pub async fn create_subscription(&self, customer_id: &str, items: &BTreeMap<String, i64>) -> Result<(String, BTreeMap<String, String>)>`
- `POST /v1/subscriptions` with `collection_method: charge_automatically` and one `items[n][price]` / `items[n][quantity]` pair per entry
- Idempotent on the customer and the `(price, quantity)` set
- Returns the subscription id and a map from price id to the created subscription item id
## `pub async fn create_subscription_item(&self, subscription_id: &str, price_id: &str, quantity: i64) -> Result<String>`
- `POST /v1/subscription_items`
- Idempotent on `(subscription_id, price_id)`
- Returns the new subscription item id
## `pub async fn set_subscription_item_quantity(&self, item_id: &str, quantity: i64) -> Result<()>`
- `POST /v1/subscription_items/:id` with `quantity`
- No idempotency key (reconcile-to-target write)
## `pub async fn delete_subscription_item(&self, item_id: &str) -> Result<()>`
- `DELETE /v1/subscription_items/:id`
## `pub async fn cancel_subscription(&self, subscription_id: &str) -> Result<()>`
- `DELETE /v1/subscriptions/:id`
## `pub async fn list_invoices(&self, customer_id: &str) -> Result<Value>`
- `GET /v1/invoices?customer=…`
- Returns the `data` array
## `pub async fn get_invoice(&self, invoice_id: &str) -> Result<Value, InvoiceLookupError>`
- `GET /v1/invoices/:id`
- On a 4xx response, returns `InvoiceLookupError::StripeClient { status }` (callers usually surface this as a client error, e.g. `404` "no such invoice"); other failures are `InvoiceLookupError::Internal`
- Returns the full invoice object
## `pub async fn pay_invoice(&self, invoice_id: &str) -> Result<()>`
- `POST /v1/invoices/:id/pay` (retries collection using the customer's default payment method)
- Idempotent on `invoice_id`
## `pub async fn pay_invoice_out_of_band(&self, invoice_id: &str) -> Result<()>`
- `POST /v1/invoices/:id/pay` with `paid_out_of_band: true` — used when payment was collected over Lightning rather than through Stripe
- Idempotent on `invoice_id`
## `pub async fn preview_upcoming_invoice(&self, customer_id: &str, subscription_id: Option<&str>) -> Result<Value>`
- `GET /v1/invoices/upcoming?customer=…[&subscription=…]`
- Used to validate proration when a subscription is downgraded
## `pub async fn has_payment_method(&self, customer_id: &str) -> Result<bool>`
- `GET /v1/payment_methods?customer=…&type=card`
- Returns whether the customer has at least one card on file
## `pub async fn create_portal_session(&self, customer_id: &str, return_url: Option<&str>) -> Result<String>`
- `POST /v1/billing_portal/sessions` with `customer` and optional `return_url`
- Returns the portal session URL
## `pub fn construct_event(&self, payload: &str, sig_header: &str) -> Result<Event>`
Verifies the `Stripe-Signature` header against `webhook_secret` and parses the body.
- Parse `t=` (timestamp) and `v1=` (signature) from the header
- Compute `HMAC-SHA256(webhook_secret, "{t}.{payload}")`, hex-encode it, and compare to `v1=`; error on mismatch
- Error if the timestamp is more than 300 seconds from now
- Returns the deserialized `Event` (`{ event_type, data: { object } }`)
# `pub enum InvoiceLookupError`
- `StripeClient { status: reqwest::StatusCode }` - Stripe returned a 4xx for an invoice lookup
- `Internal(anyhow::Error)` - any other failure
Implements `Display`/`Error` and `From<anyhow::Error>` / `From<reqwest::Error>` (both mapping to `Internal`).
# `pub struct Event` / `pub struct EventData`
The verified, parsed webhook event: `Event { event_type: String, data: EventData }`, `EventData { object: serde_json::Value }`.
+23
View File
@@ -0,0 +1,23 @@
# `pub struct Wallet`
A handle to a single Nostr Wallet Connect (NWC) wallet. `Billing` holds one as its system wallet (receives — issues and looks up invoices); tenant wallets (pay invoices) are constructed ad-hoc from the decrypted `tenant.nwc_url` at the call site.
Member:
- `url: NostrWalletConnectURI` — the parsed `nostr+walletconnect://…` URI
## `pub fn from_url(url: &str) -> Result<Self>`
Parses an `nostr+walletconnect://` URI.
## `pub async fn make_invoice(&self, amount_msats: u64, description: &str) -> Result<String>`
Issues a bolt11 invoice for `amount_msats` and returns it.
## `pub async fn pay_invoice(&self, bolt11: String) -> Result<()>`
Pays a bolt11 invoice.
## `pub async fn is_settled(&self, bolt11: &str) -> Result<bool>`
Returns whether a bolt11 invoice (previously issued by this wallet) has settled.
+134 -1094
View File
File diff suppressed because it is too large Load Diff
+241 -922
View File
File diff suppressed because it is too large Load Diff
+48
View File
@@ -0,0 +1,48 @@
use anyhow::{Result, anyhow};
pub async fn fiat_to_msats(amount_fiat_minor: i64, currency: &str) -> Result<u64> {
let price = get_bitcoin_price(&currency.to_uppercase()).await?;
let divisor = 10_f64.powi(currency_minor_exponent(currency)? as i32);
let amount_fiat = (amount_fiat_minor as f64) / divisor;
let amount_msats = (amount_fiat / price * 100_000_000_000.0).round();
Ok(amount_msats as u64)
}
#[derive(serde::Deserialize)]
struct CoinbaseSpotPriceResponse {
data: CoinbaseSpotPriceData,
}
#[derive(serde::Deserialize)]
struct CoinbaseSpotPriceData {
amount: String,
}
pub async fn get_bitcoin_price(currency: &str) -> Result<f64> {
let http = reqwest::Client::new();
let url = format!("https://api.coinbase.com/v2/prices/BTC-{currency}/spot");
let resp = http.get(url).send().await?;
let body: CoinbaseSpotPriceResponse = resp.error_for_status()?.json().await?;
Ok(body
.data
.amount
.parse::<f64>()
.map_err(|e| anyhow!("invalid BTC spot quote for {currency}: {e}"))?)
}
/// Number of decimal places in `currency`'s minor unit, following Stripe's
/// currency conventions (most are 2, JPY/KRW/… are 0, BHD/KWD/… are 3).
fn currency_minor_exponent(currency: &str) -> Result<u8> {
let normalized = currency.to_uppercase();
let exponent = match normalized.as_str() {
// Zero-decimal currencies in Stripe.
"BIF" | "CLP" | "DJF" | "GNF" | "JPY" | "KMF" | "KRW" | "MGA" | "PYG" | "RWF" | "UGX"
| "VND" | "VUV" | "XAF" | "XOF" | "XPF" => 0,
// Three-decimal currencies in Stripe.
"BHD" | "JOD" | "KWD" | "OMR" | "TND" => 3,
_ if normalized.chars().all(|c| c.is_ascii_alphabetic()) && normalized.len() == 3 => 2,
_ => return Err(anyhow!("invalid currency code: {currency}")),
};
Ok(exponent)
}
-28
View File
@@ -1,28 +0,0 @@
use anyhow::{Result, anyhow};
use nostr_sdk::prelude::*;
pub fn encrypt(plaintext: &str) -> Result<String> {
let keys = load_key()?;
nip44::encrypt(
keys.secret_key(),
&keys.public_key(),
plaintext,
nip44::Version::V2,
)
.map_err(|e| anyhow!("encryption failed: {e}"))
}
pub fn decrypt(ciphertext: &str) -> Result<String> {
let keys = load_key()?;
nip44::decrypt(keys.secret_key(), &keys.public_key(), ciphertext)
.map_err(|e| anyhow!("decryption failed: {e}"))
}
fn load_key() -> Result<Keys> {
let secret = std::env::var("ENCRYPTION_SECRET")
.map_err(|_| anyhow!("missing ENCRYPTION_SECRET environment variable"))?;
if secret.trim().is_empty() {
return Err(anyhow!("ENCRYPTION_SECRET is empty"));
}
Keys::parse(&secret).map_err(|e| anyhow!("invalid ENCRYPTION_SECRET: {e}"))
}
+126
View File
@@ -0,0 +1,126 @@
use anyhow::{Result, anyhow};
use nostr_sdk::prelude::*;
#[derive(Clone)]
pub struct Env {
pub server_host: String,
pub server_port: u16,
pub server_admin_pubkeys: Vec<String>,
pub server_allow_origins: Vec<String>,
pub database_url: String,
pub robot_name: String,
pub robot_wallet: String,
pub robot_picture: String,
pub robot_description: String,
pub robot_outbox_relays: Vec<String>,
pub robot_indexer_relays: Vec<String>,
pub robot_messaging_relays: Vec<String>,
pub blossom_s3_region: String,
pub blossom_s3_bucket: String,
pub blossom_s3_endpoint: String,
pub blossom_s3_access_key: String,
pub blossom_s3_secret_key: String,
pub zooid_api_url: String,
pub relay_domain: String,
pub livekit_url: String,
pub livekit_api_key: String,
pub livekit_api_secret: String,
pub stripe_secret_key: String,
pub stripe_webhook_secret: String,
pub stripe_price_basic: String,
pub stripe_price_growth: String,
/// Parsed from `robot_secret`; used for nostr signing and nip44 encryption.
pub keys: Keys,
}
impl Env {
pub fn load() -> Self {
let keys = Keys::parse(&require_str("ROBOT_SECRET"))
.expect("ROBOT_SECRET is not a valid nostr secret key");
Self {
server_host: require_str("SERVER_HOST"),
server_port: require_u16("SERVER_PORT"),
server_admin_pubkeys: require_csv("SERVER_ADMIN_PUBKEYS"),
server_allow_origins: require_csv("SERVER_ALLOW_ORIGINS"),
database_url: require_str("DATABASE_URL"),
robot_name: require_str("ROBOT_NAME"),
robot_wallet: require_str("ROBOT_WALLET"),
robot_picture: require_str("ROBOT_PICTURE"),
robot_description: require_str("ROBOT_DESCRIPTION"),
robot_outbox_relays: require_csv("ROBOT_OUTBOX_RELAYS"),
robot_indexer_relays: require_csv("ROBOT_INDEXER_RELAYS"),
robot_messaging_relays: require_csv("ROBOT_MESSAGING_RELAYS"),
blossom_s3_region: require_str("BLOSSOM_S3_REGION"),
blossom_s3_bucket: require_str("BLOSSOM_S3_BUCKET"),
blossom_s3_endpoint: require_str("BLOSSOM_S3_ENDPOINT"),
blossom_s3_access_key: require_str("BLOSSOM_S3_ACCESS_KEY"),
blossom_s3_secret_key: require_str("BLOSSOM_S3_SECRET_KEY"),
zooid_api_url: require_str("ZOOID_API_URL"),
relay_domain: require_str("RELAY_DOMAIN"),
livekit_url: require_str("LIVEKIT_URL"),
livekit_api_key: require_str("LIVEKIT_API_KEY"),
livekit_api_secret: require_str("LIVEKIT_API_SECRET"),
stripe_secret_key: require_str("STRIPE_SECRET_KEY"),
stripe_webhook_secret: require_str("STRIPE_WEBHOOK_SECRET"),
stripe_price_basic: require_str("STRIPE_PRICE_BASIC"),
stripe_price_growth: require_str("STRIPE_PRICE_GROWTH"),
keys,
}
}
pub fn encrypt(&self, plaintext: &str) -> Result<String> {
nip44::encrypt(
self.keys.secret_key(),
&self.keys.public_key(),
plaintext,
nip44::Version::V2,
)
.map_err(|e| anyhow!("encryption failed: {e}"))
}
pub fn decrypt(&self, ciphertext: &str) -> Result<String> {
nip44::decrypt(self.keys.secret_key(), &self.keys.public_key(), ciphertext)
.map_err(|e| anyhow!("decryption failed: {e}"))
}
pub async fn make_auth(&self, url: &str, method: HttpMethod) -> Result<String> {
let server_url = Url::parse(url)?;
let auth = HttpData::new(server_url, method)
.to_authorization(&self.keys)
.await?;
Ok(auth)
}
}
fn require_str(key: &str) -> String {
let v = std::env::var(key)
.unwrap_or_else(|_| panic!("{key} is required"))
.trim()
.to_string();
if v.is_empty() {
panic!("{key} is required")
}
v
}
fn require_u16(key: &str) -> u16 {
require_str(key)
.parse()
.unwrap_or_else(|_| panic!("{key} is invalid"))
}
fn require_csv(key: &str) -> Vec<String> {
let v: Vec<String> = std::env::var(key)
.unwrap_or_default()
.split(',')
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.collect();
if v.is_empty() {
panic!("{key} is required");
}
v
}
+161 -244
View File
@@ -3,6 +3,7 @@ use nostr_sdk::prelude::*;
use std::time::Duration;
use crate::command::Command;
use crate::env::Env;
use crate::models::{Activity, RELAY_STATUS_DELINQUENT, RELAY_STATUS_INACTIVE, Relay};
use crate::query::Query;
@@ -12,42 +13,18 @@ const RELAY_SYNC_RETRY_MAX_ATTEMPTS: usize = 6;
#[derive(Clone)]
pub struct Infra {
api_url: String,
relay_domain: String,
livekit_url: String,
livekit_api_key: String,
livekit_api_secret: String,
api_secret: String,
env: Env,
query: Query,
command: Command,
}
impl Infra {
pub fn new(query: Query, command: Command) -> Result<Self> {
let api_url = std::env::var("ZOOID_API_URL").unwrap_or_default();
let relay_domain = std::env::var("RELAY_DOMAIN").unwrap_or_default();
let livekit_url = std::env::var("LIVEKIT_URL").unwrap_or_default();
let livekit_api_key = std::env::var("LIVEKIT_API_KEY").unwrap_or_default();
let livekit_api_secret = std::env::var("LIVEKIT_API_SECRET").unwrap_or_default();
let api_secret = std::env::var("ZOOID_API_SECRET").unwrap_or_default();
if api_url.trim().is_empty() {
anyhow::bail!("missing ZOOID_API_URL");
}
if api_secret.trim().is_empty() {
anyhow::bail!("missing ZOOID_API_SECRET");
}
Ok(Self {
api_url,
relay_domain,
livekit_url,
livekit_api_key,
livekit_api_secret,
api_secret,
pub fn new(query: Query, command: Command, env: &Env) -> Self {
Self {
env: env.clone(),
query,
command,
})
}
}
pub async fn start(self) {
@@ -77,15 +54,17 @@ impl Infra {
}
async fn handle_activity(&self, activity: &Activity) -> Result<()> {
let needs_sync = should_sync_relay_activity(activity.activity_type.as_str());
let needs_sync = matches!(
activity.activity_type.as_str(),
"create_relay" | "update_relay" | "activate_relay" | "deactivate_relay" | "fail_relay_sync"
);
if !needs_sync || activity.resource_type != "relay" {
if activity.resource_type != "relay" || !needs_sync {
return Ok(());
}
if activity.activity_type == "fail_relay_sync" {
self.schedule_relay_sync_retry(&activity.resource_id, "activity")
.await?;
self.schedule_relay_sync_retry(&activity.resource_id, "activity").await?;
return Ok(());
}
@@ -93,9 +72,7 @@ impl Infra {
return Ok(());
};
let is_new = self.relay_sync_is_new(&relay).await?;
self.sync_and_report(&relay, is_new).await;
self.sync_relay(&relay).await;
Ok(())
}
@@ -110,8 +87,7 @@ impl Infra {
for relay in relays {
if relay.sync_error.trim().is_empty() {
let is_new = self.relay_sync_is_new(&relay).await?;
self.sync_and_report(&relay, is_new).await;
self.sync_relay(&relay).await;
} else {
self.schedule_relay_sync_retry(&relay.id, source).await?;
}
@@ -121,10 +97,28 @@ impl Infra {
}
async fn schedule_relay_sync_retry(&self, relay_id: &str, source: &str) -> Result<()> {
let activities = self.query.list_activity_for_relay(relay_id).await?;
let consecutive_failures = consecutive_sync_failures(&activities);
fn get_retry_delay(consecutive_failures: usize) -> Option<Duration> {
let retry_attempt = consecutive_failures.max(1);
if retry_attempt > RELAY_SYNC_RETRY_MAX_ATTEMPTS {
return None;
}
let Some(delay) = relay_sync_retry_delay(consecutive_failures) else {
let exponent = (retry_attempt - 1).min(31);
let multiplier = 1u64 << exponent;
let delay_secs = RELAY_SYNC_RETRY_BASE_DELAY_SECS
.saturating_mul(multiplier)
.min(RELAY_SYNC_RETRY_MAX_DELAY_SECS);
Some(Duration::from_secs(delay_secs))
}
let activities = self.query.list_activity_for_relay(relay_id).await?;
let consecutive_failures = activities
.iter()
.take_while(|activity| activity.activity_type == "fail_relay_sync")
.count();
let Some(delay) = get_retry_delay(consecutive_failures) else {
tracing::warn!(
relay = relay_id,
consecutive_failures,
@@ -148,40 +142,20 @@ impl Infra {
tokio::spawn(async move {
tokio::time::sleep(delay).await;
if let Err(e) = infra.retry_relay_sync(&relay_id).await {
tracing::error!(relay = %relay_id, error = %e, "relay sync retry task failed");
match infra.query.get_relay(&relay_id).await {
Ok(Some(relay)) => infra.sync_relay(&relay).await,
Ok(None) => {}
Err(e) => {
tracing::error!(relay = %relay_id, error = %e, "relay sync retry task failed");
}
}
});
Ok(())
}
async fn retry_relay_sync(&self, relay_id: &str) -> Result<()> {
let Some(relay) = self.query.get_relay(relay_id).await? else {
return Ok(());
};
if relay.sync_error.trim().is_empty() {
tracing::debug!(relay = %relay.id, "skip relay sync retry; relay has no sync_error");
return Ok(());
}
let is_new = self.relay_sync_is_new(&relay).await?;
self.sync_and_report(&relay, is_new).await;
Ok(())
}
async fn relay_sync_is_new(&self, relay: &Relay) -> Result<bool> {
if relay.synced == 1 {
return Ok(false);
}
let has_completed_sync = self.query.relay_has_completed_sync(&relay.id).await?;
Ok(!has_completed_sync)
}
async fn sync_and_report(&self, relay: &Relay, is_new: bool) {
match self.sync_relay(relay, is_new).await {
async fn sync_relay(&self, relay: &Relay) {
match self.try_sync_relay(relay).await {
Ok(()) => {
tracing::info!(relay = %relay.id, "relay sync succeeded");
if let Err(e) = self.command.complete_relay_sync(&relay.id).await {
@@ -197,191 +171,134 @@ impl Infra {
}
}
async fn nip98_auth(&self, url: &str, method: HttpMethod) -> Result<String> {
let keys = Keys::parse(&self.api_secret)?;
let server_url = Url::parse(url)?;
let auth = HttpData::new(server_url, method)
.to_authorization(&keys)
async fn try_sync_relay(&self, relay: &Relay) -> Result<()> {
// A relay is "new" (POST with a freshly generated secret) only if it has
// never completed a sync. `synced == 1` short-circuits the activity lookup;
// otherwise check the activity history so that a re-sync after an update
// (which resets `synced` to 0) PATCHes instead of clobbering the secret.
let is_new = relay.synced != 1
&& !self.query.relay_has_completed_sync(&relay.id).await?;
let mut body = serde_json::json!({
"host": format!("{}.{}", relay.subdomain, self.env.relay_domain),
"schema": relay.schema,
"inactive": relay.status == RELAY_STATUS_INACTIVE
|| relay.status == RELAY_STATUS_DELINQUENT,
"info": {
"name": relay.info_name,
"icon": relay.info_icon,
"description": relay.info_description,
"pubkey": relay.tenant,
},
"policy": {
"public_join": relay.policy_public_join == 1,
"strip_signatures": relay.policy_strip_signatures == 1,
},
"groups": { "enabled": relay.groups_enabled == 1 },
"management": { "enabled": relay.management_enabled == 1 },
"blossom": if relay.blossom_enabled == 1 {
serde_json::json!({
"enabled": true,
"adapter": "s3",
"s3": {
"endpoint": self.env.blossom_s3_endpoint,
"region": self.env.blossom_s3_region,
"bucket": self.env.blossom_s3_bucket,
"access_key": self.env.blossom_s3_access_key,
"secret_key": self.env.blossom_s3_secret_key,
"key_prefix": relay.schema,
},
})
} else {
serde_json::json!({ "enabled": false })
},
"livekit": if relay.livekit_enabled == 1 {
serde_json::json!({
"enabled": true,
"server_url": self.env.livekit_url,
"api_key": self.env.livekit_api_key,
"api_secret": self.env.livekit_api_secret,
})
} else {
serde_json::json!({ "enabled": false })
},
"push": { "enabled": relay.push_enabled == 1 },
"roles": {
"admin": { "can_manage": true, "can_invite": true },
"member": { "can_invite": true },
},
});
// Only provide a secret if the relay is new. This allows us to not store the relay secrets on our side.
if is_new {
if let Some(obj) = body.as_object_mut() {
obj.insert(
"secret".to_string(),
serde_json::Value::String(Keys::generate().secret_key().to_secret_hex()),
);
}
}
let method = if is_new { HttpMethod::POST } else { HttpMethod::PATCH };
self.request(method, &format!("relay/{}", relay.id), Some(&body))
.await?;
Ok(auth)
Ok(())
}
pub async fn list_relay_members(&self, relay_id: &str) -> Result<Vec<String>> {
let client = reqwest::Client::new();
let base = self.api_url.trim_end_matches('/');
let url = format!("{base}/relay/{relay_id}/members");
let auth = self.nip98_auth(&url, HttpMethod::GET).await?;
#[derive(serde::Deserialize)]
struct MembersResponse {
members: Vec<String>,
}
let response = client
.get(&url)
.header("Authorization", auth)
.send()
let response = self
.request(HttpMethod::GET, &format!("relay/{relay_id}/members"), None)
.await?;
let parsed: MembersResponse = response.json().await?;
Ok(parsed.members)
}
// Internal utilities
/// Sends an authenticated request to the zooid API at `path` (relative to
/// `env.zooid_api_url`). Returns the response on 2xx; bails with the body
/// text otherwise.
async fn request(
&self,
method: HttpMethod,
path: &str,
body: Option<&serde_json::Value>,
) -> Result<reqwest::Response> {
let client = reqwest::Client::builder()
.timeout(Duration::from_secs(5))
.build()?;
let base = self.env.zooid_api_url.trim_end_matches('/');
let path = path.trim_start_matches('/');
let url = format!("{base}/{path}");
let auth = self.env.make_auth(&url, method).await?;
let reqwest_method = match method {
HttpMethod::GET => reqwest::Method::GET,
HttpMethod::POST => reqwest::Method::POST,
HttpMethod::PUT => reqwest::Method::PUT,
HttpMethod::PATCH => reqwest::Method::PATCH,
};
let mut req = client
.request(reqwest_method, &url)
.header("Authorization", auth);
if let Some(body) = body {
req = req.json(body);
}
let response = req.send().await?;
if !response.status().is_success() {
let status = response.status();
let body = response.text().await.unwrap_or_default();
anyhow::bail!("zooid members returned {status}: {body}");
let text = response.text().await.unwrap_or_default();
anyhow::bail!("zooid {method} {path} returned {status}: {text}");
}
let body = response.text().await?;
parse_relay_members_response(&body)
}
async fn sync_relay(&self, relay: &Relay, is_new: bool) -> Result<()> {
let client = reqwest::Client::new();
let base = self.api_url.trim_end_matches('/');
let host = if self.relay_domain.is_empty() {
relay.subdomain.clone()
} else {
format!("{}.{}", relay.subdomain, self.relay_domain)
};
let livekit = if relay.livekit_enabled == 1 {
serde_json::json!({
"enabled": true,
"server_url": self.livekit_url,
"api_key": self.livekit_api_key,
"api_secret": self.livekit_api_secret,
})
} else {
serde_json::json!({ "enabled": false })
};
let body = relay_sync_body(
relay,
host,
livekit,
is_new.then(|| Keys::generate().secret_key().to_secret_hex()),
);
let url = format!("{}/relay/{}", base, relay.id);
let auth = self
.nip98_auth(&url, zooid_sync_http_method(is_new))
.await?;
let request = if is_new {
client.post(&url)
} else {
client.patch(&url)
};
let response = request
.header("Authorization", auth)
.json(&body)
.send()
.await?;
if !response.status().is_success() {
let status = response.status();
let body = response.text().await.unwrap_or_default();
anyhow::bail!("zooid sync returned {status}: {body}")
}
Ok(())
Ok(response)
}
}
fn zooid_sync_http_method(is_new: bool) -> HttpMethod {
if is_new {
HttpMethod::POST
} else {
HttpMethod::PATCH
}
}
fn parse_relay_members_response(body: &str) -> Result<Vec<String>> {
let value: serde_json::Value = serde_json::from_str(body)?;
if let Some(members) = members_from_value(&value) {
return Ok(members);
}
if let Some(members) = value.get("members").and_then(members_from_value) {
return Ok(members);
}
if let Some(members) = value
.get("data")
.and_then(|data| data.get("members"))
.and_then(members_from_value)
{
return Ok(members);
}
anyhow::bail!("zooid members response missing members array")
}
fn members_from_value(value: &serde_json::Value) -> Option<Vec<String>> {
let values = value.as_array()?;
values
.iter()
.map(|value| value.as_str().map(ToString::to_string))
.collect()
}
fn relay_sync_body(
relay: &Relay,
host: String,
livekit: serde_json::Value,
secret: Option<String>,
) -> serde_json::Value {
let mut body = serde_json::json!({
"host": host,
"schema": relay.schema,
"inactive": relay.status == RELAY_STATUS_INACTIVE
|| relay.status == RELAY_STATUS_DELINQUENT,
"info": {
"name": relay.info_name,
"icon": relay.info_icon,
"description": relay.info_description,
"pubkey": relay.tenant,
},
"policy": {
"public_join": relay.policy_public_join == 1,
"strip_signatures": relay.policy_strip_signatures == 1,
},
"groups": { "enabled": relay.groups_enabled == 1 },
"management": { "enabled": relay.management_enabled == 1 },
"blossom": { "enabled": relay.blossom_enabled == 1 },
"livekit": livekit,
"push": { "enabled": relay.push_enabled == 1 },
"roles": {
"admin": { "can_manage": true, "can_invite": true },
"member": { "can_invite": true },
},
});
if let (Some(secret), Some(body_obj)) = (secret, body.as_object_mut()) {
body_obj.insert("secret".to_string(), serde_json::Value::String(secret));
}
body
}
fn should_sync_relay_activity(activity_type: &str) -> bool {
matches!(
activity_type,
"create_relay" | "update_relay" | "activate_relay" | "deactivate_relay" | "fail_relay_sync"
)
}
fn consecutive_sync_failures(activities: &[Activity]) -> usize {
activities
.iter()
.take_while(|activity| activity.activity_type == "fail_relay_sync")
.count()
}
fn relay_sync_retry_delay(consecutive_failures: usize) -> Option<Duration> {
let retry_attempt = consecutive_failures.max(1);
if retry_attempt > RELAY_SYNC_RETRY_MAX_ATTEMPTS {
return None;
}
let exponent = (retry_attempt - 1).min(31);
let multiplier = 1u64 << exponent;
let delay_secs = RELAY_SYNC_RETRY_BASE_DELAY_SECS
.saturating_mul(multiplier)
.min(RELAY_SYNC_RETRY_MAX_DELAY_SECS);
Some(Duration::from_secs(delay_secs))
}
+6 -1
View File
@@ -1,9 +1,14 @@
pub mod api;
pub mod billing;
pub mod cipher;
pub mod bitcoin;
pub mod command;
pub mod env;
pub mod infra;
pub mod models;
pub mod pool;
pub mod query;
pub mod robot;
pub mod routes;
pub mod stripe;
pub mod wallet;
pub mod web;
+20 -22
View File
@@ -1,12 +1,17 @@
mod api;
mod billing;
mod cipher;
mod bitcoin;
mod command;
mod env;
mod infra;
mod models;
mod pool;
mod query;
mod robot;
mod routes;
mod stripe;
mod wallet;
mod web;
use anyhow::Result;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
@@ -16,6 +21,7 @@ use tower_http::cors::{AllowOrigin, CorsLayer};
use crate::api::Api;
use crate::billing::Billing;
use crate::command::Command;
use crate::env::Env;
use crate::infra::Infra;
use crate::query::Query;
use crate::robot::Robot;
@@ -29,30 +35,21 @@ async fn main() -> Result<()> {
.with(tracing_subscriber::fmt::layer())
.init();
let pool = pool::create_pool().await?;
let robot = Robot::new().await?;
let query = Query::new(pool.clone());
let env = Env::load();
let pool = pool::create_pool(&env.database_url).await?;
let robot = Robot::new(&env).await?;
let query = Query::new(pool.clone(), &env);
let command = Command::new(pool);
let billing = Billing::new(query.clone(), command.clone(), robot.clone());
let infra = Infra::new(query.clone(), command.clone())?;
let api = Api::new(query, command, billing.clone(), infra.clone());
let billing = Billing::new(query.clone(), command.clone(), robot.clone(), &env);
let infra = Infra::new(query.clone(), command.clone(), &env);
let api = Api::new(query, command, billing.clone(), infra.clone(), &env);
let host = std::env::var("HOST").unwrap_or_else(|_| "127.0.0.1".to_string());
let port: u16 = std::env::var("PORT")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(2892);
let origins: Vec<String> = std::env::var("ALLOW_ORIGINS")
.unwrap_or_default()
.split(',')
.map(|v| v.trim().to_string())
.filter(|v| !v.is_empty())
.collect();
let cors = if origins.is_empty() {
let cors = if env.server_allow_origins.is_empty() {
CorsLayer::permissive()
} else {
let parsed = origins
let parsed = env
.server_allow_origins
.iter()
.filter_map(|o| o.parse::<axum::http::HeaderValue>().ok())
.collect::<Vec<_>>();
@@ -69,7 +66,8 @@ async fn main() -> Result<()> {
billing.start().await;
});
let listener = tokio::net::TcpListener::bind(format!("{host}:{port}")).await?;
let listener =
tokio::net::TcpListener::bind(format!("{}:{}", env.server_host, env.server_port)).await?;
axum::serve(listener, app).await?;
Ok(())
}
+40
View File
@@ -36,6 +36,20 @@ pub struct Tenant {
pub past_due_at: Option<i64>,
}
impl Default for Tenant {
fn default() -> Self {
Self {
pubkey: String::new(),
nwc_url: String::new(),
nwc_error: None,
created_at: 0,
stripe_customer_id: String::new(),
stripe_subscription_id: None,
past_due_at: None,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct Relay {
pub id: String,
@@ -58,3 +72,29 @@ pub struct Relay {
pub push_enabled: i64,
pub synced: i64,
}
impl Default for Relay {
fn default() -> Self {
Self {
id: String::new(),
tenant: String::new(),
schema: String::new(),
subdomain: String::new(),
plan: String::new(),
stripe_subscription_item_id: None,
status: RELAY_STATUS_ACTIVE.to_string(),
sync_error: String::new(),
info_name: String::new(),
info_icon: String::new(),
info_description: String::new(),
policy_public_join: 0,
policy_strip_signatures: 0,
groups_enabled: 1,
management_enabled: 1,
blossom_enabled: 0,
livekit_enabled: 0,
push_enabled: 1,
synced: 0,
}
}
}
+2 -4
View File
@@ -7,10 +7,8 @@ use sqlx::{
sqlite::{SqliteConnectOptions, SqlitePoolOptions},
};
pub async fn create_pool() -> Result<SqlitePool> {
let raw_database_url = std::env::var("DATABASE_URL")
.unwrap_or_else(|_| format!("sqlite://{}/data/caravel.db", env!("CARGO_MANIFEST_DIR")));
let database_url = normalize_sqlite_url(&raw_database_url);
pub async fn create_pool(database_url: &str) -> Result<SqlitePool> {
let database_url = normalize_sqlite_url(database_url);
if let Some(path) = database_url.strip_prefix("sqlite://")
&& !path.is_empty()
+15 -21
View File
@@ -1,16 +1,21 @@
use anyhow::Result;
use sqlx::SqlitePool;
use crate::env::Env;
use crate::models::{Activity, Plan, Relay, Tenant};
#[derive(Clone)]
pub struct Query {
pool: SqlitePool,
env: Env,
}
impl Query {
pub fn new(pool: SqlitePool) -> Self {
Self { pool }
pub fn new(pool: SqlitePool, env: &Env) -> Self {
Self {
pool,
env: env.clone(),
}
}
pub async fn list_tenants(&self) -> Result<Vec<Tenant>> {
@@ -36,7 +41,7 @@ impl Query {
Ok(row)
}
pub fn list_plans() -> Vec<Plan> {
pub fn list_plans(&self) -> Vec<Plan> {
vec![
Plan {
id: "free".to_string(),
@@ -54,7 +59,7 @@ impl Query {
members: Some(100),
blossom: true,
livekit: true,
stripe_price_id: Some(std::env::var("STRIPE_PRICE_BASIC").unwrap_or_default()),
stripe_price_id: Some(self.env.stripe_price_basic.clone()),
},
Plan {
id: "growth".to_string(),
@@ -63,19 +68,19 @@ impl Query {
members: None,
blossom: true,
livekit: true,
stripe_price_id: Some(std::env::var("STRIPE_PRICE_GROWTH").unwrap_or_default()),
stripe_price_id: Some(self.env.stripe_price_growth.clone()),
},
]
}
pub fn get_plan(plan_id: &str) -> Option<Plan> {
Self::list_plans().into_iter().find(|p| p.id == plan_id)
pub fn get_plan(&self, plan_id: &str) -> Option<Plan> {
self.list_plans().into_iter().find(|p| p.id == plan_id)
}
/// True for any plan that costs money. Doesn't require an instance because
/// the answer doesn't depend on Stripe price ids — only the canonical plan id.
pub fn is_paid_plan(plan_id: &str) -> bool {
Self::get_plan(plan_id)
.map(|p| p.id != "free")
.unwrap_or(false)
matches!(plan_id, "basic" | "growth")
}
pub async fn list_relays(&self) -> Result<Vec<Relay>> {
@@ -184,17 +189,6 @@ impl Query {
Ok(bolt11)
}
pub async fn has_active_paid_relays(&self, tenant_id: &str) -> Result<bool> {
let plans = sqlx::query_scalar::<_, String>(
"SELECT plan FROM relay WHERE tenant = ? AND status = 'active'",
)
.bind(tenant_id)
.fetch_all(&self.pool)
.await?;
Ok(plans.into_iter().any(|plan| Self::is_paid_plan(&plan)))
}
pub async fn list_activity_for_relay(&self, relay_id: &str) -> Result<Vec<Activity>> {
let rows = sqlx::query_as::<_, Activity>(
"SELECT id, tenant, created_at, activity_type, resource_type, resource_id
+51 -127
View File
@@ -5,15 +5,11 @@ use anyhow::{Result, anyhow};
use nostr_sdk::prelude::*;
use tokio::sync::Mutex;
use crate::env::Env;
#[derive(Clone)]
pub struct Robot {
secret: String,
name: String,
description: String,
picture: String,
outbox_client: Client,
indexer_client: Client,
messaging_client: Client,
env: Env,
outbox_cache: std::sync::Arc<Mutex<HashMap<String, CacheEntry>>>,
dm_cache: std::sync::Arc<Mutex<HashMap<String, CacheEntry>>>,
}
@@ -25,84 +21,61 @@ struct CacheEntry {
}
impl Robot {
pub async fn new() -> Result<Self> {
let secret = std::env::var("ROBOT_SECRET").unwrap_or_default();
if secret.trim().is_empty() {
return Err(anyhow!("ROBOT_SECRET is required"));
}
let name = std::env::var("ROBOT_NAME").unwrap_or_default();
let description = std::env::var("ROBOT_DESCRIPTION").unwrap_or_default();
let picture = std::env::var("ROBOT_PICTURE").unwrap_or_default();
let outbox_relays = split_relays("ROBOT_OUTBOX_RELAYS");
let indexer_relays = split_relays("ROBOT_INDEXER_RELAYS");
let messaging_relays = split_relays("ROBOT_MESSAGING_RELAYS");
if outbox_relays.is_empty() {
return Err(anyhow!("ROBOT_OUTBOX_RELAYS is required"));
}
if indexer_relays.is_empty() {
return Err(anyhow!("ROBOT_INDEXER_RELAYS is required"));
}
if messaging_relays.is_empty() {
return Err(anyhow!("ROBOT_MESSAGING_RELAYS is required"));
}
let outbox_client = client_with_relays(&secret, &outbox_relays).await?;
let indexer_client = client_with_relays(&secret, &indexer_relays).await?;
let messaging_client = client_with_relays(&secret, &messaging_relays).await?;
pub async fn new(env: &Env) -> Result<Self> {
let robot = Self {
secret,
name,
description,
picture,
outbox_client,
indexer_client,
messaging_client,
env: env.clone(),
outbox_cache: std::sync::Arc::new(Mutex::new(HashMap::new())),
dm_cache: std::sync::Arc::new(Mutex::new(HashMap::new())),
};
robot
.publish_identity(&outbox_relays, &messaging_relays)
.await?;
robot.publish_identity().await?;
Ok(robot)
}
async fn make_client(&self, relays: &[String]) -> Result<Client> {
let client = Client::new(self.env.keys.clone());
for relay in relays {
client.add_relay(relay).await?;
}
client.connect().await;
Ok(client)
}
async fn publish_identity(
&self,
outbox_relays: &[String],
messaging_relays: &[String],
) -> Result<()> {
let mut metadata = Metadata::new();
if !self.name.is_empty() {
metadata = metadata.name(&self.name);
if !self.env.robot_name.is_empty() {
metadata = metadata.name(&self.env.robot_name);
}
if !self.description.is_empty() {
metadata = metadata.about(&self.description);
if !self.env.robot_description.is_empty() {
metadata = metadata.about(&self.env.robot_description);
}
if !self.picture.is_empty() {
metadata = metadata.picture(Url::parse(&self.picture)?);
if !self.env.robot_picture.is_empty() {
metadata = metadata.picture(Url::parse(&self.env.robot_picture)?);
}
self.outbox_client
let outbox_client = self.make_client(&self.env.robot_outbox_relays).await?;
let indexer_client = self.make_client(&self.env.robot_indexer_relays).await?;
outbox_client
.send_event_builder(EventBuilder::metadata(&metadata))
.await?;
let outbox_tags = outbox_relays
let outbox_tags = self.env.robot_outbox_relays
.iter()
.map(|r| Tag::parse(["r", r.as_str()]))
.collect::<std::result::Result<Vec<_>, _>>()?;
self.outbox_client
outbox_client
.send_event_builder(EventBuilder::new(Kind::Custom(10002), "").tags(outbox_tags))
.await?;
let messaging_tags = messaging_relays
let messaging_tags = self.env.robot_messaging_relays
.iter()
.map(|r| Tag::parse(["relay", r.as_str()]))
.collect::<std::result::Result<Vec<_>, _>>()?;
self.indexer_client
indexer_client
.send_event_builder(EventBuilder::new(Kind::Custom(10050), "").tags(messaging_tags))
.await?;
@@ -123,14 +96,8 @@ impl Robot {
}
let recipient_pubkey = PublicKey::parse(recipient)?;
let client = self.messaging_client.clone();
for relay in dm_relays {
let _ = client.add_relay(relay).await;
}
client.connect().await;
client
.send_private_msg(recipient_pubkey, message, [])
.await?;
let client = self.make_client(&dm_relays).await?;
client.send_private_msg(recipient_pubkey, message, []).await?;
Ok(())
}
@@ -141,10 +108,8 @@ impl Robot {
let pubkey = PublicKey::parse(recipient)?;
let filter = Filter::new().author(pubkey).kind(Kind::Custom(10002));
let events = self
.indexer_client
.fetch_events(filter, Duration::from_secs(5))
.await?;
let client = self.make_client(&self.env.robot_indexer_relays).await?;
let events = client.fetch_events(filter, Duration::from_secs(5)).await?;
let mut relays = Vec::new();
if let Some(event) = events.into_iter().max_by_key(|e| e.created_at) {
@@ -160,6 +125,22 @@ impl Robot {
Ok(relays)
}
pub async fn fetch_nostr_name(&self, pubkey: &str) -> Option<String> {
let pubkey = PublicKey::parse(pubkey).ok()?;
let filter = Filter::new().author(pubkey).kind(Kind::Metadata).limit(1);
let client = self.make_client(&self.env.robot_indexer_relays).await.ok()?;
let events = client.fetch_events(filter, Duration::from_secs(5)).await.ok()?;
let event = events.into_iter().max_by_key(|e| e.created_at)?;
let content: serde_json::Value = serde_json::from_str(&event.content).ok()?;
let name = content
.get("display_name")
.or_else(|| content.get("name"))
.and_then(|v| v.as_str())
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())?;
Some(name)
}
async fn fetch_messaging_relays_from_outbox(
&self,
recipient: &str,
@@ -170,13 +151,7 @@ impl Robot {
}
let pubkey = PublicKey::parse(recipient)?;
let keys = Keys::parse(&self.secret)?;
let client = Client::new(keys);
for relay in outbox_relays {
client.add_relay(relay).await?;
}
client.connect().await;
let client = self.make_client(outbox_relays).await?;
let filter = Filter::new().author(pubkey).kind(Kind::Custom(10050));
let events = client.fetch_events(filter, Duration::from_secs(5)).await?;
@@ -195,37 +170,6 @@ impl Robot {
}
}
fn split_relays(key: &str) -> Vec<String> {
std::env::var(key)
.unwrap_or_default()
.split(',')
.map(|v| normalize_relay_url(v.trim()))
.filter(|v| !v.is_empty())
.collect()
}
fn normalize_relay_url(url: &str) -> String {
if url.is_empty() {
return String::new();
}
if url.starts_with("ws://") || url.starts_with("wss://") {
url.to_string()
} else {
format!("wss://{url}")
}
}
async fn client_with_relays(secret: &str, relays: &[String]) -> Result<Client> {
let keys = Keys::parse(secret)?;
let client = Client::new(keys);
for relay in relays {
client.add_relay(relay).await?;
}
client.connect().await;
Ok(client)
}
async fn get_cached(
cache: &std::sync::Arc<Mutex<HashMap<String, CacheEntry>>>,
key: &str,
@@ -254,23 +198,3 @@ async fn set_cached(
},
);
}
#[cfg(test)]
impl Robot {
pub fn test_stub() -> Self {
let keys = Keys::generate();
let client = Client::new(keys);
Self {
secret: String::new(),
name: String::new(),
description: String::new(),
picture: String::new(),
outbox_client: client.clone(),
indexer_client: client.clone(),
messaging_client: client,
outbox_cache: std::sync::Arc::new(Mutex::new(HashMap::new())),
dm_cache: std::sync::Arc::new(Mutex::new(HashMap::new())),
}
}
}
+21
View File
@@ -0,0 +1,21 @@
use std::sync::Arc;
use axum::extract::State;
use serde::Serialize;
use crate::api::{Api, AuthedPubkey};
use crate::web::{ApiResult, ok};
#[derive(Serialize)]
struct IdentityResponse {
pubkey: String,
is_admin: bool,
}
pub async fn get_identity(
State(api): State<Arc<Api>>,
AuthedPubkey(pubkey): AuthedPubkey,
) -> ApiResult {
let is_admin = api.is_admin(&pubkey);
ok(IdentityResponse { pubkey, is_admin })
}
+92
View File
@@ -0,0 +1,92 @@
use std::sync::Arc;
use axum::extract::{Path, State};
use reqwest::StatusCode;
use crate::api::{Api, AuthedPubkey};
use crate::stripe::InvoiceLookupError;
use crate::web::{ApiError, ApiResult, bad_request, internal, not_found, ok};
pub async fn list_tenant_invoices(
State(api): State<Arc<Api>>,
AuthedPubkey(auth): AuthedPubkey,
Path(pubkey): Path<String>,
) -> ApiResult {
api.require_admin_or_tenant(&auth, &pubkey)?;
let tenant = api.get_tenant_or_404(&pubkey).await?;
let invoices = api
.billing
.stripe_list_invoices(&tenant.stripe_customer_id)
.await
.map_err(internal)?;
ok(invoices)
}
pub async fn get_invoice(
State(api): State<Arc<Api>>,
AuthedPubkey(auth): AuthedPubkey,
Path(id): Path<String>,
) -> ApiResult {
let (invoice, tenant) = api
.billing
.get_invoice_with_tenant(&id)
.await
.map_err(map_invoice_lookup_error)?;
api.require_admin_or_tenant(&auth, &tenant.pubkey)?;
let invoice = api
.billing
.reconcile_manual_lightning_invoice(&id, &invoice)
.await
.map_err(map_invoice_lookup_error)?;
ok(invoice)
}
pub async fn get_invoice_bolt11(
State(api): State<Arc<Api>>,
AuthedPubkey(auth): AuthedPubkey,
Path(id): Path<String>,
) -> ApiResult {
let (invoice, tenant) = api
.billing
.get_invoice_with_tenant(&id)
.await
.map_err(map_invoice_lookup_error)?;
api.require_admin_or_tenant(&auth, &tenant.pubkey)?;
let invoice = api
.billing
.reconcile_manual_lightning_invoice(&id, &invoice)
.await
.map_err(map_invoice_lookup_error)?;
let status = invoice["status"].as_str().unwrap_or_default();
if status != "open" {
return Err(bad_request("invoice-not-open", "invoice is not open"));
}
let amount_due = invoice["amount_due"].as_i64().unwrap_or(0);
let currency = invoice["currency"].as_str().unwrap_or("usd");
let bolt11 = api
.billing
.get_or_create_manual_lightning_bolt11(&id, &tenant.pubkey, amount_due, currency)
.await
.map_err(internal)?;
ok(serde_json::json!({ "bolt11": bolt11 }))
}
fn map_invoice_lookup_error(error: InvoiceLookupError) -> ApiError {
match error {
InvoiceLookupError::StripeClient { status } => match status {
StatusCode::NOT_FOUND => not_found("invoice not found"),
_ => {
tracing::warn!(%status, "stripe invoice request returned unexpected status");
internal("invoice request rejected")
}
},
InvoiceLookupError::Internal(error) => internal(error),
}
}
+6
View File
@@ -0,0 +1,6 @@
pub mod identity;
pub mod invoices;
pub mod plans;
pub mod relays;
pub mod stripe;
pub mod tenants;
+17
View File
@@ -0,0 +1,17 @@
use std::sync::Arc;
use axum::extract::{Path, State};
use crate::api::Api;
use crate::web::{ApiResult, not_found, ok};
pub async fn list_plans(State(api): State<Arc<Api>>) -> ApiResult {
ok(api.query.list_plans())
}
pub async fn get_plan(State(api): State<Arc<Api>>, Path(id): Path<String>) -> ApiResult {
match api.query.get_plan(&id) {
Some(plan) => ok(plan),
None => Err(not_found("plan not found")),
}
}
+315
View File
@@ -0,0 +1,315 @@
use std::sync::{Arc, LazyLock};
use anyhow::Result;
use axum::{
Json,
extract::{Path, State},
};
use regex::Regex;
use serde::Deserialize;
use crate::api::{Api, AuthedPubkey};
use crate::models::{
RELAY_STATUS_ACTIVE, RELAY_STATUS_DELINQUENT, RELAY_STATUS_INACTIVE, Relay,
};
use crate::web::{
ApiError, ApiResult, bad_request, created, internal, map_unique_error, ok,
parse_bool_default, unprocessable,
};
#[derive(Deserialize)]
pub struct CreateRelayRequest {
pub tenant: String,
pub subdomain: String,
pub plan: String,
pub info_name: String,
pub info_icon: String,
pub info_description: String,
pub policy_public_join: i64,
pub policy_strip_signatures: i64,
pub groups_enabled: i64,
pub management_enabled: i64,
pub blossom_enabled: i64,
pub livekit_enabled: i64,
pub push_enabled: i64,
}
#[derive(Deserialize)]
pub struct UpdateRelayRequest {
pub subdomain: Option<String>,
pub plan: Option<String>,
pub info_name: Option<String>,
pub info_icon: Option<String>,
pub info_description: Option<String>,
pub policy_public_join: Option<i64>,
pub policy_strip_signatures: Option<i64>,
pub groups_enabled: Option<i64>,
pub management_enabled: Option<i64>,
pub blossom_enabled: Option<i64>,
pub livekit_enabled: Option<i64>,
pub push_enabled: Option<i64>,
}
pub async fn list_relays(
State(api): State<Arc<Api>>,
AuthedPubkey(auth): AuthedPubkey,
) -> ApiResult {
api.require_admin(&auth)?;
let relays = api.query.list_relays().await.map_err(internal)?;
ok(relays)
}
pub async fn get_relay(
State(api): State<Arc<Api>>,
AuthedPubkey(auth): AuthedPubkey,
Path(id): Path<String>,
) -> ApiResult {
let relay = api.get_relay_or_404(&id).await?;
api.require_admin_or_tenant(&auth, &relay.tenant)?;
ok(relay)
}
pub async fn list_relay_activity(
State(api): State<Arc<Api>>,
AuthedPubkey(auth): AuthedPubkey,
Path(id): Path<String>,
) -> ApiResult {
let relay = api.get_relay_or_404(&id).await?;
api.require_admin_or_tenant(&auth, &relay.tenant)?;
let activity = api
.query
.list_activity_for_relay(&id)
.await
.map_err(internal)?;
ok(serde_json::json!({ "activity": activity }))
}
pub async fn list_relay_members(
State(api): State<Arc<Api>>,
AuthedPubkey(auth): AuthedPubkey,
Path(id): Path<String>,
) -> ApiResult {
let relay = api.get_relay_or_404(&id).await?;
api.require_admin_or_tenant(&auth, &relay.tenant)?;
let members = fetch_relay_members(&api, &relay).await.map_err(internal)?;
ok(serde_json::json!({ "members": members }))
}
pub async fn create_relay(
State(api): State<Arc<Api>>,
AuthedPubkey(auth): AuthedPubkey,
Json(payload): Json<CreateRelayRequest>,
) -> ApiResult {
api.require_admin_or_tenant(&auth, &payload.tenant)?;
let relay_id = format!(
"{}_{}",
payload.subdomain.replace('-', "_"),
&uuid::Uuid::new_v4().simple().to_string()[..8]
);
let relay = Relay {
id: relay_id.clone(),
tenant: payload.tenant,
schema: relay_id.clone(),
subdomain: payload.subdomain,
plan: payload.plan,
info_name: payload.info_name,
info_icon: payload.info_icon,
info_description: payload.info_description,
policy_public_join: payload.policy_public_join,
policy_strip_signatures: payload.policy_strip_signatures,
groups_enabled: payload.groups_enabled,
management_enabled: payload.management_enabled,
blossom_enabled: payload.blossom_enabled,
livekit_enabled: payload.livekit_enabled,
push_enabled: payload.push_enabled,
..Default::default()
};
let relay = prepare_relay(&api, relay)?;
api.command
.create_relay(&relay)
.await
.map_err(map_relay_write_error)?;
created(relay)
}
pub async fn update_relay(
State(api): State<Arc<Api>>,
AuthedPubkey(auth): AuthedPubkey,
Path(id): Path<String>,
Json(payload): Json<UpdateRelayRequest>,
) -> ApiResult {
let mut relay = api.get_relay_or_404(&id).await?;
api.require_admin_or_tenant(&auth, &relay.tenant)?;
let current_plan = relay.plan.clone();
let requested_plan = payload.plan.clone();
if let Some(v) = payload.subdomain {
relay.subdomain = v;
}
if let Some(v) = requested_plan.clone() {
relay.plan = v;
}
if let Some(v) = payload.info_name {
relay.info_name = v;
}
if let Some(v) = payload.info_icon {
relay.info_icon = v;
}
if let Some(v) = payload.info_description {
relay.info_description = v;
}
if let Some(v) = payload.policy_public_join {
relay.policy_public_join = v;
}
if let Some(v) = payload.policy_strip_signatures {
relay.policy_strip_signatures = v;
}
if let Some(v) = payload.groups_enabled {
relay.groups_enabled = v;
}
if let Some(v) = payload.management_enabled {
relay.management_enabled = v;
}
if let Some(v) = payload.blossom_enabled {
relay.blossom_enabled = v;
}
if let Some(v) = payload.livekit_enabled {
relay.livekit_enabled = v;
}
if let Some(v) = payload.push_enabled {
relay.push_enabled = v;
}
let relay = prepare_relay(&api, relay)?;
let plan_changed = requested_plan
.as_deref()
.is_some_and(|requested| requested != current_plan);
if plan_changed {
let selected_plan = api
.query
.get_plan(&relay.plan)
.expect("validated plan must exist");
if let Some(limit) = selected_plan.members {
let current_members = fetch_relay_members(&api, &relay)
.await
.map_err(internal)?
.len() as i64;
if current_members > limit {
let message = format!(
"relay has {current_members} members, which exceeds the {} plan limit of {limit}",
selected_plan.name.to_lowercase()
);
return Err(unprocessable("member-limit-exceeded", &message));
}
}
}
api.command
.update_relay(&relay)
.await
.map_err(map_relay_write_error)?;
ok(relay)
}
pub async fn deactivate_relay(
State(api): State<Arc<Api>>,
AuthedPubkey(auth): AuthedPubkey,
Path(id): Path<String>,
) -> ApiResult {
let relay = api.get_relay_or_404(&id).await?;
api.require_admin_or_tenant(&auth, &relay.tenant)?;
if relay.status == RELAY_STATUS_DELINQUENT {
return Err(bad_request("relay-is-delinquent", "relay is delinquent"));
}
if relay.status == RELAY_STATUS_INACTIVE {
return Err(bad_request("relay-is-inactive", "relay is already inactive"));
}
api.command
.deactivate_relay(&relay)
.await
.map_err(internal)?;
ok(())
}
pub async fn reactivate_relay(
State(api): State<Arc<Api>>,
AuthedPubkey(auth): AuthedPubkey,
Path(id): Path<String>,
) -> ApiResult {
let relay = api.get_relay_or_404(&id).await?;
api.require_admin_or_tenant(&auth, &relay.tenant)?;
if relay.status == RELAY_STATUS_DELINQUENT {
return Err(bad_request("relay-is-delinquent", "relay is delinquent"));
}
if relay.status == RELAY_STATUS_ACTIVE {
return Err(bad_request("relay-is-active", "relay is already active"));
}
api.command.activate_relay(&relay).await.map_err(internal)?;
ok(())
}
// --- helpers ----------------------------------------------------------------
async fn fetch_relay_members(api: &Api, relay: &Relay) -> Result<Vec<String>> {
if relay.synced == 0 {
return Ok(Vec::new());
}
api.infra.list_relay_members(&relay.id).await
}
const RESERVED_SUBDOMAINS: [&str; 3] = ["api", "admin", "internal"];
static SUBDOMAIN_RE: LazyLock<Regex> =
LazyLock::new(|| Regex::new(r"^[a-z0-9][a-z0-9-]{1,61}[a-z0-9]$").unwrap());
fn prepare_relay(api: &Api, mut relay: Relay) -> Result<Relay, ApiError> {
if !SUBDOMAIN_RE.is_match(&relay.subdomain)
|| RESERVED_SUBDOMAINS.contains(&relay.subdomain.as_str()) {
return Err(unprocessable("invalid-subdomain", "subdomain is invalid"));
}
let plan = api
.query
.get_plan(&relay.plan)
.ok_or_else(|| unprocessable("invalid-plan", "plan not found"))?;
if (!plan.blossom && relay.blossom_enabled == 1) || (!plan.livekit && relay.livekit_enabled == 1) {
return Err(unprocessable("premium-feature", "feature requires a paid plan"));
}
relay.policy_public_join = parse_bool_default(relay.policy_public_join, 0);
relay.policy_strip_signatures = parse_bool_default(relay.policy_strip_signatures, 0);
relay.groups_enabled = parse_bool_default(relay.groups_enabled, 1);
relay.management_enabled = parse_bool_default(relay.management_enabled, 1);
relay.blossom_enabled = parse_bool_default(relay.blossom_enabled, 0);
relay.livekit_enabled = parse_bool_default(relay.livekit_enabled, 0);
relay.push_enabled = parse_bool_default(relay.push_enabled, 1);
Ok(relay)
}
fn map_relay_write_error(e: anyhow::Error) -> ApiError {
if matches!(map_unique_error(&e), Some("subdomain-exists")) {
unprocessable("subdomain-exists", "subdomain already exists")
} else {
internal(e)
}
}
+55
View File
@@ -0,0 +1,55 @@
use std::sync::Arc;
use axum::{
body::Bytes,
extract::{Path, Query as QueryParams, State},
http::HeaderMap,
};
use serde::Deserialize;
use crate::api::{Api, AuthedPubkey};
use crate::web::{ApiResult, bad_request, internal, ok};
#[derive(Deserialize)]
pub struct StripeSessionParams {
return_url: Option<String>,
}
pub async fn create_stripe_session(
State(api): State<Arc<Api>>,
AuthedPubkey(auth): AuthedPubkey,
Path(pubkey): Path<String>,
QueryParams(params): QueryParams<StripeSessionParams>,
) -> ApiResult {
api.require_admin_or_tenant(&auth, &pubkey)?;
let tenant = api.get_tenant_or_404(&pubkey).await?;
let url = api
.billing
.stripe_create_portal_session(&tenant.stripe_customer_id, params.return_url.as_deref())
.await
.map_err(internal)?;
ok(serde_json::json!({ "url": url }))
}
/// Stripe webhook endpoint. Authenticated via `Stripe-Signature` verification
/// on the raw body, not via NIP-98, so it does not use `AuthedPubkey`.
pub async fn stripe_webhook(
State(api): State<Arc<Api>>,
headers: HeaderMap,
body: Bytes,
) -> ApiResult {
let signature = headers
.get("Stripe-Signature")
.and_then(|v| v.to_str().ok())
.unwrap_or("");
let payload = std::str::from_utf8(&body)
.map_err(|_| bad_request("bad-request", "invalid payload"))?;
api.billing
.handle_webhook(payload, signature)
.await
.map_err(|e| bad_request("webhook-error", &e.to_string()))?;
ok(())
}
+139
View File
@@ -0,0 +1,139 @@
use std::sync::Arc;
use axum::{
Json,
extract::{Path, State},
};
use chrono::Utc;
use serde::{Deserialize, Serialize};
use crate::api::{Api, AuthedPubkey};
use crate::models::Tenant;
use crate::web::{ApiResult, internal, map_unique_error, ok};
#[derive(Serialize)]
pub struct TenantResponse {
pub pubkey: String,
pub nwc_is_set: bool,
pub nwc_error: Option<String>,
pub created_at: i64,
pub stripe_customer_id: String,
pub stripe_subscription_id: Option<String>,
pub past_due_at: Option<i64>,
}
impl From<Tenant> for TenantResponse {
fn from(t: Tenant) -> Self {
TenantResponse {
nwc_is_set: !t.nwc_url.is_empty(),
pubkey: t.pubkey,
nwc_error: t.nwc_error,
created_at: t.created_at,
stripe_customer_id: t.stripe_customer_id,
stripe_subscription_id: t.stripe_subscription_id,
past_due_at: t.past_due_at,
}
}
}
#[derive(Deserialize)]
pub struct UpdateTenantRequest {
pub nwc_url: Option<String>,
}
pub async fn list_tenants(
State(api): State<Arc<Api>>,
AuthedPubkey(auth): AuthedPubkey,
) -> ApiResult {
api.require_admin(&auth)?;
let tenants = api.query.list_tenants().await.map_err(internal)?;
ok(tenants
.into_iter()
.map(TenantResponse::from)
.collect::<Vec<_>>())
}
/// Creates the tenant row for the calling pubkey. Idempotent: if the tenant
/// already exists (including a unique-constraint race) we return the existing
/// row.
pub async fn create_tenant(
State(api): State<Arc<Api>>,
AuthedPubkey(pubkey): AuthedPubkey,
) -> ApiResult {
if let Some(t) = api.query.get_tenant(&pubkey).await.map_err(internal)? {
return ok(TenantResponse::from(t));
}
let stripe_customer_id = api
.billing
.stripe_create_customer(&pubkey)
.await
.map_err(internal)?;
let tenant = Tenant {
pubkey: pubkey.clone(),
created_at: Utc::now().timestamp(),
stripe_customer_id,
..Default::default()
};
match api.command.create_tenant(&tenant).await {
Ok(()) => ok(TenantResponse::from(tenant)),
Err(e) if matches!(map_unique_error(&e), Some("pubkey-exists")) => {
match api.query.get_tenant(&pubkey).await {
Ok(Some(t)) => ok(TenantResponse::from(t)),
Ok(None) => Err(internal("tenant row missing after unique-constraint race")),
Err(e) => Err(internal(e)),
}
}
Err(e) => Err(internal(e)),
}
}
pub async fn get_tenant(
State(api): State<Arc<Api>>,
AuthedPubkey(auth): AuthedPubkey,
Path(pubkey): Path<String>,
) -> ApiResult {
api.require_admin_or_tenant(&auth, &pubkey)?;
let tenant = api.get_tenant_or_404(&pubkey).await?;
ok(TenantResponse::from(tenant))
}
pub async fn update_tenant(
State(api): State<Arc<Api>>,
AuthedPubkey(auth): AuthedPubkey,
Path(pubkey): Path<String>,
Json(payload): Json<UpdateTenantRequest>,
) -> ApiResult {
api.require_admin_or_tenant(&auth, &pubkey)?;
let mut tenant = api.get_tenant_or_404(&pubkey).await?;
// Encrypt tenant's nwc_url at rest
if let Some(nwc_url) = payload.nwc_url {
if nwc_url.is_empty() {
tenant.nwc_url = String::new();
} else {
tenant.nwc_url = api.env.encrypt(&nwc_url).map_err(internal)?;
}
}
api.command.update_tenant(&tenant).await.map_err(internal)?;
ok(TenantResponse::from(tenant))
}
pub async fn list_tenant_relays(
State(api): State<Arc<Api>>,
AuthedPubkey(auth): AuthedPubkey,
Path(pubkey): Path<String>,
) -> ApiResult {
api.require_admin_or_tenant(&auth, &pubkey)?;
let relays = api
.query
.list_relays_for_tenant(&pubkey)
.await
.map_err(internal)?;
ok(relays)
}
+468
View File
@@ -0,0 +1,468 @@
//! A thin async wrapper around the subset of the Stripe REST API this service uses.
//!
//! Nothing here knows about relays, tenants, or our database — it just speaks HTTP
//! to Stripe and hands back `serde_json::Value` (or small typed results). The
//! domain logic lives in [`crate::billing`].
use anyhow::{Result, anyhow};
use hmac::{Hmac, Mac};
use sha2::Sha256;
use std::collections::BTreeMap;
type HmacSha256 = Hmac<Sha256>;
const STRIPE_API: &str = "https://api.stripe.com/v1";
const WEBHOOK_TOLERANCE_SECS: i64 = 300;
/// Error returned by invoice lookups, distinguishing a Stripe 4xx (e.g. "no such
/// invoice") — which callers usually want to surface as a client error — from an
/// internal failure.
#[derive(Debug)]
pub enum InvoiceLookupError {
StripeClient { status: reqwest::StatusCode },
Internal(anyhow::Error),
}
impl std::fmt::Display for InvoiceLookupError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::StripeClient { status } => {
write!(
f,
"stripe invoice lookup failed with status {}",
status.as_u16()
)
}
Self::Internal(error) => write!(f, "{error}"),
}
}
}
impl std::error::Error for InvoiceLookupError {}
impl From<anyhow::Error> for InvoiceLookupError {
fn from(value: anyhow::Error) -> Self {
Self::Internal(value)
}
}
impl From<reqwest::Error> for InvoiceLookupError {
fn from(value: reqwest::Error) -> Self {
Self::Internal(value.into())
}
}
/// A Stripe webhook event with its signature already verified.
#[derive(serde::Deserialize)]
pub struct Event {
#[serde(rename = "type")]
pub event_type: String,
pub data: EventData,
}
#[derive(serde::Deserialize)]
pub struct EventData {
pub object: serde_json::Value,
}
#[derive(Clone)]
pub struct Stripe {
pub(crate) secret_key: String,
pub(crate) webhook_secret: String,
http: reqwest::Client,
}
impl Stripe {
pub fn new(secret_key: String, webhook_secret: String) -> Self {
Self {
secret_key,
webhook_secret,
http: reqwest::Client::new(),
}
}
// --- Customers ---
/// Creates a customer with the given display name, tagging it with the tenant
/// pubkey in metadata. Idempotent on the tenant pubkey.
pub async fn create_customer(&self, name: &str, tenant_pubkey: &str) -> Result<String> {
let resp = self
.http
.post(format!("{STRIPE_API}/customers"))
.bearer_auth(&self.secret_key)
.header(
"Idempotency-Key",
self.idempotency_key(&["create_customer", tenant_pubkey]),
)
.form(&[("name", name), ("metadata[tenant_pubkey]", tenant_pubkey)])
.send()
.await?;
let body: serde_json::Value = error_for_status(resp).await?.json().await?;
let customer_id = body["id"]
.as_str()
.ok_or_else(|| anyhow!("missing customer id"))?;
if !customer_id.starts_with("cus_") {
return Err(anyhow!("unexpected customer id format"));
}
Ok(customer_id.to_string())
}
// --- Subscriptions ---
/// Fetches a subscription, returning `None` if Stripe no longer knows about it
/// (so callers can recover from a stale subscription id).
pub async fn get_subscription(
&self,
subscription_id: &str,
) -> Result<Option<serde_json::Value>> {
let resp = self
.http
.get(format!("{STRIPE_API}/subscriptions/{subscription_id}"))
.bearer_auth(&self.secret_key)
.send()
.await?;
if resp.status() == reqwest::StatusCode::NOT_FOUND {
return Ok(None);
}
let body: serde_json::Value = error_for_status(resp).await?.json().await?;
Ok(Some(body))
}
/// Creates a subscription with one item per `(price_id, quantity)` entry, billed
/// automatically. Returns the subscription id and a map from price id to the
/// created subscription item id. Idempotent on the customer and the item set.
pub async fn create_subscription(
&self,
customer_id: &str,
items: &BTreeMap<String, i64>,
) -> Result<(String, BTreeMap<String, String>)> {
let mut form: Vec<(String, String)> = vec![
("customer".to_string(), customer_id.to_string()),
(
"collection_method".to_string(),
"charge_automatically".to_string(),
),
];
let mut key_parts: Vec<String> =
vec!["create_subscription".to_string(), customer_id.to_string()];
for (index, (price_id, quantity)) in items.iter().enumerate() {
form.push((format!("items[{index}][price]"), price_id.clone()));
form.push((format!("items[{index}][quantity]"), quantity.to_string()));
key_parts.push(format!("{price_id}={quantity}"));
}
let key_refs: Vec<&str> = key_parts.iter().map(String::as_str).collect();
let resp = self
.http
.post(format!("{STRIPE_API}/subscriptions"))
.bearer_auth(&self.secret_key)
.header("Idempotency-Key", self.idempotency_key(&key_refs))
.form(&form)
.send()
.await?;
let body: serde_json::Value = error_for_status(resp).await?.json().await?;
let subscription_id = body["id"]
.as_str()
.ok_or_else(|| anyhow!("missing subscription id"))?
.to_string();
let mut price_to_item = BTreeMap::new();
for item in body["items"]["data"]
.as_array()
.ok_or_else(|| anyhow!("missing subscription items"))?
{
let item_id = item["id"]
.as_str()
.ok_or_else(|| anyhow!("missing subscription item id"))?;
let price_id = item["price"]["id"]
.as_str()
.ok_or_else(|| anyhow!("missing subscription item price id"))?;
price_to_item.insert(price_id.to_string(), item_id.to_string());
}
Ok((subscription_id, price_to_item))
}
pub async fn create_subscription_item(
&self,
subscription_id: &str,
price_id: &str,
quantity: i64,
) -> Result<String> {
let quantity = quantity.to_string();
let resp = self
.http
.post(format!("{STRIPE_API}/subscription_items"))
.bearer_auth(&self.secret_key)
.header(
"Idempotency-Key",
self.idempotency_key(&["create_subscription_item", subscription_id, price_id]),
)
.form(&[
("subscription", subscription_id),
("price", price_id),
("quantity", quantity.as_str()),
])
.send()
.await?;
let body: serde_json::Value = error_for_status(resp).await?.json().await?;
body["id"]
.as_str()
.map(str::to_string)
.ok_or_else(|| anyhow!("missing subscription item id"))
}
/// Sets a subscription item's quantity. No idempotency key: this is a
/// reconcile-to-desired-state write, and re-applying the same target is a no-op.
pub async fn set_subscription_item_quantity(&self, item_id: &str, quantity: i64) -> Result<()> {
let resp = self
.http
.post(format!("{STRIPE_API}/subscription_items/{item_id}"))
.bearer_auth(&self.secret_key)
.form(&[("quantity", quantity.to_string())])
.send()
.await?;
error_for_status(resp).await?;
Ok(())
}
pub async fn delete_subscription_item(&self, item_id: &str) -> Result<()> {
let resp = self
.http
.delete(format!("{STRIPE_API}/subscription_items/{item_id}"))
.bearer_auth(&self.secret_key)
.send()
.await?;
error_for_status(resp).await?;
Ok(())
}
pub async fn cancel_subscription(&self, subscription_id: &str) -> Result<()> {
let resp = self
.http
.delete(format!("{STRIPE_API}/subscriptions/{subscription_id}"))
.bearer_auth(&self.secret_key)
.send()
.await?;
error_for_status(resp).await?;
Ok(())
}
// --- Invoices ---
/// Returns the `data` array of the customer's invoices.
pub async fn list_invoices(&self, customer_id: &str) -> Result<serde_json::Value> {
let resp = self
.http
.get(format!("{STRIPE_API}/invoices"))
.bearer_auth(&self.secret_key)
.query(&[("customer", customer_id)])
.send()
.await?;
let body: serde_json::Value = error_for_status(resp).await?.json().await?;
Ok(body["data"].clone())
}
pub async fn get_invoice(
&self,
invoice_id: &str,
) -> std::result::Result<serde_json::Value, InvoiceLookupError> {
let resp = self
.http
.get(format!("{STRIPE_API}/invoices/{invoice_id}"))
.bearer_auth(&self.secret_key)
.send()
.await?;
if resp.status().is_client_error() {
return Err(InvoiceLookupError::StripeClient {
status: resp.status(),
});
}
let body: serde_json::Value = error_for_status(resp).await?.json().await?;
Ok(body)
}
pub async fn pay_invoice(&self, invoice_id: &str) -> Result<()> {
let resp = self
.http
.post(format!("{STRIPE_API}/invoices/{invoice_id}/pay"))
.bearer_auth(&self.secret_key)
.header(
"Idempotency-Key",
self.idempotency_key(&["pay_invoice", invoice_id]),
)
.send()
.await?;
error_for_status(resp).await?;
Ok(())
}
/// Marks an invoice paid out of band — used when we've collected payment over
/// Lightning rather than through Stripe.
pub async fn pay_invoice_out_of_band(&self, invoice_id: &str) -> Result<()> {
let resp = self
.http
.post(format!("{STRIPE_API}/invoices/{invoice_id}/pay"))
.bearer_auth(&self.secret_key)
.header(
"Idempotency-Key",
self.idempotency_key(&["pay_invoice_oob", invoice_id]),
)
.form(&[("paid_out_of_band", "true")])
.send()
.await?;
error_for_status(resp).await?;
Ok(())
}
pub async fn preview_upcoming_invoice(
&self,
customer_id: &str,
subscription_id: Option<&str>,
) -> Result<serde_json::Value> {
let mut req = self
.http
.get(format!("{STRIPE_API}/invoices/upcoming"))
.bearer_auth(&self.secret_key)
.query(&[("customer", customer_id)]);
if let Some(subscription_id) = subscription_id {
req = req.query(&[("subscription", subscription_id)]);
}
let body: serde_json::Value = error_for_status(req.send().await?).await?.json().await?;
Ok(body)
}
// --- Payment methods ---
pub async fn has_payment_method(&self, customer_id: &str) -> Result<bool> {
let resp = self
.http
.get(format!("{STRIPE_API}/payment_methods"))
.bearer_auth(&self.secret_key)
.query(&[("customer", customer_id), ("type", "card")])
.send()
.await?;
let body: serde_json::Value = error_for_status(resp).await?.json().await?;
Ok(body["data"].as_array().is_some_and(|a| !a.is_empty()))
}
// --- Billing portal ---
pub async fn create_portal_session(
&self,
customer_id: &str,
return_url: Option<&str>,
) -> Result<String> {
let mut params = vec![("customer", customer_id.to_string())];
if let Some(url) = return_url {
params.push(("return_url", url.to_string()));
}
let resp = self
.http
.post(format!("{STRIPE_API}/billing_portal/sessions"))
.bearer_auth(&self.secret_key)
.form(&params)
.send()
.await?;
let body: serde_json::Value = error_for_status(resp).await?.json().await?;
body["url"]
.as_str()
.map(str::to_string)
.ok_or_else(|| anyhow!("missing portal session url"))
}
// --- Webhooks ---
/// Verifies the `Stripe-Signature` header against the configured webhook secret
/// (including the timestamp tolerance check) and parses the event body.
pub fn construct_event(&self, payload: &str, sig_header: &str) -> Result<Event> {
self.verify_webhook_signature(payload, sig_header)?;
Ok(serde_json::from_str(payload)?)
}
fn verify_webhook_signature(&self, payload: &str, sig_header: &str) -> Result<()> {
let mut timestamp = None;
let mut signature = None;
for part in sig_header.split(',') {
if let Some(t) = part.strip_prefix("t=") {
timestamp = Some(t);
} else if let Some(v) = part.strip_prefix("v1=") {
signature = Some(v);
}
}
let timestamp = timestamp.ok_or_else(|| anyhow!("missing webhook timestamp"))?;
let signature = signature.ok_or_else(|| anyhow!("missing webhook signature"))?;
let signed_payload = format!("{timestamp}.{payload}");
let mut mac = HmacSha256::new_from_slice(self.webhook_secret.as_bytes())
.map_err(|e| anyhow!("invalid webhook secret: {e}"))?;
mac.update(signed_payload.as_bytes());
let expected = hex::encode(mac.finalize().into_bytes());
if expected != signature {
return Err(anyhow!("webhook signature mismatch"));
}
let ts: i64 = timestamp
.parse()
.map_err(|_| anyhow!("bad webhook timestamp"))?;
let now = chrono::Utc::now().timestamp();
if (now - ts).abs() > WEBHOOK_TOLERANCE_SECS {
return Err(anyhow!("webhook timestamp outside tolerance"));
}
Ok(())
}
// --- Internals ---
/// Derives a stable idempotency key by HMAC-ing `parts` with the secret key.
fn idempotency_key(&self, parts: &[&str]) -> String {
let mut mac = HmacSha256::new_from_slice(self.secret_key.as_bytes())
.expect("HMAC accepts any key length");
for (i, part) in parts.iter().enumerate() {
if i > 0 {
mac.update(b":");
}
mac.update(part.as_bytes());
}
hex::encode(mac.finalize().into_bytes())
}
}
/// Like [`reqwest::Response::error_for_status`], but on a 4xx/5xx response it reads
/// the body and folds Stripe's JSON error payload (`error.message`/`code`/`param`)
/// into the returned error, so callers get an actionable message instead of a bare
/// "400 Bad Request" with only the URL.
async fn error_for_status(resp: reqwest::Response) -> Result<reqwest::Response> {
let status = resp.status();
if !status.is_client_error() && !status.is_server_error() {
return Ok(resp);
}
let url = resp.url().clone();
let body = resp.text().await.unwrap_or_default();
let detail = serde_json::from_str::<serde_json::Value>(&body)
.ok()
.and_then(|json| {
let error = &json["error"];
let message = error["message"].as_str()?.to_string();
let mut detail = message;
if let Some(code) = error["type"].as_str().or_else(|| error["code"].as_str()) {
detail.push_str(&format!(" [{code}]"));
}
if let Some(param) = error["param"].as_str() {
detail.push_str(&format!(" (param: {param})"));
}
Some(detail)
})
.unwrap_or_else(|| {
if body.trim().is_empty() {
"<empty response body>".to_string()
} else {
body
}
});
Err(anyhow!(
"Stripe API request to {url} failed with status {status}: {detail}"
))
}
+55
View File
@@ -0,0 +1,55 @@
use anyhow::{Result, anyhow};
use nwc::prelude::{
LookupInvoiceRequest, MakeInvoiceRequest, NWC, NostrWalletConnectURI, PayInvoiceRequest,
TransactionState,
};
#[derive(Clone)]
pub struct Wallet {
url: NostrWalletConnectURI,
}
impl Wallet {
pub fn from_url(url: &str) -> Result<Self> {
let url = url
.parse::<NostrWalletConnectURI>()
.map_err(|_| anyhow!("invalid NWC URL"))?;
Ok(Self { url })
}
pub async fn make_invoice(&self, amount_msats: u64, description: &str) -> Result<String> {
let nwc = NWC::new(self.url.clone());
let result = nwc
.make_invoice(MakeInvoiceRequest {
amount: amount_msats,
description: Some(description.to_string()),
description_hash: None,
expiry: None,
})
.await;
nwc.shutdown().await;
Ok(result
.map_err(|e| anyhow!("failed to create invoice: {e}"))?
.invoice)
}
pub async fn pay_invoice(&self, bolt11: String) -> Result<()> {
let nwc = NWC::new(self.url.clone());
let result = nwc.pay_invoice(PayInvoiceRequest::new(bolt11)).await;
nwc.shutdown().await;
result.map(|_| ()).map_err(|e| anyhow!("{e}"))
}
pub async fn is_settled(&self, bolt11: &str) -> Result<bool> {
let nwc = NWC::new(self.url.clone());
let result = nwc
.lookup_invoice(LookupInvoiceRequest {
payment_hash: None,
invoice: Some(bolt11.to_string()),
})
.await;
nwc.shutdown().await;
let response = result.map_err(|e| anyhow!("failed to lookup invoice: {e}"))?;
Ok(response.state == Some(TransactionState::Settled) || response.settled_at.is_some())
}
}
+125
View File
@@ -0,0 +1,125 @@
//! General-purpose HTTP helpers shared across route handlers.
//!
//! Success builders (`res`, `ok`, `created`) return [`ApiResult`] so they
//! can sit at the end of a handler without an `Ok(..)` wrap. Error builders
//! (`err`, `not_found`, `forbidden`, …) return [`ApiError`] so they compose
//! with `.map_err(...)` and with explicit `Err(...)` returns.
use std::fmt::Display;
use axum::{
Json,
http::StatusCode,
response::{IntoResponse, Response},
};
use serde::Serialize;
pub struct ApiError(pub Box<Response>);
impl IntoResponse for ApiError {
fn into_response(self) -> Response {
*self.0
}
}
impl From<Response> for ApiError {
fn from(r: Response) -> Self {
Self(Box::new(r))
}
}
pub type ApiResult = Result<Response, ApiError>;
#[derive(Serialize)]
pub struct DataResponse<T: Serialize> {
pub data: T,
pub code: &'static str,
}
#[derive(Serialize)]
pub struct ErrorResponse {
pub error: String,
pub code: String,
}
// --- success builders (return ApiResult) ------------------------------------
pub fn res<T: Serialize>(status: StatusCode, data: T) -> ApiResult {
Ok((status, Json(DataResponse { data, code: "ok" })).into_response())
}
pub fn ok<T: Serialize>(data: T) -> ApiResult {
res(StatusCode::OK, data)
}
pub fn created<T: Serialize>(data: T) -> ApiResult {
res(StatusCode::CREATED, data)
}
// --- error builders (return ApiError) ---------------------------------------
pub fn err(status: StatusCode, code: &str, message: &str) -> ApiError {
(
status,
Json(ErrorResponse {
error: message.to_string(),
code: code.to_string(),
}),
)
.into_response()
.into()
}
pub fn unauthorized(reason: impl Display) -> ApiError {
err(StatusCode::UNAUTHORIZED, "unauthorized", &reason.to_string())
}
pub fn forbidden(message: &str) -> ApiError {
err(StatusCode::FORBIDDEN, "forbidden", message)
}
pub fn not_found(message: &str) -> ApiError {
err(StatusCode::NOT_FOUND, "not-found", message)
}
pub fn bad_request(code: &str, message: &str) -> ApiError {
err(StatusCode::BAD_REQUEST, code, message)
}
pub fn unprocessable(code: &str, message: &str) -> ApiError {
err(StatusCode::UNPROCESSABLE_ENTITY, code, message)
}
pub fn internal(reason: impl Display) -> ApiError {
err(
StatusCode::INTERNAL_SERVER_ERROR,
"internal",
&reason.to_string(),
)
}
// --- misc utilities ---------------------------------------------------------
pub fn parse_bool_default(value: i64, default: i64) -> i64 {
if value == 0 || value == 1 {
value
} else {
default
}
}
/// Recognize sqlite UNIQUE constraint violations on known columns so the
/// caller can translate them into 422 responses instead of opaque 500s.
pub fn map_unique_error(err: &anyhow::Error) -> Option<&'static str> {
let sqlx_err = err.downcast_ref::<sqlx::Error>()?;
let sqlx::Error::Database(db_err) = sqlx_err else {
return None;
};
if db_err.message().contains("pubkey") {
return Some("pubkey-exists");
}
if db_err.message().contains("subdomain") {
return Some("subdomain-exists");
}
None
}
-31
View File
@@ -1,31 +0,0 @@
use axum::{Json, Router, routing::get};
use backend::billing::{fetch_btc_spot_price_from_base, fiat_minor_to_msats_from_quote};
#[tokio::test]
async fn quote_endpoint_can_be_stubbed_deterministically() {
async fn spot() -> Json<serde_json::Value> {
Json(serde_json::json!({ "data": { "amount": "50000.00" } }))
}
let app = Router::new().route("/v2/prices/BTC-USD/spot", get(spot));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind test server");
let addr = listener.local_addr().expect("get local addr");
tokio::spawn(async move {
axum::serve(listener, app).await.expect("serve quote stub");
});
let client = reqwest::Client::new();
let base = format!("http://{addr}/v2/prices");
let btc_price = fetch_btc_spot_price_from_base(&client, &base, "USD")
.await
.expect("fetch stubbed quote");
assert_eq!(btc_price, 50_000.0);
let msats =
fiat_minor_to_msats_from_quote(100, "USD", btc_price).expect("convert quoted fiat amount");
assert_eq!(msats, 2_000_000);
}
+1 -1
View File
@@ -40,7 +40,7 @@ export default function PaymentSetup(props: PaymentSetupProps) {
setRedirecting(true)
setError("")
try {
const { url } = await createPortalSession(account()!.pubkey)
const { url } = await createPortalSession(account()!.pubkey, window.location.href)
window.location.href = url
} catch (e) {
setError(e instanceof Error ? e.message : "Failed to open billing portal")
+3 -2
View File
@@ -253,8 +253,9 @@ export function reactivateRelay(id: string) {
return callApi<undefined, void>("POST", `/relays/${id}/reactivate`)
}
export function createPortalSession(pubkey: string) {
return callApi<undefined, { url: string }>("GET", `/tenants/${pubkey}/stripe/session`)
export function createPortalSession(pubkey: string, returnUrl?: string) {
const query = returnUrl ? `?return_url=${encodeURIComponent(returnUrl)}` : ""
return callApi<undefined, { url: string }>("GET", `/tenants/${pubkey}/stripe/session${query}`)
}
export function getInvoiceBolt11(invoiceId: string) {
+9 -4
View File
@@ -1,5 +1,5 @@
import { createSignal } from "solid-js"
import { updateRelayById, deactivateRelayById, reactivateRelayById, getLatestOpenInvoice, type Relay } from "@/lib/hooks"
import { updateRelayById, deactivateRelayById, reactivateRelayById, getLatestOpenInvoice, tenantNeedsPaymentSetup, type Relay } from "@/lib/hooks"
import { setToastMessage } from "@/components/Toast"
import type { Invoice, PlanId } from "@/lib/api"
@@ -31,6 +31,7 @@ export default function useRelayToggles(
) {
const [busy, setBusy] = createSignal(false)
const [pendingInvoice, setPendingInvoice] = createSignal<Invoice | undefined>()
const [pendingPaymentSetup, setPendingPaymentSetup] = createSignal(false)
async function updateRelay(next: Relay, previous: Relay) {
mutate(next)
@@ -101,8 +102,12 @@ export default function useRelayToggles(
}
if (plan !== "free") {
const invoice = await getLatestOpenInvoice()
if (invoice) setPendingInvoice(invoice)
const needsSetup = await tenantNeedsPaymentSetup()
if (needsSetup) {
const invoice = await getLatestOpenInvoice()
if (invoice) setPendingInvoice(invoice)
setPendingPaymentSetup(true)
}
}
}
@@ -116,5 +121,5 @@ export default function useRelayToggles(
onToggleLivekitSupport: () => toggle("livekit_enabled", relay()?.plan !== "free"),
}
return { busy, handleDeactivate, handleReactivate, handleUpdatePlan, pendingInvoice, clearPendingInvoice: () => setPendingInvoice(undefined), toggles }
return { busy, handleDeactivate, handleReactivate, handleUpdatePlan, pendingInvoice, clearPendingInvoice: () => setPendingInvoice(undefined), pendingPaymentSetup, clearPendingPaymentSetup: () => setPendingPaymentSetup(false), toggles }
}
+1 -1
View File
@@ -49,7 +49,7 @@ export default function Account() {
async function openPortal() {
setPortalLoading(true)
try {
const { url } = await createPortalSession(account()!.pubkey)
const { url } = await createPortalSession(account()!.pubkey, window.location.href)
window.location.href = url
} catch (e) {
setError(e instanceof Error ? e.message : "Failed to open billing portal")
+9 -2
View File
@@ -1,5 +1,5 @@
import { useParams } from "@solidjs/router"
import { createMemo, createResource, createSignal, Show } from "solid-js"
import { createEffect, createMemo, createResource, createSignal, Show } from "solid-js"
import BackLink from "@/components/BackLink"
import PageContainer from "@/components/PageContainer"
import PaymentDialog from "@/components/PaymentDialog"
@@ -28,13 +28,20 @@ export default function RelayDetail() {
})
const loading = useMinLoading(() => relay.loading && !relay())
const [activity] = useRelayActivity(relayId)
const { busy, handleDeactivate, handleReactivate, handleUpdatePlan, pendingInvoice, clearPendingInvoice, toggles } = useRelayToggles(relayId, relay, { refetch, mutate })
const { busy, handleDeactivate, handleReactivate, handleUpdatePlan, pendingInvoice, clearPendingInvoice, pendingPaymentSetup, clearPendingPaymentSetup, toggles } = useRelayToggles(relayId, relay, { refetch, mutate })
const [tenant, { refetch: refetchTenant }] = useTenant()
const [paymentSetupOpen, setPaymentSetupOpen] = createSignal(false)
const [invoiceDialogOpen, setInvoiceDialogOpen] = createSignal(false)
const [paymentBannerDismissed, setPaymentBannerDismissed] = createSignal(false)
createEffect(() => {
if (pendingPaymentSetup() && !pendingInvoice()) {
setPaymentSetupOpen(true)
clearPendingPaymentSetup()
}
})
const isPaidRelay = createMemo(() => {
const r = relay()
if (!r) return false
+22 -6
View File
@@ -3,13 +3,15 @@ import { useNavigate } from "@solidjs/router"
import BackLink from "@/components/BackLink"
import PageContainer from "@/components/PageContainer"
import PaymentDialog from "@/components/PaymentDialog"
import PaymentSetup from "@/components/PaymentSetup"
import RelayForm, { type RelayFormValues } from "@/components/RelayForm"
import { createRelayForActiveTenant, getLatestOpenInvoice } from "@/lib/hooks"
import { createRelayForActiveTenant, getLatestOpenInvoice, tenantNeedsPaymentSetup } from "@/lib/hooks"
import type { Invoice } from "@/lib/api"
export default function RelayNew() {
const navigate = useNavigate()
const [pendingInvoice, setPendingInvoice] = createSignal<Invoice | undefined>()
const [paymentSetupOpen, setPaymentSetupOpen] = createSignal(false)
let createdRelayId = ""
async function handleSubmit(values: RelayFormValues) {
@@ -17,9 +19,14 @@ export default function RelayNew() {
createdRelayId = relay.id
if (values.plan !== "free") {
const invoice = await getLatestOpenInvoice()
if (invoice) {
setPendingInvoice(invoice)
const needsSetup = await tenantNeedsPaymentSetup()
if (needsSetup) {
const invoice = await getLatestOpenInvoice()
if (invoice) {
setPendingInvoice(invoice)
return
}
setPaymentSetupOpen(true)
return
}
}
@@ -27,8 +34,13 @@ export default function RelayNew() {
navigate(`/relays/${relay.id}`)
}
function handleDialogClose() {
function handleInvoiceClose() {
setPendingInvoice(undefined)
setPaymentSetupOpen(true)
}
function handleSetupClose() {
setPaymentSetupOpen(false)
navigate(`/relays/${createdRelayId}`)
}
@@ -47,10 +59,14 @@ export default function RelayNew() {
<PaymentDialog
invoice={inv()}
open={true}
onClose={handleDialogClose}
onClose={handleInvoiceClose}
/>
)}
</Show>
<PaymentSetup
open={paymentSetupOpen()}
onClose={handleSetupClose}
/>
</PageContainer>
)
}
+4 -1
View File
@@ -5,6 +5,9 @@ dev:
cd frontend && bun dev &
wait
dev-backend:
cd backend && onchange src -ik -- bash -c 'RUST_LOG=backend=info cargo run'
dev-frontend:
cd frontend && bun run dev
@@ -27,7 +30,7 @@ build-backend:
cd backend && cargo build
build-frontend:
cd frontend && bun run build
cd frontend && bun i && bun run build
fmt: fmt-backend
-2
View File
@@ -1,2 +0,0 @@
- [ ] Fix billing by using stripe as a backend to do proration, then mark invoices paid manually when using bitcoin.
- [ ] Send a payment link instead of an invoice so we can generate/pay on the fly