feat: add app-agnostic record links to firehose toasts

Add clickable "view record" links to firehose toast notifications that open raw record JSON from user's PDS. Also fix particle animation direction to flow from apps to PDS (correctly showing data writes).

Changes:
- Add /api/record endpoint to fetch individual records
- Refactor firehose to use DID-specific connections via manager
- Add toast link element with hover styling
- Fetch record details for richer toast messages
- Reverse particle flow direction (app → PDS)
- Add "Your PDS" label to identity circle

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[e-mail]>

src/firehose.rs

···
 /// Broadcaster for firehose events
 pub type FirehoseBroadcaster = Arc<broadcast::Sender<FirehoseEvent>>;
 
+/// Manager for DID-specific firehose connections
+pub type FirehoseManager = Arc<Mutex<HashMap<String, FirehoseBroadcaster>>>;
+
 /// A generic ingester that broadcasts all events
 struct BroadcastIngester {
     broadcaster: FirehoseBroadcaster,
···
             action: action.to_string(),
             collection: commit.collection.clone(),
             rkey: commit.rkey.clone(),
-            namespace,
+            namespace: namespace.clone(),
         };
 
+        info!(
+            "Received event: {} {} {} (namespace: {})",
+            action, message.did, commit.collection, namespace
+        );
+
         // Broadcast the event (ignore if no receivers)
-        let _ = self.broadcaster.send(firehose_event);
+        match self.broadcaster.send(firehose_event) {
+            Ok(receivers) => {
+                info!("Broadcast to {} receivers", receivers);
+            }
+            Err(_) => {
+                // No receivers, that's ok
+            }
+        }
 
         Ok(())
     }
 }
 
-/// Start the Jetstream ingester that broadcasts events to all listeners
-pub async fn start_firehose_broadcaster() -> FirehoseBroadcaster {
+/// Create a new FirehoseManager
+pub fn create_firehose_manager() -> FirehoseManager {
+    Arc::new(Mutex::new(HashMap::new()))
+}
+
+/// Get or create a firehose broadcaster for a specific DID
+pub async fn get_or_create_broadcaster(
+    manager: &FirehoseManager,
+    did: String,
+) -> FirehoseBroadcaster {
+    // Check if we already have a broadcaster for this DID
+    {
+        let broadcasters = manager.lock().unwrap();
+        if let Some(broadcaster) = broadcasters.get(&did) {
+            info!("Reusing existing firehose connection for DID: {}", did);
+            return broadcaster.clone();
+        }
+    }
+
+    info!("Creating new firehose connection for DID: {}", did);
+
     // Create a broadcast channel with a buffer of 100 events
     let (tx, _rx) = broadcast::channel::<FirehoseEvent>(100);
     let broadcaster = Arc::new(tx);
 
+    // Store in manager
+    {
+        let mut broadcasters = manager.lock().unwrap();
+        broadcasters.insert(did.clone(), broadcaster.clone());
+    }
+
+    // Clone for the spawn
     let broadcaster_clone = broadcaster.clone();
+    let did_clone = did.clone();
 
     tokio::spawn(async move {
         loop {
-            info!("Starting Jetstream connection...");
+            info!("Starting Jetstream connection for DID: {}...", did_clone);
 
-            // Configure Jetstream to receive all events (no collection filter)
-            let opts = JetstreamOptions::builder().build();
+            // Configure Jetstream to receive events ONLY for this DID
+            let opts = JetstreamOptions::builder()
+                .wanted_dids(vec![did_clone.clone()])
+                .build();
             let jetstream = JetstreamConnection::new(opts);
 
-            // Create ingesters - we use a wildcard to capture all collections
             let mut ingesters: HashMap<String, Box<dyn LexiconIngestor + Send + Sync>> =
                 HashMap::new();
 
-            // Use "*" as a catch-all for all collections
-            ingesters.insert(
-                "*".to_string(),
-                Box::new(BroadcastIngester {
-                    broadcaster: broadcaster_clone.clone(),
-                }),
-            );
+            // Register ingesters for common Bluesky collections
+            let collections = vec![
+                "app.bsky.feed.post",
+                "app.bsky.feed.like",
+                "app.bsky.feed.repost",
+                "app.bsky.graph.follow",
+                "app.bsky.actor.profile",
+            ];
+
+            for collection in collections {
+                ingesters.insert(
+                    collection.to_string(),
+                    Box::new(BroadcastIngester {
+                        broadcaster: broadcaster_clone.clone(),
+                    }),
+                );
+            }
 
             // Get channels
             let msg_rx = jetstream.get_msg_rx();
···
             let cursor: Arc<Mutex<Option<u64>>> = Arc::new(Mutex::new(None));
             let c_cursor = cursor.clone();
 
-            // Spawn task to process messages
+            // Spawn task to process messages using proper handler
             tokio::spawn(async move {
+                info!("Starting message processing loop for DID-filtered connection");
                 while let Ok(message) = msg_rx.recv_async().await {
                     if let Err(e) = rocketman::handler::handle_message(
                         message,
···
             let failed = {
                 let connect_result = jetstream.connect(cursor).await;
                 if let Err(e) = connect_result {
-                    error!("Jetstream connection failed: {}", e);
+                    error!("Jetstream connection failed for DID {}: {}", did_clone, e);
                     true
                 } else {
                     false
···
                 continue;
             }
 
-            info!("Jetstream connection dropped, reconnecting in 5 seconds...");
+            info!("Jetstream connection dropped for DID: {}, reconnecting in 5 seconds...", did_clone);
             tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
         }
     });

src/main.rs

···
 
     let client = oauth::create_oauth_client();
 
-    // Start the firehose broadcaster
-    let firehose_broadcaster = firehose::start_firehose_broadcaster().await;
+    // Create the firehose manager (connections created lazily per-DID)
+    let firehose_manager = firehose::create_firehose_manager();
 
     println!("starting server at http://localhost:8080");
 
···
                 .build(),
             )
             .app_data(web::Data::new(client.clone()))
-            .app_data(web::Data::new(firehose_broadcaster.clone()))
+            .app_data(web::Data::new(firehose_manager.clone()))
             .service(routes::index)
             .service(routes::login)
             .service(routes::callback)
···
             .service(routes::init)
             .service(routes::get_avatar)
             .service(routes::validate_url)
+            .service(routes::get_record)
             .service(routes::firehose_watch)
             .service(routes::favicon)
             .service(Files::new("/static", "./static"))

src/routes.rs

···
 use atrium_oauth::{AuthorizeOptions, CallbackParams, KnownScope, Scope};
 use serde::Deserialize;
 
-use crate::firehose::FirehoseBroadcaster;
+use crate::firehose::FirehoseManager;
 use crate::mst;
 use crate::oauth::OAuthClientType;
 use crate::templates;
···
 }
 
 #[derive(Deserialize)]
+pub struct RecordQuery {
+    pds: String,
+    did: String,
+    collection: String,
+    rkey: String,
+}
+
+#[get("/api/record")]
+pub async fn get_record(query: web::Query<RecordQuery>) -> HttpResponse {
+    let record_url = format!(
+        "{}/xrpc/com.atproto.repo.getRecord?repo={}&collection={}&rkey={}",
+        query.pds, query.did, query.collection, query.rkey
+    );
+
+    match reqwest::get(&record_url).await {
+        Ok(response) => {
+            if !response.status().is_success() {
+                return HttpResponse::Ok().json(serde_json::json!({
+                    "error": "record not found"
+                }));
+            }
+
+            match response.json::<serde_json::Value>().await {
+                Ok(data) => HttpResponse::Ok().json(data),
+                Err(e) => HttpResponse::InternalServerError().json(serde_json::json!({
+                    "error": format!("failed to parse record: {}", e)
+                })),
+            }
+        }
+        Err(e) => HttpResponse::InternalServerError().json(serde_json::json!({
+            "error": format!("failed to fetch record: {}", e)
+        })),
+    }
+}
+
+#[derive(Deserialize)]
 pub struct FirehoseQuery {
     did: String,
 }
···
 #[get("/api/firehose/watch")]
 pub async fn firehose_watch(
     query: web::Query<FirehoseQuery>,
-    broadcaster: web::Data<FirehoseBroadcaster>,
+    manager: web::Data<FirehoseManager>,
 ) -> HttpResponse {
     let did = query.did.clone();
+
+    // Get or create a broadcaster for this DID
+    let broadcaster = crate::firehose::get_or_create_broadcaster(&manager, did.clone()).await;
     let mut rx = broadcaster.subscribe();
 
+    log::info!("SSE connection established for DID: {}", did);
+
     let stream = async_stream::stream! {
         // Send initial connection message
         yield Ok::<_, actix_web::Error>(
             web::Bytes::from(format!("data: {{\"type\":\"connected\"}}\n\n"))
         );
 
-        // Stream firehose events filtered by DID
+        log::info!("Sent initial connection message to client");
+
+        // Stream firehose events (already filtered by DID at Jetstream level)
         while let Ok(event) = rx.recv().await {
-            // Only send events for this DID
-            if event.did == did {
-                let json = serde_json::to_string(&event).unwrap_or_default();
-                yield Ok(web::Bytes::from(format!("data: {}\n\n", json)));
-            }
+            log::info!("Sending event to client: {} {} {}", event.action, event.did, event.collection);
+            let json = serde_json::to_string(&event).unwrap_or_default();
+            yield Ok(web::Bytes::from(format!("data: {}\n\n", json)));
         }
     };
 

src/templates.rs

···
             letter-spacing: 0.05em;
         }}
 
+        .identity-pds-label {{
+            position: absolute;
+            bottom: clamp(-1.5rem, -3vmin, -2rem);
+            font-size: clamp(0.55rem, 1.1vmin, 0.65rem);
+            color: var(--text-light);
+            letter-spacing: 0.05em;
+            font-weight: 500;
+        }}
+
         .identity-avatar {{
             width: clamp(30px, 6vmin, 45px);
             height: clamp(30px, 6vmin, 45px);
···
         .firehose-toast.visible {{
             opacity: 1;
             transform: translateY(0);
+            pointer-events: auto;
         }}
 
         .firehose-toast-action {{
···
             margin-top: 0.25rem;
         }}
 
+        .firehose-toast-link {{
+            display: inline-block;
+            color: var(--text-light);
+            font-size: 0.6rem;
+            margin-top: 0.5rem;
+            text-decoration: none;
+            border-bottom: 1px solid transparent;
+            transition: all 0.2s ease;
+            pointer-events: auto;
+        }}
+
+        .firehose-toast-link:hover {{
+            color: var(--text);
+            border-bottom-color: var(--text);
+        }}
+
         @media (max-width: 768px) {{
             .watch-live-btn {{
                 right: clamp(1rem, 2vmin, 1.5rem);
···
     <div class="firehose-toast" id="firehoseToast">
         <div class="firehose-toast-action"></div>
         <div class="firehose-toast-collection"></div>
+        <a class='firehose-toast-link' id='firehoseToastLink' href='#' target='_blank' rel='noopener noreferrer'>view record</a>
     </div>
 
     <div class="overlay" id="overlay"></div>
···
             <div class="identity-label">@</div>
             <div class="identity-value" id="handle">loading...</div>
             <div class="identity-hint">tap for details</div>
+            <div class="identity-pds-label">Your PDS</div>
         </div>
         <div id="field" class="loading">loading...</div>
     </div>

static/app.js

···
                 let html = `
                     <button class="detail-close" id="detailClose">×</button>
                     <h3>${namespace}</h3>
-                    <div class="subtitle">records stored in your pds:</div>
+                    <div class="subtitle">records stored in your <a href="https://atproto.com/guides/self-hosting" target="_blank" rel="noopener noreferrer" style="color: var(--text); text-decoration: underline;">PDS</a>:</div>
                 `;
 
                 if (collections && collections.length > 0) {
···
         if (alive) {
             particle.draw(firehoseCtx);
         } else {
-            // Particle reached destination - pulse the app circle
-            pulseAppCircle(particle.metadata.namespace);
+            // Particle reached destination - pulse the identity/PDS
+            pulseIdentity();
         }
         return alive;
     });
···
     }
 }
 
-function pulseAppCircle(namespace) {
-    const appCircle = document.querySelector(`[data-namespace="${namespace}"]`);
-    if (appCircle) {
-        appCircle.style.transition = 'all 0.3s ease';
-        appCircle.style.transform = 'scale(1.2)';
-        appCircle.style.boxShadow = '0 0 20px rgba(255, 255, 255, 0.5)';
-        
+function pulseIdentity() {
+    const identity = document.querySelector('.identity');
+    if (identity) {
+        identity.style.transition = 'all 0.3s ease';
+        identity.style.transform = 'scale(1.15)';
+        identity.style.boxShadow = '0 0 25px rgba(255, 255, 255, 0.6)';
+
         setTimeout(() => {
-            appCircle.style.transform = '';
-            appCircle.style.boxShadow = '';
+            identity.style.transform = '';
+            identity.style.boxShadow = '';
         }, 300);
     }
 }
 
-function showFirehoseToast(action, collection) {
-    const toast = document.getElementById('firehoseToast');
-    const actionEl = toast.querySelector('.firehose-toast-action');
-    const collectionEl = toast.querySelector('.firehose-toast-collection');
+async function fetchRecordDetails(pds, did, collection, rkey) {
+    try {
+        const response = await fetch(
+            `/api/record?pds=${encodeURIComponent(pds)}&did=${encodeURIComponent(did)}&collection=${encodeURIComponent(collection)}&rkey=${encodeURIComponent(rkey)}`
+        );
+        const data = await response.json();
+        if (data.error) return null;
+        return data.value;
+    } catch (e) {
+        console.error('Error fetching record:', e);
+        return null;
+    }
+}
 
+function formatToastMessage(action, collection, record) {
     const actionText = {
         'create': 'created',
         'update': 'updated',
         'delete': 'deleted'
     }[action] || action;
 
-    actionEl.textContent = `${actionText} record`;
-    collectionEl.textContent = collection;
+    // If we don't have record details, fall back to basic message
+    if (!record) {
+        return {
+            action: `${actionText} record`,
+            details: collection
+        };
+    }
+
+    // Format based on collection type
+    if (collection === 'app.bsky.feed.post') {
+        const text = record.text || '';
+        const preview = text.length > 50 ? text.substring(0, 50) + '...' : text;
+        return {
+            action: `${actionText} post`,
+            details: preview || 'no text'
+        };
+    } else if (collection === 'app.bsky.feed.like') {
+        return {
+            action: `${actionText} like`,
+            details: ''
+        };
+    } else if (collection === 'app.bsky.feed.repost') {
+        return {
+            action: `${actionText} repost`,
+            details: ''
+        };
+    } else if (collection === 'app.bsky.graph.follow') {
+        return {
+            action: `${actionText} follow`,
+            details: ''
+        };
+    } else if (collection === 'app.bsky.actor.profile') {
+        const displayName = record.displayName || '';
+        return {
+            action: `${actionText} profile`,
+            details: displayName || 'updated profile'
+        };
+    }
+
+    // Default for unknown collections
+    return {
+        action: `${actionText} record`,
+        details: collection
+    };
+}
+
+async function showFirehoseToast(event) {
+    const toast = document.getElementById('firehoseToast');
+    const actionEl = toast.querySelector('.firehose-toast-action');
+    const collectionEl = toast.querySelector('.firehose-toast-collection');
+    const linkEl = document.getElementById('firehoseToastLink');
+
+    // Build PDS link for the record
+    if (globalPds && event.did && event.collection && event.rkey) {
+        const recordUrl = `${globalPds}/xrpc/com.atproto.repo.getRecord?repo=${encodeURIComponent(event.did)}&collection=${encodeURIComponent(event.collection)}&rkey=${encodeURIComponent(event.rkey)}`;
+        linkEl.href = recordUrl;
+    }
+
+    // Fetch record details if available (skip for deletes)
+    let record = null;
+    if (event.action !== 'delete' && event.rkey && globalPds) {
+        record = await fetchRecordDetails(globalPds, event.did, event.collection, event.rkey);
+    }
+
+    const formatted = formatToastMessage(event.action, event.collection, record);
+
+    actionEl.textContent = formatted.action;
+    collectionEl.textContent = formatted.details;
 
     toast.classList.add('visible');
     setTimeout(() => {
         toast.classList.remove('visible');
-    }, 3000);
+    }, 4000); // Slightly longer to read details
 }
 
 function getParticleColor(action) {
···
 }
 
 function createFirehoseParticle(event) {
-    // Get identity circle position
-    const identity = document.querySelector('.identity');
-    if (!identity) return;
-    
-    const identityRect = identity.getBoundingClientRect();
-    const startX = identityRect.left + identityRect.width / 2;
-    const startY = identityRect.top + identityRect.height / 2;
-
-    // Get target app circle position
+    // Get source app circle position (where the action happened)
     const appCircle = document.querySelector(`[data-namespace="${event.namespace}"]`);
     if (!appCircle) return;
 
     const appRect = appCircle.getBoundingClientRect();
-    const endX = appRect.left + appRect.width / 2;
-    const endY = appRect.top + appRect.height / 2;
+    const startX = appRect.left + appRect.width / 2;
+    const startY = appRect.top + appRect.height / 2;
 
-    // Create particle
+    // Get target identity/PDS position (where data is written)
+    const identity = document.querySelector('.identity');
+    if (!identity) return;
+
+    const identityRect = identity.getBoundingClientRect();
+    const endX = identityRect.left + identityRect.width / 2;
+    const endY = identityRect.top + identityRect.height / 2;
+
+    // Create particle (flows from app TO PDS)
     const particle = new FirehoseParticle(
         startX, startY,
         endX, endY,
···
 }
 
 function connectFirehose() {
-    if (!did || firehoseEventSource) return;
+    console.log('[Firehose] connectFirehose called, did =', did, 'existing connection?', !!firehoseEventSource);
+    if (!did || firehoseEventSource) {
+        console.warn('[Firehose] Exiting early - did:', did, 'firehoseEventSource:', firehoseEventSource);
+        return;
+    }
 
     const url = `/api/firehose/watch?did=${encodeURIComponent(did)}`;
-    console.log('Connecting to firehose:', url);
+    console.log('[Firehose] Connecting to:', url);
     
     firehoseEventSource = new EventSource(url);
 
···
 
             // Create particle animation
             createFirehoseParticle(data);
-            
+
             // Show toast notification
-            showFirehoseToast(data.action, data.collection);
+            showFirehoseToast(data);
         } catch (error) {
             console.error('Error processing firehose message:', error);
         }
···
 
 // Toggle watch live
 document.addEventListener('DOMContentLoaded', () => {
+    console.log('[Firehose] DOMContentLoaded fired, setting up watch button');
     const watchBtn = document.getElementById('watchLiveBtn');
-    if (!watchBtn) return;
+    if (!watchBtn) {
+        console.error('[Firehose] Watch button not found!');
+        return;
+    }
 
+    console.log('[Firehose] Watch button found, attaching click handler');
     const watchLabel = watchBtn.querySelector('.watch-label');
 
     watchBtn.addEventListener('click', () => {
+        console.log('[Firehose] Watch button clicked! isWatchingLive was:', isWatchingLive);
         isWatchingLive = !isWatchingLive;
+        console.log('[Firehose] isWatchingLive now:', isWatchingLive);
 
         if (isWatchingLive) {
             // Start watching
+            console.log('[Firehose] Starting watch mode');
             watchLabel.textContent = 'connecting...';
             initFirehoseCanvas();
             connectFirehose();
             animateFirehoseParticles();
         } else {
             // Stop watching
+            console.log('[Firehose] Stopping watch mode');
             watchLabel.textContent = 'watch live';
             watchBtn.classList.remove('active');
             disconnectFirehose();