jellytau_lib/commands/
favorites.rs

1//! Pushing favourite toggles made while the server was unreachable.
2//!
3//! Favouriting works offline: `storage_toggle_favorite` writes the local
4//! `user_data` row and sets `pending_sync = 1`. Until DR-120 nothing ever
5//! cleared that flag — the offline `mark_favorite`/`unmark_favorite` are no-ops
6//! and `syncService.queueFavorite` had no callers — so an offline toggle was
7//! silently lost.
8//!
9//! The drain lives in Rust, not the frontend, because it must run whether or
10//! not any view is mounted; a drain started by a component dies with it.
11//!
12//! TRACES: UR-069 | DR-120 | UT-103
13
14use std::sync::Arc;
15
16use async_trait::async_trait;
17use log::{debug, info, warn};
18use tauri::{Emitter, Listener, Manager};
19
20use crate::repository::types::RepoError;
21use crate::repository::MediaRepository;
22use crate::storage::db_service::{DatabaseService, Query, QueryParam, RusqliteService};
23
24/// The subset of the repository the drain needs.
25///
26/// Narrow on purpose: a test double for `MediaRepository` would be forty
27/// unimplemented methods, which is how a drain ends up untested.
28#[async_trait]
29pub trait FavoriteSink: Send + Sync {
30    async fn push_favorite(&self, item_id: &str, is_favorite: bool) -> Result<(), RepoError>;
31}
32
33#[async_trait]
34impl<T: MediaRepository + ?Sized> FavoriteSink for T {
35    async fn push_favorite(&self, item_id: &str, is_favorite: bool) -> Result<(), RepoError> {
36        if is_favorite {
37            self.mark_favorite(item_id).await
38        } else {
39            self.unmark_favorite(item_id).await
40        }
41    }
42}
43
44/// A local favourite change still waiting to reach the server.
45#[derive(Debug, Clone, PartialEq, Eq)]
46pub struct PendingFavorite {
47    pub item_id: String,
48    pub is_favorite: bool,
49}
50
51/// Read every favourite change this user has pending.
52async fn read_pending(
53    db: &Arc<RusqliteService>,
54    user_id: &str,
55) -> Result<Vec<PendingFavorite>, String> {
56    db.query_many(
57        Query::with_params(
58            "SELECT item_id, is_favorite FROM user_data \
59             WHERE user_id = ? AND pending_sync = 1 AND is_favorite IS NOT NULL",
60            vec![QueryParam::String(user_id.to_string())],
61        ),
62        |row| {
63            Ok(PendingFavorite {
64                item_id: row.get::<_, String>(0)?,
65                is_favorite: row.get::<_, Option<i32>>(1)?.unwrap_or(0) != 0,
66            })
67        },
68    )
69    .await
70}
71
72/// Push pending favourite changes to the server and clear their flags.
73///
74/// Returns the ids that reached the server, for the `favorites-changed` event.
75/// A row whose push fails keeps `pending_sync = 1` and is retried on the next
76/// reconnect rather than being dropped.
77///
78/// TRACES: UR-069 | DR-120 | UT-103
79pub async fn drain_pending_favorites(
80    db: &Arc<RusqliteService>,
81    sink: &dyn FavoriteSink,
82    user_id: &str,
83) -> Result<Vec<String>, String> {
84    let pending = read_pending(db, user_id).await?;
85    if pending.is_empty() {
86        return Ok(Vec::new());
87    }
88
89    info!(
90        "[Favorites] Pushing {} favourite change(s) queued while offline",
91        pending.len()
92    );
93
94    let mut pushed = Vec::new();
95    for change in pending {
96        match sink
97            .push_favorite(&change.item_id, change.is_favorite)
98            .await
99        {
100            Ok(()) => {
101                let cleared = db
102                    .execute(Query::with_params(
103                        "UPDATE user_data SET pending_sync = 0, synced_at = ? \
104                         WHERE user_id = ? AND item_id = ?",
105                        vec![
106                            QueryParam::String(chrono::Utc::now().to_rfc3339()),
107                            QueryParam::String(user_id.to_string()),
108                            QueryParam::String(change.item_id.clone()),
109                        ],
110                    ))
111                    .await;
112
113                match cleared {
114                    Ok(_) => pushed.push(change.item_id),
115                    // The server took it; failing to clear the flag only means
116                    // we push it again next time, which is harmless.
117                    Err(e) => warn!(
118                        "[Favorites] Pushed {} but could not clear pending_sync: {}",
119                        change.item_id, e
120                    ),
121                }
122            }
123            Err(e) => {
124                // Still pending — retried on the next reconnect.
125                debug!(
126                    "[Favorites] Deferring {}, server rejected the push: {:?}",
127                    change.item_id, e
128                );
129            }
130        }
131    }
132
133    Ok(pushed)
134}
135
136/// Drain on every offline→online transition.
137///
138/// Hooks the `connectivity:reconnected` event the `ConnectivityMonitor`
139/// already emits, rather than polling — reachability is derived from real
140/// traffic (DR-055) and this just reacts to it.
141///
142/// TRACES: UR-069 | DR-120
143pub fn spawn_favorites_drain(app: tauri::AppHandle) {
144    let handle = app.clone();
145    app.listen("connectivity:reconnected", move |_event| {
146        let app = handle.clone();
147        tauri::async_runtime::spawn(async move {
148            if let Err(e) = run_drain(&app).await {
149                warn!("[Favorites] Drain skipped: {}", e);
150            }
151        });
152    });
153}
154
155async fn run_drain(app: &tauri::AppHandle) -> Result<(), String> {
156    let db_service: Arc<RusqliteService> = {
157        let db = app.state::<crate::commands::storage::DatabaseWrapper>();
158        let database = db.0.lock().map_err(|e| e.to_string())?;
159        Arc::new(database.service())
160    };
161
162    let (repo, user_id) = {
163        let manager = app.state::<crate::commands::repository::RepositoryManagerWrapper>();
164        let handles = manager.0.handles();
165        let Some(handle) = handles.first() else {
166            // Not signed in — nothing to push on behalf of.
167            return Ok(());
168        };
169        let repo = manager.0.get(handle).ok_or("Repository not found")?;
170        let user_id = repo.user_id().to_string();
171        (repo, user_id)
172    };
173
174    let pushed = drain_pending_favorites(&db_service, repo.as_ref(), &user_id).await?;
175
176    if !pushed.is_empty() {
177        let event = crate::commands::repository::FavoritesChangedEvent { item_ids: pushed };
178        if let Err(e) = app.emit(crate::commands::repository::FAVORITES_CHANGED_EVENT, &event) {
179            warn!("[Favorites] Failed to emit change event: {}", e);
180        }
181    }
182
183    Ok(())
184}
185
186#[cfg(test)]
187mod tests {
188    use super::*;
189    use crate::utils::lock::MutexSafe;
190    use rusqlite::Connection;
191    use std::sync::Mutex;
192
193    /// Records what the server was asked to do, and can be told to fail.
194    struct RecordingSink {
195        calls: Mutex<Vec<(String, bool)>>,
196        fail_for: Option<String>,
197    }
198
199    impl RecordingSink {
200        fn new() -> Self {
201            Self {
202                calls: Mutex::new(Vec::new()),
203                fail_for: None,
204            }
205        }
206
207        fn failing_for(item_id: &str) -> Self {
208            Self {
209                calls: Mutex::new(Vec::new()),
210                fail_for: Some(item_id.to_string()),
211            }
212        }
213
214        fn calls(&self) -> Vec<(String, bool)> {
215            let mut calls = self.calls.lock_safe().clone();
216            calls.sort();
217            calls
218        }
219    }
220
221    #[async_trait]
222    impl FavoriteSink for RecordingSink {
223        async fn push_favorite(&self, item_id: &str, is_favorite: bool) -> Result<(), RepoError> {
224            if self.fail_for.as_deref() == Some(item_id) {
225                return Err(RepoError::Offline);
226            }
227            self.calls
228                .lock()
229                .unwrap()
230                .push((item_id.to_string(), is_favorite));
231            Ok(())
232        }
233    }
234
235    fn test_db() -> Arc<RusqliteService> {
236        let conn = Connection::open_in_memory().unwrap();
237        conn.execute_batch(
238            r#"
239            CREATE TABLE user_data (
240                user_id TEXT NOT NULL,
241                item_id TEXT NOT NULL,
242                is_favorite INTEGER,
243                synced_at TEXT,
244                pending_sync INTEGER DEFAULT 0,
245                PRIMARY KEY (user_id, item_id)
246            );
247            "#,
248        )
249        .unwrap();
250        Arc::new(RusqliteService::new(Arc::new(Mutex::new(conn))))
251    }
252
253    async fn seed(db: &Arc<RusqliteService>, rows: &[(&str, &str, i32, i32)]) {
254        for (user, item, fav, pending) in rows {
255            db.execute(Query::with_params(
256                "INSERT INTO user_data (user_id, item_id, is_favorite, pending_sync) \
257                 VALUES (?, ?, ?, ?)",
258                vec![
259                    QueryParam::String(user.to_string()),
260                    QueryParam::String(item.to_string()),
261                    QueryParam::Int(*fav),
262                    QueryParam::Int(*pending),
263                ],
264            ))
265            .await
266            .unwrap();
267        }
268    }
269
270    async fn pending_flag(db: &Arc<RusqliteService>, item_id: &str) -> Option<i32> {
271        db.query_optional(
272            Query::with_params(
273                "SELECT pending_sync FROM user_data WHERE item_id = ?",
274                vec![QueryParam::String(item_id.to_string())],
275            ),
276            |row| row.get::<_, Option<i32>>(0),
277        )
278        .await
279        .unwrap()
280        .flatten()
281    }
282
283    /// UT-103 — the core of the bug: a favourite toggled while offline reaches
284    /// the server on reconnect, and stops being pending.
285    ///
286    /// TRACES: UR-069 | DR-120 | UT-103
287    #[tokio::test]
288    async fn test_drain_pushes_pending_favorites_and_clears_the_flag() {
289        let db = test_db();
290        seed(
291            &db,
292            &[
293                ("u1", "marked-offline", 1, 1),
294                ("u1", "unmarked-offline", 0, 1),
295                // Already synced — must not be pushed again.
296                ("u1", "already-synced", 1, 0),
297            ],
298        )
299        .await;
300
301        let sink = RecordingSink::new();
302        let pushed = drain_pending_favorites(&db, &sink, "u1").await.unwrap();
303
304        assert_eq!(
305            sink.calls(),
306            vec![
307                ("marked-offline".to_string(), true),
308                ("unmarked-offline".to_string(), false),
309            ],
310            "both pending changes push, with their direction preserved"
311        );
312        assert_eq!(pushed.len(), 2);
313        assert_eq!(pending_flag(&db, "marked-offline").await, Some(0));
314        assert_eq!(pending_flag(&db, "unmarked-offline").await, Some(0));
315    }
316
317    /// A push that fails keeps its row pending, so the change is retried rather
318    /// than dropped on the floor.
319    ///
320    /// TRACES: UR-069 | DR-120 | UT-103
321    #[tokio::test]
322    async fn test_drain_leaves_failed_pushes_pending() {
323        let db = test_db();
324        seed(&db, &[("u1", "ok", 1, 1), ("u1", "boom", 1, 1)]).await;
325
326        let sink = RecordingSink::failing_for("boom");
327        let pushed = drain_pending_favorites(&db, &sink, "u1").await.unwrap();
328
329        assert_eq!(pushed, vec!["ok".to_string()]);
330        assert_eq!(pending_flag(&db, "ok").await, Some(0));
331        assert_eq!(
332            pending_flag(&db, "boom").await,
333            Some(1),
334            "a failed push must stay queued for the next reconnect"
335        );
336    }
337
338    /// Another user's queued changes are not pushed with this user's token.
339    ///
340    /// TRACES: UR-069 | DR-120 | UT-103
341    #[tokio::test]
342    async fn test_drain_only_touches_the_given_user() {
343        let db = test_db();
344        seed(&db, &[("u1", "mine", 1, 1), ("u2", "theirs", 1, 1)]).await;
345
346        let sink = RecordingSink::new();
347        let pushed = drain_pending_favorites(&db, &sink, "u1").await.unwrap();
348
349        assert_eq!(pushed, vec!["mine".to_string()]);
350        assert_eq!(pending_flag(&db, "theirs").await, Some(1));
351    }
352
353    /// Nothing pending means no server calls at all — a reconnect must not
354    /// generate traffic just because it happened.
355    ///
356    /// TRACES: UR-069 | DR-120 | UT-103
357    #[tokio::test]
358    async fn test_drain_is_a_noop_when_nothing_is_pending() {
359        let db = test_db();
360        seed(&db, &[("u1", "synced", 1, 0)]).await;
361
362        let sink = RecordingSink::new();
363        let pushed = drain_pending_favorites(&db, &sink, "u1").await.unwrap();
364
365        assert!(pushed.is_empty());
366        assert!(sink.calls().is_empty());
367    }
368}
jellytau_lib/commands/favorites.rs

jellytau_lib/commands/
favorites.rs
