From ebf1c5b5e19dce3be1b0817b744434a52599448c Mon Sep 17 00:00:00 2001 From: Alexander Date: Mon, 29 Jun 2026 21:16:55 +0200 Subject: [PATCH] Create server side --- Cargo.lock | 342 +++++++++++++++ Cargo.toml | 23 +- devenv.nix | 3 + proto/buf.gen.yaml | 6 + proto/buf.yaml | 9 + proto/musicfs.proto | 75 ++++ src/bin/musicfs-server.rs | 76 ++++ src/lib.rs | 2 + src/music/mod.rs | 1 + src/music/parse.rs | 41 ++ src/origins/local/snapshot.rs | 12 +- src/proto/generated/musicfs/musicfs.rs | 84 ++++ src/proto/generated/musicfs/musicfs.tonic.rs | 419 +++++++++++++++++++ src/proto/mod.rs | 17 + src/server/manifest.rs | 80 ++++ src/server/mod.rs | 5 + src/server/range.rs | 175 ++++++++ src/server/state.rs | 179 ++++++++ src/server/transport/grpc.rs | 214 ++++++++++ src/server/transport/mod.rs | 39 ++ src/server/watcher.rs | 99 +++++ 21 files changed, 1890 insertions(+), 11 deletions(-) create mode 100644 proto/buf.gen.yaml create mode 100644 proto/buf.yaml create mode 100644 proto/musicfs.proto create mode 100644 src/bin/musicfs-server.rs create mode 100644 src/music/parse.rs create mode 100644 src/proto/generated/musicfs/musicfs.rs create mode 100644 src/proto/generated/musicfs/musicfs.tonic.rs create mode 100644 src/proto/mod.rs create mode 100644 src/server/manifest.rs create mode 100644 src/server/mod.rs create mode 100644 src/server/range.rs create mode 100644 src/server/state.rs create mode 100644 src/server/transport/grpc.rs create mode 100644 src/server/transport/mod.rs create mode 100644 src/server/watcher.rs diff --git a/Cargo.lock b/Cargo.lock index c1f4955..5a47d51 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -107,6 +107,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "anyhow" +version = "1.0.103" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" + [[package]] name = "arrayvec" version = "0.7.7" @@ -314,12 +320,61 @@ dependencies = [ "num-traits", ] +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + [[package]] name = "autocfg" version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" +[[package]] +name = "axum" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +dependencies = [ + "axum-core", + "bytes", + "futures-util", + "http", + "http-body", + "http-body-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "serde_core", + "sync_wrapper", + "tower", + "tower-layer", + "tower-service", +] + +[[package]] +name = "axum-core" +version = "0.5.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "sync_wrapper", + "tower-layer", + "tower-service", +] + [[package]] name = "base64" version = "0.22.1" @@ -1028,6 +1083,25 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "h2" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155" +dependencies = [ + "atomic-waker", + "bytes", + "fnv", + "futures-core", + "futures-sink", + "http", + "indexmap", + "slab", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "half" version = "2.7.1" @@ -1111,6 +1185,51 @@ dependencies = [ "digest 0.11.3", ] +[[package]] +name = "http" +version = "1.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "http-body" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b021d93e26becf5dc7e1b75b1bed1fd93124b374ceb73f43d4d4eafec896a64a" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "pin-project-lite", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + [[package]] name = "hybrid-array" version = "0.4.12" @@ -1120,6 +1239,61 @@ dependencies = [ "typenum", ] +[[package]] +name = "hyper" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "55281c53a1894c864990125767da440a4e630446785086f52523b20033b74498" +dependencies = [ + "atomic-waker", + "bytes", + "futures-channel", + "futures-core", + "h2", + "http", + "http-body", + "httparse", + "httpdate", + "itoa", + "pin-project-lite", + "smallvec", + "tokio", + "want", +] + +[[package]] +name = "hyper-timeout" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b90d566bffbce6a75bd8b09a05aa8c2cb1fabb6cb348f8840c9e4c90a0d83b0" +dependencies = [ + "hyper", + "hyper-util", + "pin-project-lite", + "tokio", + "tower-service", +] + +[[package]] +name = "hyper-util" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" +dependencies = [ + "bytes", + "futures-channel", + "futures-util", + "http", + "http-body", + "hyper", + "libc", + "pin-project-lite", + "socket2", + "tokio", + "tower-service", + "tracing", +] + [[package]] name = "iana-time-zone" version = "0.1.65" @@ -1458,6 +1632,12 @@ dependencies = [ "winapi", ] +[[package]] +name = "matchit" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" + [[package]] name = "md-5" version = "0.11.0" @@ -1483,6 +1663,12 @@ dependencies = [ "autocfg", ] +[[package]] +name = "mime" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" + [[package]] name = "mio" version = "1.2.1" @@ -1499,15 +1685,23 @@ dependencies = [ name = "musicfs" version = "0.1.0" dependencies = [ + "anyhow", + "async-trait", + "bytes", "clap", "fuser", + "http", "libc", "notify", + "prost", "sea-orm", "symphonia", "tempfile", "time", "tokio", + "tokio-stream", + "tonic", + "tonic-prost", "twox-hash", ] @@ -1729,6 +1923,26 @@ dependencies = [ "serde", ] +[[package]] +name = "pin-project" +version = "1.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2466b2336ed02bcdca6b294417127b90ec92038d1d5c4fbeac971a922e0e0924" +dependencies = [ + "pin-project-internal", +] + +[[package]] +name = "pin-project-internal" +version = "1.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c96395f0a926bc13b1c17622aaddda1ecb55d49c8f1bf9777e4d877800a43f8b" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "pin-project-lite" version = "0.2.17" @@ -1828,6 +2042,29 @@ dependencies = [ "yansi", ] +[[package]] +name = "prost" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "528ac67416ff8646872a3c02cad9cc4ee5dc9f9540c9b10771855c95cb2e5ae1" +dependencies = [ + "bytes", + "prost-derive", +] + +[[package]] +name = "prost-derive" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" +dependencies = [ + "anyhow", + "itertools", + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "ptr_meta" version = "0.1.4" @@ -2772,6 +3009,12 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "sync_wrapper" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" + [[package]] name = "synstructure" version = "0.13.2" @@ -2924,6 +3167,19 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-util" +version = "0.7.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "pin-project-lite", + "tokio", +] + [[package]] name = "toml_datetime" version = "1.1.1+spec-1.1.0" @@ -2954,6 +3210,77 @@ dependencies = [ "winnow", ] +[[package]] +name = "tonic" +version = "0.14.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef" +dependencies = [ + "async-trait", + "axum", + "base64", + "bytes", + "h2", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-timeout", + "hyper-util", + "percent-encoding", + "pin-project", + "socket2", + "sync_wrapper", + "tokio", + "tokio-stream", + "tower", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tonic-prost" +version = "0.14.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "50849f68853be452acf590cde0b146665b8d507b3b8af17261df47e02c209ea0" +dependencies = [ + "bytes", + "prost", + "tonic", +] + +[[package]] +name = "tower" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" +dependencies = [ + "futures-core", + "futures-util", + "indexmap", + "pin-project-lite", + "slab", + "sync_wrapper", + "tokio", + "tokio-util", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tower-layer" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e" + +[[package]] +name = "tower-service" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" + [[package]] name = "tracing" version = "0.1.44" @@ -2986,6 +3313,12 @@ dependencies = [ "once_cell", ] +[[package]] +name = "try-lock" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" + [[package]] name = "twox-hash" version = "2.1.2" @@ -3091,6 +3424,15 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "want" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bfa7760aed19e106de2c7c0b581b509f2f25d3dacaf737cb82ac61bc6d760b0e" +dependencies = [ + "try-lock", +] + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" diff --git a/Cargo.toml b/Cargo.toml index d7e6c0e..57baef5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,6 +3,18 @@ name = "musicfs" version = "0.1.0" edition = "2024" +[lib] +name = "musicfs" +path = "src/lib.rs" + +[[bin]] +name = "musicfs" +path = "src/main.rs" + +[[bin]] +name = "musicfs-server" +path = "src/bin/musicfs-server.rs" + [dependencies] clap = { version = "4.6.1", features = ["derive"] } fuser = "0.17.0" @@ -15,7 +27,16 @@ symphonia = { version = "0.5", default-features = false, features = [ twox-hash = "2.1.2" sea-orm = { version = "2.0.0-rc", features = [ "sqlx-postgres", "runtime-tokio", "macros" ] } -tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal"] } +tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal", "fs", "io-util", "sync"] } + +anyhow = "1" +async-trait = "0.1" +bytes = "1" +tonic = "0.14" +tonic-prost = "0.14" +prost = "0.14" +tokio-stream = "0.1" +http = "1" [dev-dependencies] tempfile = "3" diff --git a/devenv.nix b/devenv.nix index cbb521b..b2e360c 100644 --- a/devenv.nix +++ b/devenv.nix @@ -29,6 +29,9 @@ just flac ffmpeg + protobuf + buf + grpcurl opencode inputs.pinix.packages.${pkgs.system}.default diff --git a/proto/buf.gen.yaml b/proto/buf.gen.yaml new file mode 100644 index 0000000..5b2d709 --- /dev/null +++ b/proto/buf.gen.yaml @@ -0,0 +1,6 @@ +version: v2 +plugins: + - remote: buf.build/community/neoeinstein-prost + out: ../src/proto/generated + - remote: buf.build/community/neoeinstein-tonic + out: ../src/proto/generated diff --git a/proto/buf.yaml b/proto/buf.yaml new file mode 100644 index 0000000..f74da98 --- /dev/null +++ b/proto/buf.yaml @@ -0,0 +1,9 @@ +version: v2 +modules: + - path: . +lint: + use: + - STANDARD +breaking: + use: + - FILE diff --git a/proto/musicfs.proto b/proto/musicfs.proto new file mode 100644 index 0000000..7e0f7f1 --- /dev/null +++ b/proto/musicfs.proto @@ -0,0 +1,75 @@ +syntax = "proto3"; + +package musicfs; + +// MusicFs service exposes a server-side music library to a remote musicfs +// FUSE mount. Three RPCs, all server-streaming so the client can pull +// incrementally and the server can keep large bodies off the heap. +service MusicFs { + // Stream every file entry in the library. The client diffs each entry's + // hash against its DB to decide what changed. + rpc GetManifest(GetManifestRequest) returns (stream ManifestEntry); + + // Stream the bytes of one file. If `start` and `length` are both set on + // the request, only that byte range is sent; otherwise the whole file. + // The first chunk carries `total_size` and `offset`; subsequent chunks + // continue from there. + rpc GetFile(GetFileRequest) returns (stream FileChunk); + + // Long-lived stream of filesystem change events. The client uses these as + // wake-ups to refetch the manifest; correctness always rests on the hash + // diff, never on the event payload. + rpc SubscribeEvents(SubscribeEventsRequest) returns (stream ChangeEvent); +} + +message GetManifestRequest {} + +message GetFileRequest { + uint64 id = 1; + // When both are zero the server streams the entire file. + uint64 start = 2; + uint64 length = 3; +} + +message SubscribeEventsRequest {} + +message ManifestEntry { + uint64 id = 1; + string rel_path = 2; + uint64 size = 3; + uint64 mtime = 4; + uint64 ctime = 5; + uint64 crtime = 6; + MusicMetadata music_metadata = 7; +} + +message MusicMetadata { + repeated string artist = 1; + optional string album_artist = 2; + string album = 3; + int32 track_number = 4; + string track_title = 5; + repeated string other_tags = 6; + bytes header = 7; + repeated bytes picture_block_headers = 8; + repeated PictureDataRange picture_data_ranges = 9; + uint64 real_audio_start = 10; + uint64 vorbis_comment_offset = 11; + uint64 vorbis_comment_length = 12; +} + +message PictureDataRange { + uint64 offset = 1; + uint64 length = 2; +} + +message FileChunk { + bytes data = 1; + uint64 offset = 2; + uint64 total_size = 3; +} + +message ChangeEvent { + // "create", "modify", or "remove". + string kind = 1; +} diff --git a/src/bin/musicfs-server.rs b/src/bin/musicfs-server.rs new file mode 100644 index 0000000..e313b0f --- /dev/null +++ b/src/bin/musicfs-server.rs @@ -0,0 +1,76 @@ +use std::{net::SocketAddr, path::PathBuf}; + +use anyhow::{Context, Result}; +use clap::Parser; +use musicfs::server::{ + state::ServerState, + transport::{self, TransportArgs}, + watcher::ServerWatcher, +}; +use tokio::signal::unix::{SignalKind, signal}; + +#[derive(Parser, Debug)] +#[command(version, about, long_about = None)] +struct Args { + /// Directory containing the music library to serve. + #[arg(short, long, required = true)] + source: PathBuf, + + /// Address:port the transport should listen on (e.g. 0.0.0.0:50051). + #[arg(short, long, default_value = "0.0.0.0:50051")] + listen: SocketAddr, + + /// Name of the transport implementation to use. Today only "grpc". + #[arg(short, long, default_value = "grpc")] + transport: String, +} + +#[tokio::main] +async fn main() -> Result<()> { + let args = Args::parse(); + + let source = args.source.clone(); + if !source.is_dir() { + return Err(anyhow::anyhow!( + "source is not a readable directory: {}", + source.display() + )); + } + + let state = ServerState::new(); + state + .replace_all(&source) + .with_context(|| format!("initial scan of {}", source.display()))?; + + let (_watcher, events_rx) = ServerWatcher::spawn(source.clone(), state.clone()); + + let transport = transport::build( + &args.transport, + TransportArgs { + listen: args.listen, + }, + ) + .with_context(|| format!("building transport {:?}", args.transport))?; + + println!( + "musicfs-server: transport={}, listen={}, initial scan complete", + args.transport, args.listen + ); + + let server_task = tokio::spawn(async move { + if let Err(e) = transport.run(state, events_rx).await { + eprintln!("transport ended with error: {e}"); + } + }); + + let mut sigint = signal(SignalKind::interrupt()).expect("register SIGINT"); + let mut sigterm = signal(SignalKind::terminate()).expect("register SIGTERM"); + + tokio::select! { + _ = sigint.recv() => println!("received SIGINT, shutting down"), + _ = sigterm.recv() => println!("received SIGTERM, shutting down"), + _ = server_task => println!("transport task exited"), + } + + return Ok(()); +} diff --git a/src/lib.rs b/src/lib.rs index f639c59..7ca09f7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -2,4 +2,6 @@ pub mod db; pub mod item; pub mod music; pub mod origins; +pub mod proto; +pub mod server; pub mod virtual_dirs; diff --git a/src/music/mod.rs b/src/music/mod.rs index f8435dc..4277032 100644 --- a/src/music/mod.rs +++ b/src/music/mod.rs @@ -3,4 +3,5 @@ pub mod encoder; pub mod flac; pub mod metadata; pub mod mp3; +pub mod parse; pub mod parser; diff --git a/src/music/parse.rs b/src/music/parse.rs new file mode 100644 index 0000000..689416d --- /dev/null +++ b/src/music/parse.rs @@ -0,0 +1,41 @@ +use std::path::Path; + +use crate::music::encoder::MusicMetadataEncoderFactory; +use crate::music::metadata::MusicMetadata; +use crate::music::parser::MusicMetadataParserFactory; + +/// Parse a single media file into a fully-encoded [`MusicMetadata`]. +/// +/// `None` is returned for files musicfs does not treat as music (unknown +/// extension or unparseable container). When metadata is produced, the +/// matching [`MusicMetadataEncoder`] has already baked the tag fields into +/// the in-memory `header`, so callers can serve virtualized bytes directly. +/// +/// This helper is the single source of truth for "parse + encode" — used by +/// the local FUSE origin's snapshot builder, the future network origin's +/// manifest builder, and the HTTP/3 server's manifest endpoint. +pub fn parse_music_metadata_for_path(path: &Path) -> Option { + let mut metadata = MusicMetadataParserFactory::for_path(path)?.parse(path)?; + if let Some(encoder) = MusicMetadataEncoderFactory::for_path(path) { + encoder.encode(&mut metadata); + } + return Some(metadata); +} + +#[cfg(test)] +mod tests { + use super::*; + use std::path::Path; + + #[test] + fn none_for_non_music_extension() { + let path = Path::new("notes.txt"); + assert!(parse_music_metadata_for_path(path).is_none()); + } + + #[test] + fn none_for_missing_file_with_music_extension() { + let path = Path::new("/nonexistent/track.flac"); + assert!(parse_music_metadata_for_path(path).is_none()); + } +} diff --git a/src/origins/local/snapshot.rs b/src/origins/local/snapshot.rs index f799e47..17161fd 100644 --- a/src/origins/local/snapshot.rs +++ b/src/origins/local/snapshot.rs @@ -9,8 +9,7 @@ use fuser::INodeNo; use std::os::unix::fs::MetadataExt; use crate::item::{FileType, Item}; -use crate::music::encoder::MusicMetadataEncoderFactory; -use crate::music::parser::MusicMetadataParserFactory; +use crate::music::parse::parse_music_metadata_for_path; use crate::origins::attrs::FileAttrs; use crate::virtual_dirs::ensure_virtual_dirs; @@ -72,14 +71,7 @@ pub fn read_into_map( let name = entry.file_name().to_string_lossy().into_owned(); let metadata = entry.metadata()?; - let music_metadata = MusicMetadataParserFactory::for_path(&item_path) - .and_then(|parser| parser.parse(&item_path)) - .map(|mut mm| { - if let Some(encoder) = MusicMetadataEncoderFactory::for_path(&item_path) { - encoder.encode(&mut mm); - } - mm - }); + let music_metadata = parse_music_metadata_for_path(&item_path); let mut local_path = PathBuf::new(); if let Some(ref mm) = music_metadata { diff --git a/src/proto/generated/musicfs/musicfs.rs b/src/proto/generated/musicfs/musicfs.rs new file mode 100644 index 0000000..1dde6bb --- /dev/null +++ b/src/proto/generated/musicfs/musicfs.rs @@ -0,0 +1,84 @@ +// @generated +// This file is @generated by prost-build. +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct GetManifestRequest {} +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct GetFileRequest { + #[prost(uint64, tag = "1")] + pub id: u64, + /// When both are zero the server streams the entire file. + #[prost(uint64, tag = "2")] + pub start: u64, + #[prost(uint64, tag = "3")] + pub length: u64, +} +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct SubscribeEventsRequest {} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct ManifestEntry { + #[prost(uint64, tag = "1")] + pub id: u64, + #[prost(string, tag = "2")] + pub rel_path: ::prost::alloc::string::String, + #[prost(uint64, tag = "3")] + pub size: u64, + #[prost(uint64, tag = "4")] + pub mtime: u64, + #[prost(uint64, tag = "5")] + pub ctime: u64, + #[prost(uint64, tag = "6")] + pub crtime: u64, + #[prost(message, optional, tag = "7")] + pub music_metadata: ::core::option::Option, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct MusicMetadata { + #[prost(string, repeated, tag = "1")] + pub artist: ::prost::alloc::vec::Vec<::prost::alloc::string::String>, + #[prost(string, optional, tag = "2")] + pub album_artist: ::core::option::Option<::prost::alloc::string::String>, + #[prost(string, tag = "3")] + pub album: ::prost::alloc::string::String, + #[prost(int32, tag = "4")] + pub track_number: i32, + #[prost(string, tag = "5")] + pub track_title: ::prost::alloc::string::String, + #[prost(string, repeated, tag = "6")] + pub other_tags: ::prost::alloc::vec::Vec<::prost::alloc::string::String>, + #[prost(bytes = "vec", tag = "7")] + pub header: ::prost::alloc::vec::Vec, + #[prost(bytes = "vec", repeated, tag = "8")] + pub picture_block_headers: ::prost::alloc::vec::Vec<::prost::alloc::vec::Vec>, + #[prost(message, repeated, tag = "9")] + pub picture_data_ranges: ::prost::alloc::vec::Vec, + #[prost(uint64, tag = "10")] + pub real_audio_start: u64, + #[prost(uint64, tag = "11")] + pub vorbis_comment_offset: u64, + #[prost(uint64, tag = "12")] + pub vorbis_comment_length: u64, +} +#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)] +pub struct PictureDataRange { + #[prost(uint64, tag = "1")] + pub offset: u64, + #[prost(uint64, tag = "2")] + pub length: u64, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct FileChunk { + #[prost(bytes = "vec", tag = "1")] + pub data: ::prost::alloc::vec::Vec, + #[prost(uint64, tag = "2")] + pub offset: u64, + #[prost(uint64, tag = "3")] + pub total_size: u64, +} +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct ChangeEvent { + /// "create", "modify", or "remove". + #[prost(string, tag = "1")] + pub kind: ::prost::alloc::string::String, +} +include!("musicfs.tonic.rs"); +// @@protoc_insertion_point(module) diff --git a/src/proto/generated/musicfs/musicfs.tonic.rs b/src/proto/generated/musicfs/musicfs.tonic.rs new file mode 100644 index 0000000..1f903e9 --- /dev/null +++ b/src/proto/generated/musicfs/musicfs.tonic.rs @@ -0,0 +1,419 @@ +// @generated +/// Generated client implementations. +pub mod music_fs_client { + #![allow( + unused_variables, + dead_code, + missing_docs, + clippy::wildcard_imports, + clippy::let_unit_value + )] + use tonic::codegen::http::Uri; + use tonic::codegen::*; + #[derive(Debug, Clone)] + pub struct MusicFsClient { + inner: tonic::client::Grpc, + } + impl MusicFsClient { + /// Attempt to create a new client by connecting to a given endpoint. + pub async fn connect(dst: D) -> Result + where + D: TryInto, + D::Error: Into, + { + let conn = tonic::transport::Endpoint::new(dst)?.connect().await?; + Ok(Self::new(conn)) + } + } + impl MusicFsClient + where + T: tonic::client::GrpcService, + T::Error: Into, + T::ResponseBody: Body + std::marker::Send + 'static, + ::Error: Into + std::marker::Send, + { + pub fn new(inner: T) -> Self { + let inner = tonic::client::Grpc::new(inner); + Self { inner } + } + pub fn with_origin(inner: T, origin: Uri) -> Self { + let inner = tonic::client::Grpc::with_origin(inner, origin); + Self { inner } + } + pub fn with_interceptor( + inner: T, + interceptor: F, + ) -> MusicFsClient> + where + F: tonic::service::Interceptor, + T::ResponseBody: Default, + T: tonic::codegen::Service< + http::Request, + Response = http::Response< + >::ResponseBody, + >, + >, + >>::Error: + Into + std::marker::Send + std::marker::Sync, + { + MusicFsClient::new(InterceptedService::new(inner, interceptor)) + } + /// Compress requests with the given encoding. + /// + /// This requires the server to support it otherwise it might respond with an + /// error. + #[must_use] + pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self { + self.inner = self.inner.send_compressed(encoding); + self + } + /// Enable decompressing responses. + #[must_use] + pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self { + self.inner = self.inner.accept_compressed(encoding); + self + } + /// Limits the maximum size of a decoded message. + /// + /// Default: `4MB` + #[must_use] + pub fn max_decoding_message_size(mut self, limit: usize) -> Self { + self.inner = self.inner.max_decoding_message_size(limit); + self + } + /// Limits the maximum size of an encoded message. + /// + /// Default: `usize::MAX` + #[must_use] + pub fn max_encoding_message_size(mut self, limit: usize) -> Self { + self.inner = self.inner.max_encoding_message_size(limit); + self + } + pub async fn get_manifest( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response>, + tonic::Status, + > { + self.inner.ready().await.map_err(|e| { + tonic::Status::unknown(format!("Service was not ready: {}", e.into())) + })?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static("/musicfs.MusicFs/GetManifest"); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("musicfs.MusicFs", "GetManifest")); + self.inner.server_streaming(req, path, codec).await + } + pub async fn get_file( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response>, + tonic::Status, + > { + self.inner.ready().await.map_err(|e| { + tonic::Status::unknown(format!("Service was not ready: {}", e.into())) + })?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static("/musicfs.MusicFs/GetFile"); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("musicfs.MusicFs", "GetFile")); + self.inner.server_streaming(req, path, codec).await + } + pub async fn subscribe_events( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result< + tonic::Response>, + tonic::Status, + > { + self.inner.ready().await.map_err(|e| { + tonic::Status::unknown(format!("Service was not ready: {}", e.into())) + })?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static("/musicfs.MusicFs/SubscribeEvents"); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("musicfs.MusicFs", "SubscribeEvents")); + self.inner.server_streaming(req, path, codec).await + } + } +} +/// Generated server implementations. +pub mod music_fs_server { + #![allow( + unused_variables, + dead_code, + missing_docs, + clippy::wildcard_imports, + clippy::let_unit_value + )] + use tonic::codegen::*; + /// Generated trait containing gRPC methods that should be implemented for use with MusicFsServer. + #[async_trait] + pub trait MusicFs: std::marker::Send + std::marker::Sync + 'static { + /// Server streaming response type for the GetManifest method. + type GetManifestStream: tonic::codegen::tokio_stream::Stream< + Item = std::result::Result, + > + std::marker::Send + + 'static; + async fn get_manifest( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; + /// Server streaming response type for the GetFile method. + type GetFileStream: tonic::codegen::tokio_stream::Stream< + Item = std::result::Result, + > + std::marker::Send + + 'static; + async fn get_file( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; + /// Server streaming response type for the SubscribeEvents method. + type SubscribeEventsStream: tonic::codegen::tokio_stream::Stream< + Item = std::result::Result, + > + std::marker::Send + + 'static; + async fn subscribe_events( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; + } + #[derive(Debug)] + pub struct MusicFsServer { + inner: Arc, + accept_compression_encodings: EnabledCompressionEncodings, + send_compression_encodings: EnabledCompressionEncodings, + max_decoding_message_size: Option, + max_encoding_message_size: Option, + } + impl MusicFsServer { + pub fn new(inner: T) -> Self { + Self::from_arc(Arc::new(inner)) + } + pub fn from_arc(inner: Arc) -> Self { + Self { + inner, + accept_compression_encodings: Default::default(), + send_compression_encodings: Default::default(), + max_decoding_message_size: None, + max_encoding_message_size: None, + } + } + pub fn with_interceptor(inner: T, interceptor: F) -> InterceptedService + where + F: tonic::service::Interceptor, + { + InterceptedService::new(Self::new(inner), interceptor) + } + /// Enable decompressing requests with the given encoding. + #[must_use] + pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self { + self.accept_compression_encodings.enable(encoding); + self + } + /// Compress responses with the given encoding, if the client supports it. + #[must_use] + pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self { + self.send_compression_encodings.enable(encoding); + self + } + /// Limits the maximum size of a decoded message. + /// + /// Default: `4MB` + #[must_use] + pub fn max_decoding_message_size(mut self, limit: usize) -> Self { + self.max_decoding_message_size = Some(limit); + self + } + /// Limits the maximum size of an encoded message. + /// + /// Default: `usize::MAX` + #[must_use] + pub fn max_encoding_message_size(mut self, limit: usize) -> Self { + self.max_encoding_message_size = Some(limit); + self + } + } + impl tonic::codegen::Service> for MusicFsServer + where + T: MusicFs, + B: Body + std::marker::Send + 'static, + B::Error: Into + std::marker::Send + 'static, + { + type Response = http::Response; + type Error = std::convert::Infallible; + type Future = BoxFuture; + fn poll_ready( + &mut self, + _cx: &mut Context<'_>, + ) -> Poll> { + Poll::Ready(Ok(())) + } + fn call(&mut self, req: http::Request) -> Self::Future { + match req.uri().path() { + "/musicfs.MusicFs/GetManifest" => { + #[allow(non_camel_case_types)] + struct GetManifestSvc(pub Arc); + impl + tonic::server::ServerStreamingService + for GetManifestSvc + { + type Response = super::ManifestEntry; + type ResponseStream = T::GetManifestStream; + type Future = + BoxFuture, tonic::Status>; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = + async move { ::get_manifest(&inner, request).await }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = GetManifestSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.server_streaming(method, req).await; + Ok(res) + }; + Box::pin(fut) + } + "/musicfs.MusicFs/GetFile" => { + #[allow(non_camel_case_types)] + struct GetFileSvc(pub Arc); + impl tonic::server::ServerStreamingService for GetFileSvc { + type Response = super::FileChunk; + type ResponseStream = T::GetFileStream; + type Future = + BoxFuture, tonic::Status>; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = + async move { ::get_file(&inner, request).await }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = GetFileSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.server_streaming(method, req).await; + Ok(res) + }; + Box::pin(fut) + } + "/musicfs.MusicFs/SubscribeEvents" => { + #[allow(non_camel_case_types)] + struct SubscribeEventsSvc(pub Arc); + impl + tonic::server::ServerStreamingService + for SubscribeEventsSvc + { + type Response = super::ChangeEvent; + type ResponseStream = T::SubscribeEventsStream; + type Future = + BoxFuture, tonic::Status>; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { + ::subscribe_events(&inner, request).await + }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = SubscribeEventsSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.server_streaming(method, req).await; + Ok(res) + }; + Box::pin(fut) + } + _ => Box::pin(async move { + let mut response = http::Response::new(tonic::body::Body::default()); + let headers = response.headers_mut(); + headers.insert( + tonic::Status::GRPC_STATUS, + (tonic::Code::Unimplemented as i32).into(), + ); + headers.insert( + http::header::CONTENT_TYPE, + tonic::metadata::GRPC_CONTENT_TYPE, + ); + Ok(response) + }), + } + } + } + impl Clone for MusicFsServer { + fn clone(&self) -> Self { + let inner = self.inner.clone(); + Self { + inner, + accept_compression_encodings: self.accept_compression_encodings, + send_compression_encodings: self.send_compression_encodings, + max_decoding_message_size: self.max_decoding_message_size, + max_encoding_message_size: self.max_encoding_message_size, + } + } + } + /// Generated gRPC service name + pub const SERVICE_NAME: &str = "musicfs.MusicFs"; + impl tonic::server::NamedService for MusicFsServer { + const NAME: &'static str = SERVICE_NAME; + } +} diff --git a/src/proto/mod.rs b/src/proto/mod.rs new file mode 100644 index 0000000..ab182fc --- /dev/null +++ b/src/proto/mod.rs @@ -0,0 +1,17 @@ +//! Generated protobuf + tonic bindings for the musicfs service. +//! +//! `buf generate` writes prost messages to `generated/musicfs/musicfs.rs` and +//! the tonic service to `generated/musicfs/musicfs.tonic.rs`. The prost plugin +//! appends `include!("musicfs.tonic.rs")` to the end of `musicfs.rs`, so +//! including that one file here also pulls in the service. The only glue buf +//! does not generate is the module named after the proto package (`musicfs`), +//! which we provide below before re-exporting the types callers use. +pub mod musicfs { + include!("generated/musicfs/musicfs.rs"); +} + +pub use musicfs::{ + ChangeEvent, FileChunk, GetFileRequest, GetManifestRequest, ManifestEntry, MusicMetadata, + PictureDataRange, SubscribeEventsRequest, + music_fs_server::{MusicFs, MusicFsServer}, +}; diff --git a/src/server/manifest.rs b/src/server/manifest.rs new file mode 100644 index 0000000..75e9b8f --- /dev/null +++ b/src/server/manifest.rs @@ -0,0 +1,80 @@ +use crate::music::metadata::MusicMetadata; + +/// One row of the in-memory manifest: every field the client needs to +/// reconstruct an `Item` whose hash matches the server's hash. +/// +/// Field semantics: +/// - `id` is the server filesystem inode. The client uses it as the FUSE +/// inode, which keeps `compute_hash` inputs identical on both sides. +/// - `rel_path` is the path relative to the server's `--source` root, using +/// `/` as separator. The client stores it as `Item.original_path` and uses +/// it as the byte-source locator. +/// - `mtime` / `ctime` / `crtime` are seconds since the Unix epoch. These are +/// the exact three time fields `compute_hash` consumes. +/// - `size` is the real on-disk file size in bytes (the client's virtual size +/// is derived from `music_metadata.virtual_size(size)` when present). +/// - `music_metadata` is the fully-encoded metadata produced server-side via +/// `parse_music_metadata_for_path`. The client never parses audio. +/// +/// Wire format conversion lives in `transport/grpc.rs` (`From` +/// for the generated proto type). +#[derive(Debug, Clone)] +pub struct ManifestEntry { + pub id: u64, + pub rel_path: String, + pub size: u64, + pub mtime: u64, + pub ctime: u64, + pub crtime: u64, + pub music_metadata: Option, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn manifest_entry_clones_fields_into_copy() { + let entry = ManifestEntry { + id: 42, + rel_path: "Artist/Album/track.flac".to_string(), + size: 12345, + mtime: 1_700_000_000, + ctime: 1_699_999_000, + crtime: 1_699_990_000, + music_metadata: None, + }; + + let cloned = entry.clone(); + assert_eq!(cloned.id, entry.id); + assert_eq!(cloned.rel_path, entry.rel_path); + assert_eq!(cloned.size, entry.size); + assert_eq!(cloned.mtime, entry.mtime); + assert_eq!(cloned.ctime, entry.ctime); + assert_eq!(cloned.crtime, entry.crtime); + assert!(cloned.music_metadata.is_none()); + } + + #[test] + fn manifest_entry_carries_music_metadata_when_present() { + let mm = MusicMetadata { + artist: vec!["Test Artist".to_string()], + album: "Test Album".to_string(), + track_title: "Test Title".to_string(), + track_number: 3, + ..MusicMetadata::default() + }; + let entry = ManifestEntry { + id: 1, + rel_path: "a/b.flac".to_string(), + size: 100, + mtime: 0, + ctime: 0, + crtime: 0, + music_metadata: Some(mm.clone()), + }; + + assert_eq!(entry.music_metadata.as_ref().unwrap().album, "Test Album"); + assert_eq!(entry.music_metadata.as_ref().unwrap().track_number, 3); + } +} diff --git a/src/server/mod.rs b/src/server/mod.rs new file mode 100644 index 0000000..ecc3409 --- /dev/null +++ b/src/server/mod.rs @@ -0,0 +1,5 @@ +pub mod manifest; +pub mod range; +pub mod state; +pub mod transport; +pub mod watcher; diff --git a/src/server/range.rs b/src/server/range.rs new file mode 100644 index 0000000..75b829b --- /dev/null +++ b/src/server/range.rs @@ -0,0 +1,175 @@ +/// Parsed `Range: bytes=...` request header. +/// +/// Supported forms (RFC 7233): +/// - `bytes=a-b` → `StartEnd(a, b)` (inclusive) +/// - `bytes=a-` → `Start(a)` +/// - `bytes=-N` → `Suffix(N)` (last N bytes) +/// +/// Returns `None` if the header is missing, uses a unit other than `bytes`, +/// or fails to parse. Multiple ranges (`bytes=a-b,c-d`) are not supported — +/// the first is used and the rest ignored. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ByteRange { + StartEnd(u64, u64), + Start(u64), + Suffix(u64), +} + +pub fn parse_range_header(header: &str) -> Option { + let header = header.trim(); + let rest = header.strip_prefix("bytes=")?; + let first = rest.split(',').next()?.trim(); + let (left, right) = first.split_once('-')?; + let left = left.trim(); + let right = right.trim(); + + return match (left.is_empty(), right.is_empty()) { + (false, false) => Some(ByteRange::StartEnd(left.parse().ok()?, right.parse().ok()?)), + (false, true) => Some(ByteRange::Start(left.parse().ok()?)), + (true, false) => Some(ByteRange::Suffix(right.parse().ok()?)), + (true, true) => None, + }; +} + +/// Resolve a parsed range against a real file size, yielding an absolute +/// `(start, length)` pair suitable for `seek + read`. +/// +/// Returns `None` if the resolved range is unsatisfiable (e.g. start past +/// end of file). The returned `start` is clamped to `[0, size]` and `length` +/// is clamped to not exceed `size - start`. +pub fn resolve_range(range: ByteRange, size: u64) -> Option<(u64, u64)> { + let (start, end_inclusive) = match range { + ByteRange::StartEnd(a, b) => { + if a >= size || a > b { + return None; + } + (a, b.min(size - 1)) + } + ByteRange::Start(a) => { + if a >= size { + return None; + } + (a, size - 1) + } + ByteRange::Suffix(n) => { + if n == 0 || size == 0 { + return None; + } + // A suffix larger than the file clamps to the whole file. + let start = size.saturating_sub(n); + (start, size - 1) + } + }; + return Some((start, end_inclusive - start + 1)); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_start_end() { + assert_eq!( + parse_range_header("bytes=0-499"), + Some(ByteRange::StartEnd(0, 499)) + ); + assert_eq!( + parse_range_header("bytes=500-999"), + Some(ByteRange::StartEnd(500, 999)) + ); + } + + #[test] + fn parse_start_only() { + assert_eq!( + parse_range_header("bytes=9500-"), + Some(ByteRange::Start(9500)) + ); + } + + #[test] + fn parse_suffix() { + assert_eq!( + parse_range_header("bytes=-500"), + Some(ByteRange::Suffix(500)) + ); + } + + #[test] + fn parse_ignores_additional_ranges_after_first() { + assert_eq!( + parse_range_header("bytes=0-499,1000-1499"), + Some(ByteRange::StartEnd(0, 499)) + ); + } + + #[test] + fn parse_returns_none_for_non_bytes_unit() { + assert_eq!(parse_range_header("items=0-4"), None); + } + + #[test] + fn parse_returns_none_for_malformed() { + assert_eq!(parse_range_header("not a range"), None); + assert_eq!(parse_range_header("bytes="), None); + assert_eq!(parse_range_header("bytes=-"), None); + } + + #[test] + fn resolve_start_end_in_bounds() { + assert_eq!( + resolve_range(ByteRange::StartEnd(0, 499), 1000), + Some((0, 500)) + ); + assert_eq!( + resolve_range(ByteRange::StartEnd(100, 199), 1000), + Some((100, 100)) + ); + } + + #[test] + fn resolve_start_end_clamps_end_to_size() { + assert_eq!( + resolve_range(ByteRange::StartEnd(900, 2000), 1000), + Some((900, 100)) + ); + } + + #[test] + fn resolve_start_end_unsatisfiable_when_start_past_size() { + assert_eq!(resolve_range(ByteRange::StartEnd(1500, 2000), 1000), None); + } + + #[test] + fn resolve_start_only() { + assert_eq!(resolve_range(ByteRange::Start(900), 1000), Some((900, 100))); + assert_eq!(resolve_range(ByteRange::Start(0), 1000), Some((0, 1000))); + } + + #[test] + fn resolve_start_unsatisfiable_when_past_size() { + assert_eq!(resolve_range(ByteRange::Start(1500), 1000), None); + } + + #[test] + fn resolve_suffix() { + assert_eq!( + resolve_range(ByteRange::Suffix(500), 1000), + Some((500, 500)) + ); + assert_eq!(resolve_range(ByteRange::Suffix(1), 1000), Some((999, 1))); + } + + #[test] + fn resolve_suffix_larger_than_size_clamps_to_full_file() { + assert_eq!( + resolve_range(ByteRange::Suffix(2000), 1000), + Some((0, 1000)) + ); + } + + #[test] + fn resolve_suffix_zero_unsatisfiable() { + assert_eq!(resolve_range(ByteRange::Suffix(0), 1000), None); + } +} diff --git a/src/server/state.rs b/src/server/state.rs new file mode 100644 index 0000000..3907804 --- /dev/null +++ b/src/server/state.rs @@ -0,0 +1,179 @@ +use std::{ + collections::HashMap, + fs, io, + path::{Path, PathBuf}, + sync::{Arc, Mutex}, + time::{SystemTime, UNIX_EPOCH}, +}; + +use std::os::unix::fs::MetadataExt; + +use crate::music::parse::parse_music_metadata_for_path; +use crate::origins::attrs::FileAttrs; +use crate::server::manifest::ManifestEntry; + +/// Server-side entry for one file. Built once on startup from a directory +/// scan and refreshed by the watcher on inotify events. +#[derive(Debug, Clone)] +pub struct FileEntry { + pub abs_path: PathBuf, + pub rel_path: String, + pub attrs: FileAttrs, + pub music_metadata: Option, +} + +#[derive(Clone)] +pub struct ServerState { + inner: Arc>>, +} + +impl ServerState { + pub fn new() -> Self { + return ServerState { + inner: Arc::new(Mutex::new(HashMap::new())), + }; + } + + /// Replace the entire map with a fresh scan of `source`. Used on startup + /// and on watcher-driven reconciliations. + pub fn replace_all(&self, source: &Path) -> io::Result<()> { + let entries = scan_directory(source)?; + let mut map = self.inner.lock().unwrap(); + map.clear(); + for (id, entry) in entries { + map.insert(id, entry); + } + return Ok(()); + } + + /// Snapshot the current state into a manifest. Order is by inode ascending + /// so two scans of an unchanged library serialize identically. + pub fn manifest(&self) -> Vec { + let map = self.inner.lock().unwrap(); + let mut entries: Vec = map + .iter() + .map(|(id, file)| ManifestEntry { + id: *id, + rel_path: file.rel_path.clone(), + size: file.attrs.size, + mtime: secs(file.attrs.mtime), + ctime: secs(file.attrs.ctime), + crtime: secs(file.attrs.crtime), + music_metadata: file.music_metadata.clone(), + }) + .collect(); + entries.sort_by_key(|e| e.id); + return entries; + } + + pub fn lookup(&self, id: u64) -> Option { + return self.inner.lock().unwrap().get(&id).cloned(); + } +} + +fn secs(time: SystemTime) -> u64 { + return time + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); +} + +fn scan_directory(source: &Path) -> io::Result> { + let mut out = Vec::new(); + walk(source, source, &mut out)?; + return Ok(out); +} + +fn walk(source: &Path, dir: &Path, out: &mut Vec<(u64, FileEntry)>) -> io::Result<()> { + for entry in fs::read_dir(dir)? { + let entry = entry?; + let path = entry.path(); + let file_type = entry.file_type()?; + if file_type.is_dir() { + walk(source, &path, out)?; + continue; + } + if !file_type.is_file() { + continue; + } + + let metadata = entry.metadata()?; + let rel_path = path + .strip_prefix(source) + .map(|p| p.to_path_buf()) + .unwrap_or_else(|_| path.clone()) + .to_string_lossy() + .replace('\\', "/"); + + let inode = metadata.ino(); + let music_metadata = parse_music_metadata_for_path(&path); + let file_entry = FileEntry { + abs_path: path, + rel_path, + attrs: FileAttrs::from(&metadata), + music_metadata, + }; + out.push((inode, file_entry)); + } + return Ok(()); +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write; + + #[test] + fn replace_all_loads_files_into_state() { + let tmp = tempfile::tempdir().unwrap(); + let source = tmp.path(); + + let mut f = fs::File::create(source.join("a.txt")).unwrap(); + f.write_all(b"hello").unwrap(); + drop(f); + let mut f = fs::File::create(source.join("b.txt")).unwrap(); + f.write_all(b"world!").unwrap(); + drop(f); + + let state = ServerState::new(); + state.replace_all(source).unwrap(); + + let manifest = state.manifest(); + assert_eq!(manifest.len(), 2); + assert!(manifest.iter().any(|e| e.rel_path == "a.txt")); + assert!(manifest.iter().any(|e| e.rel_path == "b.txt")); + } + + #[test] + fn manifest_entries_sorted_by_id_ascending() { + let tmp = tempfile::tempdir().unwrap(); + let source = tmp.path(); + // Create files in any order; inode order is determined by the FS. + for name in ["z.txt", "a.txt", "m.txt"] { + fs::write(source.join(name), b"x").unwrap(); + } + + let state = ServerState::new(); + state.replace_all(source).unwrap(); + + let ids: Vec = state.manifest().iter().map(|e| e.id).collect(); + let mut sorted = ids.clone(); + sorted.sort(); + assert_eq!(ids, sorted); + } + + #[test] + fn replace_all_clears_existing_entries() { + let tmp = tempfile::tempdir().unwrap(); + let source = tmp.path(); + fs::write(source.join("a.txt"), b"x").unwrap(); + + let state = ServerState::new(); + state.replace_all(source).unwrap(); + assert_eq!(state.manifest().len(), 1); + + fs::remove_file(source.join("a.txt")).unwrap(); + state.replace_all(source).unwrap(); + assert_eq!(state.manifest().len(), 0); + } +} diff --git a/src/server/transport/grpc.rs b/src/server/transport/grpc.rs new file mode 100644 index 0000000..b5c318a --- /dev/null +++ b/src/server/transport/grpc.rs @@ -0,0 +1,214 @@ +use std::{ + fs, + io::{Read, Seek, SeekFrom}, + path::PathBuf, + pin::Pin, +}; + +use anyhow::Result; +use async_trait::async_trait; +use tokio::sync::broadcast::Receiver; +use tonic::{Request, Response, Status, transport::Server}; + +use crate::music::metadata::MusicMetadata; +use crate::proto as proto_types; +use crate::proto::MusicFs as MusicFsTrait; +use crate::proto::{ + ChangeEvent as ProtoChangeEvent, FileChunk, GetFileRequest, GetManifestRequest, ManifestEntry, + MusicFsServer, MusicMetadata as ProtoMusicMetadata, PictureDataRange, SubscribeEventsRequest, +}; +use crate::server::manifest::ManifestEntry as DomainManifestEntry; +use crate::server::state::ServerState; +use crate::server::transport::{MusicTransport, TransportArgs}; +use crate::server::watcher::{ChangeEvent, ChangeKind}; + +pub struct GrpcTransport { + args: TransportArgs, +} + +impl GrpcTransport { + pub fn new(args: TransportArgs) -> Self { + return GrpcTransport { args }; + } +} + +#[async_trait] +impl MusicTransport for GrpcTransport { + async fn run(self: Box, state: ServerState, events: Receiver) -> Result<()> { + let service = MusicFsService { + state, + events: tokio::sync::Mutex::new(events), + }; + let listen = self.args.listen; + println!("musicfs-server gRPC listening on {listen}"); + Server::builder() + .add_service(MusicFsServer::new(service)) + .serve(listen) + .await?; + return Ok(()); + } +} + +struct MusicFsService { + state: ServerState, + events: tokio::sync::Mutex>, +} + +type BoxStream = Pin> + Send>>; + +#[tonic::async_trait] +impl MusicFsTrait for MusicFsService { + type GetManifestStream = BoxStream; + type GetFileStream = BoxStream; + type SubscribeEventsStream = BoxStream; + + async fn get_manifest( + &self, + _request: Request, + ) -> Result, Status> { + let entries: Vec = self.state.manifest(); + let stream = tokio_stream::iter( + entries + .into_iter() + .map(|e| Ok::(e.into())), + ); + return Ok(Response::new(Box::pin(stream))); + } + + async fn get_file( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let id = req.id; + let entry = self + .state + .lookup(id) + .ok_or_else(|| Status::not_found(format!("no file with id {id}")))?; + let abs_path: PathBuf = entry.abs_path.clone(); + let total_size = entry.attrs.size; + let start = if req.start == 0 && req.length == 0 { + 0u64 + } else { + req.start + }; + let length = if req.start == 0 && req.length == 0 { + total_size + } else if req.length == 0 { + total_size.saturating_sub(start) + } else { + req.length + }; + + let chunk = + tokio::task::spawn_blocking(move || read_range_blocking(&abs_path, start, length)) + .await + .map_err(|e| Status::internal(format!("join blocking read: {e}")))? + .map_err(|e| Status::internal(format!("read file range: {e}")))?; + + let chunk_stream = tokio_stream::iter(vec![Ok::(FileChunk { + data: chunk, + offset: start, + total_size, + })]); + return Ok(Response::new(Box::pin(chunk_stream))); + } + + async fn subscribe_events( + &self, + _request: Request, + ) -> Result, Status> { + // Take an owned receiver (the guard is dropped at the end of this + // statement) so it can move into the 'static spawned task, and so each + // subscriber gets its own receiver rather than contending on one. + let mut rx = self.events.lock().await.resubscribe(); + let (tx, rx_stream) = + tokio::sync::mpsc::unbounded_channel::>(); + tokio::spawn(async move { + loop { + match rx.recv().await { + Ok(event) => { + let proto = ProtoChangeEvent { + kind: kind_to_string(event.kind), + }; + if tx.send(Ok(proto)).is_err() { + break; + } + } + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + } + }); + let stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx_stream); + return Ok(Response::new(Box::pin(stream))); + } +} + +fn read_range_blocking( + path: &std::path::Path, + start: u64, + length: u64, +) -> std::io::Result> { + let mut file = fs::File::open(path)?; + file.seek(SeekFrom::Start(start))?; + let mut buf = vec![0u8; length as usize]; + let mut filled = 0usize; + while filled < buf.len() { + let n = file.read(&mut buf[filled..])?; + if n == 0 { + buf.truncate(filled); + break; + } + filled += n; + } + return Ok(buf); +} + +fn kind_to_string(kind: ChangeKind) -> String { + return match kind { + ChangeKind::Create => "create".to_string(), + ChangeKind::Modify => "modify".to_string(), + ChangeKind::Remove => "remove".to_string(), + }; +} + +impl From for ManifestEntry { + fn from(entry: DomainManifestEntry) -> Self { + return ManifestEntry { + id: entry.id, + rel_path: entry.rel_path, + size: entry.size, + mtime: entry.mtime, + ctime: entry.ctime, + crtime: entry.crtime, + music_metadata: entry.music_metadata.map(Into::into), + }; + } +} + +impl From for ProtoMusicMetadata { + fn from(mm: MusicMetadata) -> Self { + return ProtoMusicMetadata { + artist: mm.artist, + album_artist: mm.album_artist, + album: mm.album, + track_number: mm.track_number, + track_title: mm.track_title, + other_tags: mm.other_tags, + header: mm.header, + picture_block_headers: mm.picture_block_headers, + picture_data_ranges: mm + .picture_data_ranges + .into_iter() + .map(|(offset, length)| PictureDataRange { offset, length }) + .collect(), + real_audio_start: mm.real_audio_start, + vorbis_comment_offset: mm.vorbis_comment_offset, + vorbis_comment_length: mm.vorbis_comment_length, + }; + } +} + +#[allow(unused_imports)] +use proto_types as _proto_types_anchor; diff --git a/src/server/transport/mod.rs b/src/server/transport/mod.rs new file mode 100644 index 0000000..dcee191 --- /dev/null +++ b/src/server/transport/mod.rs @@ -0,0 +1,39 @@ +pub mod grpc; + +use std::net::SocketAddr; + +use anyhow::{Result, anyhow}; +use async_trait::async_trait; +use tokio::sync::broadcast::Receiver; + +use crate::server::state::ServerState; +use crate::server::watcher::ChangeEvent; + +/// Transport-agnostic musicfs server. A single implementation is wired in +/// today (`grpc`); the factory in [`build`] is the seam where additional +/// transports (HTTP/3, raw QUIC, in-process for tests, ...) get plugged in +/// without touching the binary. +/// +/// `run` consumes `self` — a transport serves exactly once. It receives the +/// shared [`ServerState`] and a fresh broadcast receiver for change events. +#[async_trait] +pub trait MusicTransport: Send + Sync + 'static { + async fn run(self: Box, state: ServerState, events: Receiver) -> Result<()>; +} + +/// Arguments every transport understands. Transports may extend this with +/// their own configuration via their constructors. +#[derive(Debug, Clone, Copy)] +pub struct TransportArgs { + pub listen: SocketAddr, +} + +/// Construct the named transport. The list of accepted names is intentionally +/// discoverable (a single match arm) so adding a transport means adding an +/// arm here plus a module under `transport/`. +pub fn build(name: &str, args: TransportArgs) -> Result> { + return match name { + "grpc" => Ok(Box::new(grpc::GrpcTransport::new(args))), + other => Err(anyhow!("unknown transport: {other}")), + }; +} diff --git a/src/server/watcher.rs b/src/server/watcher.rs new file mode 100644 index 0000000..a704e71 --- /dev/null +++ b/src/server/watcher.rs @@ -0,0 +1,99 @@ +use std::{ + path::PathBuf, + sync::{Arc, Mutex}, + thread, +}; + +use notify::{EventKind, RecursiveMode, Watcher}; +use tokio::sync::broadcast; + +use crate::server::state::ServerState; + +/// A change observed by the watcher. Pushed onto the broadcast channel for +/// `/events` subscribers. The client treats these as wake-ups: correctness +/// always rests on the subsequent `/manifest` hash diff. +#[derive(Debug, Clone)] +pub struct ChangeEvent { + pub kind: ChangeKind, +} + +#[derive(Debug, Clone, Copy)] +pub enum ChangeKind { + Create, + Modify, + Remove, +} + +/// Server-side file watcher. Owns a background std thread driving `notify` +/// (inotify on Linux). On any filesystem event under `source` it: +/// 1. Rebuilds the shared [`ServerState`] from a fresh scan. +/// 2. Broadcasts a [`ChangeEvent`] so connected `/events` clients wake up. +/// +/// The state rebuild is full-scan rather than incremental. This matches the +/// existing `LocalOriginFileWatcher` pattern, keeps the watcher simple, and +/// is cheap on a LAN-scale library. Hash-diff reconciliation on the client +/// absorbs any over-reporting. +pub struct ServerWatcher { + _events_tx: broadcast::Sender, + _worker: Arc>>>, +} + +impl ServerWatcher { + /// Spawn the watcher. Returns the broadcast receiver that `/events` + /// handlers subscribe to. + pub fn spawn(source: PathBuf, state: ServerState) -> (Self, broadcast::Receiver) { + let (events_tx, events_rx) = broadcast::channel(64); + let worker_tx = events_tx.clone(); + let handle = thread::spawn(move || { + run_watcher_loop(source, state, worker_tx); + }); + let watcher = ServerWatcher { + _events_tx: events_tx, + _worker: Arc::new(Mutex::new(Some(handle))), + }; + return (watcher, events_rx); + } +} + +fn run_watcher_loop( + source: PathBuf, + state: ServerState, + events_tx: broadcast::Sender, +) { + let (tx, rx) = std::sync::mpsc::channel(); + let mut watcher = match notify::recommended_watcher(tx) { + Ok(w) => w, + Err(e) => { + eprintln!("server watcher: failed to create inotify watcher: {e}"); + return; + } + }; + + if let Err(e) = watcher.watch(&source, RecursiveMode::Recursive) { + eprintln!("server watcher: failed to watch {}: {e}", source.display()); + return; + } + + println!("server watcher: scanning {source:?} for changes"); + for res in rx { + match res { + Ok(event) => match event.kind { + EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_) => { + let kind = match event.kind { + EventKind::Create(_) => ChangeKind::Create, + EventKind::Modify(_) => ChangeKind::Modify, + EventKind::Remove(_) => ChangeKind::Remove, + _ => continue, + }; + if let Err(e) = state.replace_all(&source) { + eprintln!("server watcher: state refresh failed: {e}"); + continue; + } + let _ = events_tx.send(ChangeEvent { kind }); + } + _ => {} + }, + Err(e) => eprintln!("server watcher: inotify error: {e}"), + } + } +}