tsiry-sandratraina.com
A tiny Subsonic/Jellyfin/S3 server in Rust
navidrome
subsonic
s3
emby
jellyfin
Add optional Typesense search backend
Free-text search (search3/search2/Jellyfin ?searchTerm=) now routes
through Typesense when [typesense] is present in smolsonic.toml,
otherwise keeps using the existing SQLite FTS5 index. Scanner and
watcher mirror inserts/updates/deletes into Typesense; a fresh node
self-heals via a bulk reindex from SQLite on startup. Any Typesense
error transparently falls back to FTS5 so search stays working.
author
Tsiry Sandratraina
date
(Jul 3, 2026, 10:58 AM +0300)
commit
83fb4532
83fb4532fedbc4adef2ea4cd5c9c96684e5181d7
parent
cb96f7bc
cb96f7bc37e8c9aa40ac87d67c18f7d3cf5659f7
+880
-54
+29
README.md
+29
README.md
···
374
374
avahi-browse -r _subsonic._tcp
375
375
```
376
376
377
+
### Search backends
378
+
379
+
Free-text search (`/rest/search3`, `/rest/search2`, Jellyfin `?searchTerm=`)
380
+
defaults to SQLite's built-in FTS5 index. Add a `[typesense]` block to swap
381
+
in [Typesense](https://typesense.org/) for typo-tolerant search and better
382
+
ranking without changing any client:
383
+
384
+
```toml
385
+
[typesense]
386
+
url = "http://localhost:8108"
387
+
api_key = "changeme"
388
+
# collection_prefix = "smolsonic" # optional
389
+
```
390
+
391
+
Boot a local Typesense in Docker:
392
+
393
+
```sh
394
+
docker run -d --name typesense -p 8108:8108 -v typesense-data:/data \
395
+
typesense/typesense:27.1 --data-dir /data --api-key changeme
396
+
```
397
+
398
+
On startup smolsonic creates the three collections (`{prefix}_songs`,
399
+
`{prefix}_albums`, `{prefix}_artists`) and — if the songs collection is
400
+
empty — bulk-imports every row from SQLite. From then on, scanner and
401
+
watcher mirror inserts/updates/deletes into Typesense automatically.
402
+
If Typesense is unreachable or misbehaves, search transparently falls back
403
+
to FTS5 so queries keep working. Remove the `[typesense]` block and restart
404
+
to go back to FTS5-only.
405
+
377
406
## CLI
378
407
379
408
```
+10
smolsonic.example.toml
+10
smolsonic.example.toml
···
40
40
# agent per their TOS — include app name, version, and contact info.
41
41
# [musicbrainz]
42
42
# user_agent = "smolsonic/0.7.0 ( you@example.com )"
43
+
44
+
# Optional Typesense search backend. When set, free-text search
45
+
# (/rest/search3, /rest/search2, Jellyfin ?searchTerm=) uses Typesense
46
+
# instead of the built-in SQLite FTS5 index — typo-tolerance, better ranking.
47
+
# Omit the block to keep using FTS5 (the default). Fresh Typesense nodes are
48
+
# reindexed automatically from SQLite on first startup.
49
+
# [typesense]
50
+
# url = "http://localhost:8108"
51
+
# api_key = "changeme"
52
+
# collection_prefix = "smolsonic" # optional; lets multiple instances share a node
+30
src/config.rs
+30
src/config.rs
···
34
34
/// Supplements Last.fm with band-member / collaborator links.
35
35
#[serde(default)]
36
36
pub musicbrainz: Option<MusicbrainzConfig>,
37
+
/// Optional Typesense search backend. When present, the free-text
38
+
/// `search3` / `search2` / Jellyfin `?searchTerm=` endpoints use Typesense
39
+
/// instead of the built-in SQLite FTS5 index. Omit the block to keep
40
+
/// using FTS5.
41
+
#[serde(default)]
42
+
pub typesense: Option<TypesenseConfig>,
37
43
}
38
44
39
45
#[derive(Debug, Clone, Deserialize)]
···
44
50
#[derive(Debug, Clone, Deserialize)]
45
51
pub struct MusicbrainzConfig {
46
52
pub user_agent: String,
53
+
}
54
+
55
+
/// Optional Typesense search backend. `url` points at the Typesense HTTP
56
+
/// endpoint (no trailing slash); `api_key` is the admin key so smolsonic can
57
+
/// create collections and import documents. `collection_prefix` lets multiple
58
+
/// smolsonic instances share a single Typesense node without colliding.
59
+
#[derive(Debug, Clone, Deserialize)]
60
+
pub struct TypesenseConfig {
61
+
pub url: String,
62
+
pub api_key: String,
63
+
#[serde(default = "default_typesense_prefix")]
64
+
pub collection_prefix: String,
65
+
}
66
+
67
+
fn default_typesense_prefix() -> String {
68
+
"smolsonic".to_string()
47
69
}
48
70
49
71
/// Optional video library. Enabled only when this block is present in the
···
174
196
if s3.secret_key.is_empty() {
175
197
anyhow::bail!("config: s3.secret_key must not be empty");
176
198
}
199
+
}
200
+
}
201
+
if let Some(ts) = &cfg.typesense {
202
+
if ts.url.is_empty() {
203
+
anyhow::bail!("config: typesense.url must not be empty");
204
+
}
205
+
if ts.api_key.is_empty() {
206
+
anyhow::bail!("config: typesense.api_key must not be empty");
177
207
}
178
208
}
179
209
Ok(cfg)
+10
-7
src/jellyfin/handlers.rs
+10
-7
src/jellyfin/handlers.rs
···
956
956
let limit = q.limit.unwrap_or(50).max(1);
957
957
let no_filter = q.include_item_types.is_none();
958
958
let mut dtos: Vec<BaseItemDto> = Vec::new();
959
+
let ts = state.typesense.as_deref();
959
960
if no_filter || includes(&q.include_item_types, "MusicArtist") {
960
-
if let Ok(rows) = repo::search_artists(&state.pool, term, limit, 0).await {
961
+
if let Ok(rows) = repo::search_artists(&state.pool, term, limit, 0, ts).await {
961
962
for a in rows {
962
963
dtos.push(artist_to_dto(&state, &a).await);
963
964
}
964
965
}
965
966
}
966
967
if no_filter || includes(&q.include_item_types, "MusicAlbum") {
967
-
if let Ok(rows) = repo::search_albums(&state.pool, term, limit, 0).await {
968
+
if let Ok(rows) = repo::search_albums(&state.pool, term, limit, 0, ts).await {
968
969
for a in rows {
969
970
dtos.push(album_to_dto(&state, &a).await);
970
971
}
971
972
}
972
973
}
973
974
if no_filter || includes(&q.include_item_types, "Audio") {
974
-
if let Ok(rows) = repo::search_songs(&state.pool, term, limit, 0).await {
975
+
if let Ok(rows) = repo::search_songs(&state.pool, term, limit, 0, ts).await {
975
976
for s in rows {
976
977
dtos.push(song_to_dto(&state, &s).await);
977
978
}
···
4123
4124
let music_dir = state.music_dir.clone();
4124
4125
let covers_dir = state.covers_dir.clone();
4125
4126
let progress = state.music_scan_progress.clone();
4127
+
let ts = state.typesense.clone();
4126
4128
tokio::spawn(async move {
4127
4129
tracing::info!("jellyfin: triggered music scan of {}", music_dir.display());
4128
-
if let Err(e) = crate::scanner::scan(pool, music_dir, covers_dir, progress).await {
4130
+
if let Err(e) = crate::scanner::scan(pool, music_dir, covers_dir, progress, ts).await {
4129
4131
tracing::error!("jellyfin-triggered music scan failed: {e}");
4130
4132
}
4131
4133
});
···
4422
4424
4423
4425
let limit = q.limit.unwrap_or(20).max(1);
4424
4426
let mut hints: Vec<Value> = Vec::new();
4427
+
let ts = state.typesense.as_deref();
4425
4428
4426
4429
if want_artist {
4427
-
if let Ok(rows) = repo::search_artists(&state.pool, term, limit, 0).await {
4430
+
if let Ok(rows) = repo::search_artists(&state.pool, term, limit, 0, ts).await {
4428
4431
for a in rows {
4429
4432
let id = mapping::remember_artist(&state.pool, &a)
4430
4433
.await
···
4441
4444
}
4442
4445
}
4443
4446
if want_album {
4444
-
if let Ok(rows) = repo::search_albums(&state.pool, term, limit, 0).await {
4447
+
if let Ok(rows) = repo::search_albums(&state.pool, term, limit, 0, ts).await {
4445
4448
for al in rows {
4446
4449
let id = mapping::remember_album(&state.pool, &al)
4447
4450
.await
···
4460
4463
}
4461
4464
}
4462
4465
if want_song {
4463
-
if let Ok(rows) = repo::search_songs(&state.pool, term, limit, 0).await {
4466
+
if let Ok(rows) = repo::search_songs(&state.pool, term, limit, 0, ts).await {
4464
4467
for s in rows {
4465
4468
let id = mapping::remember_song(&state.pool, &s)
4466
4469
.await
+7
src/jellyfin/mod.rs
+7
src/jellyfin/mod.rs
···
9
9
use crate::config::JellyfinConfig;
10
10
use crate::db::Db;
11
11
use crate::scanner::ScanProgress;
12
+
use crate::typesense::TypesenseClient;
12
13
use crate::video_scanner::VideoScanProgress;
13
14
use actix_cors::Cors;
14
15
use actix_web::{web, App, HttpRequest, HttpResponse, HttpServer};
···
34
35
/// are `None` when their `[lastfm]` / `[musicbrainz]` blocks are absent
35
36
/// — the Similar handlers short-circuit to empty in that case.
36
37
pub similar: Arc<similar::SimilarProviders>,
38
+
/// Optional Typesense search backend for `?searchTerm=`. Falls back to
39
+
/// FTS5 on error; `None` keeps the FTS5 path.
40
+
pub typesense: Option<Arc<TypesenseClient>>,
37
41
}
38
42
39
43
pub async fn start(
···
49
53
musicbrainz: Option<crate::config::MusicbrainzConfig>,
50
54
music_scan_progress: Arc<ScanProgress>,
51
55
video_scan_progress: Arc<VideoScanProgress>,
56
+
typesense: Option<Arc<TypesenseClient>>,
52
57
) -> anyhow::Result<()> {
53
58
let server_id = auth::ensure_server_id(&pool).await?;
54
59
let user_id = mapping::user_guid(&username);
···
80
85
music_scan_progress,
81
86
video_scan_progress,
82
87
similar: similar_providers,
88
+
typesense,
83
89
});
84
90
85
91
tracing::info!(
···
714
720
music_scan_progress: Arc::new(ScanProgress::default()),
715
721
video_scan_progress: Arc::new(VideoScanProgress::default()),
716
722
similar: Arc::new(similar::SimilarProviders::new(None, None)),
723
+
typesense: None,
717
724
}
718
725
}
719
726
+54
-3
src/main.rs
+54
-3
src/main.rs
···
7
7
mod s3;
8
8
mod scanner;
9
9
mod server;
10
+
mod typesense;
10
11
mod video_scanner;
11
12
mod watcher;
12
13
···
45
46
let pool = db::init(&cfg.database_path).await?;
46
47
let scan_progress = Arc::new(scanner::ScanProgress::default());
47
48
49
+
// Optional Typesense client — construct once, share via `Arc`. On startup
50
+
// we call `bootstrap()` to create collections, then reindex from SQLite
51
+
// if the songs collection is empty (fresh Typesense node self-heals).
52
+
let typesense: Option<Arc<typesense::TypesenseClient>> =
53
+
if let Some(ts_cfg) = cfg.typesense.as_ref() {
54
+
let client = Arc::new(typesense::TypesenseClient::new(ts_cfg));
55
+
match client.bootstrap().await {
56
+
Ok(()) => {
57
+
let client_c = client.clone();
58
+
let pool_c = pool.clone();
59
+
tokio::spawn(async move {
60
+
match client_c.songs_empty().await {
61
+
Ok(true) => {
62
+
tracing::info!("typesense: songs collection empty, reindexing");
63
+
if let Err(e) = client_c.reindex_from_db(&pool_c).await {
64
+
tracing::error!("typesense reindex failed: {e}");
65
+
}
66
+
}
67
+
Ok(false) => {
68
+
tracing::info!(
69
+
"typesense: songs collection non-empty, skipping reindex"
70
+
);
71
+
}
72
+
Err(e) => {
73
+
tracing::error!("typesense describe songs: {e}");
74
+
}
75
+
}
76
+
});
77
+
Some(client)
78
+
}
79
+
Err(e) => {
80
+
tracing::error!("typesense: bootstrap failed ({e}) — falling back to FTS5");
81
+
None
82
+
}
83
+
}
84
+
} else {
85
+
None
86
+
};
87
+
48
88
if !args.no_scan {
49
89
let pool_c = pool.clone();
50
90
let music_dir = cfg.music_dir.clone();
51
91
let covers_dir = cfg.covers_dir.clone();
52
92
let progress = scan_progress.clone();
93
+
let ts_c = typesense.clone();
53
94
tokio::spawn(async move {
54
95
tracing::info!("starting library scan of {}", music_dir.display());
55
96
if let Err(e) = scanner::scan(
···
57
98
music_dir.clone(),
58
99
covers_dir.clone(),
59
100
progress,
101
+
ts_c.clone(),
60
102
)
61
103
.await
62
104
{
63
105
tracing::error!("scan failed: {e}");
64
106
}
65
-
watcher::start(pool_c, music_dir, covers_dir);
107
+
watcher::start(pool_c, music_dir, covers_dir, ts_c);
66
108
});
67
109
} else {
68
-
watcher::start(pool.clone(), cfg.music_dir.clone(), cfg.covers_dir.clone());
110
+
watcher::start(
111
+
pool.clone(),
112
+
cfg.music_dir.clone(),
113
+
cfg.covers_dir.clone(),
114
+
typesense.clone(),
115
+
);
69
116
}
70
117
71
118
if cfg.scan_interval_secs > 0 {
···
73
120
let music_dir = cfg.music_dir.clone();
74
121
let covers_dir = cfg.covers_dir.clone();
75
122
let progress = scan_progress.clone();
123
+
let ts_c = typesense.clone();
76
124
let interval = Duration::from_secs(cfg.scan_interval_secs);
77
125
tokio::spawn(async move {
78
126
let mut ticker = tokio::time::interval(interval);
···
90
138
music_dir.clone(),
91
139
covers_dir.clone(),
92
140
progress.clone(),
141
+
ts_c.clone(),
93
142
)
94
143
.await
95
144
{
···
177
226
let musicbrainz = cfg.musicbrainz.clone();
178
227
let music_progress = scan_progress.clone();
179
228
let video_progress = video_scan_progress.clone();
229
+
let ts_c = typesense.clone();
180
230
actix_web::rt::spawn(async move {
181
231
if let Err(e) = jellyfin::start(
182
232
jf_cfg_c,
···
191
241
musicbrainz,
192
242
music_progress,
193
243
video_progress,
244
+
ts_c,
194
245
)
195
246
.await
196
247
{
···
247
298
None
248
299
};
249
300
250
-
server::start(cfg, pool, scan_progress).await
301
+
server::start(cfg, pool, scan_progress, typesense).await
251
302
}
+100
-13
src/scanner.rs
+100
-13
src/scanner.rs
···
1
1
use crate::db::Db;
2
+
use crate::models::{Album, Artist, Song};
3
+
use crate::typesense::TypesenseClient;
2
4
use anyhow::{Context, Result};
3
5
use lofty::file::{AudioFile, TaggedFileExt};
4
6
use lofty::picture::{MimeType, Picture};
···
36
38
music_dir: PathBuf,
37
39
covers_dir: PathBuf,
38
40
progress: Arc<ScanProgress>,
41
+
typesense: Option<Arc<TypesenseClient>>,
39
42
) -> Result<ScanStats> {
40
43
progress.running.store(true, Ordering::SeqCst);
41
44
progress.count.store(0, Ordering::SeqCst);
···
54
57
}
55
58
stats.scanned += 1;
56
59
progress.count.store(stats.scanned, Ordering::SeqCst);
57
-
match process_file(&pool, path, &covers_dir).await {
60
+
match process_file(&pool, path, &covers_dir, typesense.as_deref()).await {
58
61
Ok(ProcessResult::Inserted) => stats.inserted += 1,
59
62
Ok(ProcessResult::Updated) => stats.updated += 1,
60
63
Ok(ProcessResult::Skipped) => stats.skipped += 1,
···
65
68
}
66
69
}
67
70
68
-
match reconcile_deletions(&pool).await {
71
+
match reconcile_deletions(&pool, typesense.as_deref()).await {
69
72
Ok(removed) => stats.removed = removed as usize,
70
73
Err(e) => tracing::warn!("reconcile deletions: {e}"),
71
74
}
···
82
85
Ok(stats)
83
86
}
84
87
85
-
pub async fn reconcile_deletions(pool: &Db) -> Result<u64> {
86
-
let rows: Vec<(String, String, String)> =
87
-
sqlx::query_as("SELECT path, album_id, artist_id FROM songs")
88
+
pub async fn reconcile_deletions(pool: &Db, typesense: Option<&TypesenseClient>) -> Result<u64> {
89
+
let rows: Vec<(String, String, String, String)> =
90
+
sqlx::query_as("SELECT id, path, album_id, artist_id FROM songs")
88
91
.fetch_all(pool)
89
92
.await?;
90
93
91
94
let mut missing_paths = Vec::new();
95
+
let mut missing_song_ids = Vec::new();
92
96
let mut album_ids = std::collections::HashSet::new();
93
97
let mut artist_ids = std::collections::HashSet::new();
94
-
for (path, album_id, artist_id) in rows {
98
+
for (id, path, album_id, artist_id) in rows {
95
99
if !Path::new(&path).exists() {
96
100
missing_paths.push(path);
101
+
missing_song_ids.push(id);
97
102
album_ids.insert(album_id);
98
103
artist_ids.insert(artist_id);
99
104
}
···
114
119
}
115
120
tx.commit().await?;
116
121
122
+
if let Some(ts) = typesense {
123
+
for id in &missing_song_ids {
124
+
if let Err(e) = ts.delete_song(id).await {
125
+
tracing::warn!("typesense delete song {id}: {e}");
126
+
}
127
+
}
128
+
}
129
+
117
130
for album_id in &album_ids {
118
-
gc_album(pool, album_id).await?;
131
+
gc_album(pool, album_id, typesense).await?;
119
132
}
120
133
for artist_id in &artist_ids {
121
-
gc_artist(pool, artist_id).await?;
134
+
gc_artist(pool, artist_id, typesense).await?;
122
135
}
123
136
124
137
Ok(deleted)
125
138
}
126
139
127
-
pub async fn gc_album(pool: &Db, album_id: &str) -> Result<()> {
140
+
pub async fn gc_album(
141
+
pool: &Db,
142
+
album_id: &str,
143
+
typesense: Option<&TypesenseClient>,
144
+
) -> Result<()> {
128
145
let remaining: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM songs WHERE album_id = ?1")
129
146
.bind(album_id)
130
147
.fetch_one(pool)
···
134
151
.bind(album_id)
135
152
.execute(pool)
136
153
.await?;
154
+
if let Some(ts) = typesense {
155
+
if let Err(e) = ts.delete_album(album_id).await {
156
+
tracing::warn!("typesense delete album {album_id}: {e}");
157
+
}
158
+
}
137
159
}
138
160
Ok(())
139
161
}
140
162
141
-
pub async fn gc_artist(pool: &Db, artist_id: &str) -> Result<()> {
163
+
pub async fn gc_artist(
164
+
pool: &Db,
165
+
artist_id: &str,
166
+
typesense: Option<&TypesenseClient>,
167
+
) -> Result<()> {
142
168
let remaining: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM songs WHERE artist_id = ?1")
143
169
.bind(artist_id)
144
170
.fetch_one(pool)
···
148
174
.bind(artist_id)
149
175
.execute(pool)
150
176
.await?;
177
+
if let Some(ts) = typesense {
178
+
if let Err(e) = ts.delete_artist(artist_id).await {
179
+
tracing::warn!("typesense delete artist {artist_id}: {e}");
180
+
}
181
+
}
151
182
}
152
183
Ok(())
153
184
}
···
165
196
}
166
197
}
167
198
168
-
pub async fn upsert_file(pool: &Db, path: &Path, covers_dir: &Path) -> Result<()> {
169
-
process_file(pool, path, covers_dir).await.map(|_| ())
199
+
pub async fn upsert_file(
200
+
pool: &Db,
201
+
path: &Path,
202
+
covers_dir: &Path,
203
+
typesense: Option<&TypesenseClient>,
204
+
) -> Result<()> {
205
+
process_file(pool, path, covers_dir, typesense)
206
+
.await
207
+
.map(|_| ())
170
208
}
171
209
172
-
async fn process_file(pool: &Db, path: &Path, covers_dir: &Path) -> Result<ProcessResult> {
210
+
async fn process_file(
211
+
pool: &Db,
212
+
path: &Path,
213
+
covers_dir: &Path,
214
+
typesense: Option<&TypesenseClient>,
215
+
) -> Result<ProcessResult> {
173
216
let path_str = path.to_string_lossy().to_string();
174
217
let meta = std::fs::metadata(path)?;
175
218
let mtime = meta
···
307
350
.await?;
308
351
309
352
tx.commit().await?;
353
+
354
+
if let Some(ts) = typesense {
355
+
let artist_row = Artist {
356
+
id: artist_id.clone(),
357
+
name: display_artist.clone(),
358
+
};
359
+
let album_row = Album {
360
+
id: album_id.clone(),
361
+
title: album.clone(),
362
+
artist: display_artist.clone(),
363
+
artist_id: artist_id.clone(),
364
+
year,
365
+
cover_art: cover_filename.clone(),
366
+
};
367
+
let song_row = Song {
368
+
id: song_id.clone(),
369
+
path: path_str.clone(),
370
+
title: title.clone(),
371
+
artist: artist.clone(),
372
+
artist_id: artist_id.clone(),
373
+
album: album.clone(),
374
+
album_id: album_id.clone(),
375
+
genre: genre.clone(),
376
+
track_number: track,
377
+
disc_number: disc,
378
+
year: year_to_opt(year),
379
+
duration_ms,
380
+
bitrate,
381
+
filesize,
382
+
suffix: suffix.clone(),
383
+
content_type: content_type.clone(),
384
+
cover_art: cover_filename.clone(),
385
+
};
386
+
if let Err(e) = ts.upsert_artist(&artist_row).await {
387
+
tracing::warn!("typesense upsert artist {}: {e}", artist_row.id);
388
+
}
389
+
if let Err(e) = ts.upsert_album(&album_row).await {
390
+
tracing::warn!("typesense upsert album {}: {e}", album_row.id);
391
+
}
392
+
if let Err(e) = ts.upsert_song(&song_row).await {
393
+
tracing::warn!("typesense upsert song {}: {e}", song_row.id);
394
+
}
395
+
}
396
+
310
397
Ok(if existed {
311
398
ProcessResult::Updated
312
399
} else {
+11
-8
src/server/handlers.rs
+11
-8
src/server/handlers.rs
···
616
616
let album_offset = q.album_offset.unwrap_or(0);
617
617
let song_offset = q.song_offset.unwrap_or(0);
618
618
619
-
let artists = repo::search_artists(&state.pool, &term, artist_limit, artist_offset)
619
+
let ts = state.typesense.as_deref();
620
+
let artists = repo::search_artists(&state.pool, &term, artist_limit, artist_offset, ts)
620
621
.await
621
622
.unwrap_or_default();
622
-
let albums = repo::search_albums(&state.pool, &term, album_limit, album_offset)
623
+
let albums = repo::search_albums(&state.pool, &term, album_limit, album_offset, ts)
623
624
.await
624
625
.unwrap_or_default();
625
-
let songs = repo::search_songs(&state.pool, &term, song_limit, song_offset)
626
+
let songs = repo::search_songs(&state.pool, &term, song_limit, song_offset, ts)
626
627
.await
627
628
.unwrap_or_default();
628
629
···
920
921
let music_dir = state.music_dir.clone();
921
922
let covers_dir = state.covers_dir.clone();
922
923
let progress = state.scan_progress.clone();
924
+
let ts = state.typesense.clone();
923
925
tokio::spawn(async move {
924
-
if let Err(e) = scanner::scan(pool, music_dir, covers_dir, progress).await {
926
+
if let Err(e) = scanner::scan(pool, music_dir, covers_dir, progress, ts).await {
925
927
tracing::error!("scan: {e}");
926
928
}
927
929
});
···
1586
1588
return response::ok_json(json!({ "topSongs": { "song": [] } }));
1587
1589
};
1588
1590
let count = q.count.unwrap_or(50).clamp(1, 500);
1589
-
let songs = repo::search_songs(&state.pool, artist, count, 0)
1591
+
let songs = repo::search_songs(&state.pool, artist, count, 0, state.typesense.as_deref())
1590
1592
.await
1591
1593
.unwrap_or_default();
1592
1594
let children: Vec<Value> = songs
···
1668
1670
let album_offset = q.album_offset.unwrap_or(0);
1669
1671
let song_offset = q.song_offset.unwrap_or(0);
1670
1672
1671
-
let artists = repo::search_artists(&state.pool, &term, artist_limit, artist_offset)
1673
+
let ts = state.typesense.as_deref();
1674
+
let artists = repo::search_artists(&state.pool, &term, artist_limit, artist_offset, ts)
1672
1675
.await
1673
1676
.unwrap_or_default();
1674
-
let albums = repo::search_albums(&state.pool, &term, album_limit, album_offset)
1677
+
let albums = repo::search_albums(&state.pool, &term, album_limit, album_offset, ts)
1675
1678
.await
1676
1679
.unwrap_or_default();
1677
-
let songs = repo::search_songs(&state.pool, &term, song_limit, song_offset)
1680
+
let songs = repo::search_songs(&state.pool, &term, song_limit, song_offset, ts)
1678
1681
.await
1679
1682
.unwrap_or_default();
1680
1683
let artist_jsons: Vec<Value> = artists.iter().map(|a| artist_to_json(a, 0)).collect();
+12
-1
src/server/mod.rs
+12
-1
src/server/mod.rs
···
6
6
use crate::config::Config;
7
7
use crate::db::Db;
8
8
use crate::scanner::ScanProgress;
9
+
use crate::typesense::TypesenseClient;
9
10
use actix_cors::Cors;
10
11
use actix_web::{web, App, HttpServer};
11
12
use std::path::PathBuf;
···
18
19
pub music_dir: PathBuf,
19
20
pub covers_dir: PathBuf,
20
21
pub scan_progress: Arc<ScanProgress>,
22
+
/// Optional Typesense client. When `Some`, `search3`/`search2` route
23
+
/// through Typesense with fallback to FTS5 on error.
24
+
pub typesense: Option<Arc<TypesenseClient>>,
21
25
}
22
26
23
-
pub async fn start(cfg: Config, pool: Db, scan_progress: Arc<ScanProgress>) -> anyhow::Result<()> {
27
+
pub async fn start(
28
+
cfg: Config,
29
+
pool: Db,
30
+
scan_progress: Arc<ScanProgress>,
31
+
typesense: Option<Arc<TypesenseClient>>,
32
+
) -> anyhow::Result<()> {
24
33
let addr = format!("{}:{}", cfg.host, cfg.port);
25
34
let state = web::Data::new(SubsonicState {
26
35
pool,
···
29
38
music_dir: cfg.music_dir.clone(),
30
39
covers_dir: cfg.covers_dir.clone(),
31
40
scan_progress,
41
+
typesense,
32
42
});
33
43
34
44
HttpServer::new(move || {
···
468
478
music_dir: music_dir.to_path_buf(),
469
479
covers_dir: covers_dir.to_path_buf(),
470
480
scan_progress: Arc::new(ScanProgress::default()),
481
+
typesense: None,
471
482
}
472
483
}
473
484
+133
-3
src/server/repo.rs
+133
-3
src/server/repo.rs
···
1
1
use crate::db::Db;
2
2
use crate::models::{Album, Artist, Playlist, Song, Video};
3
+
use crate::typesense::TypesenseClient;
3
4
use anyhow::Result;
4
5
5
6
// ── Videos ──────────────────────────────────────────────────────────────────
···
500
501
}
501
502
}
502
503
503
-
pub async fn search_artists(pool: &Db, term: &str, limit: i64, offset: i64) -> Result<Vec<Artist>> {
504
+
pub async fn search_artists(
505
+
pool: &Db,
506
+
term: &str,
507
+
limit: i64,
508
+
offset: i64,
509
+
typesense: Option<&TypesenseClient>,
510
+
) -> Result<Vec<Artist>> {
511
+
if let Some(ts) = typesense {
512
+
match ts.search_artist_ids(term, limit, offset).await {
513
+
Ok(ids) if !ids.is_empty() => {
514
+
if let Ok(mut rows) = hydrate_artists(pool, &ids).await {
515
+
reorder_by_ids(&mut rows, &ids, |a| &a.id);
516
+
return Ok(rows);
517
+
}
518
+
}
519
+
Ok(_) => return Ok(Vec::new()),
520
+
Err(e) => {
521
+
tracing::warn!("typesense search_artists('{term}'): {e} — falling back to FTS5")
522
+
}
523
+
}
524
+
}
504
525
let Some(q) = fts5_match(term) else {
505
526
let rows = sqlx::query_as::<_, Artist>(
506
527
"SELECT id, name FROM artists ORDER BY name COLLATE NOCASE LIMIT ?1 OFFSET ?2",
···
526
547
Ok(rows)
527
548
}
528
549
529
-
pub async fn search_albums(pool: &Db, term: &str, limit: i64, offset: i64) -> Result<Vec<Album>> {
550
+
pub async fn search_albums(
551
+
pool: &Db,
552
+
term: &str,
553
+
limit: i64,
554
+
offset: i64,
555
+
typesense: Option<&TypesenseClient>,
556
+
) -> Result<Vec<Album>> {
557
+
if let Some(ts) = typesense {
558
+
match ts.search_album_ids(term, limit, offset).await {
559
+
Ok(ids) if !ids.is_empty() => {
560
+
if let Ok(mut rows) = hydrate_albums(pool, &ids).await {
561
+
reorder_by_ids(&mut rows, &ids, |a| &a.id);
562
+
return Ok(rows);
563
+
}
564
+
}
565
+
Ok(_) => return Ok(Vec::new()),
566
+
Err(e) => {
567
+
tracing::warn!("typesense search_albums('{term}'): {e} — falling back to FTS5")
568
+
}
569
+
}
570
+
}
530
571
let Some(q) = fts5_match(term) else {
531
572
let rows = sqlx::query_as::<_, Album>(
532
573
"SELECT id, title, artist, artist_id, year, cover_art FROM albums
···
553
594
Ok(rows)
554
595
}
555
596
556
-
pub async fn search_songs(pool: &Db, term: &str, limit: i64, offset: i64) -> Result<Vec<Song>> {
597
+
pub async fn search_songs(
598
+
pool: &Db,
599
+
term: &str,
600
+
limit: i64,
601
+
offset: i64,
602
+
typesense: Option<&TypesenseClient>,
603
+
) -> Result<Vec<Song>> {
604
+
if let Some(ts) = typesense {
605
+
match ts.search_song_ids(term, limit, offset).await {
606
+
Ok(ids) if !ids.is_empty() => {
607
+
if let Ok(mut rows) = hydrate_songs(pool, &ids).await {
608
+
reorder_by_ids(&mut rows, &ids, |s| &s.id);
609
+
return Ok(rows);
610
+
}
611
+
}
612
+
Ok(_) => return Ok(Vec::new()),
613
+
Err(e) => {
614
+
tracing::warn!("typesense search_songs('{term}'): {e} — falling back to FTS5")
615
+
}
616
+
}
617
+
}
557
618
let Some(q) = fts5_match(term) else {
558
619
let rows = sqlx::query_as::<_, Song>(
559
620
"SELECT id, path, title, artist, artist_id, album, album_id, genre, track_number,
···
581
642
.fetch_all(pool)
582
643
.await?;
583
644
Ok(rows)
645
+
}
646
+
647
+
/// Hydrate a Typesense id list back into `Artist` rows. The result is
648
+
/// unordered — callers pass it through `reorder_by_ids` to preserve
649
+
/// Typesense's ranking.
650
+
async fn hydrate_artists(pool: &Db, ids: &[String]) -> Result<Vec<Artist>> {
651
+
if ids.is_empty() {
652
+
return Ok(Vec::new());
653
+
}
654
+
let placeholders = build_in_placeholders(ids.len());
655
+
let sql = format!("SELECT id, name FROM artists WHERE id IN ({placeholders})");
656
+
let mut q = sqlx::query_as::<_, Artist>(&sql);
657
+
for id in ids {
658
+
q = q.bind(id);
659
+
}
660
+
Ok(q.fetch_all(pool).await?)
661
+
}
662
+
663
+
async fn hydrate_albums(pool: &Db, ids: &[String]) -> Result<Vec<Album>> {
664
+
if ids.is_empty() {
665
+
return Ok(Vec::new());
666
+
}
667
+
let placeholders = build_in_placeholders(ids.len());
668
+
let sql = format!(
669
+
"SELECT id, title, artist, artist_id, year, cover_art FROM albums WHERE id IN ({placeholders})"
670
+
);
671
+
let mut q = sqlx::query_as::<_, Album>(&sql);
672
+
for id in ids {
673
+
q = q.bind(id);
674
+
}
675
+
Ok(q.fetch_all(pool).await?)
676
+
}
677
+
678
+
async fn hydrate_songs(pool: &Db, ids: &[String]) -> Result<Vec<Song>> {
679
+
if ids.is_empty() {
680
+
return Ok(Vec::new());
681
+
}
682
+
let placeholders = build_in_placeholders(ids.len());
683
+
let sql = format!(
684
+
"SELECT id, path, title, artist, artist_id, album, album_id, genre, track_number,
685
+
disc_number, year, duration_ms, bitrate, filesize, suffix, content_type, cover_art
686
+
FROM songs WHERE id IN ({placeholders})"
687
+
);
688
+
let mut q = sqlx::query_as::<_, Song>(&sql);
689
+
for id in ids {
690
+
q = q.bind(id);
691
+
}
692
+
Ok(q.fetch_all(pool).await?)
693
+
}
694
+
695
+
fn build_in_placeholders(n: usize) -> String {
696
+
(1..=n)
697
+
.map(|i| format!("?{i}"))
698
+
.collect::<Vec<_>>()
699
+
.join(", ")
700
+
}
701
+
702
+
/// Reorder `rows` in-place so their `id`s match `ids`'s order. Rows whose id
703
+
/// isn't in `ids` land at the end (shouldn't happen but is safe).
704
+
fn reorder_by_ids<T, F>(rows: &mut Vec<T>, ids: &[String], key: F)
705
+
where
706
+
F: Fn(&T) -> &String,
707
+
{
708
+
let rank: std::collections::HashMap<&str, usize> = ids
709
+
.iter()
710
+
.enumerate()
711
+
.map(|(i, id)| (id.as_str(), i))
712
+
.collect();
713
+
rows.sort_by_key(|r| rank.get(key(r).as_str()).copied().unwrap_or(usize::MAX));
584
714
}
585
715
586
716
pub async fn songs_for_album_duration(pool: &Db, album_id: &str) -> Result<i64> {
+437
src/typesense.rs
+437
src/typesense.rs
···
1
+
//! Optional Typesense search backend for the free-text search endpoints.
2
+
//!
3
+
//! When `[typesense]` is present in `smolsonic.toml`, `search3` / `search2` /
4
+
//! Jellyfin `?searchTerm=` route through Typesense instead of the built-in
5
+
//! SQLite FTS5 index. All write paths (scanner + watcher) mirror
6
+
//! `songs / albums / artists` inserts, updates, and deletes into three
7
+
//! Typesense collections. On any Typesense failure the search layer logs a
8
+
//! warning and falls back to FTS5, so a Typesense outage never breaks search.
9
+
10
+
use crate::config::TypesenseConfig;
11
+
use crate::db::Db;
12
+
use crate::models::{Album, Artist, Song};
13
+
use anyhow::{anyhow, Context, Result};
14
+
use serde::{Deserialize, Serialize};
15
+
use serde_json::json;
16
+
17
+
/// Text collections we manage. Fixed set — every write path knows about all
18
+
/// three and reindex walks all three sequentially.
19
+
const SONGS: &str = "songs";
20
+
const ALBUMS: &str = "albums";
21
+
const ARTISTS: &str = "artists";
22
+
23
+
pub struct TypesenseClient {
24
+
base_url: String,
25
+
api_key: String,
26
+
prefix: String,
27
+
http: reqwest::Client,
28
+
}
29
+
30
+
impl TypesenseClient {
31
+
pub fn new(cfg: &TypesenseConfig) -> Self {
32
+
Self {
33
+
base_url: cfg.url.trim_end_matches('/').to_string(),
34
+
api_key: cfg.api_key.clone(),
35
+
prefix: cfg.collection_prefix.clone(),
36
+
http: reqwest::Client::builder()
37
+
.timeout(std::time::Duration::from_secs(8))
38
+
.build()
39
+
.expect("build reqwest client"),
40
+
}
41
+
}
42
+
43
+
fn collection(&self, kind: &str) -> String {
44
+
format!("{}_{}", self.prefix, kind)
45
+
}
46
+
47
+
/// Create the three collections if they don't already exist. Idempotent —
48
+
/// a 409 "already exists" is treated as success.
49
+
pub async fn bootstrap(&self) -> Result<()> {
50
+
self.ensure_collection(
51
+
SONGS,
52
+
&[
53
+
field("title", "string"),
54
+
field("artist", "string"),
55
+
field("album", "string"),
56
+
field_optional("genre", "string"),
57
+
field_optional("year", "int32"),
58
+
],
59
+
)
60
+
.await?;
61
+
self.ensure_collection(
62
+
ALBUMS,
63
+
&[
64
+
field("title", "string"),
65
+
field("artist", "string"),
66
+
field_optional("year", "int32"),
67
+
],
68
+
)
69
+
.await?;
70
+
self.ensure_collection(ARTISTS, &[field("name", "string")])
71
+
.await?;
72
+
Ok(())
73
+
}
74
+
75
+
async fn ensure_collection(&self, kind: &str, fields: &[serde_json::Value]) -> Result<()> {
76
+
let name = self.collection(kind);
77
+
let url = format!("{}/collections", self.base_url);
78
+
let body = json!({
79
+
"name": name,
80
+
"fields": fields,
81
+
"default_sorting_field": "",
82
+
});
83
+
let resp = self
84
+
.http
85
+
.post(&url)
86
+
.header("X-TYPESENSE-API-KEY", &self.api_key)
87
+
.json(&body)
88
+
.send()
89
+
.await
90
+
.with_context(|| format!("typesense: create collection {name}"))?;
91
+
let status = resp.status();
92
+
if status.is_success() || status.as_u16() == 409 {
93
+
return Ok(());
94
+
}
95
+
let text = resp.text().await.unwrap_or_default();
96
+
Err(anyhow!(
97
+
"typesense: create collection {name} -> {status}: {text}"
98
+
))
99
+
}
100
+
101
+
// ── Writes ──────────────────────────────────────────────────────────────
102
+
103
+
pub async fn upsert_song(&self, song: &Song) -> Result<()> {
104
+
let doc = json!({
105
+
"id": song.id,
106
+
"title": song.title,
107
+
"artist": song.artist,
108
+
"album": song.album,
109
+
"genre": song.genre.clone().unwrap_or_default(),
110
+
"year": song.year.unwrap_or(0),
111
+
});
112
+
self.upsert_doc(SONGS, doc).await
113
+
}
114
+
115
+
pub async fn upsert_album(&self, album: &Album) -> Result<()> {
116
+
let doc = json!({
117
+
"id": album.id,
118
+
"title": album.title,
119
+
"artist": album.artist,
120
+
"year": album.year,
121
+
});
122
+
self.upsert_doc(ALBUMS, doc).await
123
+
}
124
+
125
+
pub async fn upsert_artist(&self, artist: &Artist) -> Result<()> {
126
+
let doc = json!({
127
+
"id": artist.id,
128
+
"name": artist.name,
129
+
});
130
+
self.upsert_doc(ARTISTS, doc).await
131
+
}
132
+
133
+
async fn upsert_doc(&self, kind: &str, doc: serde_json::Value) -> Result<()> {
134
+
let name = self.collection(kind);
135
+
let url = format!(
136
+
"{}/collections/{}/documents?action=upsert",
137
+
self.base_url, name
138
+
);
139
+
let resp = self
140
+
.http
141
+
.post(&url)
142
+
.header("X-TYPESENSE-API-KEY", &self.api_key)
143
+
.json(&doc)
144
+
.send()
145
+
.await
146
+
.with_context(|| format!("typesense: upsert into {name}"))?;
147
+
if resp.status().is_success() {
148
+
return Ok(());
149
+
}
150
+
let status = resp.status();
151
+
let text = resp.text().await.unwrap_or_default();
152
+
Err(anyhow!("typesense: upsert {name} -> {status}: {text}"))
153
+
}
154
+
155
+
pub async fn delete_song(&self, id: &str) -> Result<()> {
156
+
self.delete_doc(SONGS, id).await
157
+
}
158
+
159
+
pub async fn delete_album(&self, id: &str) -> Result<()> {
160
+
self.delete_doc(ALBUMS, id).await
161
+
}
162
+
163
+
pub async fn delete_artist(&self, id: &str) -> Result<()> {
164
+
self.delete_doc(ARTISTS, id).await
165
+
}
166
+
167
+
async fn delete_doc(&self, kind: &str, id: &str) -> Result<()> {
168
+
let name = self.collection(kind);
169
+
let url = format!(
170
+
"{}/collections/{}/documents/{}",
171
+
self.base_url,
172
+
name,
173
+
urlencoding::encode(id)
174
+
);
175
+
let resp = self
176
+
.http
177
+
.delete(&url)
178
+
.header("X-TYPESENSE-API-KEY", &self.api_key)
179
+
.send()
180
+
.await
181
+
.with_context(|| format!("typesense: delete from {name}"))?;
182
+
let status = resp.status();
183
+
// 404 is fine: the caller might delete after we already GC'd the id
184
+
// through a prior reconcile.
185
+
if status.is_success() || status.as_u16() == 404 {
186
+
return Ok(());
187
+
}
188
+
let text = resp.text().await.unwrap_or_default();
189
+
Err(anyhow!("typesense: delete {name} -> {status}: {text}"))
190
+
}
191
+
192
+
// ── Searches ────────────────────────────────────────────────────────────
193
+
194
+
pub async fn search_song_ids(
195
+
&self,
196
+
term: &str,
197
+
limit: i64,
198
+
offset: i64,
199
+
) -> Result<Vec<String>> {
200
+
self.search(SONGS, "title,artist,album,genre", term, limit, offset)
201
+
.await
202
+
}
203
+
204
+
pub async fn search_album_ids(
205
+
&self,
206
+
term: &str,
207
+
limit: i64,
208
+
offset: i64,
209
+
) -> Result<Vec<String>> {
210
+
self.search(ALBUMS, "title,artist", term, limit, offset)
211
+
.await
212
+
}
213
+
214
+
pub async fn search_artist_ids(
215
+
&self,
216
+
term: &str,
217
+
limit: i64,
218
+
offset: i64,
219
+
) -> Result<Vec<String>> {
220
+
self.search(ARTISTS, "name", term, limit, offset).await
221
+
}
222
+
223
+
async fn search(
224
+
&self,
225
+
kind: &str,
226
+
query_by: &str,
227
+
term: &str,
228
+
limit: i64,
229
+
offset: i64,
230
+
) -> Result<Vec<String>> {
231
+
let name = self.collection(kind);
232
+
// Typesense pagination uses page/per_page (1-based). Convert
233
+
// offset/limit — offset must be a clean multiple of limit for exact
234
+
// pagination; when it isn't (rare in this app) we round down and
235
+
// clients get slightly overlapping windows, which is fine.
236
+
let per_page = limit.clamp(1, 250);
237
+
let page = if per_page > 0 {
238
+
(offset / per_page) + 1
239
+
} else {
240
+
1
241
+
};
242
+
let url = format!("{}/collections/{}/documents/search", self.base_url, name);
243
+
let resp = self
244
+
.http
245
+
.get(&url)
246
+
.header("X-TYPESENSE-API-KEY", &self.api_key)
247
+
.query(&[
248
+
("q", term),
249
+
("query_by", query_by),
250
+
("per_page", &per_page.to_string()),
251
+
("page", &page.to_string()),
252
+
("include_fields", "id"),
253
+
])
254
+
.send()
255
+
.await
256
+
.with_context(|| format!("typesense: search {name}"))?;
257
+
if !resp.status().is_success() {
258
+
let status = resp.status();
259
+
let text = resp.text().await.unwrap_or_default();
260
+
return Err(anyhow!("typesense: search {name} -> {status}: {text}"));
261
+
}
262
+
let parsed: SearchResponse = resp
263
+
.json()
264
+
.await
265
+
.with_context(|| format!("typesense: parse search response for {name}"))?;
266
+
Ok(parsed
267
+
.hits
268
+
.into_iter()
269
+
.filter_map(|h| h.document.id)
270
+
.collect())
271
+
}
272
+
273
+
// ── Startup reindex ─────────────────────────────────────────────────────
274
+
275
+
/// True iff the songs collection currently has zero documents. Used to
276
+
/// decide whether to run the initial reindex on startup.
277
+
pub async fn songs_empty(&self) -> Result<bool> {
278
+
let name = self.collection(SONGS);
279
+
let url = format!("{}/collections/{}", self.base_url, name);
280
+
let resp = self
281
+
.http
282
+
.get(&url)
283
+
.header("X-TYPESENSE-API-KEY", &self.api_key)
284
+
.send()
285
+
.await
286
+
.with_context(|| format!("typesense: describe {name}"))?;
287
+
if !resp.status().is_success() {
288
+
let status = resp.status();
289
+
let text = resp.text().await.unwrap_or_default();
290
+
return Err(anyhow!("typesense: describe {name} -> {status}: {text}"));
291
+
}
292
+
let info: CollectionInfo = resp.json().await?;
293
+
Ok(info.num_documents == 0)
294
+
}
295
+
296
+
/// Walk the SQLite tables and bulk-import every artist / album / song
297
+
/// into Typesense. Called from `main` at startup when the songs
298
+
/// collection reports 0 documents, so a fresh Typesense node self-heals
299
+
/// without operator intervention.
300
+
pub async fn reindex_from_db(&self, pool: &Db) -> Result<()> {
301
+
let artists: Vec<Artist> =
302
+
sqlx::query_as("SELECT id, name FROM artists ORDER BY name COLLATE NOCASE")
303
+
.fetch_all(pool)
304
+
.await?;
305
+
let artist_lines: Vec<String> = artists
306
+
.iter()
307
+
.filter_map(|a| serde_json::to_string(&json!({ "id": a.id, "name": a.name })).ok())
308
+
.collect();
309
+
self.import_jsonl(ARTISTS, &artist_lines).await?;
310
+
311
+
let albums: Vec<Album> = sqlx::query_as(
312
+
"SELECT id, title, artist, artist_id, year, cover_art FROM albums
313
+
ORDER BY title COLLATE NOCASE",
314
+
)
315
+
.fetch_all(pool)
316
+
.await?;
317
+
let album_lines: Vec<String> = albums
318
+
.iter()
319
+
.filter_map(|a| {
320
+
serde_json::to_string(&json!({
321
+
"id": a.id,
322
+
"title": a.title,
323
+
"artist": a.artist,
324
+
"year": a.year,
325
+
}))
326
+
.ok()
327
+
})
328
+
.collect();
329
+
self.import_jsonl(ALBUMS, &album_lines).await?;
330
+
331
+
let songs: Vec<Song> = sqlx::query_as(
332
+
"SELECT id, path, title, artist, artist_id, album, album_id, genre, track_number,
333
+
disc_number, year, duration_ms, bitrate, filesize, suffix, content_type, cover_art
334
+
FROM songs ORDER BY title COLLATE NOCASE",
335
+
)
336
+
.fetch_all(pool)
337
+
.await?;
338
+
let song_lines: Vec<String> = songs
339
+
.iter()
340
+
.filter_map(|s| {
341
+
serde_json::to_string(&json!({
342
+
"id": s.id,
343
+
"title": s.title,
344
+
"artist": s.artist,
345
+
"album": s.album,
346
+
"genre": s.genre.clone().unwrap_or_default(),
347
+
"year": s.year.unwrap_or(0),
348
+
}))
349
+
.ok()
350
+
})
351
+
.collect();
352
+
self.import_jsonl(SONGS, &song_lines).await?;
353
+
354
+
tracing::info!(
355
+
"typesense: reindexed {} artists, {} albums, {} songs",
356
+
artists.len(),
357
+
albums.len(),
358
+
songs.len()
359
+
);
360
+
Ok(())
361
+
}
362
+
363
+
/// Typesense bulk-import wire format is newline-delimited JSON, one
364
+
/// document per line. Empty payload is a no-op.
365
+
async fn import_jsonl(&self, kind: &str, lines: &[String]) -> Result<()> {
366
+
if lines.is_empty() {
367
+
return Ok(());
368
+
}
369
+
let name = self.collection(kind);
370
+
let url = format!(
371
+
"{}/collections/{}/documents/import?action=upsert",
372
+
self.base_url, name
373
+
);
374
+
let body = lines.join("\n");
375
+
let resp = self
376
+
.http
377
+
.post(&url)
378
+
.header("X-TYPESENSE-API-KEY", &self.api_key)
379
+
.header("Content-Type", "text/plain")
380
+
.body(body)
381
+
.send()
382
+
.await
383
+
.with_context(|| format!("typesense: import {name}"))?;
384
+
if !resp.status().is_success() {
385
+
let status = resp.status();
386
+
let text = resp.text().await.unwrap_or_default();
387
+
return Err(anyhow!("typesense: import {name} -> {status}: {text}"));
388
+
}
389
+
Ok(())
390
+
}
391
+
}
392
+
393
+
// ── Wire shapes ──────────────────────────────────────────────────────────────
394
+
395
+
fn field(name: &str, ty: &str) -> serde_json::Value {
396
+
json!({ "name": name, "type": ty })
397
+
}
398
+
399
+
fn field_optional(name: &str, ty: &str) -> serde_json::Value {
400
+
json!({ "name": name, "type": ty, "optional": true })
401
+
}
402
+
403
+
#[derive(Deserialize)]
404
+
struct SearchResponse {
405
+
#[serde(default)]
406
+
hits: Vec<Hit>,
407
+
}
408
+
409
+
#[derive(Deserialize)]
410
+
struct Hit {
411
+
document: HitDoc,
412
+
}
413
+
414
+
#[derive(Deserialize)]
415
+
struct HitDoc {
416
+
#[serde(default)]
417
+
id: Option<String>,
418
+
}
419
+
420
+
#[derive(Deserialize)]
421
+
struct CollectionInfo {
422
+
#[serde(default)]
423
+
num_documents: i64,
424
+
}
425
+
426
+
// The Serialize derives are here so the type is usable with `serde_json`
427
+
// without an extra roundtrip when we add richer document payloads later.
428
+
#[allow(dead_code)]
429
+
#[derive(Serialize)]
430
+
struct SongDoc<'a> {
431
+
id: &'a str,
432
+
title: &'a str,
433
+
artist: &'a str,
434
+
album: &'a str,
435
+
genre: &'a str,
436
+
year: i64,
437
+
}
+47
-19
src/watcher.rs
+47
-19
src/watcher.rs
···
1
1
use crate::db::Db;
2
2
use crate::scanner::{gc_album, gc_artist, has_audio_extension, upsert_file};
3
+
use crate::typesense::TypesenseClient;
3
4
use anyhow::Result;
4
5
use notify::{EventKind, RecursiveMode};
5
6
use notify_debouncer_full::new_debouncer;
6
7
use std::path::{Path, PathBuf};
8
+
use std::sync::Arc;
7
9
use std::time::Duration;
8
10
9
11
const DEBOUNCE: Duration = Duration::from_secs(2);
10
12
11
-
pub fn start(pool: Db, music_dir: PathBuf, covers_dir: PathBuf) {
13
+
pub fn start(
14
+
pool: Db,
15
+
music_dir: PathBuf,
16
+
covers_dir: PathBuf,
17
+
typesense: Option<Arc<TypesenseClient>>,
18
+
) {
12
19
let rt = tokio::runtime::Handle::current();
13
20
std::thread::Builder::new()
14
21
.name("library-watcher".into())
15
22
.spawn(move || {
16
-
if let Err(e) = run(pool, music_dir, covers_dir, rt) {
23
+
if let Err(e) = run(pool, music_dir, covers_dir, typesense, rt) {
17
24
tracing::error!("watcher stopped: {e}");
18
25
}
19
26
})
···
24
31
pool: Db,
25
32
music_dir: PathBuf,
26
33
covers_dir: PathBuf,
34
+
typesense: Option<Arc<TypesenseClient>>,
27
35
rt: tokio::runtime::Handle,
28
36
) -> Result<()> {
29
37
let (tx, rx) = std::sync::mpsc::channel();
···
39
47
continue;
40
48
}
41
49
for path in &ev.event.paths {
42
-
rt.block_on(apply_path(&pool, &covers_dir, path));
50
+
rt.block_on(apply_path(&pool, &covers_dir, typesense.as_deref(), path));
43
51
}
44
52
}
45
53
}
···
54
62
Ok(())
55
63
}
56
64
57
-
async fn apply_path(pool: &Db, covers_dir: &Path, path: &Path) {
65
+
async fn apply_path(
66
+
pool: &Db,
67
+
covers_dir: &Path,
68
+
typesense: Option<&TypesenseClient>,
69
+
path: &Path,
70
+
) {
58
71
if path.is_file() {
59
72
if !has_audio_extension(path) {
60
73
return;
61
74
}
62
-
match upsert_file(pool, path, covers_dir).await {
75
+
match upsert_file(pool, path, covers_dir, typesense).await {
63
76
Ok(_) => tracing::info!("watch upsert {}", path.display()),
64
77
Err(e) => tracing::warn!("watch upsert {}: {e}", path.display()),
65
78
}
···
72
85
73
86
let path_str = path.to_string_lossy().to_string();
74
87
if has_audio_extension(path) {
75
-
match delete_song(pool, &path_str).await {
88
+
match delete_song(pool, &path_str, typesense).await {
76
89
Ok(true) => tracing::info!("watch delete {path_str}"),
77
90
Ok(false) => {}
78
91
Err(e) => tracing::warn!("watch delete {path_str}: {e}"),
···
80
93
return;
81
94
}
82
95
83
-
match delete_songs_under(pool, &path_str).await {
96
+
match delete_songs_under(pool, &path_str, typesense).await {
84
97
Ok(n) if n > 0 => tracing::info!("watch delete {n} songs under {path_str}"),
85
98
Ok(_) => {}
86
99
Err(e) => tracing::warn!("watch delete dir {path_str}: {e}"),
87
100
}
88
101
}
89
102
90
-
async fn delete_song(pool: &Db, path: &str) -> Result<bool> {
91
-
let row = sqlx::query_as::<_, (String, String)>(
92
-
"SELECT album_id, artist_id FROM songs WHERE path = ?1",
103
+
async 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",
93
106
)
94
107
.bind(path)
95
108
.fetch_optional(pool)
96
109
.await?;
97
110
98
-
let Some((album_id, artist_id)) = row else {
111
+
let Some((song_id, album_id, artist_id)) = row else {
99
112
return Ok(false);
100
113
};
101
114
···
104
117
.execute(pool)
105
118
.await?;
106
119
107
-
gc_album(pool, &album_id).await?;
108
-
gc_artist(pool, &artist_id).await?;
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?;
109
128
Ok(true)
110
129
}
111
130
112
-
async fn delete_songs_under(pool: &Db, dir: &str) -> Result<u64> {
131
+
async fn delete_songs_under(
132
+
pool: &Db,
133
+
dir: &str,
134
+
typesense: Option<&TypesenseClient>,
135
+
) -> Result<u64> {
113
136
let mut prefix = dir.trim_end_matches(std::path::MAIN_SEPARATOR).to_string();
114
137
prefix.push(std::path::MAIN_SEPARATOR);
115
138
let pattern = format!("{}%", escape_like(&prefix));
116
139
117
-
let rows: Vec<(String, String)> =
118
-
sqlx::query_as("SELECT album_id, artist_id FROM songs WHERE path LIKE ?1 ESCAPE '\\'")
140
+
let rows: Vec<(String, String, String)> =
141
+
sqlx::query_as("SELECT id, album_id, artist_id FROM songs WHERE path LIKE ?1 ESCAPE '\\'")
119
142
.bind(&pattern)
120
143
.fetch_all(pool)
121
144
.await?;
···
132
155
133
156
let mut seen_albums = std::collections::HashSet::new();
134
157
let mut seen_artists = std::collections::HashSet::new();
135
-
for (album_id, artist_id) in rows {
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
+
}
136
164
if seen_albums.insert(album_id.clone()) {
137
-
gc_album(pool, &album_id).await?;
165
+
gc_album(pool, &album_id, typesense).await?;
138
166
}
139
167
if seen_artists.insert(artist_id.clone()) {
140
-
gc_artist(pool, &artist_id).await?;
168
+
gc_artist(pool, &artist_id, typesense).await?;
141
169
}
142
170
}
143
171
Ok(deleted)