feat: add basic HTTP, websocket functionality

- add ntex and serde as requirements
- add HTTP server with index/ and ws/ endpoints
- add websocket types and systems, basic Ping and GetTracks endpoints
- add serialize/deserialize support to models::Tracks
- add handle and get_tracks method to library
- clean up types.rs

Cargo.toml

@@ -8,3 +8,6 @@ diesel = { version = "2.3.14", features = ["sqlite", "returning_clauses_for_sqli
 diesel_migrations = { version = "2.3.2", features = ["sqlite"] }
 dotenvy = "0.15.7"
 lofty = "0.25.4"
+ntex = { version = "4.4.0", features = ["tokio"] }
+serde = { version = "1.0.229", features = ["derive"] }
+serde_json = "1.0.154"

src/library.rs

@@ -2,6 +2,7 @@ use std::{
     error::Error,
     fs::{self, File},
     path::{Path, PathBuf},
+    sync::{Arc, Mutex},
 };
 
 use diesel::{prelude::*, sqlite::SqliteConnection};
@@ -25,6 +26,13 @@ impl Library {
         Self { conn }
     }
 
+    // TODO: implement similar limit/offset/count behavior
+    pub fn get_tracks(&mut self) -> QueryResult<Vec<Track>> {
+        schema::tracks::table
+            .order(schema::tracks::id.asc())
+            .load::<Track>(&mut self.conn)
+    }
+
     pub fn print(&mut self) -> QueryResult<()> {
         let artists = schema::artists::table
             .order(schema::artists::id.asc())
@@ -46,6 +54,20 @@ impl Library {
     }
 }
 
+#[derive(Clone)]
+pub struct Handle(Arc<Mutex<Library>>);
+
+impl Handle {
+    pub fn new(library: Library) -> Self {
+        Self(Arc::new(Mutex::new(library)))
+    }
+
+    pub fn with_library<R>(&self, f: impl FnOnce(&mut Library) -> R) -> R {
+        let mut library = self.0.lock().expect("Library handle poisoned!");
+        f(&mut library)
+    }
+}
+
 pub struct Importer;
 
 impl Importer {

src/main.rs

@@ -10,10 +10,11 @@ use std::env;
 
 mod library;
 mod models;
+mod net;
 mod schema;
 mod types;
 
-use library::{Importer, Library};
+use library::{Handle, Importer, Library};
 
 fn open_db_file() -> SqliteConnection {
     let database_url = env::var("DATABASE_URL").expect("DATABASE_URL missing");
@@ -47,4 +48,8 @@ fn main() {
         .expect("Error importing files");
 
     library_instance.print().expect("Error printing library");
+
+    let library_handle = Handle::new(library_instance);
+
+    net::http::start(library_handle).expect("HTTP server failed to start");
 }

src/models.rs

@@ -1,4 +1,5 @@
 use diesel::prelude::*;
+use serde::{Deserialize, Serialize};
 
 use crate::schema::{artists, collections, tracks};
 
@@ -22,7 +23,10 @@ pub struct Collection {
     pub display_name: String,
 }
 
-#[derive(Debug, Clone, Queryable, Selectable, Identifiable, Associations)]
+// TODO: make this more ergonomic (nested artist object instead of artist ID, etc)
+#[derive(
+    Debug, Clone, Queryable, Selectable, Identifiable, Associations, Serialize, Deserialize,
+)]
 #[diesel(table_name = tracks)]
 #[diesel(belongs_to(Artist, foreign_key = artist_id))]
 #[diesel(belongs_to(Collection, foreign_key = collection_id))]
@@ -68,4 +72,4 @@ pub struct NewTrack<'a> {
     pub genre: Option<&'a str>,
     pub track_datetime: Option<&'a str>,
     pub length_sec: i32,
-}
+}

src/net.rs

@@ -0,0 +1,250 @@
+pub mod http {
+    use std::{
+        env, io,
+        sync::{self, atomic},
+    };
+
+    use dotenvy::dotenv;
+    use ntex::{SharedCfg, server, web};
+
+    use crate::library;
+    use crate::net::ws;
+
+    // This is per-worker state
+    #[derive(Clone)]
+    struct AppState {
+        worker_id: usize,
+        library_handle: library::Handle,
+    }
+
+    impl web::State for AppState {
+        type Error = web::DefaultError;
+    }
+
+    // This is global application state
+    struct AppBuilder {
+        next_worker_id: atomic::AtomicUsize,
+        library_handle: library::Handle,
+    }
+
+    impl AppBuilder {
+        // This runs when creating the application, so we pass the database handle into it (which will be cloned later)
+        fn new(library_handle: library::Handle) -> Self {
+            Self {
+                next_worker_id: atomic::AtomicUsize::new(0),
+                library_handle,
+            }
+        }
+    }
+
+    impl server::ServerAppConfig for AppBuilder {
+        type State = AppState;
+
+        // This runs whenever a worker is created, and uses the global state created on application creation
+        async fn create(&self) -> io::Result<Self::State> {
+            let worker_id = self.next_worker_id.fetch_add(1, atomic::Ordering::Relaxed);
+            println!("Creating state for worker {worker_id}");
+
+            Ok(AppState {
+                worker_id,
+                library_handle: self.library_handle.clone(),
+            })
+        }
+    }
+
+    async fn index() -> web::HttpResponse {
+        println!("received request");
+
+        // TODO: add basic stats like database size, taken from a counter updated upon import task");
+        web::HttpResponse::Ok().body("<h1>tuneparty is running!</h1>")
+    }
+
+    async fn websocket(state: &AppState, _: (), req: web::HttpRequest) -> web::HttpResponse {
+        println!("starting websocket connection");
+
+        let cloned_handle = state.library_handle.clone();
+
+        let result = web::ws::start(&req, None, async move |frame: web::ws::Frame| {
+            let message = match frame {
+                web::ws::Frame::Text(text) => ws::handle_message(&cloned_handle, text),
+                web::ws::Frame::Binary(data) => Some(web::ws::Message::Binary(data)),
+                web::ws::Frame::Close(_) => Some(web::ws::Message::Close(Some(
+                    web::ws::CloseCode::Normal.into(),
+                ))),
+                _ => None,
+            };
+
+            Ok::<_, std::io::Error>(message)
+        })
+        .await;
+
+        if let Err(error) = result {
+            eprintln!("WebSocket connection failed: {error}");
+        }
+
+        web::HttpResponse::Ok().body("websocket connected")
+    }
+
+    #[ntex::main]
+    pub async fn start(library_handle: library::Handle) -> std::io::Result<()> {
+        dotenv().ok();
+
+        let address = env::var("HTTP_ADDRESS").unwrap_or_else(|_| {
+            eprintln!("HTTP_ADDRESS not set; using default 127.0.0.1");
+            "127.0.0.1".to_string()
+        });
+        let port: u16 = env::var("HTTP_PORT")
+            .unwrap_or_else(|_| {
+                eprintln!("HTTP_PORT not set; using default 8080");
+                "8080".to_string()
+            })
+            .parse()
+            .expect("port must be a number between 1 and 65536");
+
+        let cfg = AppBuilder::new(library_handle);
+
+        web::server_with_config(cfg, async |_: &AppState| {
+            web::App::new()
+                .service(web::resource("/").to(index))
+                .service(web::resource("/ws").to_with_state(websocket))
+        })
+        .bind((address, port), ntex::SharedCfg::new("STATE"))?
+        .workers(2)
+        .run()
+        .await
+    }
+}
+
+mod ws {
+    use ntex::{util::Bytes, web, web::ws::Message as WsMessage};
+    use serde::{Deserialize, Serialize};
+
+    use crate::{library, models};
+
+    // TODO: custom deserializer that tries the id if possible in order to send potential errors back to the correct device
+    // taking notes from https://github.com/clangen/musikcube/wiki/remote-api-documentation
+    #[derive(Serialize, Deserialize)]
+    struct Message {
+        pub id: i32,
+        pub direction: MessageDirection,
+        pub params: MessageParams,
+    }
+
+    #[derive(Serialize, Deserialize)]
+    #[serde(tag = "msg_type")]
+    enum MessageParams {
+        Error(ErrorParams),
+
+        GetTracks(GetTracks),
+        GetTracksResponse(GetTracksResponse),
+
+        Ping(()),
+        Pong(()),
+    }
+
+    #[derive(Serialize, Deserialize)]
+    enum MessageDirection {
+        Request,
+        Response,
+        Broadcast,
+    }
+
+    #[derive(Serialize, Deserialize)]
+    struct ErrorParams {
+        reason: String,
+    }
+
+    // doesn't take any params for now and queries all tracks period
+    #[derive(Serialize, Deserialize)]
+    struct GetTracks {}
+
+    // TODO: implement similar limit/offset/count behavior
+    #[derive(Serialize, Deserialize)]
+    struct GetTracksResponse {
+        items: Vec<models::Track>,
+    }
+
+    pub fn handle_message(library_handle: &library::Handle, msg_raw: Bytes) -> Option<WsMessage> {
+        print!("received websocket message! ");
+
+        let msg = match decode_message(msg_raw) {
+            Ok(msg) => msg,
+            Err(e) => {
+                println!("couldn't decode it :(\nwhy: {:?}", e);
+                return Some(web::ws::Message::Text("received malformed data".into()));
+            }
+        };
+
+        print!("message has id {}\n", &msg.id);
+
+        // this only handles incoming message types
+        match msg.params {
+            MessageParams::Ping(_) => generate_pong_response(&msg),
+            MessageParams::GetTracks(ref p) => get_tracks(library_handle, &msg, &p),
+            _ => generate_error_response("invalid msg_type", None),
+        }
+
+        //Some(web::ws::Message::Text("ok".into()))
+    }
+
+    fn decode_message(msg_raw: Bytes) -> Result<Message, serde_json::Error> {
+        let msg_str = String::from_utf8_lossy(&msg_raw);
+
+        serde_json::from_str::<Message>(&msg_str)
+    }
+
+    fn generate_pong_response(msg: &Message) -> Option<WsMessage> {
+        let msg = Message {
+            id: msg.id,
+            direction: MessageDirection::Response,
+            params: MessageParams::Pong(()),
+        };
+
+        let payload = serde_json::to_string(&msg)
+            .expect("unable to serialize pong message for whatever reason...");
+
+        Some(web::ws::Message::Text(payload.into()))
+    }
+
+    // TODO: make error messages more flexible/descriptive
+    fn generate_error_response(error_text: &str, id: Option<i32>) -> Option<WsMessage> {
+        let id = id.unwrap_or(-1);
+
+        let msg = Message {
+            id,
+            direction: MessageDirection::Response,
+            params: MessageParams::Error(ErrorParams {
+                reason: error_text.into(),
+            }),
+        };
+
+        let payload = serde_json::to_string(&msg)
+            .expect("unable to serialize error message for whatever reason...");
+
+        Some(web::ws::Message::Text(payload.into()))
+    }
+
+    fn get_tracks(
+        library_handle: &library::Handle,
+        msg: &Message,
+        params: &GetTracks,
+    ) -> Option<WsMessage> {
+        // TODO: perhaps more descriptive errors
+        let tracks = library_handle.with_library(|lib| lib.get_tracks());
+
+        let Ok(tracks) = tracks else {
+            return generate_error_response("error getting tracks from database", Some(msg.id));
+        };
+
+        let msg = Message {
+            id: msg.id,
+            direction: MessageDirection::Response,
+            params: MessageParams::GetTracksResponse(GetTracksResponse { items: tracks }),
+        };
+
+        let payload = serde_json::to_string(&msg)
+            .expect("unable to serialize error message for whatever reason...");
+
+        Some(web::ws::Message::Text(payload.into()))
+    }
+}

src/types.rs

@@ -5,18 +5,6 @@ pub mod library {
 
     use crate::types::errors;
 
-    #[derive(Debug)]
-    struct Artist {
-        id: u32,
-        name: String,
-    }
-
-    #[derive(Debug)]
-    struct Collection {
-        id: u32,
-        name: String,
-    }
-
     #[derive(Debug)]
     pub struct ScannedTrackMetadata {
         pub artist: Option<String>,