tsiry-sandratraina.com
A tiny Subsonic/Jellyfin/S3 server in Rust
navidrome
subsonic
s3
emby
jellyfin
1use crate::db::Db;
2use crate::scanner::{gc_album, gc_artist, has_audio_extension, upsert_file};
3use crate::typesense::TypesenseClient;
4use anyhow::Result;
5use notify::{EventKind, RecursiveMode};
6use notify_debouncer_full::new_debouncer;
7use std::path::{Path, PathBuf};
8use std::sync::Arc;
9use std::time::Duration;
10
11const DEBOUNCE: Duration = Duration::from_secs(2);
12
13pub fn start(
14 pool: Db,
15 music_dir: PathBuf,
16 covers_dir: PathBuf,
17 typesense: Option<Arc<TypesenseClient>>,
18) {
19 let rt = tokio::runtime::Handle::current();
20 std::thread::Builder::new()
21 .name("library-watcher".into())
22 .spawn(move || {
23 if let Err(e) = run(pool, music_dir, covers_dir, typesense, rt) {
24 tracing::error!("watcher stopped: {e}");
25 }
26 })
27 .expect("spawn watcher thread");
28}
29
30fn run(
31 pool: Db,
32 music_dir: PathBuf,
33 covers_dir: PathBuf,
34 typesense: Option<Arc<TypesenseClient>>,
35 rt: tokio::runtime::Handle,
36) -> Result<()> {
37 let (tx, rx) = std::sync::mpsc::channel();
38 let mut debouncer = new_debouncer(DEBOUNCE, None, tx)?;
39 debouncer.watch(&music_dir, RecursiveMode::Recursive)?;
40 tracing::info!("watching {} for changes", music_dir.display());
41
42 for result in rx {
43 match result {
44 Ok(events) => {
45 for ev in events {
46 if matches!(ev.event.kind, EventKind::Access(_) | EventKind::Other) {
47 continue;
48 }
49 for path in &ev.event.paths {
50 rt.block_on(apply_path(&pool, &covers_dir, typesense.as_deref(), path));
51 }
52 }
53 }
54 Err(errors) => {
55 for e in errors {
56 tracing::warn!("watcher error: {e}");
57 }
58 }
59 }
60 }
61 drop(debouncer);
62 Ok(())
63}
64
65async fn apply_path(
66 pool: &Db,
67 covers_dir: &Path,
68 typesense: Option<&TypesenseClient>,
69 path: &Path,
70) {
71 if path.is_file() {
72 if !has_audio_extension(path) {
73 return;
74 }
75 match upsert_file(pool, path, covers_dir, typesense).await {
76 Ok(_) => tracing::info!("watch upsert {}", path.display()),
77 Err(e) => tracing::warn!("watch upsert {}: {e}", path.display()),
78 }
79 return;
80 }
81
82 if path.exists() {
83 return;
84 }
85
86 let path_str = path.to_string_lossy().to_string();
87 if has_audio_extension(path) {
88 match delete_song(pool, &path_str, typesense).await {
89 Ok(true) => tracing::info!("watch delete {path_str}"),
90 Ok(false) => {}
91 Err(e) => tracing::warn!("watch delete {path_str}: {e}"),
92 }
93 return;
94 }
95
96 match delete_songs_under(pool, &path_str, typesense).await {
97 Ok(n) if n > 0 => tracing::info!("watch delete {n} songs under {path_str}"),
98 Ok(_) => {}
99 Err(e) => tracing::warn!("watch delete dir {path_str}: {e}"),
100 }
101}
102
103async fn delete_song(pool: &Db, path: &str, typesense: Option<&TypesenseClient>) -> Result<bool> {
104 let row = sqlx::query_as::<_, (String, String, String)>(
105 "SELECT id, album_id, artist_id FROM songs WHERE path = ?1",
106 )
107 .bind(path)
108 .fetch_optional(pool)
109 .await?;
110
111 let Some((song_id, album_id, artist_id)) = row else {
112 return Ok(false);
113 };
114
115 sqlx::query("DELETE FROM songs WHERE path = ?1")
116 .bind(path)
117 .execute(pool)
118 .await?;
119
120 if let Some(ts) = typesense {
121 if let Err(e) = ts.delete_song(&song_id).await {
122 tracing::warn!("typesense delete song {song_id}: {e}");
123 }
124 }
125
126 gc_album(pool, &album_id, typesense).await?;
127 gc_artist(pool, &artist_id, typesense).await?;
128 Ok(true)
129}
130
131async fn delete_songs_under(
132 pool: &Db,
133 dir: &str,
134 typesense: Option<&TypesenseClient>,
135) -> Result<u64> {
136 let mut prefix = dir.trim_end_matches(std::path::MAIN_SEPARATOR).to_string();
137 prefix.push(std::path::MAIN_SEPARATOR);
138 let pattern = format!("{}%", escape_like(&prefix));
139
140 let rows: Vec<(String, String, String)> =
141 sqlx::query_as("SELECT id, album_id, artist_id FROM songs WHERE path LIKE ?1 ESCAPE '\\'")
142 .bind(&pattern)
143 .fetch_all(pool)
144 .await?;
145
146 if rows.is_empty() {
147 return Ok(0);
148 }
149
150 let deleted = sqlx::query("DELETE FROM songs WHERE path LIKE ?1 ESCAPE '\\'")
151 .bind(&pattern)
152 .execute(pool)
153 .await?
154 .rows_affected();
155
156 let mut seen_albums = std::collections::HashSet::new();
157 let mut seen_artists = std::collections::HashSet::new();
158 for (song_id, album_id, artist_id) in rows {
159 if let Some(ts) = typesense {
160 if let Err(e) = ts.delete_song(&song_id).await {
161 tracing::warn!("typesense delete song {song_id}: {e}");
162 }
163 }
164 if seen_albums.insert(album_id.clone()) {
165 gc_album(pool, &album_id, typesense).await?;
166 }
167 if seen_artists.insert(artist_id.clone()) {
168 gc_artist(pool, &artist_id, typesense).await?;
169 }
170 }
171 Ok(deleted)
172}
173
174fn escape_like(s: &str) -> String {
175 let mut out = String::with_capacity(s.len());
176 for ch in s.chars() {
177 match ch {
178 '\\' | '%' | '_' => {
179 out.push('\\');
180 out.push(ch);
181 }
182 _ => out.push(ch),
183 }
184 }
185 out
186}