more Stuff

2023-05-08 16:49:04 +01:00 · 2023-05-08 16:49:04 +01:00 · 6aff9f52b0
parent fd0d9127c8
commit 6aff9f52b0
8 changed files with 226 additions and 24 deletions
--- a/.editorconfig
+++ b/.editorconfig
@ -2,8 +2,11 @@ root = true

 [*]
 indent_style = space
-indent_size = 4
+indent_size = 2
 end_of_line = lf
 charset = utf-8
 trim_trailing_whitespace = false
 insert_final_newline = true
+
+[*.rs]
+indent_size = 4
--- a/Cargo.lock
+++ b/Cargo.lock
@ -225,6 +225,19 @@ dependencies = [
 "typenum",
 ]

+[[package]]
+name = "dashmap"
+version = "5.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "907076dfda823b0b36d2a1bb5f90c96660a5bbcd7729e10727f07858f22c4edc"
+dependencies = [
+ "cfg-if",
+ "hashbrown",
+ "lock_api",
+ "once_cell",
+ "parking_lot_core",
+]
+
 [[package]]
 name = "digest"
 version = "0.10.6"
@ -1442,11 +1455,13 @@ version = "0.1.0"
 dependencies = [
 "axum",
 "confique",
+ "dashmap",
 "futures",
 "futures-util",
 "miette",
 "once_cell",
 "serde",
+ "serde_json",
 "tokio",
 "tower-http",
 "uuid",
--- a/Cargo.toml
+++ b/Cargo.toml
@ -14,3 +14,5 @@ futures = "0.3"
 futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] }
 uuid = { version = "1.3.1", features = ["v4", "serde"] }
 once_cell = "1.17.1"
+serde_json = "1.0.95"
+dashmap = "5.4.0"
--- a/public/assets/main.css
+++ b/public/assets/main.css
@ -0,0 +1,38 @@
+:root {
+  --bg-rgb: 28, 23, 36;
+  --fg-rgb: 234, 234, 248;
+  --accent-rgb: 181, 127, 220;
+  --fg: rgb(var(--fg-rgb));
+  --bg: rgb(var(--bg-rgb));
+  --default-user-color: rgb(238, 126, 255);
+  --accent: rgb(var(--accent-rgb));
+  --fg-transparent: rgba(var(--fg-rgb), 0.25);
+  --bg-transparent: rgba(var(--bg-rgb), 0.25);
+  --autocomplete-bg: linear-gradient(
+      var(--fg-transparent),
+      var(--fg-transparent)
+    ),
+    linear-gradient(var(--bg), var(--bg));
+  --chip-bg: linear-gradient(
+      var(--accent-transparent),
+      var(--accent-transparent)
+    ),
+    linear-gradient(var(--bg), var(--bg));
+  --accent-transparent: rgba(var(--accent-rgb), 0.25);
+}
+
+html {
+  background-color: var(--bg);
+  color: var(--fg);
+  font-family: sans-serif;
+  font-size: 1.125rem;
+}
+
+html, body {
+  margin: 0;
+  padding: 0;
+}
+
+.hidden {
+  display: none;
+}
--- a/public/index.html
+++ b/public/index.html
@ -0,0 +1,14 @@
+<!DOCTYPE html>
+<html>
+  <head>
+    <title>watch party</title>
+    <link rel="stylesheet" href="/assets/main.css">
+  </head>
+
+  <body>
+    <main id="pre-join">
+      <h1>watch party</h1>
+    </main>
+    <main class="hidden" id=""></main>
+  </body>
+</html>
--- a/src/main.rs
+++ b/src/main.rs
@ -1,15 +1,28 @@
-use std::net::{IpAddr, SocketAddr};
+use std::{
+    net::{IpAddr, SocketAddr},
+    sync::{Arc, Mutex},
+};

 use axum::{
-    extract::{ws::WebSocket, Query, WebSocketUpgrade},
+    extract::{
+        ws::{Message, WebSocket},
+        Query, WebSocketUpgrade,
+    },
+    http::StatusCode,
    response::{IntoResponse, Response},
-    routing::get,
-    Router, Server,
+    routing::{get, post},
+    Json, Router, Server,
 };
 use confique::Config;
+use futures::{SinkExt, StreamExt};
 use miette::{Context, IntoDiagnostic, Result};
+use rooms::create_room;
 use serde::Deserialize;
+use sources::MediaSource;
 use tower_http::services::ServeDir;
+use uuid::Uuid;
+
+use crate::rooms::{find_room, Member, RoomRef};

 mod rooms;
 mod sources;
@ -25,12 +38,61 @@ struct WatchPartyConfig {
 #[derive(Deserialize)]
 struct WsQuery {
    room: String,
+    rejoin_id: Option<String>,
 }

-async fn handle_socket(socket: WebSocket, room: String) {}
-
 async fn ws_handler(ws: WebSocketUpgrade, Query(q): Query<WsQuery>) -> Response {
-    ws.on_upgrade(move |s| handle_socket(s, q.room))
+    let Ok(room_id) = Uuid::parse_str(&q.room) else {
+        return (StatusCode::BAD_REQUEST, "Malformed Room ID").into_response()
+    };
+    let Some(room) = find_room(room_id).await else {
+        return (StatusCode::NOT_FOUND, "Room does not exist").into_response()
+    };
+    let rejoin_id = q.rejoin_id.and_then(|s| Uuid::parse_str(&s).ok());
+
+    ws.on_upgrade(move |s| handle_socket(s, room, rejoin_id))
+}
+
+async fn handle_socket(mut socket: WebSocket, room: RoomRef, rejoin_id: Option<Uuid>) {
+    let (mut tx, mut rx) = socket.split();
+
+    let mut member: Member = match rejoin_id {
+        Some(_rejoin_id) => todo!("find existing member"),
+        None => Member {
+            id: Uuid::new_v4(),
+            chat_identity: None,
+            connections: vec![],
+        },
+    };
+    member.connections.push(tx);
+    let member_id = member.id;
+    room.add_member(member).await;
+
+    while let Some(Ok(message)) = rx.next().await {
+        match message {
+            Message::Text(text) => {
+                // TODO: parse a proto message
+            }
+            Message::Close(_) => break,
+            _ => continue,
+        }
+    }
+}
+
+#[derive(Deserialize)]
+struct CreateRoomPayload {
+    playlist: Vec<MediaSource>,
+}
+
+async fn create_room_handler(Json(body): Json<CreateRoomPayload>) -> Response {
+    if body.playlist.is_empty() {
+        return (StatusCode::BAD_REQUEST, "").into_response();
+    }
+
+    let room = create_room(body.playlist).await;
+    let id = room.get().await.id;
+
+    id.to_string().into_response()
 }

 #[tokio::main]
@ -44,9 +106,12 @@ async fn main() -> Result<()> {

    let app = Router::new()
        .fallback_service(ServeDir::new("public"))
-        .route("/api/ws", get(ws_handler));
+        .route("/api/ws", get(ws_handler))
+        .route("/api/create", post(create_room_handler));

    let bind_address: SocketAddr = (conf.address, conf.port).into();
+
+    println!("Listening on http://{} ...", &bind_address);
    Server::bind(&bind_address)
        .serve(app.into_make_service())
        .await
--- a/src/rooms.rs
+++ b/src/rooms.rs
@ -1,29 +1,95 @@
-use std::{
-    collections::HashMap,
-    sync::{Arc, Mutex},
-};
+use std::{collections::HashMap, sync::Arc};

-use axum::extract::ws::WebSocket;
-use miette::Result;
+use axum::extract::ws::{Message, WebSocket};
+use dashmap::DashMap;
+use futures::{stream::SplitSink, SinkExt};
 use once_cell::sync::Lazy;
-use serde::Serialize;
+use tokio::sync::{Mutex, MutexGuard};
 use uuid::Uuid;

 use crate::sources::MediaSource;

-// TODO: How do we store the active rooms?
+#[derive(Clone)]
+pub struct RoomRef {
+    room: Arc<Mutex<Room>>,
+}
+
+impl RoomRef {
+    fn new(room: Room) -> Self {
+        RoomRef {
+            room: Arc::new(Mutex::new(room)),
+        }
+    }
+
+    pub async fn get(&self) -> MutexGuard<Room> {
+        self.room.lock().await
+    }
+
+    pub async fn add_member(&self, member: Member) {
+        let mut room = self.room.lock().await;
+        room.members.insert(member.id, member);
+    }
+
+    pub async fn broadcast(&self, message: Message) {
+        let mut room = self.room.lock().await;
+        for member in room.members.values_mut() {
+            member.send(message.clone()).await
+        }
+    }
+
+    pub async fn list_chat_members(&self) -> Vec<ChatIdentity> {
+        let room = self.room.lock().await;
+        room.members
+            .values()
+            .flat_map(|m| &m.chat_identity)
+            .cloned()
+            .collect()
+    }
+}
+
+static ROOMS: Lazy<DashMap<Uuid, RoomRef>> = Lazy::new(DashMap::new);
+
+pub async fn create_room(playlist: Vec<MediaSource>) -> RoomRef {
+    let id = Uuid::new_v4();
+    let room = Room {
+        id,
+        playlist,
+        currently_playing: 0,
+        members: HashMap::new(),
+    };
+
+    let room_ref = RoomRef::new(room);
+    ROOMS.insert(id, room_ref.clone());
+    room_ref
+}
+
+pub async fn find_room(id: Uuid) -> Option<RoomRef> {
+    ROOMS.get(&id).as_deref().cloned()
+}

-#[derive(Serialize)]
 pub struct Room {
    pub id: Uuid,
    pub playlist: Vec<MediaSource>,
-    // TODO: Currently playing information
-    #[serde(skip)]
-    pub members: Vec<Member>,
+    pub currently_playing: usize,
+    pub members: HashMap<Uuid, Member>,
+}
+
+#[derive(Clone)]
+pub struct ChatIdentity {
+    pub nickname: String,
+    pub color: String,
 }

 pub struct Member {
    pub id: Uuid,
-    pub nickname: Option<String>,
-    pub socket: WebSocket,
+    pub chat_identity: Option<ChatIdentity>,
+    pub connections: Vec<SplitSink<WebSocket, Message>>,
+}
+
+impl Member {
+    pub async fn send(&mut self, message: Message) {
+        for conn in self.connections.iter_mut() {
+            let _ = conn.send(message.clone()).await;
+        }
+    }
 }
--- a/src/sources.rs
+++ b/src/sources.rs
@ -1,5 +1,4 @@
 use serde::{Deserialize, Serialize};
-use uuid::Uuid;

 #[derive(Serialize, Deserialize)]
 pub struct DirectMediaSource {