Skip to main content

koan_core/remote/
sync.rs

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    /// Albums whose details could not be fetched. Non-zero means `last_sync`
26    /// was left where it was so the next sync picks them up again.
27    pub albums_failed: usize,
28}
29
30impl SyncResult {
31    /// Whether the run covered everything it set out to.
32    pub fn is_complete(&self) -> bool {
33        self.albums_failed == 0
34    }
35}
36
37/// Get the last sync timestamp for a remote server, if any.
38pub 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
54/// Update (or insert) the last sync timestamp for a remote server.
55pub 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
70/// Parse an ISO 8601 / RFC 3339 timestamp string into a unix timestamp (seconds).
71/// Returns `None` if the string can't be parsed.
72///
73/// Handles common Subsonic/Navidrome variants:
74/// - Full RFC 3339: `2024-01-15T10:30:00Z`, `2024-01-15T10:30:00+05:30`
75/// - Fractional seconds: `2024-01-15T10:30:00.123Z`
76/// - Missing timezone (assumed UTC): `2024-01-15T10:30:00`
77fn parse_iso8601_to_unix(s: &str) -> Option<i64> {
78    use chrono::{DateTime, FixedOffset, NaiveDateTime};
79
80    // Try strict RFC 3339 first (handles Z, offsets, fractional seconds).
81    if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
82        return Some(dt.timestamp());
83    }
84
85    // Subsonic sometimes omits timezone — parse as naive and assume UTC.
86    // Try with fractional seconds first, then without.
87    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    // Some servers use space instead of T.
95    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
105/// Pull the Navidrome/Subsonic library into the local DB.
106///
107/// The album list is always walked in `alphabeticalByName` order: it is the one
108/// ordering stable under concurrent server-side inserts, so an offset walk can
109/// never skip an album that was added between two pages. An incremental sync
110/// walks the same list but only fetches details for albums created after
111/// `last_sync` — the list pages are cheap, the per-album fetches are not.
112///
113/// `last_sync` only advances when every album fetch succeeded. A run that lost
114/// albums to network errors leaves the timestamp alone so the next incremental
115/// sync re-fetches them, rather than writing a permanent hole in the library.
116///
117/// Deduplication happens in `upsert_track` — if a local track already exists
118/// with the same artist + album + title + track#, the remote_id and remote_url
119/// are merged onto the existing row instead of creating a duplicate.
120pub 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    // Guards against an album appearing on two pages when the server-side list
152    // shifts under the offset walk.
153    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        // Skip albums already handled this run, and — when incremental —
164        // albums the server says predate the last sync. An unparseable
165        // `created` is treated as new: re-fetching is cheap, missing is not.
166        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
229/// Write one batch of fetched albums in a single transaction.
230fn 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        // 2024-01-15T10:30:00Z = 1705314600
293        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        // 10:30 IST (+05:30) = 05:00 UTC = 1705294800
302        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        // 10:30 EST (-05:00) = 15:30 UTC = 1705332600
311        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    // --- Sync → DB integration tests ---
378
379    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    /// Build a TrackMeta matching how sync_library constructs them from SubsonicSong data.
390    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        // Verify the track exists with correct remote_id.
425        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    // --- sync_library against a stub Subsonic server ---
436
437    /// Minimal Subsonic server: serves getArtists, getAlbumList2 and getAlbum
438    /// so `sync_library` can be driven end to end without a real Navidrome.
439    struct StubServer {
440        addr: std::net::SocketAddr,
441        shutdown: Arc<std::sync::atomic::AtomicBool>,
442    }
443
444    #[derive(Default)]
445    struct StubState {
446        /// Album (id, name, created) in the order the server lists them.
447        albums: Mutex<Vec<(String, String, String)>>,
448        /// Album ids whose getAlbum call fails with a 500.
449        failing: Mutex<HashSet<String>>,
450        /// Prepended to the album list once the first list page has been served,
451        /// modelling a server-side insert landing mid-pagination.
452        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                            // BSD sockets inherit O_NONBLOCK from the listener.
471                            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    /// Serve requests on one connection until the peer closes it. Keep-alive
498    /// matters here: reqwest pools connections, and a server that hangs up after
499    /// every response makes parallel fetches fail on reused sockets.
500    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, &params);
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        // Second run: the album now succeeds and is picked up because the sync
641        // still has no watermark to skip past.
642        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        // 600 albums forces a second list page; a server-side insert between
665        // pages shifts the offset window, which without de-dup replays the
666        // page boundary and, with `newest` ordering, can drop albums entirely.
667        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        // Every album present before the shift must still have been fetched.
692        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        // Watermark between the two vintages.
721        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        // First upsert.
757        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        // Second upsert with same remote_id but different metadata.
761        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        // Should be the same row (dedup by remote_id).
765        assert_eq!(id1, id2, "same remote_id should resolve to same track row");
766
767        // Verify the metadata was updated.
768        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        // Verify only one track exists.
775        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}