jellytau_lib/commands/
sync.rs

1//! Tauri commands for sync queue operations
2//!
3//! The sync queue stores mutations (favorites, playback progress, etc.)
4//! that need to be synced to the Jellyfin server when connectivity is restored.
5//! Draining it lives in `sync_drain` (DR-131); this module is the storage and
6//! read side the UI lists from (DR-132).
7//! TRACES: UR-002, UR-017, UR-025 | DR-014, DR-131, DR-132
8
9use serde::{Deserialize, Serialize};
10use std::sync::Arc;
11use tauri::State;
12
13use super::storage::DatabaseWrapper;
14use crate::storage::db_service::{DatabaseService, Query, QueryParam};
15
16/// Sync queue item returned to frontend
17#[derive(specta::Type, Debug, Clone, Serialize, Deserialize)]
18#[serde(rename_all = "camelCase")]
19pub struct SyncQueueItem {
20    pub id: i64,
21    pub user_id: String,
22    pub operation: String,
23    pub item_id: Option<String>,
24    pub payload: Option<String>,
25    pub status: String,
26    pub retry_count: i32,
27    pub created_at: Option<String>,
28    pub error_message: Option<String>,
29    /// Cached title of the item the operation is about, when the catalog knows
30    /// it. Resolved here rather than by a per-row frontend fetch — the queue
31    /// list is otherwise a wall of opaque ids.
32    ///
33    /// TRACES: UR-025 | DR-132
34    pub item_name: Option<String>,
35}
36
37/// Queue a mutation for sync to server
38#[tauri::command]
39#[specta::specta]
40pub async fn sync_queue_mutation(
41    db: State<'_, DatabaseWrapper>,
42    user_id: String,
43    operation: String,
44    item_id: Option<String>,
45    payload: Option<String>,
46) -> Result<i64, String> {
47    let db_service = {
48        let database = db.0.lock().map_err(|e| e.to_string())?;
49        Arc::new(database.service())
50    };
51
52    enqueue_mutation(&*db_service, user_id, operation, item_id, payload).await
53}
54
55/// Insert one pending mutation and return the id of *that* row.
56///
57/// TRACES: UR-002, UR-017 | DR-014
58pub(crate) async fn enqueue_mutation<S: DatabaseService>(
59    db_service: &S,
60    user_id: String,
61    operation: String,
62    item_id: Option<String>,
63    payload: Option<String>,
64) -> Result<i64, String> {
65    let query = Query::with_params(
66        "INSERT INTO sync_queue (user_id, operation, item_id, payload, status, created_at)
67         VALUES (?, ?, ?, ?, 'pending', CURRENT_TIMESTAMP)",
68        vec![
69            QueryParam::String(user_id),
70            QueryParam::String(operation),
71            item_id.map(QueryParam::String).unwrap_or(QueryParam::Null),
72            payload.map(QueryParam::String).unwrap_or(QueryParam::Null),
73        ],
74    );
75
76    // `insert`, not `execute` + `last_insert_rowid`: the id must be read in
77    // the same job as the insert, or a concurrent write hands us its row.
78    db_service.insert(query).await
79}
80
81/// Get all pending sync operations for a user
82#[tauri::command]
83#[specta::specta]
84pub async fn sync_get_pending(
85    db: State<'_, DatabaseWrapper>,
86    user_id: String,
87    limit: Option<i32>,
88) -> Result<Vec<SyncQueueItem>, String> {
89    let db_service = {
90        let database = db.0.lock().map_err(|e| e.to_string())?;
91        Arc::new(database.service())
92    };
93
94    // The `items` join names the queued item where the catalog has it; a row for
95    // an item that was never cached still lists, with a null name.
96    // `abandoned` rows (DR-131 gave up on them) are excluded here for the same
97    // reason they are excluded from the count — they are no longer waiting.
98    const SELECT: &str = "SELECT q.id, q.user_id, q.operation, q.item_id, q.payload, q.status,
99                                 COALESCE(q.retry_count, 0), q.created_at, q.error_message, i.name
100                          FROM sync_queue q
101                          LEFT JOIN items i ON i.id = q.item_id
102                          WHERE q.user_id = ? AND q.status IN ('pending', 'failed')
103                          ORDER BY q.created_at ASC, q.id ASC";
104
105    let sql = match limit {
106        Some(l) => format!("{} LIMIT {}", SELECT, l),
107        None => SELECT.to_string(),
108    };
109
110    let query = Query::with_params(sql, vec![QueryParam::String(user_id)]);
111
112    db_service
113        .query_many(query, |row| {
114            Ok(SyncQueueItem {
115                id: row.get(0)?,
116                user_id: row.get(1)?,
117                operation: row.get(2)?,
118                item_id: row.get(3)?,
119                payload: row.get(4)?,
120                status: row.get(5)?,
121                retry_count: row.get(6)?,
122                created_at: row.get(7)?,
123                error_message: row.get(8)?,
124                item_name: row.get(9)?,
125            })
126        })
127        .await
128        .map_err(|e| e.to_string())
129}
130
131/// Mark a sync operation as in progress
132#[tauri::command]
133#[specta::specta]
134pub async fn sync_mark_processing(db: State<'_, DatabaseWrapper>, id: i64) -> Result<(), String> {
135    let db_service = {
136        let database = db.0.lock().map_err(|e| e.to_string())?;
137        Arc::new(database.service())
138    };
139
140    let query = Query::with_params(
141        "UPDATE sync_queue SET status = 'processing' WHERE id = ?",
142        vec![QueryParam::Int64(id)],
143    );
144
145    db_service.execute(query).await.map_err(|e| e.to_string())?;
146    Ok(())
147}
148
149/// Mark a sync operation as completed
150#[tauri::command]
151#[specta::specta]
152pub async fn sync_mark_completed(db: State<'_, DatabaseWrapper>, id: i64) -> Result<(), String> {
153    let db_service = {
154        let database = db.0.lock().map_err(|e| e.to_string())?;
155        Arc::new(database.service())
156    };
157
158    let query = Query::with_params(
159        "UPDATE sync_queue SET status = 'completed', processed_at = CURRENT_TIMESTAMP WHERE id = ?",
160        vec![QueryParam::Int64(id)],
161    );
162
163    db_service.execute(query).await.map_err(|e| e.to_string())?;
164    Ok(())
165}
166
167/// Mark a sync operation as failed with error message
168#[tauri::command]
169#[specta::specta]
170pub async fn sync_mark_failed(
171    db: State<'_, DatabaseWrapper>,
172    id: i64,
173    error: String,
174) -> Result<(), String> {
175    let db_service = {
176        let database = db.0.lock().map_err(|e| e.to_string())?;
177        Arc::new(database.service())
178    };
179
180    let query = Query::with_params(
181        "UPDATE sync_queue
182         SET status = 'failed',
183             retry_count = retry_count + 1,
184             error_message = ?,
185             processed_at = CURRENT_TIMESTAMP
186         WHERE id = ?",
187        vec![QueryParam::String(error), QueryParam::Int64(id)],
188    );
189
190    db_service.execute(query).await.map_err(|e| e.to_string())?;
191    Ok(())
192}
193
194/// Get count of pending sync operations for a user
195#[tauri::command]
196#[specta::specta]
197pub async fn sync_get_pending_count(
198    db: State<'_, DatabaseWrapper>,
199    user_id: String,
200) -> Result<i32, String> {
201    let db_service = {
202        let database = db.0.lock().map_err(|e| e.to_string())?;
203        Arc::new(database.service())
204    };
205
206    let query = Query::with_params(
207        "SELECT COUNT(*) FROM sync_queue WHERE user_id = ? AND status IN ('pending', 'failed')",
208        vec![QueryParam::String(user_id)],
209    );
210
211    db_service
212        .query_one(query, |row| row.get(0))
213        .await
214        .map_err(|e| e.to_string())
215}
216
217/// Delete completed sync operations older than specified days
218#[tauri::command]
219#[specta::specta]
220pub async fn sync_cleanup_completed(
221    db: State<'_, DatabaseWrapper>,
222    days_old: i32,
223) -> Result<i32, String> {
224    let db_service = {
225        let database = db.0.lock().map_err(|e| e.to_string())?;
226        Arc::new(database.service())
227    };
228
229    let query = Query::with_params(
230        "DELETE FROM sync_queue
231         WHERE status = 'completed'
232         AND processed_at < datetime('now', ?)",
233        vec![QueryParam::String(format!("-{} days", days_old))],
234    );
235
236    let deleted = db_service.execute(query).await.map_err(|e| e.to_string())?;
237    Ok(deleted as i32)
238}
239
240/// Delete all sync operations for a user (used during logout)
241#[tauri::command]
242#[specta::specta]
243pub async fn sync_clear_user(
244    db: State<'_, DatabaseWrapper>,
245    user_id: String,
246) -> Result<(), String> {
247    let db_service = {
248        let database = db.0.lock().map_err(|e| e.to_string())?;
249        Arc::new(database.service())
250    };
251
252    let query = Query::with_params(
253        "DELETE FROM sync_queue WHERE user_id = ?",
254        vec![QueryParam::String(user_id)],
255    );
256
257    db_service.execute(query).await.map_err(|e| e.to_string())?;
258    Ok(())
259}
260
261#[cfg(test)]
262mod tests {
263    use super::*;
264
265    #[test]
266    fn test_sync_queue_item_serialization() {
267        let item = SyncQueueItem {
268            id: 1,
269            user_id: "user-123".to_string(),
270            operation: "favorite".to_string(),
271            item_id: Some("item-456".to_string()),
272            payload: Some(r#"{"isFavorite": true}"#.to_string()),
273            status: "pending".to_string(),
274            retry_count: 0,
275            created_at: Some("2024-02-14T08:00:00Z".to_string()),
276            error_message: None,
277            item_name: None,
278        };
279
280        // Should serialize successfully
281        let json = serde_json::to_string(&item);
282        assert!(json.is_ok());
283
284        let serialized = json.unwrap();
285        assert!(serialized.contains("user-123"));
286        assert!(serialized.contains("favorite"));
287        assert!(serialized.contains("pending"));
288    }
289
290    #[test]
291    fn test_sync_queue_item_with_error() {
292        let item = SyncQueueItem {
293            id: 2,
294            user_id: "user-789".to_string(),
295            operation: "update_progress".to_string(),
296            item_id: Some("item-999".to_string()),
297            payload: None,
298            status: "failed".to_string(),
299            retry_count: 3,
300            created_at: Some("2024-02-14T07:00:00Z".to_string()),
301            error_message: Some("Connection timeout".to_string()),
302            item_name: None,
303        };
304
305        let json = serde_json::to_string(&item).unwrap();
306        assert!(json.contains("failed"));
307        assert!(json.contains("Connection timeout"));
308        assert!(json.contains("3")); // retry_count
309    }
310
311    #[test]
312    fn test_sync_queue_item_without_optional_fields() {
313        let item = SyncQueueItem {
314            id: 3,
315            user_id: "user-000".to_string(),
316            operation: "clear_progress".to_string(),
317            item_id: None,
318            payload: None,
319            status: "completed".to_string(),
320            retry_count: 0,
321            created_at: None,
322            error_message: None,
323            item_name: None,
324        };
325
326        let json = serde_json::to_string(&item).unwrap();
327        assert!(json.contains("completed"));
328        assert!(json.contains("null") || json.contains("\"itemId\":null"));
329    }
330
331    #[test]
332    fn test_sync_status_values() {
333        // Verify all expected status values
334        let valid_statuses = vec!["pending", "processing", "completed", "failed"];
335
336        for status in valid_statuses {
337            let item = SyncQueueItem {
338                id: 1,
339                user_id: "test".to_string(),
340                operation: "test".to_string(),
341                item_id: None,
342                payload: None,
343                status: status.to_string(),
344                retry_count: 0,
345                created_at: None,
346                error_message: None,
347                item_name: None,
348            };
349
350            let json = serde_json::to_string(&item).unwrap();
351            assert!(json.contains(status));
352        }
353    }
354
355    #[test]
356    fn test_query_param_generation() {
357        // Test QueryParam generation for sync operations
358        let user_id = "user-123".to_string();
359        let operation = "favorite".to_string();
360
361        let params: Vec<QueryParam> = vec![
362            QueryParam::String(user_id.clone()),
363            QueryParam::String(operation.clone()),
364            QueryParam::Null,
365            QueryParam::Null,
366        ];
367
368        assert_eq!(params.len(), 4);
369        assert!(matches!(params[0], QueryParam::String(_)));
370        assert!(matches!(params[1], QueryParam::String(_)));
371        assert!(matches!(params[2], QueryParam::Null));
372        assert!(matches!(params[3], QueryParam::Null));
373    }
374
375    #[test]
376    fn test_retry_count_increment() {
377        // Verify retry count management
378        let mut item = SyncQueueItem {
379            id: 1,
380            user_id: "user-123".to_string(),
381            operation: "favorite".to_string(),
382            item_id: None,
383            payload: None,
384            status: "pending".to_string(),
385            retry_count: 0,
386            created_at: None,
387            error_message: None,
388            item_name: None,
389        };
390
391        // Simulate retries
392        for i in 1..=5 {
393            item.retry_count = i;
394            item.status = if i < 3 { "pending" } else { "failed" }.to_string();
395
396            assert!(item.retry_count == i);
397        }
398    }
399
400    /// Each queued mutation must get back the id of its *own* row.
401    ///
402    /// The id used to come from a separate `last_insert_rowid()` call — a
403    /// second trip to the shared connection — so another insert landing in
404    /// between handed this mutation someone else's id, and marking it synced
405    /// later completed the wrong row.
406    ///
407    /// TRACES: UR-002, UR-017 | DR-014 | UT-014
408    #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
409    async fn concurrent_enqueues_each_get_their_own_row_id() {
410        let database = crate::storage::Database::open_in_memory().unwrap();
411        let service = Arc::new(database.service());
412        service
413            .execute(Query::new(
414                "INSERT INTO servers (id, name, url) VALUES ('s', 'S', 'http://s')",
415            ))
416            .await
417            .unwrap();
418        service
419            .execute(Query::new(
420                "INSERT INTO users (id, server_id, username) VALUES ('u', 's', 'u')",
421            ))
422            .await
423            .unwrap();
424
425        let tasks: Vec<_> = (0..200)
426            .map(|i| {
427                let service = Arc::clone(&service);
428                tokio::spawn(async move {
429                    let op = format!("op-{i}");
430                    let id = enqueue_mutation(&*service, "u".into(), op.clone(), None, None)
431                        .await
432                        .unwrap();
433                    (op, id)
434                })
435            })
436            .collect();
437
438        for task in tasks {
439            let (op, id) = task.await.unwrap();
440            let stored: String = service
441                .query_one(
442                    Query::with_params(
443                        "SELECT operation FROM sync_queue WHERE id = ?",
444                        vec![QueryParam::Int64(id)],
445                    ),
446                    |row| row.get(0),
447                )
448                .await
449                .unwrap();
450            assert_eq!(stored, op, "mutation {op} was handed row {id}");
451        }
452    }
453}
jellytau_lib/commands/sync.rs

jellytau_lib/commands/
sync.rs
