1use std::collections::HashSet;
2use std::sync::atomic::{AtomicUsize, Ordering};
3
4use crate::db::connection::Database;
5use crate::db::queries::{self, TrackMeta};
6use crate::remote::client::{SubsonicAlbumFull, SubsonicClient};
7
8use rayon::prelude::*;
9use rusqlite::params;
10use thiserror::Error;
11
12#[derive(Debug, Error)]
13pub enum SyncError {
14 #[error("subsonic error: {0}")]
15 Subsonic(#[from] super::client::SubsonicError),
16 #[error("db error: {0}")]
17 Db(#[from] crate::db::connection::DbError),
18}
19
20#[derive(Debug, Default)]
21pub struct SyncResult {
22 pub artists_synced: usize,
23 pub albums_synced: usize,
24 pub tracks_synced: usize,
25 pub albums_failed: usize,
28}
29
30impl SyncResult {
31 pub fn is_complete(&self) -> bool {
33 self.albums_failed == 0
34 }
35}
36
37pub fn get_last_sync(
39 db: &Database,
40 url: &str,
41) -> Result<Option<i64>, crate::db::connection::DbError> {
42 let result = db.conn.query_row(
43 "SELECT last_sync FROM remote_servers WHERE url = ?1",
44 params![url],
45 |row| row.get::<_, Option<i64>>(0),
46 );
47 match result {
48 Ok(ts) => Ok(ts),
49 Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
50 Err(e) => Err(e.into()),
51 }
52}
53
54pub fn update_last_sync(
56 db: &Database,
57 url: &str,
58 username: &str,
59 timestamp: i64,
60) -> Result<(), crate::db::connection::DbError> {
61 db.conn.execute(
62 "INSERT INTO remote_servers (url, username, last_sync)
63 VALUES (?1, ?2, ?3)
64 ON CONFLICT(url) DO UPDATE SET last_sync = ?3",
65 params![url, username, timestamp],
66 )?;
67 Ok(())
68}
69
70fn parse_iso8601_to_unix(s: &str) -> Option<i64> {
78 use chrono::{DateTime, FixedOffset, NaiveDateTime};
79
80 if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
82 return Some(dt.timestamp());
83 }
84
85 if let Ok(naive) = NaiveDateTime::parse_from_str(s, "%Y-%m-%dT%H:%M:%S%.f") {
88 return Some(naive.and_utc().timestamp());
89 }
90 if let Ok(naive) = NaiveDateTime::parse_from_str(s, "%Y-%m-%dT%H:%M:%S") {
91 return Some(naive.and_utc().timestamp());
92 }
93
94 if let Ok(dt) = DateTime::<FixedOffset>::parse_from_str(s, "%Y-%m-%d %H:%M:%S%:z") {
96 return Some(dt.timestamp());
97 }
98 if let Ok(naive) = NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S") {
99 return Some(naive.and_utc().timestamp());
100 }
101
102 None
103}
104
105pub fn sync_library(
121 db: &Database,
122 client: &SubsonicClient,
123 full: bool,
124 server_url: &str,
125 username: &str,
126) -> Result<SyncResult, SyncError> {
127 let mut result = SyncResult::default();
128
129 let artists = client.get_artists()?;
130 result.artists_synced = artists.len();
131 log::info!("syncing {} artists from remote", artists.len());
132
133 let last_sync = if full {
134 None
135 } else {
136 get_last_sync(db, server_url)?
137 };
138
139 match last_sync {
140 Some(ts) => log::info!("incremental sync (albums created after {})", ts),
141 None => log::info!("full sync"),
142 }
143
144 let sync_start = std::time::SystemTime::now()
145 .duration_since(std::time::UNIX_EPOCH)
146 .unwrap_or_default()
147 .as_secs() as i64;
148
149 let mut offset = 0u32;
150 let page_size = 500u32;
151 let mut seen_ids: HashSet<String> = HashSet::new();
154
155 loop {
156 let page = client.get_album_list("alphabeticalByName", page_size, offset)?;
157 if page.is_empty() {
158 break;
159 }
160 let page_count = page.len();
161 offset += page_count as u32;
162
163 let to_fetch: Vec<String> = page
167 .into_iter()
168 .filter(|a| seen_ids.insert(a.id.clone()))
169 .filter(|a| match last_sync {
170 None => true,
171 Some(ts) => a
172 .created
173 .as_deref()
174 .and_then(parse_iso8601_to_unix)
175 .is_none_or(|created| created >= ts),
176 })
177 .map(|a| a.id)
178 .collect();
179
180 if !to_fetch.is_empty() {
181 let failures = AtomicUsize::new(0);
182 let fetched: Vec<SubsonicAlbumFull> = to_fetch
183 .into_par_iter()
184 .filter_map(|id| match client.get_album(&id) {
185 Ok(full) => Some(full),
186 Err(e) => {
187 log::warn!("failed to fetch album {}: {}", id, e);
188 failures.fetch_add(1, Ordering::Relaxed);
189 None
190 }
191 })
192 .collect();
193 result.albums_failed += failures.into_inner();
194
195 write_albums(db, client, &fetched, &mut result)?;
196 }
197
198 log::info!(
199 "synced {} albums ({} tracks) so far...",
200 result.albums_synced,
201 result.tracks_synced
202 );
203
204 if (page_count as u32) < page_size {
205 break;
206 }
207 }
208
209 if result.is_complete() {
210 update_last_sync(db, server_url, username, sync_start)?;
211 } else {
212 log::warn!(
213 "{} album(s) failed to fetch — leaving last_sync unchanged so the next sync retries them",
214 result.albums_failed
215 );
216 }
217
218 log::info!(
219 "sync complete: {} artists, {} albums, {} tracks, {} failed",
220 result.artists_synced,
221 result.albums_synced,
222 result.tracks_synced,
223 result.albums_failed,
224 );
225
226 Ok(result)
227}
228
229fn write_albums(
231 db: &Database,
232 client: &SubsonicClient,
233 albums: &[SubsonicAlbumFull],
234 result: &mut SyncResult,
235) -> Result<(), SyncError> {
236 db.conn
237 .execute_batch("BEGIN")
238 .map_err(crate::db::connection::DbError::from)?;
239
240 for album in albums {
241 result.albums_synced += 1;
242 let artist_name = album.artist.as_deref().unwrap_or("Unknown Artist");
243
244 for song in &album.song {
245 let meta = TrackMeta {
246 title: song.title.clone(),
247 artist: song
248 .artist
249 .clone()
250 .unwrap_or_else(|| artist_name.to_string()),
251 album_artist: album.artist.clone(),
252 album: album.name.clone(),
253 date: album.year.map(|y| y.to_string()),
254 disc: song.disc_number,
255 track_number: song.track,
256 genre: song.genre.clone().or_else(|| album.genre.clone()),
257 label: None,
258 duration_ms: song.duration.map(|d| d * 1000),
259 codec: song.suffix.clone(),
260 sample_rate: None,
261 bit_depth: None,
262 channels: None,
263 bitrate: song.bit_rate,
264 size_bytes: None,
265 mtime: None,
266 path: None,
267 source: "remote".to_string(),
268 remote_id: Some(song.id.clone()),
269 remote_url: Some(client.stream_url_template(&song.id)),
270 };
271
272 match queries::upsert_track(&db.conn, &meta) {
273 Ok(_) => result.tracks_synced += 1,
274 Err(e) => log::warn!("failed to insert remote track {}: {}", song.title, e),
275 }
276 }
277 }
278
279 db.conn
280 .execute_batch("COMMIT")
281 .map_err(crate::db::connection::DbError::from)?;
282
283 Ok(())
284}
285
286#[cfg(test)]
287mod tests {
288 use super::*;
289
290 #[test]
291 fn parse_rfc3339_with_z() {
292 assert_eq!(
294 parse_iso8601_to_unix("2024-01-15T10:30:00Z"),
295 Some(1705314600)
296 );
297 }
298
299 #[test]
300 fn parse_rfc3339_with_offset() {
301 assert_eq!(
303 parse_iso8601_to_unix("2024-01-15T10:30:00+05:30"),
304 Some(1705294800)
305 );
306 }
307
308 #[test]
309 fn parse_rfc3339_negative_offset() {
310 assert_eq!(
312 parse_iso8601_to_unix("2024-01-15T10:30:00-05:00"),
313 Some(1705332600)
314 );
315 }
316
317 #[test]
318 fn parse_fractional_seconds_z() {
319 assert_eq!(
320 parse_iso8601_to_unix("2024-01-15T10:30:00.123Z"),
321 Some(1705314600)
322 );
323 }
324
325 #[test]
326 fn parse_fractional_seconds_offset() {
327 assert_eq!(
328 parse_iso8601_to_unix("2024-01-15T10:30:00.999+00:00"),
329 Some(1705314600)
330 );
331 }
332
333 #[test]
334 fn parse_no_timezone_assumes_utc() {
335 assert_eq!(
336 parse_iso8601_to_unix("2024-01-15T10:30:00"),
337 Some(1705314600)
338 );
339 }
340
341 #[test]
342 fn parse_no_timezone_fractional() {
343 assert_eq!(
344 parse_iso8601_to_unix("2024-01-15T10:30:00.500"),
345 Some(1705314600)
346 );
347 }
348
349 #[test]
350 fn parse_space_separator_with_tz() {
351 assert_eq!(
352 parse_iso8601_to_unix("2024-01-15 10:30:00+00:00"),
353 Some(1705314600)
354 );
355 }
356
357 #[test]
358 fn parse_space_separator_no_tz() {
359 assert_eq!(
360 parse_iso8601_to_unix("2024-01-15 10:30:00"),
361 Some(1705314600)
362 );
363 }
364
365 #[test]
366 fn parse_garbage_returns_none() {
367 assert_eq!(parse_iso8601_to_unix("not-a-date"), None);
368 assert_eq!(parse_iso8601_to_unix(""), None);
369 assert_eq!(parse_iso8601_to_unix("2024"), None);
370 }
371
372 #[test]
373 fn parse_epoch() {
374 assert_eq!(parse_iso8601_to_unix("1970-01-01T00:00:00Z"), Some(0));
375 }
376
377 use crate::db::connection::Database;
380 use crate::db::queries;
381 use std::sync::{Arc, Mutex};
382
383 fn test_db() -> (Database, tempfile::TempDir) {
384 let dir = tempfile::tempdir().unwrap();
385 let db = Database::open(&dir.path().join("sync_test.db")).unwrap();
386 (db, dir)
387 }
388
389 fn remote_track_meta(remote_id: &str, title: &str, artist: &str, album: &str) -> TrackMeta {
391 TrackMeta {
392 title: title.into(),
393 artist: artist.into(),
394 album_artist: Some(artist.into()),
395 album: album.into(),
396 date: Some("2024".into()),
397 disc: Some(1),
398 track_number: Some(1),
399 genre: Some("Electronic".into()),
400 label: None,
401 duration_ms: Some(240_000),
402 codec: Some("FLAC".into()),
403 sample_rate: None,
404 bit_depth: None,
405 channels: None,
406 bitrate: Some(1000),
407 size_bytes: None,
408 mtime: None,
409 path: None,
410 source: "remote".into(),
411 remote_id: Some(remote_id.into()),
412 remote_url: Some(format!("https://example.com/stream?id={}", remote_id)),
413 }
414 }
415
416 #[test]
417 fn sync_upserts_tracks_to_database() {
418 let (db, _dir) = test_db();
419
420 let meta = remote_track_meta("remote-001", "Vordhosbn", "Aphex Twin", "Drukqs");
421 let track_id = queries::upsert_track(&db.conn, &meta).unwrap();
422 assert!(track_id > 0, "upsert should return a valid track ID");
423
424 let row = queries::get_track_row(&db.conn, track_id)
426 .unwrap()
427 .expect("track should exist in DB");
428 assert_eq!(row.title, "Vordhosbn");
429 assert_eq!(row.artist_name, "Aphex Twin");
430 assert_eq!(row.album_title, "Drukqs");
431 assert_eq!(row.remote_id.as_deref(), Some("remote-001"));
432 assert_eq!(row.source, "remote");
433 }
434
435 struct StubServer {
440 addr: std::net::SocketAddr,
441 shutdown: Arc<std::sync::atomic::AtomicBool>,
442 }
443
444 #[derive(Default)]
445 struct StubState {
446 albums: Mutex<Vec<(String, String, String)>>,
448 failing: Mutex<HashSet<String>>,
450 insert_after_first_page: Mutex<Option<(String, String, String)>>,
453 list_pages_served: AtomicUsize,
454 list_types: Mutex<Vec<String>>,
455 album_calls: Mutex<Vec<String>>,
456 }
457
458 impl StubServer {
459 fn start(state: Arc<StubState>) -> Self {
460 let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
461 listener.set_nonblocking(true).unwrap();
462 let addr = listener.local_addr().unwrap();
463 let shutdown = Arc::new(std::sync::atomic::AtomicBool::new(false));
464
465 let stop = shutdown.clone();
466 std::thread::spawn(move || {
467 while !stop.load(Ordering::Relaxed) {
468 match listener.accept() {
469 Ok((stream, _)) => {
470 let _ = stream.set_nonblocking(false);
472 let state = state.clone();
473 std::thread::spawn(move || handle(stream, state));
474 }
475 Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
476 std::thread::sleep(std::time::Duration::from_millis(2));
477 }
478 Err(_) => break,
479 }
480 }
481 });
482
483 Self { addr, shutdown }
484 }
485
486 fn url(&self) -> String {
487 format!("http://{}", self.addr)
488 }
489 }
490
491 impl Drop for StubServer {
492 fn drop(&mut self) {
493 self.shutdown.store(true, Ordering::Relaxed);
494 }
495 }
496
497 fn handle(mut stream: std::net::TcpStream, state: Arc<StubState>) {
501 use std::io::{BufRead, Write};
502
503 let Ok(peek) = stream.try_clone() else { return };
504 let mut reader = std::io::BufReader::new(peek);
505
506 loop {
507 let mut request_line = String::new();
508 if reader.read_line(&mut request_line).unwrap_or(0) == 0 {
509 return;
510 }
511 let mut line = String::new();
512 while reader.read_line(&mut line).unwrap_or(0) > 0 {
513 if line == "\r\n" || line == "\n" {
514 break;
515 }
516 line.clear();
517 }
518
519 let target = request_line.split_whitespace().nth(1).unwrap_or("/");
520 let (path, query) = target.split_once('?').unwrap_or((target, ""));
521 let params: std::collections::HashMap<&str, &str> = query
522 .split('&')
523 .filter_map(|kv| kv.split_once('='))
524 .collect();
525
526 let (status, body) = respond(&state, path, ¶ms);
527
528 let write = write!(
529 stream,
530 "HTTP/1.1 {} OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n",
531 status,
532 body.len()
533 )
534 .and_then(|()| stream.write_all(body.as_bytes()))
535 .and_then(|()| stream.flush());
536 if write.is_err() {
537 return;
538 }
539 }
540 }
541
542 fn respond(
543 state: &Arc<StubState>,
544 path: &str,
545 params: &std::collections::HashMap<&str, &str>,
546 ) -> (u16, String) {
547 match path.rsplit('/').next().unwrap_or("") {
548 "getArtists" => (
549 200,
550 r#"{"subsonic-response":{"status":"ok","artists":{"index":[{"artist":[{"id":"ar1","name":"Stub Artist"}]}]}}}"#.to_string(),
551 ),
552 "getAlbumList2" => {
553 state
554 .list_types
555 .lock()
556 .unwrap()
557 .push(params.get("type").copied().unwrap_or("").to_string());
558 let offset: usize = params.get("offset").and_then(|o| o.parse().ok()).unwrap_or(0);
559 let size: usize = params.get("size").and_then(|s| s.parse().ok()).unwrap_or(500);
560
561 let albums = state.albums.lock().unwrap();
562 let slice: Vec<String> = albums
563 .iter()
564 .skip(offset)
565 .take(size)
566 .map(|(id, name, created)| {
567 format!(
568 r#"{{"id":"{}","name":"{}","artist":"Stub Artist","created":"{}"}}"#,
569 id, name, created
570 )
571 })
572 .collect();
573 drop(albums);
574
575 if state.list_pages_served.fetch_add(1, Ordering::SeqCst) == 0
576 && let Some(new_album) = state.insert_after_first_page.lock().unwrap().take()
577 {
578 state.albums.lock().unwrap().insert(0, new_album);
579 }
580
581 (
582 200,
583 format!(
584 r#"{{"subsonic-response":{{"status":"ok","albumList2":{{"album":[{}]}}}}}}"#,
585 slice.join(",")
586 ),
587 )
588 }
589 "getAlbum" => {
590 let id = params.get("id").copied().unwrap_or("");
591 state.album_calls.lock().unwrap().push(id.to_string());
592 if state.failing.lock().unwrap().contains(id) {
593 (500, r#"{"error":"boom"}"#.to_string())
594 } else {
595 (
596 200,
597 format!(
598 r#"{{"subsonic-response":{{"status":"ok","album":{{"id":"{id}","name":"Album {id}","artist":"Stub Artist","song":[{{"id":"s{id}","title":"Song {id}","track":1,"suffix":"flac"}}]}}}}}}"#
599 ),
600 )
601 }
602 }
603 _ => (404, "{}".to_string()),
604 }
605 }
606
607 fn stub_albums(n: usize) -> Vec<(String, String, String)> {
608 (0..n)
609 .map(|i| {
610 (
611 format!("a{:04}", i),
612 format!("Album {:04}", i),
613 "2024-01-15T10:30:00Z".to_string(),
614 )
615 })
616 .collect()
617 }
618
619 #[test]
620 fn failed_album_fetch_does_not_advance_last_sync_and_next_sync_retries() {
621 let (db, _dir) = test_db();
622 let state = Arc::new(StubState {
623 albums: Mutex::new(stub_albums(4)),
624 failing: Mutex::new(["a0002".to_string()].into_iter().collect()),
625 ..Default::default()
626 });
627 let server = StubServer::start(state.clone());
628 let client = SubsonicClient::new(&server.url(), "u", "p");
629
630 let first = sync_library(&db, &client, false, &server.url(), "u").unwrap();
631 assert_eq!(first.albums_failed, 1, "the failing album must be counted");
632 assert_eq!(first.albums_synced, 3);
633 assert!(!first.is_complete());
634 assert_eq!(
635 get_last_sync(&db, &server.url()).unwrap(),
636 None,
637 "an incomplete sync must not advance last_sync"
638 );
639
640 state.failing.lock().unwrap().clear();
643 state.album_calls.lock().unwrap().clear();
644 let second = sync_library(&db, &client, false, &server.url(), "u").unwrap();
645
646 assert!(
647 state
648 .album_calls
649 .lock()
650 .unwrap()
651 .contains(&"a0002".to_string()),
652 "the previously failed album must be retried"
653 );
654 assert_eq!(second.albums_failed, 0);
655 assert!(second.is_complete());
656 assert!(
657 get_last_sync(&db, &server.url()).unwrap().is_some(),
658 "a clean sync advances last_sync"
659 );
660 }
661
662 #[test]
663 fn album_inserted_mid_pagination_is_not_fetched_twice_or_skipped() {
664 let (db, _dir) = test_db();
668 let state = Arc::new(StubState {
669 albums: Mutex::new(stub_albums(600)),
670 insert_after_first_page: Mutex::new(Some((
671 "aNEW".to_string(),
672 "AAA Brand New".to_string(),
673 "2024-06-01T00:00:00Z".to_string(),
674 ))),
675 ..Default::default()
676 });
677 let server = StubServer::start(state.clone());
678 let client = SubsonicClient::new(&server.url(), "u", "p");
679
680 let result = sync_library(&db, &client, true, &server.url(), "u").unwrap();
681 assert_eq!(result.albums_failed, 0);
682
683 let calls = state.album_calls.lock().unwrap().clone();
684 let unique: HashSet<&String> = calls.iter().collect();
685 assert_eq!(
686 calls.len(),
687 unique.len(),
688 "no album may be fetched twice after the window shifts"
689 );
690
691 for i in 0..600 {
693 let id = format!("a{:04}", i);
694 assert!(unique.contains(&id), "album {} was skipped", id);
695 }
696
697 let types = state.list_types.lock().unwrap().clone();
698 assert!(
699 types.iter().all(|t| t == "alphabeticalByName"),
700 "the paginated walk must use a stable ordering, got {:?}",
701 types
702 );
703 }
704
705 #[test]
706 fn incremental_sync_only_fetches_albums_created_after_last_sync() {
707 let (db, _dir) = test_db();
708 let mut albums = stub_albums(3);
709 albums[0].2 = "2020-01-01T00:00:00Z".into();
710 albums[1].2 = "2020-01-01T00:00:00Z".into();
711 albums[2].2 = "2030-01-01T00:00:00Z".into();
712
713 let state = Arc::new(StubState {
714 albums: Mutex::new(albums),
715 ..Default::default()
716 });
717 let server = StubServer::start(state.clone());
718 let client = SubsonicClient::new(&server.url(), "u", "p");
719
720 let watermark = parse_iso8601_to_unix("2025-01-01T00:00:00Z").unwrap();
722 update_last_sync(&db, &server.url(), "u", watermark).unwrap();
723
724 let result = sync_library(&db, &client, false, &server.url(), "u").unwrap();
725
726 assert_eq!(result.albums_synced, 1, "only the new album needs fetching");
727 assert_eq!(
728 *state.album_calls.lock().unwrap(),
729 vec!["a0002".to_string()]
730 );
731 }
732
733 #[test]
734 fn album_with_unparseable_created_is_always_fetched() {
735 let (db, _dir) = test_db();
736 let state = Arc::new(StubState {
737 albums: Mutex::new(vec![("a0000".into(), "Album".into(), "who knows".into())]),
738 ..Default::default()
739 });
740 let server = StubServer::start(state.clone());
741 let client = SubsonicClient::new(&server.url(), "u", "p");
742
743 update_last_sync(&db, &server.url(), "u", 4_000_000_000).unwrap();
744 let result = sync_library(&db, &client, false, &server.url(), "u").unwrap();
745
746 assert_eq!(
747 result.albums_synced, 1,
748 "an album with no usable timestamp must not be assumed old"
749 );
750 }
751
752 #[test]
753 fn sync_deduplicates_by_remote_id() {
754 let (db, _dir) = test_db();
755
756 let meta1 = remote_track_meta("remote-dup", "Original Title", "Artist A", "Album X");
758 let id1 = queries::upsert_track(&db.conn, &meta1).unwrap();
759
760 let meta2 = remote_track_meta("remote-dup", "Updated Title", "Artist A", "Album X");
762 let id2 = queries::upsert_track(&db.conn, &meta2).unwrap();
763
764 assert_eq!(id1, id2, "same remote_id should resolve to same track row");
766
767 let row = queries::get_track_row(&db.conn, id2)
769 .unwrap()
770 .expect("track should exist");
771 assert_eq!(row.title, "Updated Title");
772 assert_eq!(row.remote_id.as_deref(), Some("remote-dup"));
773
774 let stats = queries::library_stats(&db.conn).unwrap();
776 assert_eq!(
777 stats.total_tracks, 1,
778 "should have exactly 1 track after dedup"
779 );
780 }
781}