diff --git a/backend/src/infra.rs b/backend/src/infra.rs index 597ae8b..5e8e8cd 100644 --- a/backend/src/infra.rs +++ b/backend/src/infra.rs @@ -54,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(()); } @@ -70,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(()) } @@ -87,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?; } @@ -98,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 { + 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, @@ -125,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 { - 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 { @@ -174,229 +171,134 @@ impl Infra { } } - pub async fn list_relay_members(&self, relay_id: &str) -> Result> { - let client = reqwest::Client::new(); - let base = self.env.zooid_api_url.trim_end_matches('/'); - let url = format!("{base}/relay/{relay_id}/members"); - let auth = self.env.make_auth(&url, HttpMethod::GET).await?; + 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 response = client - .get(&url) - .header("Authorization", auth) - .send() - .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 }, + }, + }); - if !response.status().is_success() { - let status = response.status(); - let body = response.text().await.unwrap_or_default(); - anyhow::bail!("zooid members returned {status}: {body}"); + // 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 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.env.zooid_api_url.trim_end_matches('/'); - - let host = if self.env.relay_domain.is_empty() { - relay.subdomain.clone() - } else { - format!("{}.{}", relay.subdomain, self.env.relay_domain) - }; - - let 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 }) - }; - - let body = relay_sync_body( - relay, - host, - livekit, - is_new.then(|| Keys::generate().secret_key().to_secret_hex()), - &self.env, - ); - - let url = format!("{}/relay/{}", base, relay.id); - let auth = self - .env - .make_auth(&url, zooid_sync_http_method(is_new)) + let method = if is_new { HttpMethod::POST } else { HttpMethod::PATCH }; + self.request(method, &format!("relay/{}", relay.id), Some(&body)) .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(()) } -} -fn zooid_sync_http_method(is_new: bool) -> HttpMethod { - if is_new { - HttpMethod::POST - } else { - HttpMethod::PATCH + pub async fn list_relay_members(&self, relay_id: &str) -> Result> { + #[derive(serde::Deserialize)] + struct MembersResponse { + members: Vec, + } + + 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 { + 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 text = response.text().await.unwrap_or_default(); + anyhow::bail!("zooid {method} {path} returned {status}: {text}"); + } + Ok(response) } } -fn parse_relay_members_response(body: &str) -> Result> { - 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> { - 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, - env: &Env, -) -> serde_json::Value { - let blossom = blossom_sync_json(relay, env); - - 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": blossom, - "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 -} - -/// Relay sync sets `key_prefix` to the relay schema so each relay gets its own -/// blob namespace within the shared bucket. -fn blossom_sync_json(relay: &Relay, env: &Env) -> serde_json::Value { - if relay.blossom_enabled != 1 { - return serde_json::json!({ "enabled": false }); - } - - let mut s3_obj = serde_json::Map::new(); - if !env.blossom_s3_endpoint.trim().is_empty() { - s3_obj.insert( - "endpoint".to_string(), - serde_json::Value::String(env.blossom_s3_endpoint.clone()), - ); - } - s3_obj.insert( - "region".to_string(), - serde_json::Value::String(env.blossom_s3_region.clone()), - ); - s3_obj.insert( - "bucket".to_string(), - serde_json::Value::String(env.blossom_s3_bucket.clone()), - ); - s3_obj.insert( - "access_key".to_string(), - serde_json::Value::String(env.blossom_s3_access_key.clone()), - ); - s3_obj.insert( - "secret_key".to_string(), - serde_json::Value::String(env.blossom_s3_secret_key.clone()), - ); - s3_obj.insert( - "key_prefix".to_string(), - serde_json::Value::String(relay.schema.clone()), - ); - - serde_json::json!({ - "enabled": true, - "adapter": "s3", - "s3": serde_json::Value::Object(s3_obj), - }) -} - -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 { - 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)) -}