userAdityaa/caravel
1
1
Fork 0
forked from coracle/caravel
Code Pull Requests Activity

fix: invoice.paid reactivating manually deactivated relays

Browse Source
This commit is contained in:
Aditya Chaudhary
2026-04-15 00:14:13 +05:45
parent 9a8d02b286
commit 43ab49b0bd
6 changed files with 146 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/migrations/0002_relay_deactivation_source.sql
+2
View File
@@ -0,0 +1,2 @@
ALTER TABLE relay
ADD COLUMN deactivation_source TEXT CHECK (deactivation_source IN ('manual', 'billing') OR deactivation_source IS NULL);
backend/src/api.rs
+1
View File
@@ -516,6 +516,7 @@ async fn create_relay(
plan: payload.plan,
stripe_subscription_item_id: None,
status: "active".to_string(),
deactivation_source: None,
sync_error: String::new(),
info_name: payload.info_name.unwrap_or_default(),
info_icon: payload.info_icon.unwrap_or_default(),
backend/src/billing.rs
+71 -5
View File
@@ -6,7 +6,7 @@ use nwc::prelude::{
use sha2::Sha256;
use crate::command::Command;
use crate::models::Activity;
use crate::models::{Activity, RELAY_DEACTIVATION_SOURCE_BILLING, Relay};
use crate::query::Query;
use crate::robot::Robot;
@@ -348,7 +348,7 @@ impl Billing {
let relays = self.query.list_relays_for_tenant(&tenant.pubkey).await?;
for relay in relays {
if relay.status == "inactive" && relay.plan != "free" {
if Self::should_reactivate_after_payment(&relay) {
self.command.activate_relay(&relay).await?;
}
}
@@ -403,7 +403,7 @@ impl Billing {
let relays = self.query.list_relays_for_tenant(&tenant.pubkey).await?;
for relay in relays {
if relay.status == "active" && relay.plan != "free" {
self.command.deactivate_relay(&relay).await?;
self.command.deactivate_relay_for_billing(&relay).await?;
}
}
@@ -438,7 +438,7 @@ impl Billing {
let relays = self.query.list_relays_for_tenant(&tenant.pubkey).await?;
for relay in relays {
if relay.status == "active" && relay.plan != "free" {
self.command.deactivate_relay(&relay).await?;
self.command.deactivate_relay_for_billing(&relay).await?;
}
}
@@ -722,6 +722,12 @@ impl Billing {
fiat_minor_to_msats_from_quote(amount_due_minor, &normalized_currency, btc_price)
}
fn should_reactivate_after_payment(relay: &Relay) -> bool {
relay.status == "inactive"
&& relay.plan != "free"
&& relay.deactivation_source.as_deref() == Some(RELAY_DEACTIVATION_SOURCE_BILLING)
}
async fn fetch_btc_spot_price(&self, currency: &str) -> Result<f64> {
fetch_btc_spot_price_from_base(&self.http, &self.btc_quote_api_base, currency).await
}
@@ -799,7 +805,35 @@ pub fn fiat_minor_to_msats_from_quote(
#[cfg(test)]
mod tests {
use super::fiat_minor_to_msats_from_quote;
use super::{Billing, fiat_minor_to_msats_from_quote};
use crate::models::{
RELAY_DEACTIVATION_SOURCE_BILLING, RELAY_DEACTIVATION_SOURCE_MANUAL, Relay,
};
fn relay_fixture(status: &str, plan: &str, deactivation_source: Option<&str>) -> Relay {
Relay {
id: "relay-1".to_string(),
tenant: "tenant-1".to_string(),
schema: "tenant_1".to_string(),
subdomain: "relay-1".to_string(),
plan: plan.to_string(),
stripe_subscription_item_id: None,
status: status.to_string(),
deactivation_source: deactivation_source.map(ToString::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: 1,
livekit_enabled: 1,
push_enabled: 1,
synced: 1,
}
}
#[test]
fn converts_usd_minor_units_with_quote() {
@@ -814,4 +848,36 @@ mod tests {
.expect("conversion should succeed");
assert_eq!(msats, 1_000_000);
}
#[test]
fn reactivates_only_paid_relays_deactivated_for_billing() {
let billing_inactive_paid =
relay_fixture("inactive", "basic", Some(RELAY_DEACTIVATION_SOURCE_BILLING));
assert!(Billing::should_reactivate_after_payment(
&billing_inactive_paid
));
let manual_inactive_paid =
relay_fixture("inactive", "basic", Some(RELAY_DEACTIVATION_SOURCE_MANUAL));
assert!(!Billing::should_reactivate_after_payment(
&manual_inactive_paid
));
let unknown_inactive_paid = relay_fixture("inactive", "basic", None);
assert!(!Billing::should_reactivate_after_payment(
&unknown_inactive_paid
));
let billing_inactive_free =
relay_fixture("inactive", "free", Some(RELAY_DEACTIVATION_SOURCE_BILLING));
assert!(!Billing::should_reactivate_after_payment(
&billing_inactive_free
));
let billing_active_paid =
relay_fixture("active", "basic", Some(RELAY_DEACTIVATION_SOURCE_BILLING));
assert!(!Billing::should_reactivate_after_payment(
&billing_active_paid
));
}
}
backend/src/command.rs
+58 -25
View File
@@ -2,7 +2,9 @@ use anyhow::Result;
use sqlx::{Sqlite, SqlitePool, Transaction};
use tokio::sync::broadcast;
use crate::models::{Activity, Relay, Tenant};
use crate::models::{
Activity, RELAY_DEACTIVATION_SOURCE_BILLING, RELAY_DEACTIVATION_SOURCE_MANUAL, Relay, Tenant,
};
#[derive(Clone)]
pub struct Command {
@@ -77,7 +79,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 +96,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);
@@ -105,12 +109,12 @@ impl Command {
sqlx::query(
"INSERT INTO relay (
id, tenant, schema, subdomain, plan, status, sync_error,
id, tenant, schema, subdomain, plan, status, deactivation_source, sync_error,
info_name, info_icon, info_description,
policy_public_join, policy_strip_signatures,
groups_enabled, management_enabled, blossom_enabled,
livekit_enabled, push_enabled
) VALUES (?, ?, ?, ?, ?, 'active', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
) VALUES (?, ?, ?, ?, ?, 'active', NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
)
.bind(&relay.id)
.bind(&relay.tenant)
@@ -143,7 +147,7 @@ impl Command {
sqlx::query(
"UPDATE relay
SET tenant = ?, schema = ?, subdomain = ?, plan = ?, status = ?, sync_error = ?,
SET tenant = ?, schema = ?, subdomain = ?, plan = ?, status = ?, deactivation_source = ?, sync_error = ?,
info_name = ?, info_icon = ?, info_description = ?,
policy_public_join = ?, policy_strip_signatures = ?,
groups_enabled = ?, management_enabled = ?, blossom_enabled = ?,
@@ -155,6 +159,7 @@ impl Command {
.bind(&relay.subdomain)
.bind(&relay.plan)
.bind(&relay.status)
.bind(&relay.deactivation_source)
.bind(&relay.sync_error)
.bind(&relay.info_name)
.bind(&relay.info_icon)
@@ -178,14 +183,42 @@ impl Command {
}
pub async fn deactivate_relay(&self, relay: &Relay) -> Result<()> {
self.set_relay_status(
&relay.id,
"inactive",
Some(RELAY_DEACTIVATION_SOURCE_MANUAL),
"deactivate_relay",
)
.await
}
pub async fn deactivate_relay_for_billing(&self, relay: &Relay) -> Result<()> {
self.set_relay_status(
&relay.id,
"inactive",
Some(RELAY_DEACTIVATION_SOURCE_BILLING),
"deactivate_relay",
)
.await
}
async fn set_relay_status(
&self,
relay_id: &str,
status: &str,
deactivation_source: Option<&str>,
activity_type: &str,
) -> Result<()> {
let mut tx = self.pool.begin().await?;
sqlx::query("UPDATE relay SET status = 'inactive' WHERE id = ?")
.bind(&relay.id)
sqlx::query("UPDATE relay SET status = ?, deactivation_source = ? WHERE id = ?")
.bind(status)
.bind(deactivation_source)
.bind(relay_id)
.execute(&mut *tx)
.await?;
let activity = Self::insert_activity(&mut tx, "deactivate_relay", "relay", &relay.id).await?;
let activity = Self::insert_activity(&mut tx, activity_type, "relay", relay_id).await?;
tx.commit().await?;
self.emit(activity);
@@ -193,18 +226,8 @@ impl Command {
}
pub async fn activate_relay(&self, relay: &Relay) -> Result<()> {
let mut tx = self.pool.begin().await?;
sqlx::query("UPDATE relay SET status = 'active' WHERE id = ?")
.bind(&relay.id)
.execute(&mut *tx)
.await?;
let activity = Self::insert_activity(&mut tx, "activate_relay", "relay", &relay.id).await?;
tx.commit().await?;
self.emit(activity);
Ok(())
self.set_relay_status(&relay.id, "active", None, "activate_relay")
.await
}
pub async fn fail_relay_sync(&self, relay: &Relay, sync_error: String) -> Result<()> {
@@ -216,7 +239,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 +255,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 +271,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 +284,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)
backend/src/models.rs
+4
View File
@@ -1,5 +1,8 @@
use serde::{Deserialize, Serialize};
pub const RELAY_DEACTIVATION_SOURCE_MANUAL: &str = "manual";
pub const RELAY_DEACTIVATION_SOURCE_BILLING: &str = "billing";
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct Activity {
pub id: String,
@@ -41,6 +44,7 @@ pub struct Relay {
pub plan: String,
pub stripe_subscription_item_id: Option<String>,
pub status: String,
pub deactivation_source: Option<String>,
pub sync_error: String,
pub info_name: String,
pub info_icon: String,
backend/src/query.rs
+10 -7
View File
@@ -70,8 +70,8 @@ impl Query {
pub async fn list_relays(&self) -> Result<Vec<Relay>> {
let rows = sqlx::query_as::<_, Relay>(
"SELECT id, tenant, schema, subdomain, plan, stripe_subscription_item_id,
status, sync_error,
"SELECT id, tenant, schema, subdomain, plan, stripe_subscription_item_id,
status, deactivation_source, sync_error,
info_name, info_icon, info_description,
policy_public_join, policy_strip_signatures,
groups_enabled, management_enabled, blossom_enabled,
@@ -86,8 +86,8 @@ impl Query {
pub async fn list_relays_for_tenant(&self, tenant_id: &str) -> Result<Vec<Relay>> {
let rows = sqlx::query_as::<_, Relay>(
"SELECT id, tenant, schema, subdomain, plan, stripe_subscription_item_id,
status, sync_error,
"SELECT id, tenant, schema, subdomain, plan, stripe_subscription_item_id,
status, deactivation_source, sync_error,
info_name, info_icon, info_description,
policy_public_join, policy_strip_signatures,
groups_enabled, management_enabled, blossom_enabled,
@@ -104,8 +104,8 @@ impl Query {
pub async fn get_relay(&self, id: &str) -> Result<Option<Relay>> {
let row = sqlx::query_as::<_, Relay>(
"SELECT id, tenant, schema, subdomain, plan, stripe_subscription_item_id,
status, sync_error,
"SELECT id, tenant, schema, subdomain, plan, stripe_subscription_item_id,
status, deactivation_source, sync_error,
info_name, info_icon, info_description,
policy_public_join, policy_strip_signatures,
groups_enabled, management_enabled, blossom_enabled,
@@ -119,7 +119,10 @@ impl Query {
Ok(row)
}
pub async fn get_tenant_by_stripe_customer_id(&self, stripe_customer_id: &str) -> Result<Option<Tenant>> {
pub async fn get_tenant_by_stripe_customer_id(
&self,
stripe_customer_id: &str,
) -> Result<Option<Tenant>> {
let row = sqlx::query_as::<_, Tenant>(
"SELECT pubkey, nwc_url, nwc_error, created_at, stripe_customer_id, stripe_subscription_id, past_due_at
FROM tenant