priyanshu_bharti/caravel
0
0
Fork 0
forked from coracle/caravel
Code Pull Requests Activity

Update backend implementation to fit spec

Browse Source
This commit is contained in:
Jon Staab
2026-03-25 11:43:09 -07:00
parent 2e0740910c
commit cb2e37c74a
19 changed files with 1798 additions and 2341 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/api.rs
+525 -699
View File
File diff suppressed because it is too large Load Diff
backend/src/auth.rs
-61
View File
@@ -1,61 +0,0 @@
use anyhow::{Result, anyhow};
use base64::Engine;
use base64::engine::general_purpose;
use std::str::FromStr;
use nostr_sdk::JsonUtil;
use nostr_sdk::nostr::key::PublicKey;
use nostr_sdk::nostr::nips::nip98::HttpMethod;
use nostr_sdk::nostr::types::url::Url;
use nostr_sdk::nostr::{Alphabet, Event, Kind, SingleLetterTag, TagKind, TagStandard};
pub fn verify_nip98(auth_header: &str, url: &str, method: &str) -> Result<PublicKey> {
let url = Url::parse(url)?;
let method = HttpMethod::from_str(&method.to_uppercase())?;
let event = decode_auth_event(auth_header)?;
if event.kind != Kind::HttpAuth {
return Err(anyhow!("authorization event kind mismatch"));
}
let authorized_url =
match event
.tags
.find_standardized(TagKind::SingleLetter(SingleLetterTag::lowercase(
Alphabet::U,
))) {
Some(TagStandard::AbsoluteURL(url)) => url,
_ => return Err(anyhow!("authorization header missing url tag")),
};
let authorized_method = match event.tags.find_standardized(TagKind::Method) {
Some(TagStandard::Method(method)) => method,
_ => return Err(anyhow!("authorization header missing method tag")),
};
if authorized_url != &url || authorized_method != &method {
return Err(anyhow!("authorization does not match request"));
}
event.verify()?;
Ok(event.pubkey)
}
fn decode_auth_event(auth_header: &str) -> Result<Event> {
if auth_header.trim().is_empty() {
return Err(anyhow!("missing authorization header"));
}
let (prefix, encoded) = auth_header
.split_once(' ')
.ok_or_else(|| anyhow!("malformed authorization header"))?;
if prefix != "Nostr" || encoded.is_empty() {
return Err(anyhow!("malformed authorization header"));
}
let decoded = general_purpose::STANDARD.decode(encoded)?;
let json = String::from_utf8(decoded)?;
Ok(Event::from_json(json)?)
}
backend/src/billing.rs
+276 -393
View File
@@ -1,316 +1,263 @@
use anyhow::{Result, anyhow};
use chrono::{DateTime, Months, TimeZone, Utc};
use std::collections::HashMap;
use tokio::time::{Duration, sleep};
use uuid::Uuid;
use crate::models::{Invoice, InvoiceAttempt, InvoiceItem, RelayLifecycleEvent, Tenant};
use crate::notifications::Nip17Notifier;
use anyhow::Result;
use chrono::{DateTime, Datelike, Duration, Months, TimeZone, Utc};
use tokio::sync::Mutex;
use crate::models::{Activity, Invoice, InvoiceItem, Relay, Tenant};
use crate::repo::Repo;
use nostr_sdk::nips::nip47::{self, MakeInvoiceRequest, NostrWalletConnectURI, PayInvoiceRequest};
use nostr_sdk::{Client, Filter, Keys, Kind, Timestamp};
const GRACE_DAYS: i64 = 7;
const DUE_DAYS: i64 = 7;
const WORKER_INTERVAL_SECS: u64 = 300;
// ── service ───────────────────────────────────────────────────────────────────
use crate::robot::Robot;
#[derive(Clone)]
pub struct BillingService {
pub struct Billing {
nwc_url: String,
repo: Repo,
notifier: Nip17Notifier,
platform_nwc_url: String,
robot: Robot,
last_activity_at: std::sync::Arc<Mutex<i64>>,
}
impl BillingService {
pub fn new(repo: Repo, notifier: Nip17Notifier, platform_nwc_url: String) -> Self {
impl Billing {
pub fn new(repo: Repo, robot: Robot) -> Self {
let nwc_url = std::env::var("NWC_URL").unwrap_or_default();
Self {
nwc_url,
repo,
notifier,
platform_nwc_url,
robot,
last_activity_at: std::sync::Arc::new(Mutex::new(0)),
}
}
pub async fn run(self) {
pub async fn start(self) {
let mut interval = tokio::time::interval(std::time::Duration::from_secs(3600));
loop {
if let Err(err) = self.process_once().await {
tracing::error!(error = %err, "billing run failed");
interval.tick().await;
if let Err(e) = self.tick().await {
tracing::error!(error = %e, "billing tick failed");
}
sleep(Duration::from_secs(WORKER_INTERVAL_SECS)).await;
}
}
async fn process_once(&self) -> Result<()> {
pub async fn tick(&self) -> Result<()> {
let mut since_guard = self.last_activity_at.lock().await;
let since = *since_guard;
let activity = self.repo.list_activity(&since, None).await?;
for a in &activity {
if matches!(a.activity_type.as_str(), "relay_created" | "relay_updated" | "relay_activated") {
self.maybe_reset_anchor_for_first_paid_relay(a).await?;
}
*since_guard = (*since_guard).max(a.created_at);
}
drop(since_guard);
let tenants = self.repo.list_tenants().await?;
for tenant in &tenants {
if let Err(err) = self.generate_invoice_if_due(tenant).await {
tracing::error!(tenant = %tenant.pubkey, error = %err, "invoice generation failed");
}
}
for tenant in &tenants {
if let Err(err) = self.collect_outstanding(tenant).await {
tracing::error!(tenant = %tenant.pubkey, error = %err, "collection failed");
}
self.generate_invoice_if_due(tenant).await?;
self.collect_outstanding(tenant).await?;
}
Ok(())
}
// ── invoice generation ────────────────────────────────────────────────────
async fn generate_invoice_if_due(&self, tenant: &Tenant) -> Result<()> {
if tenant.status != "active" {
async fn maybe_reset_anchor_for_first_paid_relay(&self, activity: &Activity) -> Result<()> {
let relay = match self.repo.get_relay(&activity.identifier).await? {
Some(r) => r,
None => return Ok(()),
};
if relay.plan == "free" {
return Ok(());
}
let anchor = ts_to_dt(tenant.billing_anchor_at)?;
let now = Utc::now();
let (period_start, period_end) = current_billing_period(anchor, now);
let relays = self.repo.list_relays(Some(&relay.tenant)).await?;
let paid_active_count = relays
.into_iter()
.filter(|r| r.status == "active" && r.plan != "free")
.count() as i64;
// Only generate once the period has closed
if paid_active_count == 1 {
self.repo
.update_tenant_billing_anchor(&relay.tenant, activity.created_at)
.await?;
}
Ok(())
}
async fn generate_invoice_if_due(&self, tenant: &Tenant) -> Result<()> {
if self.repo.total_pending_invoices_for_tenant(&tenant.pubkey).await? > 0 {
return Ok(());
}
let relays = self.repo.list_relays(Some(&tenant.pubkey)).await?;
let active_paid_relays: Vec<Relay> = relays
.iter()
.filter(|r| r.status == "active" && r.plan != "free")
.cloned()
.collect();
if active_paid_relays.is_empty() {
return Ok(());
}
let now = Utc::now();
let anchor = ts_to_dt(tenant.billing_anchor)?;
let (period_start, period_end) = billing_window(anchor, now);
if now < period_end {
return Ok(());
}
let plans = self.repo.list_plans().await?;
let plan_amount_map: HashMap<String, i64> =
plans.into_iter().map(|p| (p.id, p.sats_per_month)).collect();
let usage_events = self.repo.list_activity(&tenant.billing_anchor, Some(&tenant.pubkey)).await?;
let invoice_id = uuid::Uuid::new_v4().to_string();
let mut items = Vec::new();
let events = self
.repo
.list_lifecycle_events_for_tenant(&tenant.pubkey, dt_to_ts(period_end))
.await?;
for relay in active_paid_relays {
let hours = relay_active_hours_in_window(&relay, &usage_events, period_start, period_end);
if hours <= 0 {
continue;
}
let plan_monthly = self.repo.get_relay_plan_amount_sats(&relay.plan).await?;
if plan_monthly <= 0 {
continue;
}
let invoice_id = Uuid::new_v4().to_string();
let items = compute_invoice_items(
&invoice_id,
&events,
&plan_amount_map,
period_start,
period_end,
);
let sats = ((plan_monthly as f64 / 30.0 / 24.0) * hours as f64).ceil() as i64;
if sats <= 0 {
continue;
}
let total: i64 = items.iter().map(|i| i.amount).sum();
items.push(InvoiceItem {
id: uuid::Uuid::new_v4().to_string(),
invoice: invoice_id.clone(),
relay: relay.id,
sats,
});
}
let total: i64 = items.iter().map(|i| i.sats).sum();
if total == 0 {
return Ok(());
}
let bolt11 = self.make_bolt11(total).await.unwrap_or_default();
let invoice = Invoice {
id: invoice_id.clone(),
tenant: tenant.pubkey.clone(),
amount: total,
status: "pending".to_string(),
created_at: dt_to_ts(now),
bolt11,
period_start: dt_to_ts(period_start),
period_end: dt_to_ts(period_end),
};
let created = self
.repo
.create_invoice_with_items(&invoice, &items)
.await?;
if created {
tracing::info!(tenant = %tenant.pubkey, invoice = %invoice_id, amount = total, "invoice generated");
}
Ok(())
}
// ── collection ────────────────────────────────────────────────────────────
async fn collect_outstanding(&self, tenant: &Tenant) -> Result<()> {
let invoices = self.repo.list_invoices_by_tenant(&tenant.pubkey).await?;
let unpaid: Vec<&Invoice> = invoices
.iter()
.filter(|inv| matches!(inv.status.as_str(), "pending" | "past_due"))
.collect();
if unpaid.is_empty() {
return Ok(());
}
for invoice in &unpaid {
self.attempt_collection(tenant, invoice).await?;
}
// Re-fetch to check if all are now paid; auto-reactivate if so
let invoices_after = self.repo.list_invoices_by_tenant(&tenant.pubkey).await?;
let still_unpaid = invoices_after
.iter()
.any(|inv| matches!(inv.status.as_str(), "pending" | "past_due"));
if !still_unpaid && tenant.status == "suspended" {
let now = now_ts();
self.repo
.update_tenant_status(&tenant.pubkey, "active")
.await?;
self.repo
.reactivate_relays_for_tenant(&tenant.pubkey, now)
.await?;
tracing::info!(tenant = %tenant.pubkey, "tenant reactivated after full balance payment");
}
Ok(())
}
async fn attempt_collection(&self, tenant: &Tenant, invoice: &Invoice) -> Result<()> {
let now = Utc::now();
let created_at = ts_to_dt(invoice.created_at)?;
let due_at = created_at + chrono::Duration::days(DUE_DAYS);
let grace_ends_at = due_at + chrono::Duration::days(GRACE_DAYS);
// Deactivate after grace period expires
if now > grace_ends_at && invoice.status != "past_due" {
let ts = now_ts();
self.repo
.update_tenant_status(&tenant.pubkey, "suspended")
.await?;
self.repo
.suspend_relays_for_tenant(&tenant.pubkey, ts)
.await?;
self.repo
.record_attempt(
&InvoiceAttempt {
id: Uuid::new_v4().to_string(),
invoice: invoice.id.clone(),
run_id: Uuid::new_v4().to_string(),
method: "system".to_string(),
outcome: "failed".to_string(),
error: "grace period expired".to_string(),
created_at: ts,
},
"past_due",
)
.await?;
return Ok(());
}
// Only retry once per 24h
let attempts = self.repo.list_attempts_for_invoice(&invoice.id).await?;
if let Some(last) = attempts.last()
&& last.method != "nip17_dm"
{
let last_at = ts_to_dt(last.created_at)?;
if now - last_at < chrono::Duration::hours(24) {
let bolt11 = match self.make_bolt11(total).await {
Ok(v) => v,
Err(e) => {
tracing::error!(tenant = %tenant.pubkey, error = %e, "bolt11 generation failed");
return Ok(());
}
}
};
let run_id = Uuid::new_v4().to_string();
let invoice = Invoice {
id: invoice_id,
tenant: tenant.pubkey.clone(),
status: "pending".to_string(),
created_at: now.timestamp(),
attempted_at: 0,
error: String::new(),
closed_at: 0,
sent_at: 0,
paid_at: 0,
bolt11,
period_start: period_start.timestamp(),
period_end: period_end.timestamp(),
};
// 1. Try NWC
if !tenant.nwc_url.trim().is_empty() {
match self
.pay_via_nwc(&tenant.nwc_url, &invoice.bolt11)
.await
{
Ok(()) => {
self.repo
.record_attempt(
&attempt(&invoice.id, &run_id, "nwc", "success", ""),
"paid",
)
.await?;
return Ok(());
}
Err(err) => {
tracing::warn!(tenant = %tenant.pubkey, error = %err, "NWC payment failed");
self.repo
.record_attempt(
&attempt(&invoice.id, &run_id, "nwc", "failed", &err.to_string()),
&invoice.status,
)
.await?;
self.repo.create_invoice(&invoice, &items).await?;
Ok(())
}
async fn collect_outstanding(&self, tenant: &Tenant) -> Result<()> {
let invoices = self.repo.list_invoices(Some(&tenant.pubkey)).await?;
let now = now_ts();
for invoice in invoices.into_iter().filter(|i| i.status == "pending") {
if invoice.attempted_at > 0 && now - invoice.attempted_at < 24 * 3600 {
continue;
}
if self.is_bolt11_paid(&invoice.bolt11).await {
self.repo.mark_invoice_paid(&invoice.id).await?;
continue;
}
let mut collected = false;
if !tenant.nwc_url.trim().is_empty() && self.pay_invoice_nwc(&tenant.nwc_url, &invoice.bolt11).await {
self.repo.mark_invoice_paid(&invoice.id).await?;
collected = true;
}
if !collected {
self.repo
.mark_invoice_attempted(&invoice.id, Some("autopay failed or unavailable"))
.await?;
if invoice.sent_at == 0 {
let amount: i64 = self
.repo
.get_invoice_items(&invoice.id)
.await?
.into_iter()
.map(|i| i.sats)
.sum();
let message = format!(
"Invoice {} is due. Amount: {} sats\n{}",
invoice.id, amount, invoice.bolt11
);
if self.robot.send_dm(&tenant.pubkey, &message).await.is_ok() {
self.repo.mark_invoice_sent(&invoice.id).await?;
}
}
}
}
// 2. Try Stripe
if !tenant.stripe_subscription_id.trim().is_empty() {
match self.pay_via_stripe(tenant, invoice).await {
Ok(()) => {
self.repo
.record_attempt(
&attempt(&invoice.id, &run_id, "stripe", "success", ""),
"paid",
)
.await?;
return Ok(());
}
Err(err) => {
tracing::warn!(tenant = %tenant.pubkey, error = %err, "Stripe payment failed");
self.repo
.record_attempt(
&attempt(&invoice.id, &run_id, "stripe", "failed", &err.to_string()),
&invoice.status,
)
.await?;
}
}
}
// 3. Fallback: Lightning invoice shown in-app; send one DM if no auto-pay configured
let dm_sent = self.repo.invoice_dm_sent(&invoice.id).await?;
if !dm_sent {
match self
.send_invoice_dm(tenant, invoice, invoice.bolt11.as_str())
.await
{
Ok(()) => {
self.repo
.record_attempt(
&attempt(&invoice.id, &run_id, "nip17_dm", "sent", ""),
&invoice.status,
)
.await?;
}
Err(err) => {
tracing::warn!(tenant = %tenant.pubkey, error = %err, "NIP-17 DM failed");
}
if now - invoice.created_at >= 7 * 24 * 3600 {
self.repo.mark_invoice_closed(&invoice.id).await?;
}
}
Ok(())
}
// ── payment providers ─────────────────────────────────────────────────────
async fn make_bolt11(&self, amount_sats: i64) -> Result<String> {
if self.platform_nwc_url.trim().is_empty() {
return Err(anyhow!("NWC_URL is required to generate invoices"));
if self.nwc_url.trim().is_empty() {
anyhow::bail!("NWC_URL not configured")
}
let uri = NostrWalletConnectURI::parse(&self.platform_nwc_url)?;
let req = nip47::Request::make_invoice(MakeInvoiceRequest {
amount: (amount_sats as u64) * 1_000,
description: Some("Relay hosting".to_string()),
description_hash: None,
expiry: None,
});
let uri = nostr_sdk::nips::nip47::NostrWalletConnectURI::parse(&self.nwc_url)?;
let req = nostr_sdk::nips::nip47::Request::make_invoice(
nostr_sdk::nips::nip47::MakeInvoiceRequest {
amount: (amount_sats as u64) * 1_000,
description: Some("Caravel relay invoice".to_string()),
description_hash: None,
expiry: None,
},
);
let resp = self.send_nwc_request(&uri, req).await?;
Ok(resp.to_make_invoice()?.invoice)
}
async fn pay_via_nwc(&self, nwc_url: &str, bolt11: &str) -> Result<()> {
let uri = NostrWalletConnectURI::parse(nwc_url)?;
let req = nip47::Request::pay_invoice(PayInvoiceRequest::new(bolt11));
self.send_nwc_request(&uri, req).await?.to_pay_invoice()?;
Ok(())
async fn is_bolt11_paid(&self, _bolt11: &str) -> bool {
false
}
async fn pay_via_stripe(&self, _tenant: &Tenant, _invoice: &Invoice) -> Result<()> {
// TODO: implement Stripe off-session charge using tenant.stripe_subscription_id
Err(anyhow!("Stripe not yet implemented"))
async fn pay_invoice_nwc(&self, nwc_url: &str, bolt11: &str) -> bool {
let uri = match nostr_sdk::nips::nip47::NostrWalletConnectURI::parse(nwc_url) {
Ok(v) => v,
Err(_) => return false,
};
let req = nostr_sdk::nips::nip47::Request::pay_invoice(
nostr_sdk::nips::nip47::PayInvoiceRequest::new(bolt11),
);
self.send_nwc_request(&uri, req)
.await
.and_then(|r| r.to_pay_invoice().map(|_| ()).map_err(anyhow::Error::from))
.is_ok()
}
async fn send_nwc_request(
&self,
uri: &NostrWalletConnectURI,
request: nip47::Request,
) -> Result<nip47::Response> {
uri: &nostr_sdk::nips::nip47::NostrWalletConnectURI,
request: nostr_sdk::nips::nip47::Request,
) -> Result<nostr_sdk::nips::nip47::Response> {
use nostr_sdk::{Client, Filter, Kind, Keys, Timestamp};
let app_keys = Keys::new(uri.secret.clone());
let app_pubkey = app_keys.public_key();
let client = Client::new(app_keys);
@@ -327,183 +274,119 @@ impl BillingService {
.pubkey(app_pubkey)
.since(started_at);
let events = client.fetch_events(filter, Duration::from_secs(10)).await?;
let events = client
.fetch_events(filter, std::time::Duration::from_secs(10))
.await?;
let event = events
.into_iter()
.max_by_key(|e| e.created_at)
.ok_or_else(|| anyhow!("no NWC response received"))?;
.ok_or_else(|| anyhow::anyhow!("no NWC response received"))?;
Ok(nip47::Response::from_event(uri, &event)?)
}
async fn send_invoice_dm(
&self,
tenant: &Tenant,
invoice: &Invoice,
bolt11: &str,
) -> Result<()> {
let due_date = ts_to_dt(invoice.created_at + DUE_DAYS * 86400)?;
let period_start = ts_to_dt(invoice.period_start)?;
let period_end = ts_to_dt(invoice.period_end)?;
let message = format!(
"You have an outstanding invoice of {} sats due by {}.\n\
Period: {} → {}\n\
Pay with Lightning:\n{}",
invoice.amount,
due_date.format("%Y-%m-%d"),
period_start.format("%Y-%m-%d"),
period_end.format("%Y-%m-%d"),
bolt11,
);
self.notifier.send(&tenant.pubkey, &message).await
Ok(nostr_sdk::nips::nip47::Response::from_event(uri, &event)?)
}
}
// ── billing math ──────────────────────────────────────────────────────────────
fn now_ts() -> i64 {
Utc::now().timestamp()
}
/// Given a billing anchor and the current time, return the current billing
/// period [start, end) based on rolling monthly windows from the anchor.
fn current_billing_period(
anchor: DateTime<Utc>,
now: DateTime<Utc>,
) -> (DateTime<Utc>, DateTime<Utc>) {
let mut period_start = anchor;
fn ts_to_dt(ts: i64) -> Result<DateTime<Utc>> {
Utc.timestamp_opt(ts, 0)
.single()
.ok_or_else(|| anyhow::anyhow!("invalid unix timestamp"))
}
fn billing_window(anchor: DateTime<Utc>, now: DateTime<Utc>) -> (DateTime<Utc>, DateTime<Utc>) {
let mut start = anchor;
loop {
let period_end = period_start + Months::new(1);
if now < period_end {
return (period_start, period_end);
let end = start + Months::new(1);
if now < end {
return (start, end);
}
period_start = period_end;
start = end;
}
}
/// Compute per-relay billable sats for a billing period from the lifecycle
/// event log. Rules:
/// - Billing starts at `provisioned`, pauses at `suspended`, resumes at
/// `unsuspended`, stops at `deactivated`.
/// - Only time within [period_start, period_end) counts.
/// - Round each relay's total billable seconds up to the next full hour.
/// - Minimum 1 billable hour per relay per period.
/// - Amount is based on the relay's current plan amount (retroactive within period).
fn compute_invoice_items(
invoice_id: &str,
events: &[RelayLifecycleEvent],
plan_amount_map: &HashMap<String, i64>,
period_start: DateTime<Utc>,
period_end: DateTime<Utc>,
) -> Vec<InvoiceItem> {
// Group events by relay, preserving sort order from the DB (relay, created_at, id)
let mut by_relay: HashMap<&str, Vec<&RelayLifecycleEvent>> = HashMap::new();
for event in events {
by_relay.entry(&event.relay).or_default().push(event);
}
let mut items = Vec::new();
for (relay_id, relay_events) in &by_relay {
// Use the latest plan for this relay (retroactive rate within period)
let Some(latest_event) = relay_events.last() else {
continue;
};
let plan_amount = *plan_amount_map.get(latest_event.plan.as_str()).unwrap_or(&0);
if plan_amount == 0 {
continue;
}
let billable_secs = billable_seconds_in_period(relay_events, period_start, period_end);
if billable_secs == 0 {
continue;
}
// Round up to next full hour, minimum 1 hour
let hours = ((billable_secs as f64) / 3600.0).ceil().max(1.0) as i64;
items.push(InvoiceItem {
id: Uuid::new_v4().to_string(),
invoice: invoice_id.to_string(),
relay: relay_id.to_string(),
amount: hours * plan_amount,
period_start: dt_to_ts(period_start),
period_end: dt_to_ts(period_end),
});
}
items
}
/// Compute total billable seconds for one relay within [period_start, period_end).
/// Replays the full event history to correctly handle events that precede the period.
fn billable_seconds_in_period(
events: &[&RelayLifecycleEvent],
period_start: DateTime<Utc>,
period_end: DateTime<Utc>,
fn relay_active_hours_in_window(
relay: &Relay,
events: &[Activity],
start: DateTime<Utc>,
end: DateTime<Utc>,
) -> i64 {
let mut total_secs: i64 = 0;
let mut billing_start: Option<DateTime<Utc>> = None;
if relay.plan == "free" {
return 0;
}
let mut marks: HashMap<&str, Vec<&Activity>> = HashMap::new();
for event in events {
let Ok(ts) = ts_to_dt(event.created_at) else {
continue;
};
if event.identifier == relay.id {
marks.entry(&relay.id).or_default().push(event);
}
}
match event.event_type.as_str() {
"provisioned" | "unsuspended" => {
if billing_start.is_none() {
billing_start = Some(ts.max(period_start));
let Some(entries) = marks.get(relay.id.as_str()) else {
if relay.status == "active" {
return ((end - start).num_seconds() as f64 / 3600.0).ceil() as i64;
}
return 0;
};
let mut active = relay.status == "active";
let mut cursor = start;
let mut secs = 0i64;
for event in entries.iter().copied() {
let ts = match Utc.timestamp_opt(event.created_at, 0).single() {
Some(v) => v,
None => continue,
};
if ts <= start || ts >= end {
continue;
}
match event.activity_type.as_str() {
"relay_created" | "relay_activated" => {
if !active {
active = true;
cursor = ts;
}
}
"suspended" | "deactivated" => {
if let Some(start) = billing_start.take() {
let end = ts.min(period_end);
if end > start {
total_secs += (end - start).num_seconds();
}
"relay_deactivated" | "relay_sync_failed" => {
if active {
active = false;
secs += (ts - cursor).num_seconds().max(0);
}
}
_ => {}
}
}
// Still billing at period end
if let Some(start) = billing_start
&& period_end > start
{
total_secs += (period_end - start).num_seconds();
if active {
secs += (end - cursor).num_seconds().max(0);
}
total_secs
let hours = (secs as f64 / 3600.0).ceil() as i64;
if hours > 0 { hours } else { 0 }
}
// ── helpers ───────────────────────────────────────────────────────────────────
pub fn now_ts() -> i64 {
Utc::now().timestamp()
#[allow(dead_code)]
fn _same_month(a: DateTime<Utc>, b: DateTime<Utc>) -> bool {
a.year() == b.year() && a.month() == b.month()
}
fn dt_to_ts(dt: DateTime<Utc>) -> i64 {
dt.timestamp()
#[allow(dead_code)]
fn _days_between(a: i64, b: i64) -> i64 {
let da = Utc.timestamp_opt(a, 0).single().unwrap_or_else(Utc::now);
let db = Utc.timestamp_opt(b, 0).single().unwrap_or_else(Utc::now);
(db - da).num_days()
}
fn ts_to_dt(ts: i64) -> Result<DateTime<Utc>> {
Utc.timestamp_opt(ts, 0)
.single()
.ok_or_else(|| anyhow!("invalid unix timestamp: {ts}"))
#[allow(dead_code)]
fn _hours_between(a: DateTime<Utc>, b: DateTime<Utc>) -> i64 {
((b - a).num_seconds() as f64 / 3600.0).ceil() as i64
}
fn attempt(
invoice_id: &str,
run_id: &str,
method: &str,
outcome: &str,
error: &str,
) -> InvoiceAttempt {
InvoiceAttempt {
id: Uuid::new_v4().to_string(),
invoice: invoice_id.to_string(),
run_id: run_id.to_string(),
method: method.to_string(),
outcome: outcome.to_string(),
error: error.to_string(),
created_at: now_ts(),
}
#[allow(dead_code)]
fn _next_day(dt: DateTime<Utc>) -> DateTime<Utc> {
dt + Duration::days(1)
}
backend/src/config.rs
-108
View File
@@ -1,108 +0,0 @@
use std::env;
use std::path::Path;
use std::str::FromStr;
use nostr_sdk::nostr::key::PublicKey;
#[derive(Debug, Clone)]
pub struct Config {
pub database_url: String,
pub host: String,
pub port: u16,
pub admin_pubkeys: Vec<String>,
pub zooid_api_url: String,
pub platform_secret: String,
pub relay_domain: String,
pub platform_nwc_url: String,
pub indexer_relays: Vec<String>,
pub platform_name: String,
pub platform_description: String,
pub platform_picture: String,
pub platform_messaging_relays: Vec<String>,
}
impl Config {
pub fn from_env() -> Self {
let database_url =
env::var("DATABASE_URL").unwrap_or_else(|_| "sqlite://data/caravel.db".to_string());
let database_url = resolve_database_url(database_url);
let host = env::var("HOST").unwrap_or_else(|_| "127.0.0.1".to_string());
let port = env::var("PORT")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(3000);
let admin_pubkeys = env::var("PLATFORM_ADMIN_PUBKEYS")
.unwrap_or_default()
.split(',')
.filter_map(normalize_pubkey)
.filter(|v| !v.is_empty())
.collect::<Vec<_>>();
let zooid_api_url =
env::var("ZOOID_API_URL").unwrap_or_else(|_| "http://127.0.0.1:8032".to_string());
let platform_secret = env::var("PLATFORM_SECRET").unwrap_or_default();
let relay_domain =
env::var("RELAY_DOMAIN").unwrap_or_else(|_| "spaces.coracle.social".to_string());
let platform_nwc_url = env::var("NWC_URL").unwrap_or_default();
let indexer_relays = env::var("NOSTR_INDEXER_RELAYS")
.unwrap_or_default()
.split(',')
.map(|v| v.trim().to_string())
.filter(|v| !v.is_empty())
.collect::<Vec<_>>();
let platform_name = env::var("PLATFORM_NAME").unwrap_or_default();
let platform_description = env::var("PLATFORM_DESCRIPTION").unwrap_or_default();
let platform_picture = env::var("PLATFORM_PICTURE").unwrap_or_default();
let platform_messaging_relays = env::var("PLATFORM_MESSAGING_RELAYS")
.unwrap_or_default()
.split(',')
.map(|v| v.trim().to_string())
.filter(|v| !v.is_empty())
.collect::<Vec<_>>();
Self {
database_url,
host,
port,
admin_pubkeys,
zooid_api_url,
platform_secret,
relay_domain,
platform_nwc_url,
indexer_relays,
platform_name,
platform_description,
platform_picture,
platform_messaging_relays,
}
}
}
fn normalize_pubkey(value: &str) -> Option<String> {
let trimmed = value.trim();
if trimmed.is_empty() {
return None;
}
match PublicKey::from_str(trimmed) {
Ok(pubkey) => Some(pubkey.to_hex()),
Err(_) => Some(trimmed.to_lowercase()),
}
}
fn resolve_database_url(database_url: String) -> String {
const PREFIX: &str = "sqlite://";
if !database_url.starts_with(PREFIX) {
return database_url;
}
let path = &database_url[PREFIX.len()..];
if path.is_empty() || path.starts_with('/') || path == ":memory:" {
return database_url;
}
let absolute = Path::new(env!("CARGO_MANIFEST_DIR")).join(path);
format!(
"sqlite:///{}",
absolute.to_string_lossy().trim_start_matches('/')
)
}
backend/src/db.rs
-45
View File
@@ -1,45 +0,0 @@
use anyhow::Result;
use sqlx::{
SqlitePool, migrate::Migrator, sqlite::SqliteConnectOptions, sqlite::SqlitePoolOptions,
};
use std::fs;
use std::path::Path;
use std::str::FromStr;
static MIGRATOR: Migrator = sqlx::migrate!("./migrations");
pub async fn init_pool(database_url: &str) -> Result<SqlitePool> {
ensure_sqlite_directory(database_url)?;
let options = SqliteConnectOptions::from_str(database_url)?.create_if_missing(true);
let pool = SqlitePoolOptions::new()
.max_connections(5)
.connect_with(options)
.await?;
sqlx::query("PRAGMA journal_mode = WAL;")
.execute(&pool)
.await?;
MIGRATOR.run(&pool).await?;
Ok(pool)
}
fn ensure_sqlite_directory(database_url: &str) -> Result<()> {
let Some(path) = sqlite_path(database_url) else {
return Ok(());
};
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)?;
}
Ok(())
}
fn sqlite_path(database_url: &str) -> Option<&Path> {
const PREFIX: &str = "sqlite://";
let path = database_url.strip_prefix(PREFIX)?;
if path.is_empty() || path == ":memory:" {
return None;
}
Some(Path::new(path))
}
backend/src/infra.rs
+126
View File
@@ -0,0 +1,126 @@
use anyhow::Result;
use tokio::sync::Mutex;
use crate::repo::Repo;
#[derive(Clone)]
pub struct Infra {
api_url: String,
relay_domain: String,
livekit_url: String,
livekit_api_key: String,
livekit_api_secret: String,
repo: Repo,
last_activity_at: std::sync::Arc<Mutex<i64>>,
}
impl Infra {
pub fn new(repo: Repo) -> 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();
Self {
api_url,
relay_domain,
livekit_url,
livekit_api_key,
livekit_api_secret,
repo,
last_activity_at: std::sync::Arc::new(Mutex::new(0)),
}
}
pub async fn start(self) {
let mut interval = tokio::time::interval(std::time::Duration::from_secs(10));
loop {
interval.tick().await;
if let Err(e) = self.tick().await {
tracing::error!(error = %e, "infra tick failed");
}
}
}
pub async fn tick(&self) -> Result<()> {
let mut since_guard = self.last_activity_at.lock().await;
let since = *since_guard;
let activity = self.repo.list_activity(&since, None).await?;
for a in activity {
if matches!(
a.activity_type.as_str(),
"relay_created" | "relay_updated" | "relay_deactivated"
) {
let Some(relay) = self.repo.get_relay(&a.identifier).await? else {
continue;
};
if let Err(e) = self.sync_relay(&relay).await {
tracing::warn!(relay = %relay.id, error = %e, "relay sync failed");
self.repo.fail_relay_sync(&relay, e.to_string()).await?;
}
}
*since_guard = (*since_guard).max(a.created_at);
}
Ok(())
}
async fn sync_relay(&self, relay: &crate::models::Relay) -> Result<()> {
let client = reqwest::Client::new();
let url = format!("{}/relay/{}", self.api_url.trim_end_matches('/'), relay.id);
let host = if self.relay_domain.is_empty() {
relay.subdomain.clone()
} else {
format!("{}.{}", relay.subdomain, self.relay_domain)
};
let secret = uuid::Uuid::new_v4().to_string();
let livekit = if relay.livekit_enabled == 1 {
serde_json::json!({
"enabled": true,
"url": self.livekit_url,
"api_key": self.livekit_api_key,
"api_secret": self.livekit_api_secret,
})
} else {
serde_json::json!({ "enabled": false })
};
let body = serde_json::json!({
"host": host,
"schema": relay.schema,
"secret": secret,
"inactive": relay.status == "inactive",
"info": {
"name": relay.info_name,
"icon": relay.info_icon,
"description": relay.info_description,
},
"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": [
{ "name": "admin", "permissions": ["read", "write", "admin"] },
{ "name": "member", "permissions": ["read", "write"] },
{ "name": "guest", "permissions": ["read"] },
],
});
let response = client.put(url).json(&body).send().await?;
if !response.status().is_success() {
anyhow::bail!("zooid sync returned {}", response.status())
}
Ok(())
}
}
backend/src/main.rs
+19 -71
View File
@@ -1,29 +1,18 @@
mod api;
mod auth;
mod billing;
mod config;
mod db;
mod infra;
mod models;
mod notifications;
mod platform;
mod provisioning;
mod repo;
use std::net::SocketAddr;
mod robot;
use anyhow::Result;
use axum::{Router, routing::get};
use tokio::net::TcpListener;
use tower_http::cors::CorsLayer;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
use crate::billing::BillingService;
use crate::config::Config;
use crate::db::init_pool;
use crate::notifications::Nip17Notifier;
use crate::platform::publish_platform_identity;
use crate::provisioning::Provisioner;
use crate::api::Api;
use crate::billing::Billing;
use crate::infra::Infra;
use crate::repo::Repo;
use crate::robot::Robot;
#[tokio::main]
async fn main() -> Result<()> {
@@ -34,61 +23,20 @@ async fn main() -> Result<()> {
.with(tracing_subscriber::fmt::layer())
.init();
let config = Config::from_env();
ensure_sqlite_dir(&config.database_url)?;
let repo = Repo::new().await?;
repo.migrate().await?;
let robot = Robot::new().await?;
let billing = Billing::new(repo.clone(), robot.clone());
let infra = Infra::new(repo.clone());
let api = Api::new(repo);
let pool = init_pool(&config.database_url).await?;
let repo = Repo::new(pool);
publish_platform_identity(
&config.platform_secret,
&config.indexer_relays,
&config.platform_name,
&config.platform_description,
&config.platform_picture,
&config.platform_messaging_relays,
)
.await?;
let notifier = Nip17Notifier::new(
config.platform_secret.clone(),
config.indexer_relays.clone(),
)
.await?;
let billing = BillingService::new(repo.clone(), notifier, config.platform_nwc_url.clone());
tokio::spawn(billing.run());
let provisioner = Provisioner::new(
config.zooid_api_url.clone(),
config.relay_domain.clone(),
config.platform_secret.clone(),
)?;
let state = api::AppState {
repo,
admin_pubkeys: std::sync::Arc::new(config.admin_pubkeys.clone()),
provisioner,
};
tokio::spawn(async move {
billing.start().await;
});
let app = Router::new()
.merge(api::router(state))
.route("/healthz", get(healthz))
.layer(CorsLayer::permissive());
tokio::spawn(async move {
infra.start().await;
});
let addr: SocketAddr = format!("{}:{}", config.host, config.port).parse()?;
let listener = TcpListener::bind(addr).await?;
tracing::info!("listening on {}", addr);
axum::serve(listener, app).await?;
Ok(())
}
fn ensure_sqlite_dir(database_url: &str) -> Result<()> {
if let Some(path) = database_url.strip_prefix("sqlite://")
&& let Some(dir) = std::path::Path::new(path).parent()
&& !dir.as_os_str().is_empty()
{
std::fs::create_dir_all(dir)?;
}
Ok(())
}
async fn healthz() -> &'static str {
"ok"
api.serve().await
}
backend/src/models.rs
+26 -59
View File
@@ -1,68 +1,53 @@
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RelayConfig {
pub policy: Option<serde_json::Value>,
pub groups: Option<serde_json::Value>,
pub management: Option<serde_json::Value>,
pub blossom: Option<serde_json::Value>,
pub livekit: Option<serde_json::Value>,
pub push: Option<serde_json::Value>,
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct Activity {
pub id: String,
pub created_at: i64,
pub activity_type: String,
pub identifier: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct Tenant {
pub pubkey: String,
pub status: String,
pub nwc_url: String,
pub created_at: i64,
pub billing_anchor_at: i64,
pub stripe_customer_id: String,
pub stripe_subscription_id: String,
pub billing_anchor: i64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct Relay {
pub id: String,
pub tenant: String,
pub name: String,
pub schema: String,
pub subdomain: String,
pub icon: String,
pub description: String,
pub plan: String,
pub status: String,
pub config: Option<RelayConfig>,
}
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct Plan {
pub id: String,
pub sats_per_month: i64,
}
/// Append-only record of relay lifecycle transitions (provisioned, suspended,
/// unsuspended, deactivated). Used as the source of truth for usage metering.
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct RelayLifecycleEvent {
pub id: String,
pub relay: String,
pub tenant: String,
/// One of: "provisioned", "suspended", "unsuspended", "deactivated"
pub event_type: String,
/// Plan active on the relay at the time of the event
pub plan: String,
pub created_at: i64,
pub sync_error: 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(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct Invoice {
pub id: String,
pub tenant: String,
pub amount: i64,
/// One of: "pending", "past_due", "paid", "void"
pub status: String,
pub created_at: i64,
/// bolt11 invoice string (may be refreshed in-app when expired)
pub attempted_at: i64,
pub error: String,
pub closed_at: i64,
pub sent_at: i64,
pub paid_at: i64,
pub bolt11: String,
pub period_start: i64,
pub period_end: i64,
@@ -73,23 +58,5 @@ pub struct InvoiceItem {
pub id: String,
pub invoice: String,
pub relay: String,
pub amount: i64,
pub period_start: i64,
pub period_end: i64,
}
/// Canonical history of payment attempts. `invoices.status` is a synchronous
/// projection updated in the same transaction as each new attempt row.
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct InvoiceAttempt {
pub id: String,
pub invoice: String,
/// Groups all method attempts within a single collection run
pub run_id: String,
/// One of: "nwc", "stripe", "lightning", "nip17_dm"
pub method: String,
/// One of: "success", "failed", "sent" (for DM)
pub outcome: String,
pub error: String,
pub created_at: i64,
pub sats: i64,
}
backend/src/notifications.rs
-106
View File
@@ -1,106 +0,0 @@
use anyhow::{Result, anyhow};
use nostr_sdk::prelude::*;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::Mutex;
#[derive(Clone)]
pub struct Nip17Notifier {
keys: Keys,
indexer_client: Client,
indexer_enabled: bool,
cache: Arc<Mutex<HashMap<String, CacheEntry>>>,
}
impl Nip17Notifier {
pub async fn new(platform_secret: String, relays: Vec<String>) -> Result<Self> {
if platform_secret.trim().is_empty() {
return Err(anyhow!(
"PLATFORM_SECRET is required for NIP-17 notifications"
));
}
let keys = Keys::parse(&platform_secret)?;
let indexer_client = Client::new(keys.clone());
for relay in &relays {
indexer_client.add_relay(relay).await?;
}
let indexer_enabled = !relays.is_empty();
if indexer_enabled {
indexer_client.connect().await;
}
Ok(Self {
keys,
indexer_client,
indexer_enabled,
cache: Arc::new(Mutex::new(HashMap::new())),
})
}
pub async fn send(&self, recipient: &str, message: &str) -> Result<()> {
if !self.indexer_enabled {
return Ok(());
}
let relays = self.fetch_dm_relays(recipient).await?;
if relays.is_empty() {
return Ok(());
}
let pubkey = PublicKey::parse(recipient)?;
let client = Client::new(self.keys.clone());
for relay in relays {
client.add_relay(relay).await?;
}
client.connect().await;
client.send_private_msg(pubkey, message, []).await?;
Ok(())
}
async fn fetch_dm_relays(&self, recipient: &str) -> Result<Vec<String>> {
let mut cache = self.cache.lock().await;
if let Some(entry) = cache.get(recipient)
&& entry.fetched_at.elapsed() < Duration::from_secs(300)
{
return Ok(entry.relays.clone());
}
let pubkey = PublicKey::parse(recipient)?;
let filter = Filter::new().kind(Kind::Custom(10050)).author(pubkey);
let events = self
.indexer_client
.fetch_events(filter, Duration::from_secs(5))
.await?;
let mut relays = Vec::new();
if let Some(event) = events.into_iter().max_by_key(|event| event.created_at) {
for tag in event.tags.iter() {
if tag.as_slice().first().is_some_and(|t| t == "relay")
&& let Some(value) = tag.as_slice().get(1)
{
relays.push(value.to_string());
}
}
}
cache.insert(
recipient.to_string(),
CacheEntry {
relays: relays.clone(),
fetched_at: Instant::now(),
},
);
Ok(relays)
}
}
#[derive(Clone)]
struct CacheEntry {
relays: Vec<String>,
fetched_at: Instant,
}
backend/src/platform.rs
-57
View File
@@ -1,57 +0,0 @@
use anyhow::{Result, anyhow};
use nostr_sdk::prelude::*;
pub async fn publish_platform_identity(
platform_secret: &str,
indexer_relays: &[String],
name: &str,
description: &str,
picture: &str,
messaging_relays: &[String],
) -> Result<()> {
if indexer_relays.is_empty() {
return Ok(());
}
if platform_secret.trim().is_empty() {
return Err(anyhow!("PLATFORM_SECRET is required for platform identity"));
}
let keys = Keys::parse(platform_secret)?;
let client = Client::new(keys);
for relay in indexer_relays {
client.add_relay(relay).await?;
}
client.connect().await;
let mut metadata = Metadata::new();
if !name.is_empty() {
metadata = metadata.name(name);
}
if !description.is_empty() {
metadata = metadata.about(description);
}
if !picture.is_empty() {
metadata = metadata.picture(Url::parse(picture)?);
}
let metadata_builder = EventBuilder::metadata(&metadata);
client.send_event_builder(metadata_builder).await?;
if messaging_relays.is_empty() {
return Ok(());
}
let mut tags = Vec::new();
for relay in messaging_relays {
let tag = Tag::parse(["relay", relay.as_str()])?;
tags.push(tag);
}
let relay_builder = EventBuilder::new(Kind::Custom(10050), "").tags(tags);
client.send_event_builder(relay_builder).await?;
Ok(())
}
backend/src/provisioning.rs
-181
View File
@@ -1,181 +0,0 @@
use anyhow::{Result, anyhow};
use rand::RngCore;
use rand::rngs::OsRng;
use reqwest::Client;
use serde_json::{Value, json};
use nostr_sdk::nostr::Keys;
use nostr_sdk::nostr::nips::nip98::{HttpData, HttpMethod};
use nostr_sdk::nostr::types::url::Url;
use crate::models::{Relay, RelayConfig};
#[derive(Clone)]
pub struct Provisioner {
base_url: String,
relay_domain: String,
admin_keys: Keys,
client: Client,
}
impl Provisioner {
pub fn new(base_url: String, relay_domain: String, admin_secret: String) -> Result<Self> {
if admin_secret.trim().is_empty() {
return Err(anyhow!("PLATFORM_SECRET is required"));
}
let admin_keys = Keys::parse(&admin_secret)?;
let client = Client::new();
Ok(Self {
base_url,
relay_domain,
admin_keys,
client,
})
}
/// Create a relay in zooid.
///
/// POSTs the full config (including a generated secret and host).
pub async fn create_relay(&self, relay: &Relay) -> Result<()> {
let url = format!("{}/relay/{}", self.base_url.trim_end_matches('/'), relay.id);
let blossom_default = relay.plan != "free";
let livekit_default = relay.plan != "free";
let cfg = relay.config.as_ref();
let host = format!("{}.{}", relay.subdomain, self.relay_domain);
let secret = generate_secret_hex();
let payload = json!({
"host": host,
"schema": relay.id,
"secret": secret,
"info": {
"name": relay.name,
"icon": relay.icon,
"pubkey": relay.tenant,
"description": relay.description,
},
"policy": {
"public_join": cfg_bool(cfg, |c| &c.policy, "public_join", false),
"strip_signatures": cfg_bool(cfg, |c| &c.policy, "strip_signatures", false),
},
"groups": {
"enabled": cfg_bool(cfg, |c| &c.groups, "enabled", true),
"auto_join": cfg_bool(cfg, |c| &c.groups, "auto_join", true),
},
"push": {
"enabled": cfg_bool(cfg, |c| &c.push, "enabled", true),
},
"management": {
"enabled": cfg_bool(cfg, |c| &c.management, "enabled", true),
},
"blossom": {
"enabled": cfg_bool(cfg, |c| &c.blossom, "enabled", blossom_default),
},
"livekit": {
"enabled": cfg_bool(cfg, |c| &c.livekit, "enabled", livekit_default),
},
"roles": {
"member": { "pubkeys": [], "can_invite": true, "can_manage": false }
},
});
let auth = self.build_auth_header(&url, HttpMethod::POST).await?;
let res = self
.client
.post(&url)
.header(reqwest::header::AUTHORIZATION, auth)
.json(&payload)
.send()
.await?;
if !res.status().is_success() {
let status = res.status();
let body = res.text().await.unwrap_or_default();
return Err(anyhow!("zooid create failed: {} {}", status, body));
}
Ok(())
}
/// Update a relay in zooid.
///
/// PATCHes only the mutable fields (info + config sections).
pub async fn update_relay(&self, relay: &Relay) -> Result<()> {
let url = format!("{}/relay/{}", self.base_url.trim_end_matches('/'), relay.id);
let host = format!("{}.{}", relay.subdomain, self.relay_domain);
let blossom_default = relay.plan != "free";
let livekit_default = relay.plan != "free";
let cfg = relay.config.as_ref();
let patch = json!({
"host": host,
"info": {
"name": relay.name,
"icon": relay.icon,
"description": relay.description,
},
"policy": {
"public_join": cfg_bool(cfg, |c| &c.policy, "public_join", false),
"strip_signatures": cfg_bool(cfg, |c| &c.policy, "strip_signatures", false),
},
"groups": {
"enabled": cfg_bool(cfg, |c| &c.groups, "enabled", true),
"auto_join": cfg_bool(cfg, |c| &c.groups, "auto_join", true),
},
"push": {
"enabled": cfg_bool(cfg, |c| &c.push, "enabled", true),
},
"management": {
"enabled": cfg_bool(cfg, |c| &c.management, "enabled", true),
},
"blossom": {
"enabled": cfg_bool(cfg, |c| &c.blossom, "enabled", blossom_default),
},
"livekit": {
"enabled": cfg_bool(cfg, |c| &c.livekit, "enabled", livekit_default),
},
});
let auth = self.build_auth_header(&url, HttpMethod::PATCH).await?;
let res = self
.client
.patch(&url)
.header(reqwest::header::AUTHORIZATION, auth)
.json(&patch)
.send()
.await?;
if !res.status().is_success() {
let status = res.status();
let body = res.text().await.unwrap_or_default();
return Err(anyhow!("zooid patch failed: {} {}", status, body));
}
Ok(())
}
async fn build_auth_header(&self, url: &str, method: HttpMethod) -> Result<String> {
let url = Url::parse(url)?;
let data = HttpData::new(url, method);
let header = data.to_authorization(&self.admin_keys).await?;
Ok(header)
}
}
fn cfg_bool(
cfg: Option<&RelayConfig>,
section: impl Fn(&RelayConfig) -> &Option<Value>,
key: &str,
default: bool,
) -> bool {
cfg.and_then(|c| section(c).as_ref())
.and_then(|v| v[key].as_bool())
.unwrap_or(default)
}
fn generate_secret_hex() -> String {
let mut bytes = [0u8; 32];
OsRng.fill_bytes(&mut bytes);
hex::encode(bytes)
}
backend/src/repo.rs
+542 -487
View File
File diff suppressed because it is too large Load Diff
backend/src/robot.rs
+239
View File
@@ -0,0 +1,239 @@
use std::collections::HashMap;
use std::time::{Duration, Instant};
use anyhow::{Result, anyhow};
use nostr_sdk::prelude::*;
use tokio::sync::Mutex;
#[derive(Clone)]
pub struct Robot {
secret: String,
name: String,
description: String,
picture: String,
outbox_relays: Vec<String>,
indexer_relays: Vec<String>,
messaging_relays: Vec<String>,
client: Client,
outbox_cache: std::sync::Arc<Mutex<HashMap<String, CacheEntry>>>,
dm_cache: std::sync::Arc<Mutex<HashMap<String, CacheEntry>>>,
}
#[derive(Clone)]
struct CacheEntry {
values: Vec<String>,
fetched_at: Instant,
}
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_env("ROBOT_OUTBOX_RELAYS");
let indexer_relays = split_env("ROBOT_INDEXER_RELAYS");
let messaging_relays = split_env("ROBOT_MESSAGING_RELAYS");
let keys = Keys::parse(&secret)?;
let client = Client::new(keys);
for relay in &outbox_relays {
client.add_relay(relay).await?;
}
for relay in &indexer_relays {
client.add_relay(relay).await?;
}
for relay in &messaging_relays {
client.add_relay(relay).await?;
}
client.connect().await;
let robot = Self {
secret,
name,
description,
picture,
outbox_relays,
indexer_relays,
messaging_relays,
client,
outbox_cache: std::sync::Arc::new(Mutex::new(HashMap::new())),
dm_cache: std::sync::Arc::new(Mutex::new(HashMap::new())),
};
robot.publish_identity().await?;
Ok(robot)
}
async fn publish_identity(&self) -> Result<()> {
let mut metadata = Metadata::new();
if !self.name.is_empty() {
metadata = metadata.name(&self.name);
}
if !self.description.is_empty() {
metadata = metadata.about(&self.description);
}
if !self.picture.is_empty() {
metadata = metadata.picture(Url::parse(&self.picture)?);
}
self.client
.send_event_builder(EventBuilder::metadata(&metadata))
.await?;
let outbox_tags = self
.outbox_relays
.iter()
.map(|r| Tag::parse(["r", r.as_str()]))
.collect::<std::result::Result<Vec<_>, _>>()?;
self.client
.send_event_builder(EventBuilder::new(Kind::Custom(10002), "").tags(outbox_tags))
.await?;
let mut selection_tags = Vec::new();
for relay in &self.messaging_relays {
selection_tags.push(Tag::parse(["relay", relay.as_str()])?);
}
self.client
.send_event_builder(EventBuilder::new(Kind::Custom(10050), "").tags(selection_tags))
.await?;
Ok(())
}
pub async fn send_dm(&self, recipient: &str, message: &str) -> Result<()> {
let outbox = self.fetch_outbox_relays(recipient).await?;
if outbox.is_empty() {
return Err(anyhow!("no outbox relays found for recipient"));
}
let dm_relays = self.fetch_messaging_relays_from_outbox(recipient, &outbox).await?;
if dm_relays.is_empty() {
return Err(anyhow!("no messaging relays found for recipient"));
}
let recipient_pubkey = PublicKey::parse(recipient)?;
let keys = Keys::parse(&self.secret)?;
let client = Client::new(keys);
for relay in dm_relays {
client.add_relay(relay).await?;
}
client.connect().await;
client.send_private_msg(recipient_pubkey, message, []).await?;
Ok(())
}
async fn fetch_outbox_relays(&self, recipient: &str) -> Result<Vec<String>> {
if let Some(values) = get_cached(&self.outbox_cache, recipient).await {
return Ok(values);
}
let pubkey = PublicKey::parse(recipient)?;
let client = indexer_client(&self.secret, &self.indexer_relays).await?;
let filter = Filter::new().author(pubkey).kind(Kind::Custom(10002));
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) {
for tag in event.tags.iter() {
let values = tag.as_slice();
if values.len() >= 2 && values[0] == "r" {
relays.push(values[1].to_string());
}
}
}
set_cached(&self.outbox_cache, recipient, relays.clone()).await;
Ok(relays)
}
async fn fetch_messaging_relays_from_outbox(
&self,
recipient: &str,
outbox_relays: &[String],
) -> Result<Vec<String>> {
if let Some(values) = get_cached(&self.dm_cache, recipient).await {
return Ok(values);
}
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 filter = Filter::new().author(pubkey).kind(Kind::Custom(10050));
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) {
for tag in event.tags.iter() {
let values = tag.as_slice();
if values.len() >= 2 && values[0] == "relay" {
relays.push(values[1].to_string());
}
}
}
set_cached(&self.dm_cache, recipient, relays.clone()).await;
Ok(relays)
}
}
fn split_env(key: &str) -> Vec<String> {
std::env::var(key)
.unwrap_or_default()
.split(',')
.map(|v| v.trim().to_string())
.filter(|v| !v.is_empty())
.collect()
}
async fn indexer_client(secret: &str, indexer_relays: &[String]) -> Result<Client> {
let keys = Keys::parse(secret)?;
let client = Client::new(keys);
for relay in indexer_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,
) -> Option<Vec<String>> {
let guard = cache.lock().await;
guard.get(key).and_then(|entry| {
if entry.fetched_at.elapsed() < Duration::from_secs(300) {
Some(entry.values.clone())
} else {
None
}
})
}
async fn set_cached(
cache: &std::sync::Arc<Mutex<HashMap<String, CacheEntry>>>,
key: &str,
values: Vec<String>,
) {
let mut guard = cache.lock().await;
guard.insert(
key.to_string(),
CacheEntry {
values,
fetched_at: Instant::now(),
},
);
}