feat: Add support for Jetstream and PLC fallback URLs

This commit adds support for fallback URLs for both Jetstream and PLC.
This provides resilience in case of primary server outages. Also
includes retry logic for Jetstream with exponential backoff and max
delay.

src/config/mod.rs

···
 #[derive(Debug, Clone)]
 pub struct JetstreamConfig {
     pub url: String,
+    pub fallback_urls: Vec<String>,
     pub wanted_collections: Vec<String>,
     pub cursor_update_interval: u64,
+    pub retry_delay_secs: u64,
+    pub max_retry_delay_secs: u64,
 }
 
 #[derive(Debug, Clone)]
···
 #[derive(Debug, Clone)]
 pub struct PlcConfig {
     pub endpoint: String,
+    pub fallback_endpoints: Vec<String>,
 }
 
 #[derive(Debug, Clone)]
···
                     "JETSTREAM_URL",
                     Some("wss://jetstream.atproto.tools/subscribe"),
                 )?,
+                fallback_urls: get_env_list(
+                    "JETSTREAM_FALLBACK_URLS",
+                    vec![
+                        "wss://jetstream1.us-east.fire.hose.cam/subscribe".to_string(),
+                        "wss://jetstream2.us-east.fire.hose.cam/subscribe".to_string(),
+                    ],
+                ),
                 wanted_collections: vec!["app.bsky.feed.post".to_string()],
                 cursor_update_interval: get_env_u64("CURSOR_UPDATE_INTERVAL", 10_000),
+                retry_delay_secs: get_env_u64("JETSTREAM_RETRY_DELAY_SECS", 5),
+                max_retry_delay_secs: get_env_u64("JETSTREAM_MAX_RETRY_DELAY_SECS", 300),
             },
             redis: RedisConfig {
                 url: get_env("REDIS_URL", Some("redis://localhost:6379"))?,
···
             },
             plc: PlcConfig {
                 endpoint: get_env("PLC_ENDPOINT", Some("https://plc.directory"))?,
+                fallback_endpoints: get_env_list(
+                    "PLC_FALLBACK_ENDPOINTS",
+                    vec!["https://plc.directory".to_string()],
+                ),
             },
             automod: AutomodConfig {
                 handle: get_env("AUTOMOD_HANDLE", None)
···
 
 /// Get environment variable with optional default
 fn get_env(key: &str, default: Option<&str>) -> Result<String> {
-    env::var(key)
-        .into_diagnostic()
-        .or_else(|_| {
-            default
-                .ok_or_else(|| miette::miette!("Missing required environment variable: {}", key))
-                .map(String::from)
-        })
+    env::var(key).into_diagnostic().or_else(|_| {
+        default
+            .ok_or_else(|| miette::miette!("Missing required environment variable: {}", key))
+            .map(String::from)
+    })
 }
 
 /// Get environment variable as u32 with default
···
         .map(|v| {
             let v = v.to_lowercase();
             v == "true" || v == "1" || v == "yes"
+        })
+        .unwrap_or(default)
+}
+
+/// Get environment variable as comma-separated list with default
+fn get_env_list(key: &str, default: Vec<String>) -> Vec<String> {
+    env::var(key)
+        .ok()
+        .map(|v| {
+            v.split(',')
+                .map(|s| s.trim().to_string())
+                .filter(|s| !s.is_empty())
+                .collect()
         })
         .unwrap_or(default)
 }

src/jetstream/events.rs

···
 #[cfg(test)]
 mod tests {
     use super::*;
+    use jacquard_common::types::string::AtprotoStr;
     use jacquard_common::types::value::{Array, Object};
     use jacquard_common::CowStr;
     use std::collections::BTreeMap;

src/jetstream/mod.rs

···
     pub async fn subscribe(
         self,
         job_sender: mpsc::UnboundedSender<ImageJob>,
-        mut shutdown_rx: tokio::sync::oneshot::Receiver<()>,
+        mut shutdown_rx: tokio::sync::broadcast::Receiver<()>,
     ) -> Result<()> {
         info!("Connecting to Jetstream: {}", self.url);
 
···
                         }
                     }
                 }
-                _ = &mut shutdown_rx => {
+                _ = shutdown_rx.recv() => {
                     info!("Shutting down Jetstream client");
                     info!("Processed {} total messages", message_count);
 

src/lib.rs

···
 // Metrics
 pub mod metrics;
 
+// PLC Directory client
+pub mod plc;
+
 // Re-export commonly used types
 pub use config::Config;
 pub use types::{BlobCheck, BlobReference, ImageJob, MatchResult};

src/main.rs

···
 
     // Create shutdown channels
     let (_shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
-    let (jetstream_shutdown_tx, jetstream_shutdown_rx) = tokio::sync::oneshot::channel::<()>();
+    let (jetstream_shutdown_tx, _jetstream_shutdown_rx) = tokio::sync::broadcast::channel::<()>(1);
 
     // Create job channel for jetstream -> queue
     let (job_tx, mut job_rx) = mpsc::unbounded_channel();
···
         info!("Resuming from cursor: {}", c);
     }
 
-    // Start jetstream subscriber
-    info!("Starting Jetstream subscriber...");
+    // Start jetstream subscriber with retry logic
+    info!("Starting Jetstream subscriber with auto-retry...");
     let jetstream_config = config.clone();
-    let jetstream_handle = tokio::spawn(async move {
-        let jetstream = JetstreamClient::new(jetstream_config.jetstream.url.clone(), cursor)
-            .expect("Failed to create Jetstream client");
+    let jetstream_shutdown_broadcast = jetstream_shutdown_tx.clone();
+    let _jetstream_handle = tokio::spawn(async move {
+        let mut retry_delay = jetstream_config.jetstream.retry_delay_secs;
+        let max_retry_delay = jetstream_config.jetstream.max_retry_delay_secs;
+        let mut url_index = 0;
+
+        // Build list of URLs to try (primary + fallbacks)
+        let mut urls = vec![jetstream_config.jetstream.url.clone()];
+        urls.extend(jetstream_config.jetstream.fallback_urls.clone());
+
+        info!("Jetstream URLs configured: {} total (1 primary + {} fallbacks)",
+            urls.len(), urls.len() - 1);
+
+        loop {
+            let current_url = &urls[url_index];
+            let cursor = skywatch_phash_rs::jetstream::cursor::read_cursor();
+
+            info!("Attempting Jetstream connection to: {} (retry delay: {}s)", current_url, retry_delay);
 
-        if let Err(e) = jetstream.subscribe(job_tx, jetstream_shutdown_rx).await {
-            error!("Jetstream subscriber failed: {}", e);
+            let jetstream = match JetstreamClient::new(current_url.clone(), cursor) {
+                Ok(client) => client,
+                Err(e) => {
+                    error!("Failed to create Jetstream client: {}", e);
+                    tokio::time::sleep(Duration::from_secs(retry_delay)).await;
+                    retry_delay = (retry_delay * 2).min(max_retry_delay);
+                    url_index = (url_index + 1) % urls.len();
+                    continue;
+                }
+            };
+
+            // Create new shutdown receiver for this attempt
+            let shutdown_rx = jetstream_shutdown_broadcast.subscribe();
+
+            match jetstream.subscribe(job_tx.clone(), shutdown_rx).await {
+                Ok(_) => {
+                    info!("Jetstream subscriber completed normally (shutdown received)");
+                    break;
+                }
+                Err(e) => {
+                    error!("Jetstream connection failed: {} - Retrying in {}s", e, retry_delay);
+                    tokio::time::sleep(Duration::from_secs(retry_delay)).await;
+
+                    // Exponential backoff with max cap
+                    retry_delay = (retry_delay * 2).min(max_retry_delay);
+
+                    // Rotate to next URL
+                    url_index = (url_index + 1) % urls.len();
+                    if url_index == 0 {
+                        info!("Tried all Jetstream URLs, cycling back to primary");
+                    }
+                }
+            }
         }
     });
 
···
         }
         _ = all_workers_future => {
             info!("All workers completed");
-        }
-        _ = jetstream_handle => {
-            info!("Jetstream subscriber completed");
         }
     }