initial vibe coded app

src/hub.rs

@@ -0,0 +1,133 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use tokio::sync::{Mutex, broadcast};
+
+#[derive(Clone, Debug)]
+pub struct PresenceUser {
+    pub user_id: String,
+    pub display_name: String,
+}
+
+#[derive(Clone, Debug)]
+pub enum HubEvent {
+    ListChanged { list_id: String, revision: i64 },
+    PresenceChanged { list_id: String },
+}
+
+#[derive(Debug)]
+struct ConnectionInfo {
+    user_id: String,
+    display_name: String,
+}
+
+struct Room {
+    sender: broadcast::Sender<HubEvent>,
+    connections: HashMap<String, ConnectionInfo>,
+}
+
+pub struct Subscription {
+    pub connection_id: String,
+    pub receiver: broadcast::Receiver<HubEvent>,
+    pub presence: Vec<PresenceUser>,
+}
+
+#[derive(Clone, Default)]
+pub struct Hub {
+    rooms: Arc<Mutex<HashMap<String, Room>>>,
+}
+
+impl Hub {
+    pub async fn join(
+        &self,
+        list_id: String,
+        user_id: String,
+        display_name: String,
+    ) -> Subscription {
+        let mut rooms = self.rooms.lock().await;
+        let room = rooms.entry(list_id.clone()).or_insert_with(|| {
+            let (sender, _) = broadcast::channel(64);
+            Room {
+                sender,
+                connections: HashMap::new(),
+            }
+        });
+
+        let connection_id = crate::db::new_secret();
+        let already_present = room
+            .connections
+            .values()
+            .any(|connection| connection.user_id == user_id);
+        room.connections.insert(
+            connection_id.clone(),
+            ConnectionInfo {
+                user_id,
+                display_name,
+            },
+        );
+        let presence = current_presence(room);
+
+        if !already_present {
+            let _ = room.sender.send(HubEvent::PresenceChanged { list_id });
+        }
+        let receiver = room.sender.subscribe();
+
+        Subscription {
+            connection_id,
+            receiver,
+            presence,
+        }
+    }
+
+    pub async fn leave(&self, list_id: &str, connection_id: &str) {
+        let mut rooms = self.rooms.lock().await;
+        let mut remove_room = false;
+        if let Some(room) = rooms.get_mut(list_id) {
+            let removed = room.connections.remove(connection_id);
+            if let Some(removed) = removed {
+                let still_present = room
+                    .connections
+                    .values()
+                    .any(|connection| connection.user_id == removed.user_id);
+                if !still_present {
+                    let _ = room.sender.send(HubEvent::PresenceChanged {
+                        list_id: list_id.to_owned(),
+                    });
+                }
+            }
+            remove_room = room.connections.is_empty();
+        }
+        if remove_room {
+            rooms.remove(list_id);
+        }
+    }
+
+    pub async fn publish_list_changed(&self, list_id: String, revision: i64) {
+        let rooms = self.rooms.lock().await;
+        if let Some(room) = rooms.get(&list_id) {
+            let _ = room
+                .sender
+                .send(HubEvent::ListChanged { list_id, revision });
+        }
+    }
+
+    pub async fn presence(&self, list_id: &str) -> Vec<PresenceUser> {
+        let rooms = self.rooms.lock().await;
+        rooms.get(list_id).map(current_presence).unwrap_or_default()
+    }
+}
+
+fn current_presence(room: &Room) -> Vec<PresenceUser> {
+    let mut users = HashMap::<String, PresenceUser>::new();
+    for connection in room.connections.values() {
+        users
+            .entry(connection.user_id.clone())
+            .or_insert_with(|| PresenceUser {
+                user_id: connection.user_id.clone(),
+                display_name: connection.display_name.clone(),
+            });
+    }
+    let mut users = users.into_values().collect::<Vec<_>>();
+    users.sort_by_key(|user| user.display_name.to_lowercase());
+    users
+}