diff --git a/Cargo.toml b/Cargo.toml index a110bd0..3b523cd 100644 --- a/Cargo.toml +++ b/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" diff --git a/src/library.rs b/src/library.rs index 9025155..a8d8fc6 100644 --- a/src/library.rs +++ b/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> { + schema::tracks::table + .order(schema::tracks::id.asc()) + .load::(&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>); + +impl Handle { + pub fn new(library: Library) -> Self { + Self(Arc::new(Mutex::new(library))) + } + + pub fn with_library(&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 { diff --git a/src/main.rs b/src/main.rs index 3ecd649..08774c3 100644 --- a/src/main.rs +++ b/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"); } diff --git a/src/models.rs b/src/models.rs index bb6aee7..4fbe310 100644 --- a/src/models.rs +++ b/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, -} \ No newline at end of file +} diff --git a/src/net.rs b/src/net.rs new file mode 100644 index 0000000..008e843 --- /dev/null +++ b/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 { + 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("

tuneparty is running!

") + } + + 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, + } + + pub fn handle_message(library_handle: &library::Handle, msg_raw: Bytes) -> Option { + 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 { + let msg_str = String::from_utf8_lossy(&msg_raw); + + serde_json::from_str::(&msg_str) + } + + fn generate_pong_response(msg: &Message) -> Option { + 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) -> Option { + 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 { + // 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())) + } +} diff --git a/src/types.rs b/src/types.rs index c0707bc..1776de1 100644 --- a/src/types.rs +++ b/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,