Refactor infra

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<Duration> {
+            let retry_attempt = consecutive_failures.max(1);
+            if retry_attempt > RELAY_SYNC_RETRY_MAX_ATTEMPTS {
+                return None;
+            }
 
-        let Some(delay) = relay_sync_retry_delay(consecutive_failures) else {
+            let exponent = (retry_attempt - 1).min(31);
+            let multiplier = 1u64 << exponent;
+            let delay_secs = RELAY_SYNC_RETRY_BASE_DELAY_SECS
+                .saturating_mul(multiplier)
+                .min(RELAY_SYNC_RETRY_MAX_DELAY_SECS);
+
+            Some(Duration::from_secs(delay_secs))
+        }
+
+        let activities = self.query.list_activity_for_relay(relay_id).await?;
+        let consecutive_failures = activities
+            .iter()
+            .take_while(|activity| activity.activity_type == "fail_relay_sync")
+            .count();
+
+        let Some(delay) = get_retry_delay(consecutive_failures) else {
             tracing::warn!(
                 relay = relay_id,
                 consecutive_failures,
@@ -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<bool> {
-        if relay.synced == 1 {
-            return Ok(false);
-        }
-
-        let has_completed_sync = self.query.relay_has_completed_sync(&relay.id).await?;
-        Ok(!has_completed_sync)
-    }
-
-    async fn sync_and_report(&self, relay: &Relay, is_new: bool) {
-        match self.sync_relay(relay, is_new).await {
+    async fn sync_relay(&self, relay: &Relay) {
+        match self.try_sync_relay(relay).await {
             Ok(()) => {
                 tracing::info!(relay = %relay.id, "relay sync succeeded");
                 if let Err(e) = self.command.complete_relay_sync(&relay.id).await {
@@ -174,229 +171,134 @@ impl Infra {
         }
     }
 
-    pub async fn list_relay_members(&self, relay_id: &str) -> Result<Vec<String>> {
-        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<Vec<String>> {
+        #[derive(serde::Deserialize)]
+        struct MembersResponse {
+            members: Vec<String>,
+        }
+
+        let response = self
+            .request(HttpMethod::GET, &format!("relay/{relay_id}/members"), None)
+            .await?;
+        let parsed: MembersResponse = response.json().await?;
+        Ok(parsed.members)
+    }
+
+    // Internal utilities
+
+    /// Sends an authenticated request to the zooid API at `path` (relative to
+    /// `env.zooid_api_url`). Returns the response on 2xx; bails with the body
+    /// text otherwise.
+    async fn request(
+        &self,
+        method: HttpMethod,
+        path: &str,
+        body: Option<&serde_json::Value>,
+    ) -> Result<reqwest::Response> {
+        let client = reqwest::Client::builder()
+            .timeout(Duration::from_secs(5))
+            .build()?;
+        let base = self.env.zooid_api_url.trim_end_matches('/');
+        let path = path.trim_start_matches('/');
+        let url = format!("{base}/{path}");
+        let auth = self.env.make_auth(&url, method).await?;
+
+        let reqwest_method = match method {
+            HttpMethod::GET => reqwest::Method::GET,
+            HttpMethod::POST => reqwest::Method::POST,
+            HttpMethod::PUT => reqwest::Method::PUT,
+            HttpMethod::PATCH => reqwest::Method::PATCH,
+        };
+
+        let mut req = client
+            .request(reqwest_method, &url)
+            .header("Authorization", auth);
+        if let Some(body) = body {
+            req = req.json(body);
+        }
+
+        let response = req.send().await?;
+
+        if !response.status().is_success() {
+            let status = response.status();
+            let 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<Vec<String>> {
-    let value: serde_json::Value = serde_json::from_str(body)?;
-
-    if let Some(members) = members_from_value(&value) {
-        return Ok(members);
-    }
-    if let Some(members) = value.get("members").and_then(members_from_value) {
-        return Ok(members);
-    }
-    if let Some(members) = value
-        .get("data")
-        .and_then(|data| data.get("members"))
-        .and_then(members_from_value)
-    {
-        return Ok(members);
-    }
-
-    anyhow::bail!("zooid members response missing members array")
-}
-
-fn members_from_value(value: &serde_json::Value) -> Option<Vec<String>> {
-    let values = value.as_array()?;
-    values
-        .iter()
-        .map(|value| value.as_str().map(ToString::to_string))
-        .collect()
-}
-
-fn relay_sync_body(
-    relay: &Relay,
-    host: String,
-    livekit: serde_json::Value,
-    secret: Option<String>,
-    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<Duration> {
-    let retry_attempt = consecutive_failures.max(1);
-    if retry_attempt > RELAY_SYNC_RETRY_MAX_ATTEMPTS {
-        return None;
-    }
-
-    let exponent = (retry_attempt - 1).min(31);
-    let multiplier = 1u64 << exponent;
-    let delay_secs = RELAY_SYNC_RETRY_BASE_DELAY_SECS
-        .saturating_mul(multiplier)
-        .min(RELAY_SYNC_RETRY_MAX_DELAY_SECS);
-
-    Some(Duration::from_secs(delay_secs))
-}