jellytau/src-tauri/src/commands/favorites.rs

//! Pushing favourite toggles made while the server was unreachable.
//!
//! Favouriting works offline: `storage_toggle_favorite` writes the local
//! `user_data` row and sets `pending_sync = 1`. Until DR-120 nothing ever
//! cleared that flag — the offline `mark_favorite`/`unmark_favorite` are no-ops
//! and `syncService.queueFavorite` had no callers — so an offline toggle was
//! silently lost.
//!
//! The drain lives in Rust, not the frontend, because it must run whether or
//! not any view is mounted; a drain started by a component dies with it.
//!
//! TRACES: UR-069 | DR-120 | UT-103

use std::sync::Arc;

use async_trait::async_trait;
use log::{debug, info, warn};
use tauri::{Emitter, Listener, Manager};

use crate::repository::types::RepoError;
use crate::repository::MediaRepository;
use crate::storage::db_service::{DatabaseService, Query, QueryParam, RusqliteService};

/// The subset of the repository the drain needs.
///
/// Narrow on purpose: a test double for `MediaRepository` would be forty
/// unimplemented methods, which is how a drain ends up untested.
#[async_trait]
pub trait FavoriteSink: Send + Sync {
    async fn push_favorite(&self, item_id: &str, is_favorite: bool) -> Result<(), RepoError>;
}

#[async_trait]
impl<T: MediaRepository + ?Sized> FavoriteSink for T {
    async fn push_favorite(&self, item_id: &str, is_favorite: bool) -> Result<(), RepoError> {
        if is_favorite {
            self.mark_favorite(item_id).await
        } else {
            self.unmark_favorite(item_id).await
        }
    }
}

/// A local favourite change still waiting to reach the server.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingFavorite {
    pub item_id: String,
    pub is_favorite: bool,
}

/// Read every favourite change this user has pending.
async fn read_pending(
    db: &Arc<RusqliteService>,
    user_id: &str,
) -> Result<Vec<PendingFavorite>, String> {
    db.query_many(
        Query::with_params(
            "SELECT item_id, is_favorite FROM user_data \
             WHERE user_id = ? AND pending_sync = 1 AND is_favorite IS NOT NULL",
            vec![QueryParam::String(user_id.to_string())],
        ),
        |row| {
            Ok(PendingFavorite {
                item_id: row.get::<_, String>(0)?,
                is_favorite: row.get::<_, Option<i32>>(1)?.unwrap_or(0) != 0,
            })
        },
    )
    .await
}

/// Push pending favourite changes to the server and clear their flags.
///
/// Returns the ids that reached the server, for the `favorites-changed` event.
/// A row whose push fails keeps `pending_sync = 1` and is retried on the next
/// reconnect rather than being dropped.
///
/// TRACES: UR-069 | DR-120 | UT-103
pub async fn drain_pending_favorites(
    db: &Arc<RusqliteService>,
    sink: &dyn FavoriteSink,
    user_id: &str,
) -> Result<Vec<String>, String> {
    let pending = read_pending(db, user_id).await?;
    if pending.is_empty() {
        return Ok(Vec::new());
    }

    info!(
        "[Favorites] Pushing {} favourite change(s) queued while offline",
        pending.len()
    );

    let mut pushed = Vec::new();
    for change in pending {
        match sink
            .push_favorite(&change.item_id, change.is_favorite)
            .await
        {
            Ok(()) => {
                let cleared = db
                    .execute(Query::with_params(
                        "UPDATE user_data SET pending_sync = 0, synced_at = ? \
                         WHERE user_id = ? AND item_id = ?",
                        vec![
                            QueryParam::String(chrono::Utc::now().to_rfc3339()),
                            QueryParam::String(user_id.to_string()),
                            QueryParam::String(change.item_id.clone()),
                        ],
                    ))
                    .await;

                match cleared {
                    Ok(_) => pushed.push(change.item_id),
                    // The server took it; failing to clear the flag only means
                    // we push it again next time, which is harmless.
                    Err(e) => warn!(
                        "[Favorites] Pushed {} but could not clear pending_sync: {}",
                        change.item_id, e
                    ),
                }
            }
            Err(e) => {
                // Still pending — retried on the next reconnect.
                debug!(
                    "[Favorites] Deferring {}, server rejected the push: {:?}",
                    change.item_id, e
                );
            }
        }
    }

    Ok(pushed)
}

/// Drain on every offline→online transition.
///
/// Hooks the `connectivity:reconnected` event the `ConnectivityMonitor`
/// already emits, rather than polling — reachability is derived from real
/// traffic (DR-055) and this just reacts to it.
///
/// TRACES: UR-069 | DR-120
pub fn spawn_favorites_drain(app: tauri::AppHandle) {
    let handle = app.clone();
    app.listen("connectivity:reconnected", move |_event| {
        let app = handle.clone();
        tauri::async_runtime::spawn(async move {
            if let Err(e) = run_drain(&app).await {
                warn!("[Favorites] Drain skipped: {}", e);
            }
        });
    });
}

async fn run_drain(app: &tauri::AppHandle) -> Result<(), String> {
    let db_service: Arc<RusqliteService> = {
        let db = app.state::<crate::commands::storage::DatabaseWrapper>();
        let database = db.0.lock().map_err(|e| e.to_string())?;
        Arc::new(database.service())
    };

    let (repo, user_id) = {
        let manager = app.state::<crate::commands::repository::RepositoryManagerWrapper>();
        let handles = manager.0.handles();
        let Some(handle) = handles.first() else {
            // Not signed in — nothing to push on behalf of.
            return Ok(());
        };
        let repo = manager.0.get(handle).ok_or("Repository not found")?;
        let user_id = repo.user_id().to_string();
        (repo, user_id)
    };

    let pushed = drain_pending_favorites(&db_service, repo.as_ref(), &user_id).await?;

    if !pushed.is_empty() {
        let event = crate::commands::repository::FavoritesChangedEvent { item_ids: pushed };
        if let Err(e) = app.emit(crate::commands::repository::FAVORITES_CHANGED_EVENT, &event) {
            warn!("[Favorites] Failed to emit change event: {}", e);
        }
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::utils::lock::MutexSafe;
    use rusqlite::Connection;
    use std::sync::Mutex;

    /// Records what the server was asked to do, and can be told to fail.
    struct RecordingSink {
        calls: Mutex<Vec<(String, bool)>>,
        fail_for: Option<String>,
    }

    impl RecordingSink {
        fn new() -> Self {
            Self {
                calls: Mutex::new(Vec::new()),
                fail_for: None,
            }
        }

        fn failing_for(item_id: &str) -> Self {
            Self {
                calls: Mutex::new(Vec::new()),
                fail_for: Some(item_id.to_string()),
            }
        }

        fn calls(&self) -> Vec<(String, bool)> {
            let mut calls = self.calls.lock_safe().clone();
            calls.sort();
            calls
        }
    }

    #[async_trait]
    impl FavoriteSink for RecordingSink {
        async fn push_favorite(&self, item_id: &str, is_favorite: bool) -> Result<(), RepoError> {
            if self.fail_for.as_deref() == Some(item_id) {
                return Err(RepoError::Offline);
            }
            self.calls
                .lock()
                .unwrap()
                .push((item_id.to_string(), is_favorite));
            Ok(())
        }
    }

    fn test_db() -> Arc<RusqliteService> {
        let conn = Connection::open_in_memory().unwrap();
        conn.execute_batch(
            r#"
            CREATE TABLE user_data (
                user_id TEXT NOT NULL,
                item_id TEXT NOT NULL,
                is_favorite INTEGER,
                synced_at TEXT,
                pending_sync INTEGER DEFAULT 0,
                PRIMARY KEY (user_id, item_id)
            );
            "#,
        )
        .unwrap();
        Arc::new(RusqliteService::new(Arc::new(Mutex::new(conn))))
    }

    async fn seed(db: &Arc<RusqliteService>, rows: &[(&str, &str, i32, i32)]) {
        for (user, item, fav, pending) in rows {
            db.execute(Query::with_params(
                "INSERT INTO user_data (user_id, item_id, is_favorite, pending_sync) \
                 VALUES (?, ?, ?, ?)",
                vec![
                    QueryParam::String(user.to_string()),
                    QueryParam::String(item.to_string()),
                    QueryParam::Int(*fav),
                    QueryParam::Int(*pending),
                ],
            ))
            .await
            .unwrap();
        }
    }

    async fn pending_flag(db: &Arc<RusqliteService>, item_id: &str) -> Option<i32> {
        db.query_optional(
            Query::with_params(
                "SELECT pending_sync FROM user_data WHERE item_id = ?",
                vec![QueryParam::String(item_id.to_string())],
            ),
            |row| row.get::<_, Option<i32>>(0),
        )
        .await
        .unwrap()
        .flatten()
    }

    /// UT-103 — the core of the bug: a favourite toggled while offline reaches
    /// the server on reconnect, and stops being pending.
    ///
    /// TRACES: UR-069 | DR-120 | UT-103
    #[tokio::test]
    async fn test_drain_pushes_pending_favorites_and_clears_the_flag() {
        let db = test_db();
        seed(
            &db,
            &[
                ("u1", "marked-offline", 1, 1),
                ("u1", "unmarked-offline", 0, 1),
                // Already synced — must not be pushed again.
                ("u1", "already-synced", 1, 0),
            ],
        )
        .await;

        let sink = RecordingSink::new();
        let pushed = drain_pending_favorites(&db, &sink, "u1").await.unwrap();

        assert_eq!(
            sink.calls(),
            vec![
                ("marked-offline".to_string(), true),
                ("unmarked-offline".to_string(), false),
            ],
            "both pending changes push, with their direction preserved"
        );
        assert_eq!(pushed.len(), 2);
        assert_eq!(pending_flag(&db, "marked-offline").await, Some(0));
        assert_eq!(pending_flag(&db, "unmarked-offline").await, Some(0));
    }

    /// A push that fails keeps its row pending, so the change is retried rather
    /// than dropped on the floor.
    ///
    /// TRACES: UR-069 | DR-120 | UT-103
    #[tokio::test]
    async fn test_drain_leaves_failed_pushes_pending() {
        let db = test_db();
        seed(&db, &[("u1", "ok", 1, 1), ("u1", "boom", 1, 1)]).await;

        let sink = RecordingSink::failing_for("boom");
        let pushed = drain_pending_favorites(&db, &sink, "u1").await.unwrap();

        assert_eq!(pushed, vec!["ok".to_string()]);
        assert_eq!(pending_flag(&db, "ok").await, Some(0));
        assert_eq!(
            pending_flag(&db, "boom").await,
            Some(1),
            "a failed push must stay queued for the next reconnect"
        );
    }

    /// Another user's queued changes are not pushed with this user's token.
    ///
    /// TRACES: UR-069 | DR-120 | UT-103
    #[tokio::test]
    async fn test_drain_only_touches_the_given_user() {
        let db = test_db();
        seed(&db, &[("u1", "mine", 1, 1), ("u2", "theirs", 1, 1)]).await;

        let sink = RecordingSink::new();
        let pushed = drain_pending_favorites(&db, &sink, "u1").await.unwrap();

        assert_eq!(pushed, vec!["mine".to_string()]);
        assert_eq!(pending_flag(&db, "theirs").await, Some(1));
    }

    /// Nothing pending means no server calls at all — a reconnect must not
    /// generate traffic just because it happened.
    ///
    /// TRACES: UR-069 | DR-120 | UT-103
    #[tokio::test]
    async fn test_drain_is_a_noop_when_nothing_is_pending() {
        let db = test_db();
        seed(&db, &[("u1", "synced", 1, 0)]).await;

        let sink = RecordingSink::new();
        let pushed = drain_pending_favorites(&db, &sink, "u1").await.unwrap();

        assert!(pushed.is_empty());
        assert!(sink.calls().is_empty());
    }
}