Compare commits

..

2 Commits

7 changed files with 322 additions and 74 deletions
+1 -1
View File
@@ -4,7 +4,7 @@ PORT=3000
ALLOW_ORIGINS= # Optional comma-separated allowed CORS origins; empty = permissive
# Auth
ADMINS= # Comma-separated admin keys (hex pubkey or npub)
ADMINS= # Comma-separated hex pubkeys with admin access
# Database
DATABASE_URL=sqlite://data/caravel.db
+1 -1
View File
@@ -35,7 +35,7 @@ Environment variables:
| `DATABASE_URL` | SQLite URL. Relative paths are resolved under `backend/`. | `sqlite://<backend>/data/caravel.db` |
| `HOST` | API bind host (also used for NIP-98 `u` host check) | `127.0.0.1` |
| `PORT` | API bind port | `3000` |
| `ADMINS` | Comma-separated admin pubkeys (`hex` or `npub`) | _optional_ |
| `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_ |
+1 -1
View File
@@ -12,7 +12,7 @@ CREATE TABLE IF NOT EXISTS tenant (
nwc_url TEXT NOT NULL DEFAULT '',
nwc_error TEXT,
created_at INTEGER NOT NULL,
stripe_customer_id TEXT NOT NULL DEFAULT '',
stripe_customer_id TEXT NOT NULL,
stripe_subscription_id TEXT,
past_due_at INTEGER
);
+45 -44
View File
@@ -9,7 +9,7 @@ use axum::{
routing::{get, post},
};
use base64::Engine;
use nostr_sdk::{Event, JsonUtil, Keys, Kind, PublicKey};
use nostr_sdk::{Event, JsonUtil, Kind};
use serde::{Deserialize, Serialize};
use crate::billing::Billing;
@@ -90,7 +90,8 @@ impl Api {
let admins = std::env::var("ADMINS")
.unwrap_or_default()
.split(',')
.filter_map(parse_admin_pubkey)
.map(|v| v.trim().to_lowercase())
.filter(|v| !v.is_empty())
.collect();
Self {
host,
@@ -254,24 +255,6 @@ impl Api {
}
}
fn parse_admin_pubkey(value: &str) -> Option<String> {
let value = value.trim();
if value.is_empty() {
return None;
}
if let Ok(pubkey) = PublicKey::parse(value) {
return Some(pubkey.to_hex());
}
// Allow nsec values by deriving their pubkey so admin matching still works.
if let Ok(keys) = Keys::parse(value) {
return Some(keys.public_key().to_hex());
}
None
}
fn ok<T: Serialize>(status: StatusCode, data: T) -> Response {
(status, Json(OkResponse { data, code: "ok" })).into_response()
}
@@ -385,32 +368,50 @@ async fn get_identity(
let pubkey = state.api.extract_auth_pubkey(&headers)?;
let is_admin = state.api.admins.iter().any(|a| a == &pubkey);
// Only create if tenant doesn't exist yet
if let Ok(None) = state.api.query.get_tenant(&pubkey).await {
// TODO: Call Stripe API to create a new customer
let stripe_customer_id = String::new();
// Ensure tenant exists.
match state.api.query.get_tenant(&pubkey).await {
Ok(Some(_)) => {}
Ok(None) => {
let stripe_customer_id = match state.api.billing.stripe_create_customer(&pubkey).await {
Ok(id) => id,
Err(e) => {
return Ok(err(
StatusCode::INTERNAL_SERVER_ERROR,
"stripe-customer-create-failed",
&e.to_string(),
));
}
};
let tenant = Tenant {
pubkey: pubkey.clone(),
nwc_url: String::new(),
nwc_error: None,
created_at: now_ts(),
stripe_customer_id,
stripe_subscription_id: None,
past_due_at: None,
};
let tenant = Tenant {
pubkey: pubkey.clone(),
nwc_url: String::new(),
nwc_error: None,
created_at: now_ts(),
stripe_customer_id,
stripe_subscription_id: None,
past_due_at: None,
};
match state.api.command.create_tenant(&tenant).await {
Ok(()) => {}
Err(e) if matches!(map_unique_error(&e), Some("pubkey-exists")) => {}
Err(e) => {
return Ok(err(
StatusCode::INTERNAL_SERVER_ERROR,
"internal",
&e.to_string(),
));
}
};
match state.api.command.create_tenant(&tenant).await {
Ok(()) => {}
Err(e) if matches!(map_unique_error(&e), Some("pubkey-exists")) => {}
Err(e) => {
return Ok(err(
StatusCode::INTERNAL_SERVER_ERROR,
"internal",
&e.to_string(),
));
}
};
}
Err(e) => {
return Ok(err(
StatusCode::INTERNAL_SERVER_ERROR,
"internal",
&e.to_string(),
));
}
}
Ok(ok(StatusCode::OK, IdentityResponse { pubkey, is_admin }))
+177 -20
View File
@@ -1,7 +1,7 @@
use anyhow::{Result, anyhow};
use hmac::{Hmac, Mac};
use nwc::prelude::{
MakeInvoiceRequest, NostrWalletConnectURI, PayInvoiceRequest as NwcPayInvoiceRequest, NWC,
MakeInvoiceRequest, NWC, NostrWalletConnectURI, PayInvoiceRequest as NwcPayInvoiceRequest,
};
use sha2::Sha256;
@@ -75,8 +75,12 @@ impl Billing {
async fn handle_activity(&self, activity: &Activity) -> Result<()> {
let needs_billing_sync = matches!(
activity.activity_type.as_str(),
"create_relay" | "update_relay" | "activate_relay" | "deactivate_relay"
| "fail_relay_sync" | "complete_relay_sync"
"create_relay"
| "update_relay"
| "activate_relay"
| "deactivate_relay"
| "fail_relay_sync"
| "complete_relay_sync"
);
if needs_billing_sync {
@@ -99,7 +103,9 @@ impl Billing {
if relay.plan == "free" {
if let Some(ref item_id) = relay.stripe_subscription_item_id {
self.stripe_delete_subscription_item(item_id).await?;
self.command.delete_relay_subscription_item(&relay.id).await?;
self.command
.delete_relay_subscription_item(&relay.id)
.await?;
}
self.cleanup_empty_subscription(&tenant.pubkey).await?;
return Ok(());
@@ -109,16 +115,16 @@ impl Billing {
if relay.status == "inactive" {
if let Some(ref item_id) = relay.stripe_subscription_item_id {
self.stripe_delete_subscription_item(item_id).await?;
self.command.delete_relay_subscription_item(&relay.id).await?;
self.command
.delete_relay_subscription_item(&relay.id)
.await?;
}
self.cleanup_empty_subscription(&tenant.pubkey).await?;
return Ok(());
}
// Active relay on a paid plan
let plan = Query::list_plans()
.into_iter()
.find(|p| p.id == relay.plan);
let plan = Query::list_plans().into_iter().find(|p| p.id == relay.plan);
let Some(plan) = plan else {
return Ok(());
@@ -128,6 +134,13 @@ impl Billing {
return Ok(());
};
if tenant.stripe_customer_id.trim().is_empty() {
return Err(anyhow!(
"tenant {} has no stripe_customer_id",
tenant.pubkey
));
}
// Ensure subscription exists
if tenant.stripe_subscription_id.is_none() {
let (subscription_id, item_id) = self
@@ -170,7 +183,9 @@ impl Billing {
if let Some(ref subscription_id) = tenant.stripe_subscription_id {
self.stripe_cancel_subscription(subscription_id).await?;
self.command.clear_tenant_subscription(tenant_pubkey).await?;
self.command
.clear_tenant_subscription(tenant_pubkey)
.await?;
}
Ok(())
@@ -240,7 +255,9 @@ impl Billing {
return Err(anyhow!("webhook signature mismatch"));
}
let ts: i64 = timestamp.parse().map_err(|_| anyhow!("bad webhook timestamp"))?;
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"));
@@ -272,9 +289,7 @@ impl Billing {
match self.nwc_pay_invoice(amount_due, &tenant.nwc_url).await {
Ok(()) => {
self.stripe_pay_invoice_out_of_band(invoice_id).await?;
self.command
.clear_tenant_nwc_error(&tenant.pubkey)
.await?;
self.command.clear_tenant_nwc_error(&tenant.pubkey).await?;
return Ok(());
}
Err(e) => {
@@ -442,6 +457,37 @@ impl Billing {
Ok((invoice, tenant))
}
pub async fn stripe_create_customer(&self, tenant_pubkey: &str) -> Result<String> {
if self.stripe_secret_key.trim().is_empty() {
return Err(anyhow!("missing STRIPE_SECRET_KEY"));
}
let short_pubkey: String = tenant_pubkey.chars().take(12).collect();
let display_name = format!("Caravel tenant {short_pubkey}");
let resp = self
.http
.post(format!("{STRIPE_API}/customers"))
.bearer_auth(&self.stripe_secret_key)
.form(&[
("name", display_name.as_str()),
("metadata[tenant_pubkey]", tenant_pubkey),
])
.send()
.await?;
let body: serde_json::Value = resp.error_for_status()?.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())
}
pub async fn stripe_list_invoices(&self, customer_id: &str) -> Result<serde_json::Value> {
let resp = self
.http
@@ -470,7 +516,9 @@ impl Billing {
pub async fn create_bolt11(&self, amount_due_cents: i64) -> Result<String> {
let amount_msats = (amount_due_cents as u64) * 1000; // placeholder conversion
let system_uri: NostrWalletConnectURI = self.nwc_url.parse()
let system_uri: NostrWalletConnectURI = self
.nwc_url
.parse()
.map_err(|_| anyhow!("invalid system NWC URL"))?;
let system_nwc = NWC::new(system_uri);
@@ -550,10 +598,7 @@ impl Billing {
.http
.post(format!("{STRIPE_API}/subscription_items"))
.bearer_auth(&self.stripe_secret_key)
.form(&[
("subscription", subscription_id),
("price", price_id),
])
.form(&[("subscription", subscription_id), ("price", price_id)])
.send()
.await?;
@@ -649,7 +694,9 @@ impl Billing {
let amount_msats = (amount_due_cents as u64) * 1000; // placeholder conversion, actual rate would come from exchange
// Create a bolt11 invoice using the system wallet (self.nwc_url)
let system_uri: NostrWalletConnectURI = self.nwc_url.parse()
let system_uri: NostrWalletConnectURI = self
.nwc_url
.parse()
.map_err(|_| anyhow!("invalid system NWC URL"))?;
let system_nwc = NWC::new(system_uri);
@@ -668,7 +715,8 @@ impl Billing {
system_nwc.shutdown().await;
// Pay the bolt11 invoice using the tenant's wallet
let tenant_uri: NostrWalletConnectURI = tenant_nwc_url.parse()
let tenant_uri: NostrWalletConnectURI = tenant_nwc_url
.parse()
.map_err(|_| anyhow!("invalid tenant NWC URL"))?;
let tenant_nwc = NWC::new(tenant_uri);
@@ -684,3 +732,112 @@ impl Billing {
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use sqlx::SqlitePool;
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
use std::str::FromStr;
use crate::models::Activity;
async fn test_pool() -> SqlitePool {
let connect_options = SqliteConnectOptions::from_str("sqlite::memory:")
.expect("valid sqlite memory url")
.create_if_missing(true);
let pool = SqlitePoolOptions::new()
.max_connections(1)
.connect_with(connect_options)
.await
.expect("connect sqlite memory db");
sqlx::migrate!("./migrations")
.run(&pool)
.await
.expect("run migrations");
pool
}
fn test_billing(pool: SqlitePool, stripe_secret_key: &str) -> Billing {
Billing {
nwc_url: String::new(),
stripe_secret_key: stripe_secret_key.to_string(),
stripe_webhook_secret: String::new(),
http: reqwest::Client::new(),
query: Query::new(pool.clone()),
command: Command::new(pool),
robot: Robot::test_stub(),
}
}
#[tokio::test]
async fn stripe_create_customer_requires_secret_key() {
let pool = test_pool().await;
let billing = test_billing(pool, "");
let err = billing
.stripe_create_customer("tenant_pubkey")
.await
.expect_err("missing key should fail before HTTP call");
assert!(
err.to_string().contains("missing STRIPE_SECRET_KEY"),
"unexpected error: {err}"
);
}
#[tokio::test]
async fn sync_relay_subscription_fails_for_empty_tenant_customer_id() {
let pool = test_pool().await;
sqlx::query(
"INSERT INTO tenant (pubkey, nwc_url, created_at, stripe_customer_id)
VALUES (?, ?, ?, ?)",
)
.bind("tenant_pubkey")
.bind("")
.bind(0_i64)
.bind("")
.execute(&pool)
.await
.expect("insert tenant");
sqlx::query(
"INSERT INTO relay (id, tenant, schema, subdomain, plan, status)
VALUES (?, ?, ?, ?, ?, ?)",
)
.bind("relay_1")
.bind("tenant_pubkey")
.bind("relay_1")
.bind("relay-1")
.bind("basic")
.bind("active")
.execute(&pool)
.await
.expect("insert relay");
let billing = test_billing(pool, "sk_test_dummy");
let activity = Activity {
id: "activity_1".to_string(),
tenant: "tenant_pubkey".to_string(),
created_at: 0,
activity_type: "create_relay".to_string(),
resource_type: "relay".to_string(),
resource_id: "relay_1".to_string(),
};
let err = billing
.sync_relay_subscription(&activity)
.await
.expect_err("empty tenant customer id should fail clearly");
assert!(
err.to_string()
.contains("tenant tenant_pubkey has no stripe_customer_id"),
"unexpected error: {err}"
);
}
}
+77 -7
View File
@@ -64,6 +64,10 @@ impl Command {
}
pub async fn create_tenant(&self, tenant: &Tenant) -> Result<()> {
if tenant.stripe_customer_id.trim().is_empty() {
anyhow::bail!("stripe_customer_id is required");
}
let mut tx = self.pool.begin().await?;
sqlx::query(
@@ -77,7 +81,8 @@ impl Command {
.execute(&mut *tx)
.await?;
let activity = Self::insert_activity(&mut tx, "create_tenant", "tenant", &tenant.pubkey).await?;
let activity =
Self::insert_activity(&mut tx, "create_tenant", "tenant", &tenant.pubkey).await?;
tx.commit().await?;
self.emit(activity);
@@ -93,7 +98,8 @@ impl Command {
.execute(&mut *tx)
.await?;
let activity = Self::insert_activity(&mut tx, "update_tenant", "tenant", &tenant.pubkey).await?;
let activity =
Self::insert_activity(&mut tx, "update_tenant", "tenant", &tenant.pubkey).await?;
tx.commit().await?;
self.emit(activity);
@@ -185,7 +191,8 @@ impl Command {
.execute(&mut *tx)
.await?;
let activity = Self::insert_activity(&mut tx, "deactivate_relay", "relay", &relay.id).await?;
let activity =
Self::insert_activity(&mut tx, "deactivate_relay", "relay", &relay.id).await?;
tx.commit().await?;
self.emit(activity);
@@ -216,7 +223,8 @@ impl Command {
.execute(&mut *tx)
.await?;
let activity = Self::insert_activity(&mut tx, "fail_relay_sync", "relay", &relay.id).await?;
let activity =
Self::insert_activity(&mut tx, "fail_relay_sync", "relay", &relay.id).await?;
tx.commit().await?;
self.emit(activity);
@@ -231,7 +239,8 @@ impl Command {
.execute(&mut *tx)
.await?;
let activity = Self::insert_activity(&mut tx, "complete_relay_sync", "relay", relay_id).await?;
let activity =
Self::insert_activity(&mut tx, "complete_relay_sync", "relay", relay_id).await?;
tx.commit().await?;
self.emit(activity);
@@ -246,7 +255,11 @@ impl Command {
Ok(())
}
pub async fn set_relay_subscription_item(&self, relay_id: &str, stripe_subscription_item_id: &str) -> Result<()> {
pub async fn set_relay_subscription_item(
&self,
relay_id: &str,
stripe_subscription_item_id: &str,
) -> Result<()> {
sqlx::query("UPDATE relay SET stripe_subscription_item_id = ? WHERE id = ?")
.bind(stripe_subscription_item_id)
.bind(relay_id)
@@ -255,7 +268,11 @@ impl Command {
Ok(())
}
pub async fn set_tenant_subscription(&self, pubkey: &str, stripe_subscription_id: &str) -> Result<()> {
pub async fn set_tenant_subscription(
&self,
pubkey: &str,
stripe_subscription_id: &str,
) -> Result<()> {
sqlx::query("UPDATE tenant SET stripe_subscription_id = ? WHERE pubkey = ?")
.bind(stripe_subscription_id)
.bind(pubkey)
@@ -307,3 +324,56 @@ impl Command {
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use sqlx::SqlitePool;
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
use std::str::FromStr;
async fn test_pool() -> SqlitePool {
let connect_options = SqliteConnectOptions::from_str("sqlite::memory:")
.expect("valid sqlite memory url")
.create_if_missing(true);
let pool = SqlitePoolOptions::new()
.max_connections(1)
.connect_with(connect_options)
.await
.expect("connect sqlite memory db");
sqlx::migrate!("./migrations")
.run(&pool)
.await
.expect("run migrations");
pool
}
#[tokio::test]
async fn create_tenant_rejects_empty_stripe_customer_id() {
let pool = test_pool().await;
let command = Command::new(pool);
let tenant = Tenant {
pubkey: "tenant_pubkey".to_string(),
nwc_url: String::new(),
nwc_error: None,
created_at: 0,
stripe_customer_id: " ".to_string(),
stripe_subscription_id: None,
past_due_at: None,
};
let err = command
.create_tenant(&tenant)
.await
.expect_err("empty customer id must be rejected");
assert!(
err.to_string().contains("stripe_customer_id is required"),
"unexpected error: {err}"
);
}
}
+20
View File
@@ -254,3 +254,23 @@ 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())),
}
}
}