SSE initial implementation

[calebbourg]

Description

This PR implements the foundational infrastructure for Server-Sent Events (SSE) to enable real-time, one-way communication from the backend to authenticated users. This establishes the core architecture needed for future real-time features like action notifications, coaching session updates, and system announcements.

GitHub Issue: [Closes|Fixes|Resolves] #your GitHub issue number here

Changes

• New sse crate: Standalone crate with generic types (String IDs, JSON payloads) to avoid circular dependencies
  • ConnectionRegistry: Dual-index architecture using DashMap for O(1) concurrent lookups by connection_id and user_id
  • Manager: High-level API for connection lifecycle and message routing
  • Message types: Event enum with action, agreement, goal, and system event variants
  • MessageScope: User-targeted and broadcast message delivery
• SSE HTTP endpoint (web/src/sse/handler.rs): GET /sse endpoint for establishing long-lived SSE connections
  • Async streaming using async_stream::stream! and Tokio channels
  • Automatic connection cleanup on disconnect
  • One connection per authenticated user, persisting across page navigation
• AppState integration: Added sse_manager: Arc<sse::Manager> to service layer's AppState
  • Enables controllers to send SSE events via app_state.sse_manager.send_message()
  • Manager initialized at application startup in main.rs and seed_db.rs
• Infrastructure configuration:
  • Nginx configuration for /api/sse endpoint (24h timeout, no buffering, chunked encoding)
  • Docker Compose warning about single-instance limitation (in-memory connection tracking)
• Testing tool (sse-test-client): CLI tool for integration testing SSE functionality
  • Connection stability testing
  • Force logout scenario
  • Multiple test scenarios with varying permission requirements
• Architecture documentation updates: Updated crate dependency graph, system architecture diagram, and network flow diagram
  to reflect SSE components

Testing Strategy

1. Manual testing with sse-test-client:

   cd sse-test-client
   cargo run
   # Select "Connection Test" to verify basic SSE connectivity

2. Unit tests: Updated auth middleware tests to include SSE manager initialization
3. Integration testing: Use sse-test-client scenarios to verify:

• Connection establishment and stability
• Multiple concurrent user connections
• Graceful connection cleanup on disconnect

Concerns

• Single-instance limitation: SSE connections are tracked in-memory using DashMap, which means the application cannot scale horizontally without implementing Redis Pub/Sub for cross-instance event distribution
  • Warning added to docker-compose.yaml
  • Symptom if violated: events randomly fail to deliver with multiple replicas
• No event persistence: All SSE events are ephemeral - offline users miss events and see fresh data on next page load
• No message delivery guarantees: If channel send fails, events are dropped (logged but not retried)
• Generic types in SSE crate: Using String for IDs and serde_json::Value for payloads to avoid circular dependencies - type safety enforced at web layer boundaries

[jhodapp]

@calebbourg I don't seem to be able to run the sse-test-client successfully on my local machine, I might be missing something but I am running with the examples provided under sse-test-client/README.md.

Here's what I'm getting:

> cargo run -p sse-test-client -- \
  --base-url http://localhost:4000 \
  --user1 "user1@example.com:password123" \
  --user2 "user2@example.com:password456" \
  --scenario connection-test

=== SETUP PHASE ===
→ Authenticating users...
Error: Login failed: 401 Unauthorized - Response: UNAUTHORIZED

I'm guessing that I need to use a real test user from our seeded data for dev. If this is what I'm missing, I recommend updating the documentation in this README.md to just use the actual users instead of non-existent ones since this data isn't for production anyway and poses no security threat.

[jhodapp]

Let's use the real seeded dev data here since it poses no security threat and helps a developer trying to use these examples to already have them ready to copy + paste from here.

[jhodapp]

For this test client, do we always need to test with a coach and a coachee?

[calebbourg]

The way things are set up, I believe we do need both users for all test scenarios

[jhodapp]

Now I'm seeing the following error once I use two valid users (me as a user, and you as a user) when running the sse-test-client:

> cargo run -p sse-test-client -- \
  --base-url http://localhost:4000 \
  --user1 "jim@refactorcoach.com:password" \
  --user2 "calebbourg2@gmail.com:password" \
  --scenario action-create

14:49:39 [INFO] summary="SELECT DISTINCT \"organizations\".\"id\", \"organizations\".\"name\", …" db.statement="\n\nSELECT DISTINCT \"organizations\".\"id\", \"organizations\".\"name\", \"organizations\".\"logo\", \"organizations\".\"slug\", \"organizations\".\"created_at\", \"organizations\".\"updated_at\" FROM \"refactor_platform\".\"organizations\" INNER JOIN \"refactor_platform\".\"user_roles\" ON \"organizations\".\"id\" = \"user_roles\".\"organization_id\" WHERE \"user_roles\".\"user_id\" = $1\n" rows_affected=1 rows_returned=1 elapsed=224.417µs elapsed_secs=0.000224417
14:49:39 [DEBUG] (2) sea_orm::driver::sqlx_postgres: SELECT "coaching_relationships"."id", "coaching_relationships"."organization_id", "coaching_relationships"."coach_id", "coaching_relationships"."coachee_id", "coaching_relationships"."slug", "coaching_relationships"."created_at", "coaching_relationships"."updated_at" FROM "refactor_platform"."coaching_relationships" WHERE "coaching_relationships"."organization_id" IN (SELECT "organizations"."id" FROM "refactor_platform"."organizations" WHERE "organizations"."id" = '7254d83b-75bb-4523-89e6-7c17005c4a7e')
14:49:39 [INFO] summary="SELECT \"coaching_relationships\".\"id\", \"coaching_relationships\".\"organization_id\", \"coaching_relationships\".\"coach_id\", …" db.statement="\n\nSELECT \"coaching_relationships\".\"id\", \"coaching_relationships\".\"organization_id\", \"coaching_relationships\".\"coach_id\", \"coaching_relationships\".\"coachee_id\", \"coaching_relationships\".\"slug\", \"coaching_relationships\".\"created_at\", \"coaching_relationships\".\"updated_at\" FROM \"refactor_platform\".\"coaching_relationships\" WHERE \"coaching_relationships\".\"organization_id\" IN (SELECT \"organizations\".\"id\" FROM \"refactor_platform\".\"organizations\" WHERE \"organizations\".\"id\" = $1)\n" rows_affected=5 rows_returned=5 elapsed=2.769666ms elapsed_secs=0.002769666
14:49:39 [ERROR] Coaching relationship already exists for coach: a37817a8-ad03-4f04-a6a4-2fe9ca070172 and coachee: 9b66b36a-62f3-4815-a6bb-0a62756537c2 in organization: 00000000-0000-0000-0000-000000000000
14:49:39 [WARN] EntityErrorKind::Other: Responding with 500 Internal Server Error. Error: Domain(Error { source: Some(Error { source: None, error_kind: ValidationError }), error_kind: Internal(Entity(Other("EntityErrorKind"))) })

[jhodapp]

@calebbourg As we discussed in our sync, consider moving sse-test-client into something like a tools directory or test-tools or something like it.

[calebbourg]

Updated in a4c5870550ff06b346a37fce327402ec4bddabf9

[calebbourg]

Can we add monitoring to this somehow in DO?

[jhodapp]

Looking very good! This is just an initial Rust-focused review primarily from Claude using the /rust-review Claude Code command I just added in my last PR.

[jhodapp]

Bug: Connections not cleaned up on send failure

When sender.send() fails (receiver dropped), the connection is logged but remains in the registry, causing a resource leak over time.

Suggested fix:

pub fn send_to_user(&self, user_id: &UserId, event: Event) {
    let Some(connection_ids) = self.user_index.get(user_id) else {
        return;
    };

    let failed: Vec<ConnectionId> = connection_ids
        .iter()
        .filter_map(|conn_id| {
            let info = self.connections.get(conn_id)?;
            info.sender.send(Ok(event.clone())).err().map(|e| {
                warn!(
                    "Failed to send event to connection {}: {}. Marking for cleanup.",
                    conn_id.as_str(),
                    e
                );
                conn_id.clone()
            })
        })
        .collect();

    drop(connection_ids); // Release lock before cleanup

    for conn_id in failed {
        self.unregister(&conn_id);
    }
}

[jhodapp]

Bug: Same connection leak issue in broadcast

Suggested fix:

pub fn broadcast(&self, event: Event) {
    let failed: Vec<ConnectionId> = self
        .connections
        .iter()
        .filter_map(|entry| {
            if let Err(e) = entry.value().sender.send(Ok(event.clone())) {
                warn!(
                    "Failed to send broadcast to connection {}: {}. Marking for cleanup.",
                    entry.key().as_str(),
                    e
                );
                return Some(entry.key().clone());
            }
            None
        })
        .collect();

    for conn_id in failed {
        self.unregister(&conn_id);
    }
}

[jhodapp]

Bug: Connection cleanup may not run if stream is dropped early

The cleanup code at the end of the stream! block won't execute if the client disconnects abruptly. Use a Drop guard to ensure cleanup.

Suggested fix:

use crate::extractors::authenticated_user::AuthenticatedUser;
use async_stream::stream;
use axum::extract::State;
use axum::response::sse::{Event, KeepAlive, Sse};
use futures::Stream;
use log::*;
use service::AppState;
use sse::connection::ConnectionId;
use std::convert::Infallible;
use std::sync::Arc;
use tokio::sync::mpsc;

/// Drop guard to ensure connection cleanup happens even if stream is dropped
struct ConnectionCleanupGuard {
    manager: Arc<sse::Manager>,
    connection_id: ConnectionId,
    user_id: domain::Id,
}

impl Drop for ConnectionCleanupGuard {
    fn drop(&mut self) {
        debug!("SSE connection closed for user {}, cleaning up", self.user_id);
        self.manager.unregister_connection(&self.connection_id);
    }
}

/// SSE handler that establishes a long-lived connection for real-time updates.
/// One connection per authenticated user, stays open across page navigation.
pub(crate) async fn sse_handler(
    AuthenticatedUser(user): AuthenticatedUser,
    State(app_state): State<AppState>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    debug!("Establishing SSE connection for user {}", user.id);

    let (tx, mut rx) = mpsc::unbounded_channel();

    let connection_id = app_state
        .sse_manager
        .register_connection(user.id.to_string(), tx);

    let guard = ConnectionCleanupGuard {
        manager: app_state.sse_manager.clone(),
        connection_id,
        user_id: user.id,
    };

    let stream = stream! {
        let _guard = guard; // Move guard into stream; Drop runs when stream ends
        while let Some(event) = rx.recv().await {
            yield event;
        }
    };

    Sse::new(stream).keep_alive(KeepAlive::default())
}

[jhodapp]

Issue: Serialization errors are logged but not propagated

Callers have no way to know if send_message failed. Consider returning a Result.

Suggested fix - Add error type to sse/src/lib.rs:

pub mod connection;
pub mod manager;
pub mod message;

pub use manager::Manager;

/// Errors that can occur in SSE operations
#[derive(Debug, thiserror::Error)]
pub enum SseError {
    #[error("Failed to serialize event: {0}")]
    Serialization(#[from] serde_json::Error),
}

Update sse/Cargo.toml:

[dependencies]
thiserror = "1.0"

Update send_message in sse/src/manager.rs:

use crate::SseError;

/// Send a message based on its scope
pub fn send_message(&self, message: SseMessage) -> Result<(), SseError> {
    let event_type = message.event.event_type();

    let event_data = serde_json::to_string(&message.event)?;

    let event = Event::default().event(event_type).data(event_data);

    match message.scope {
        MessageScope::User { user_id } => {
            self.registry.send_to_user(&user_id, event);
        }
        MessageScope::Broadcast => {
            self.registry.broadcast(event);
        }
    }

    Ok(())
}

[jhodapp]

Issue: Missing unit tests for ConnectionRegistry

The core SSE logic lacks test coverage. Consider adding these tests:

Suggested addition:

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_register_and_unregister() {
        let registry = ConnectionRegistry::new();
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();

        let user_id = "user123".to_string();
        let conn_id = registry.register(user_id.clone(), tx);

        assert_eq!(registry.connection_count(), 1);
        assert_eq!(registry.active_user_count(), 1);
        assert_eq!(registry.connections_per_user(&user_id), 1);

        registry.unregister(&conn_id);

        assert_eq!(registry.connection_count(), 0);
        assert_eq!(registry.active_user_count(), 0);
    }

    #[test]
    fn test_multiple_connections_per_user() {
        let registry = ConnectionRegistry::new();
        let (tx1, _) = tokio::sync::mpsc::unbounded_channel();
        let (tx2, _) = tokio::sync::mpsc::unbounded_channel();

        let user_id = "user123".to_string();
        let _conn1 = registry.register(user_id.clone(), tx1);
        let _conn2 = registry.register(user_id.clone(), tx2);

        assert_eq!(registry.connection_count(), 2);
        assert_eq!(registry.active_user_count(), 1);
        assert_eq!(registry.connections_per_user(&user_id), 2);
    }

    #[test]
    fn test_unregister_nonexistent_connection() {
        let registry = ConnectionRegistry::new();
        let fake_id = ConnectionId::new();

        // Should not panic
        registry.unregister(&fake_id);
        assert_eq!(registry.connection_count(), 0);
    }
}

[jhodapp]

Ergonomics: Implement Display for ConnectionId

This allows using {} in format strings instead of .as_str().

Suggested addition:

impl std::fmt::Display for ConnectionId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        self.0.fmt(f)
    }
}

Then logging becomes cleaner:

// Before: warn!("Failed for {}", conn_id.as_str());
// After:  warn!("Failed for {}", conn_id);

[jhodapp]

Documentation: Add doc comments to public methods

Suggested fix:

impl Manager {
    /// Creates a new SSE manager with an empty connection registry.
    pub fn new() -> Self {
        Self {
            registry: Arc::new(ConnectionRegistry::new()),
        }
    }

    /// Registers a new SSE connection for the given user.
    ///
    /// # Arguments
    /// * `user_id` - The authenticated user's ID
    /// * `sender` - Channel for pushing events to this connection
    ///
    /// # Returns
    /// A unique `ConnectionId` for later cleanup via `unregister_connection`.
    pub fn register_connection(
        &self,
        user_id: UserId,
        sender: tokio::sync::mpsc::UnboundedSender<Result<Event, std::convert::Infallible>>,
    ) -> ConnectionId {
        // ... existing implementation
    }

    /// Unregisters a connection, removing it from the registry.
    ///
    /// Safe to call multiple times or with invalid IDs.
    pub fn unregister_connection(&self, connection_id: &ConnectionId) {
        // ... existing implementation
    }

    /// Sends a message to recipients based on the message scope.
    ///
    /// # Arguments
    /// * `message` - Contains both the event data and routing scope
    pub fn send_message(&self, message: SseMessage) {
        // ... existing implementation
    }

    /// Returns the total number of active connections.
    pub fn connection_count(&self) -> usize {
        self.registry.connection_count()
    }

    /// Returns the number of unique users with active connections.
    pub fn active_user_count(&self) -> usize {
        self.registry.active_user_count()
    }
}

[jhodapp]

Robustness: Add #[must_use] to prevent accidental misuse

Suggested fix:

impl ConnectionId {
    #[must_use]
    pub fn new() -> Self {
        Self(uuid::Uuid::new_v4().to_string())
    }

    // ...
}

[jhodapp]

Bug: Passwords containing colons will break parsing

split(':') will fail for passwords like p@ss:word.

Suggested fix:

impl UserCredentials {
    pub fn parse(input: &str) -> Result<Self> {
        let mut parts = input.splitn(2, ':'); // Only split on first colon
        let email = parts
            .next()
            .context("Missing email in credentials")?;
        let password = parts
            .next()
            .context("Missing password in credentials (expected email:password)")?;

        Ok(Self {
            email: email.to_string(),
            password: password.to_string(),
        })
    }
}

[jhodapp]

Robustness: Replace .unwrap() with proper error handling

Even in test code, better errors help debugging.

Example at line 32:

// Before:
let action_id = action["id"].as_str().unwrap();

// After:
let action_id = action["id"]
    .as_str()
    .context("Response missing 'id' field")?;

Example at line 47:

// Before:
let received_action_id = event.data["data"]["action"]["id"].as_str().unwrap();

// After:
let received_action_id = event.data["data"]["action"]["id"]
    .as_str()
    .context("Event data missing action ID")?;

Apply similar changes throughout the file.