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    let query = Query::with_params(
53        "INSERT INTO sync_queue (user_id, operation, item_id, payload, status, created_at)
54         VALUES (?, ?, ?, ?, 'pending', CURRENT_TIMESTAMP)",
55        vec![
56            QueryParam::String(user_id),
57            QueryParam::String(operation),
58            item_id.map(QueryParam::String).unwrap_or(QueryParam::Null),
59            payload.map(QueryParam::String).unwrap_or(QueryParam::Null),
60        ],
61    );
62
63    db_service.execute(query).await.map_err(|e| e.to_string())?;
64    let id = db_service
65        .last_insert_rowid()
66        .await
67        .map_err(|e| e.to_string())?;
68
69    Ok(id)
70}
71
72/// Get all pending sync operations for a user
73#[tauri::command]
74#[specta::specta]
75pub async fn sync_get_pending(
76    db: State<'_, DatabaseWrapper>,
77    user_id: String,
78    limit: Option<i32>,
79) -> Result<Vec<SyncQueueItem>, String> {
80    let db_service = {
81        let database = db.0.lock().map_err(|e| e.to_string())?;
82        Arc::new(database.service())
83    };
84
85    // The `items` join names the queued item where the catalog has it; a row for
86    // an item that was never cached still lists, with a null name.
87    // `abandoned` rows (DR-131 gave up on them) are excluded here for the same
88    // reason they are excluded from the count — they are no longer waiting.
89    const SELECT: &str = "SELECT q.id, q.user_id, q.operation, q.item_id, q.payload, q.status,
90                                 COALESCE(q.retry_count, 0), q.created_at, q.error_message, i.name
91                          FROM sync_queue q
92                          LEFT JOIN items i ON i.id = q.item_id
93                          WHERE q.user_id = ? AND q.status IN ('pending', 'failed')
94                          ORDER BY q.created_at ASC, q.id ASC";
95
96    let sql = match limit {
97        Some(l) => format!("{} LIMIT {}", SELECT, l),
98        None => SELECT.to_string(),
99    };
100
101    let query = Query::with_params(sql, vec![QueryParam::String(user_id)]);
102
103    db_service
104        .query_many(query, |row| {
105            Ok(SyncQueueItem {
106                id: row.get(0)?,
107                user_id: row.get(1)?,
108                operation: row.get(2)?,
109                item_id: row.get(3)?,
110                payload: row.get(4)?,
111                status: row.get(5)?,
112                retry_count: row.get(6)?,
113                created_at: row.get(7)?,
114                error_message: row.get(8)?,
115                item_name: row.get(9)?,
116            })
117        })
118        .await
119        .map_err(|e| e.to_string())
120}
121
122/// Mark a sync operation as in progress
123#[tauri::command]
124#[specta::specta]
125pub async fn sync_mark_processing(db: State<'_, DatabaseWrapper>, id: i64) -> Result<(), String> {
126    let db_service = {
127        let database = db.0.lock().map_err(|e| e.to_string())?;
128        Arc::new(database.service())
129    };
130
131    let query = Query::with_params(
132        "UPDATE sync_queue SET status = 'processing' WHERE id = ?",
133        vec![QueryParam::Int64(id)],
134    );
135
136    db_service.execute(query).await.map_err(|e| e.to_string())?;
137    Ok(())
138}
139
140/// Mark a sync operation as completed
141#[tauri::command]
142#[specta::specta]
143pub async fn sync_mark_completed(db: State<'_, DatabaseWrapper>, id: i64) -> Result<(), String> {
144    let db_service = {
145        let database = db.0.lock().map_err(|e| e.to_string())?;
146        Arc::new(database.service())
147    };
148
149    let query = Query::with_params(
150        "UPDATE sync_queue SET status = 'completed', processed_at = CURRENT_TIMESTAMP WHERE id = ?",
151        vec![QueryParam::Int64(id)],
152    );
153
154    db_service.execute(query).await.map_err(|e| e.to_string())?;
155    Ok(())
156}
157
158/// Mark a sync operation as failed with error message
159#[tauri::command]
160#[specta::specta]
161pub async fn sync_mark_failed(
162    db: State<'_, DatabaseWrapper>,
163    id: i64,
164    error: String,
165) -> Result<(), String> {
166    let db_service = {
167        let database = db.0.lock().map_err(|e| e.to_string())?;
168        Arc::new(database.service())
169    };
170
171    let query = Query::with_params(
172        "UPDATE sync_queue
173         SET status = 'failed',
174             retry_count = retry_count + 1,
175             error_message = ?,
176             processed_at = CURRENT_TIMESTAMP
177         WHERE id = ?",
178        vec![QueryParam::String(error), QueryParam::Int64(id)],
179    );
180
181    db_service.execute(query).await.map_err(|e| e.to_string())?;
182    Ok(())
183}
184
185/// Get count of pending sync operations for a user
186#[tauri::command]
187#[specta::specta]
188pub async fn sync_get_pending_count(
189    db: State<'_, DatabaseWrapper>,
190    user_id: String,
191) -> Result<i32, String> {
192    let db_service = {
193        let database = db.0.lock().map_err(|e| e.to_string())?;
194        Arc::new(database.service())
195    };
196
197    let query = Query::with_params(
198        "SELECT COUNT(*) FROM sync_queue WHERE user_id = ? AND status IN ('pending', 'failed')",
199        vec![QueryParam::String(user_id)],
200    );
201
202    db_service
203        .query_one(query, |row| row.get(0))
204        .await
205        .map_err(|e| e.to_string())
206}
207
208/// Delete completed sync operations older than specified days
209#[tauri::command]
210#[specta::specta]
211pub async fn sync_cleanup_completed(
212    db: State<'_, DatabaseWrapper>,
213    days_old: i32,
214) -> Result<i32, String> {
215    let db_service = {
216        let database = db.0.lock().map_err(|e| e.to_string())?;
217        Arc::new(database.service())
218    };
219
220    let query = Query::with_params(
221        "DELETE FROM sync_queue
222         WHERE status = 'completed'
223         AND processed_at < datetime('now', ?)",
224        vec![QueryParam::String(format!("-{} days", days_old))],
225    );
226
227    let deleted = db_service.execute(query).await.map_err(|e| e.to_string())?;
228    Ok(deleted as i32)
229}
230
231/// Delete all sync operations for a user (used during logout)
232#[tauri::command]
233#[specta::specta]
234pub async fn sync_clear_user(
235    db: State<'_, DatabaseWrapper>,
236    user_id: String,
237) -> Result<(), String> {
238    let db_service = {
239        let database = db.0.lock().map_err(|e| e.to_string())?;
240        Arc::new(database.service())
241    };
242
243    let query = Query::with_params(
244        "DELETE FROM sync_queue WHERE user_id = ?",
245        vec![QueryParam::String(user_id)],
246    );
247
248    db_service.execute(query).await.map_err(|e| e.to_string())?;
249    Ok(())
250}
251
252#[cfg(test)]
253mod tests {
254    use super::*;
255
256    #[test]
257    fn test_sync_queue_item_serialization() {
258        let item = SyncQueueItem {
259            id: 1,
260            user_id: "user-123".to_string(),
261            operation: "favorite".to_string(),
262            item_id: Some("item-456".to_string()),
263            payload: Some(r#"{"isFavorite": true}"#.to_string()),
264            status: "pending".to_string(),
265            retry_count: 0,
266            created_at: Some("2024-02-14T08:00:00Z".to_string()),
267            error_message: None,
268            item_name: None,
269        };
270
271        // Should serialize successfully
272        let json = serde_json::to_string(&item);
273        assert!(json.is_ok());
274
275        let serialized = json.unwrap();
276        assert!(serialized.contains("user-123"));
277        assert!(serialized.contains("favorite"));
278        assert!(serialized.contains("pending"));
279    }
280
281    #[test]
282    fn test_sync_queue_item_with_error() {
283        let item = SyncQueueItem {
284            id: 2,
285            user_id: "user-789".to_string(),
286            operation: "update_progress".to_string(),
287            item_id: Some("item-999".to_string()),
288            payload: None,
289            status: "failed".to_string(),
290            retry_count: 3,
291            created_at: Some("2024-02-14T07:00:00Z".to_string()),
292            error_message: Some("Connection timeout".to_string()),
293            item_name: None,
294        };
295
296        let json = serde_json::to_string(&item).unwrap();
297        assert!(json.contains("failed"));
298        assert!(json.contains("Connection timeout"));
299        assert!(json.contains("3")); // retry_count
300    }
301
302    #[test]
303    fn test_sync_queue_item_without_optional_fields() {
304        let item = SyncQueueItem {
305            id: 3,
306            user_id: "user-000".to_string(),
307            operation: "clear_progress".to_string(),
308            item_id: None,
309            payload: None,
310            status: "completed".to_string(),
311            retry_count: 0,
312            created_at: None,
313            error_message: None,
314            item_name: None,
315        };
316
317        let json = serde_json::to_string(&item).unwrap();
318        assert!(json.contains("completed"));
319        assert!(json.contains("null") || json.contains("\"itemId\":null"));
320    }
321
322    #[test]
323    fn test_sync_status_values() {
324        // Verify all expected status values
325        let valid_statuses = vec!["pending", "processing", "completed", "failed"];
326
327        for status in valid_statuses {
328            let item = SyncQueueItem {
329                id: 1,
330                user_id: "test".to_string(),
331                operation: "test".to_string(),
332                item_id: None,
333                payload: None,
334                status: status.to_string(),
335                retry_count: 0,
336                created_at: None,
337                error_message: None,
338                item_name: None,
339            };
340
341            let json = serde_json::to_string(&item).unwrap();
342            assert!(json.contains(status));
343        }
344    }
345
346    #[test]
347    fn test_query_param_generation() {
348        // Test QueryParam generation for sync operations
349        let user_id = "user-123".to_string();
350        let operation = "favorite".to_string();
351
352        let params: Vec<QueryParam> = vec![
353            QueryParam::String(user_id.clone()),
354            QueryParam::String(operation.clone()),
355            QueryParam::Null,
356            QueryParam::Null,
357        ];
358
359        assert_eq!(params.len(), 4);
360        assert!(matches!(params[0], QueryParam::String(_)));
361        assert!(matches!(params[1], QueryParam::String(_)));
362        assert!(matches!(params[2], QueryParam::Null));
363        assert!(matches!(params[3], QueryParam::Null));
364    }
365
366    #[test]
367    fn test_retry_count_increment() {
368        // Verify retry count management
369        let mut item = SyncQueueItem {
370            id: 1,
371            user_id: "user-123".to_string(),
372            operation: "favorite".to_string(),
373            item_id: None,
374            payload: None,
375            status: "pending".to_string(),
376            retry_count: 0,
377            created_at: None,
378            error_message: None,
379            item_name: None,
380        };
381
382        // Simulate retries
383        for i in 1..=5 {
384            item.retry_count = i;
385            item.status = if i < 3 { "pending" } else { "failed" }.to_string();
386
387            assert!(item.retry_count == i);
388        }
389    }
390}
jellytau_lib/commands/sync.rs

jellytau_lib/commands/
sync.rs
