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

//! Draining the offline mutation queue (`sync_queue`) to the server.
//!
//! `sync_queue` had producers but no consumer: `PlaybackReporter::queue_for_sync`
//! inserts a row whenever a start/stop/mark-played cannot reach the server, and
//! nothing ever pushed one. `sync_mark_processing`/`_completed`/`_failed` were
//! registered commands with no callers, so the queue only grew — the offline
//! banner's "N pending" climbed forever and the watch positions those rows stood
//! for never reached Jellyfin.
//!
//! Same shape as the favourites drain (DR-120), and for the same reason: a drain
//! started by a component dies with it, so it lives in Rust and hangs off the
//! `connectivity:reconnected` transition the `ConnectivityMonitor` already emits.
//!
//! TRACES: UR-025, UR-002 | DR-131 | UT-122

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};

/// How many times a row may fail before it stops being retried.
///
/// A row that can never succeed (a deleted item, an operation this build does
/// not know how to push) must eventually leave the queue, or it re-creates the
/// bug this module fixes: a count that only ever goes up.
pub const MAX_SYNC_ATTEMPTS: i32 = 5;

/// Emitted after a drain so open views can re-read the queue instead of waiting
/// for the frontend's 10s poll.
pub const SYNC_QUEUE_CHANGED_EVENT: &str = "sync-queue-changed";

/// A queued mutation, resolved from its stored `operation` + JSON `payload`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum QueuedOp {
    PlaybackStart {
        item_id: String,
        position_ticks: i64,
    },
    /// Also where `update_progress` lands: replaying a mid-playback progress
    /// report long after the fact would tell the server we are still playing.
    /// What the row actually carries is a resume position, and "stopped at N"
    /// is how that reaches Jellyfin's `UserData`.
    PlaybackStopped {
        item_id: String,
        position_ticks: i64,
    },
    MarkPlayed {
        item_id: String,
    },
    /// The inverse, queued by the watched toggle. Pushes as `clear_watch_history`
    /// (Jellyfin's mark-unplayed), which also zeroes the resume position — so an
    /// item un-marked offline does not come back carrying a stale position.
    MarkUnplayed {
        item_id: String,
    },
    /// Legacy rows only — live favourite toggles drain via `user_data.pending_sync`
    /// (DR-120). Supported so a row written by an older build still lands.
    Favorite {
        item_id: String,
        is_favorite: bool,
    },
}

/// Turn a stored row into something pushable.
///
/// Payload keys differ by producer: the Rust reporter writes `position_ticks`,
/// while `syncService.queuePlaybackProgress` writes camelCase `positionMs`.
/// Both are accepted rather than normalised at the producer, because rows
/// already in users' databases were written by both.
///
/// TRACES: UR-025 | DR-131 | UT-122
pub fn parse_queued_op(
    operation: &str,
    item_id: Option<&str>,
    payload: Option<&str>,
) -> Result<QueuedOp, String> {
    let json: serde_json::Value = match payload {
        Some(raw) if !raw.trim().is_empty() => {
            serde_json::from_str(raw).map_err(|e| format!("Unreadable payload: {}", e))?
        }
        _ => serde_json::Value::Null,
    };

    let item_id = item_id
        .filter(|id| !id.is_empty())
        .ok_or_else(|| format!("Operation '{}' has no item id", operation))?
        .to_string();

    let ticks = || -> i64 {
        if let Some(t) = json.get("position_ticks").and_then(|v| v.as_i64()) {
            return t;
        }
        if let Some(ms) = json.get("positionMs").and_then(|v| v.as_i64()) {
            return ms * 10_000; // ms → Jellyfin ticks (100ns)
        }
        0
    };

    match operation {
        "report_playback_start" => Ok(QueuedOp::PlaybackStart {
            item_id,
            position_ticks: ticks(),
        }),
        "report_playback_stopped" | "update_progress" => Ok(QueuedOp::PlaybackStopped {
            item_id,
            position_ticks: ticks(),
        }),
        "mark_played" => Ok(QueuedOp::MarkPlayed { item_id }),
        "mark_unplayed" => Ok(QueuedOp::MarkUnplayed { item_id }),
        "mark_favorite" => Ok(QueuedOp::Favorite {
            item_id,
            is_favorite: true,
        }),
        "unmark_favorite" => Ok(QueuedOp::Favorite {
            item_id,
            is_favorite: false,
        }),
        other => Err(format!("Unsupported operation '{}'", other)),
    }
}

/// The slice of the repository the drain needs — narrow so it can be doubled in
/// a test without forty `unimplemented!()` methods.
#[async_trait]
pub trait SyncSink: Send + Sync {
    async fn push(&self, op: &QueuedOp) -> Result<(), RepoError>;
}

#[async_trait]
impl<T: MediaRepository + ?Sized> SyncSink for T {
    async fn push(&self, op: &QueuedOp) -> Result<(), RepoError> {
        match op {
            QueuedOp::PlaybackStart {
                item_id,
                position_ticks,
            } => self.report_playback_start(item_id, *position_ticks).await,
            QueuedOp::PlaybackStopped {
                item_id,
                position_ticks,
            } => self.report_playback_stopped(item_id, *position_ticks).await,
            QueuedOp::MarkPlayed { item_id } => self.mark_played(item_id).await,
            QueuedOp::MarkUnplayed { item_id } => self.clear_watch_history(item_id).await,
            QueuedOp::Favorite {
                item_id,
                is_favorite,
            } => {
                if *is_favorite {
                    self.mark_favorite(item_id).await
                } else {
                    self.unmark_favorite(item_id).await
                }
            }
        }
    }
}

/// What a drain did, for logging and for the frontend's "Sync now" button.
#[derive(
    Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize, specta::Type,
)]
#[serde(rename_all = "camelCase")]
pub struct DrainReport {
    /// Rows that reached the server and are now `completed`.
    pub pushed: i32,
    /// Rows that failed and will be retried on the next reconnect.
    pub deferred: i32,
    /// Rows that exhausted `MAX_SYNC_ATTEMPTS` and were given up on.
    pub abandoned: i32,
    /// Rows still waiting afterwards (what the badge counts).
    pub remaining: i32,
}

/// Why a push failed, and whether the row should be charged an attempt for it.
struct PushFailure {
    reason: String,
    /// The server could not be reached at all — retry later, free of charge.
    transient: bool,
}

#[derive(Debug, Clone)]
struct QueuedRow {
    id: i64,
    operation: String,
    item_id: Option<String>,
    payload: Option<String>,
    retry_count: i32,
}

async fn read_queue(db: &Arc<RusqliteService>, user_id: &str) -> Result<Vec<QueuedRow>, String> {
    db.query_many(
        Query::with_params(
            "SELECT id, operation, item_id, payload, COALESCE(retry_count, 0) \
             FROM sync_queue \
             WHERE user_id = ? AND status IN ('pending', 'failed') \
             ORDER BY created_at ASC, id ASC",
            vec![QueryParam::String(user_id.to_string())],
        ),
        |row| {
            Ok(QueuedRow {
                id: row.get(0)?,
                operation: row.get(1)?,
                item_id: row.get(2)?,
                payload: row.get(3)?,
                retry_count: row.get(4)?,
            })
        },
    )
    .await
}

async fn remaining_count(db: &Arc<RusqliteService>, user_id: &str) -> Result<i32, String> {
    db.query_one(
        Query::with_params(
            "SELECT COUNT(*) FROM sync_queue WHERE user_id = ? AND status IN ('pending', 'failed')",
            vec![QueryParam::String(user_id.to_string())],
        ),
        |row| row.get(0),
    )
    .await
}

/// Push every queued mutation for this user, oldest first.
///
/// Chronological order matters: a stale start replayed after a later stop would
/// otherwise move the server's resume position backwards.
///
/// A row that fails keeps its place and is retried on the next reconnect, until
/// `MAX_SYNC_ATTEMPTS` — after which it is abandoned, because a row nothing can
/// ever push is exactly what turned this queue into a counter that only grew.
///
/// TRACES: UR-025, UR-002 | DR-131 | UT-122
pub async fn drain_sync_queue(
    db: &Arc<RusqliteService>,
    sink: &dyn SyncSink,
    user_id: &str,
) -> Result<DrainReport, String> {
    let queued = read_queue(db, user_id).await?;
    if queued.is_empty() {
        return Ok(DrainReport::default());
    }

    info!(
        "[SyncQueue] Pushing {} operation(s) queued while offline",
        queued.len()
    );

    let mut report = DrainReport::default();

    for row in queued {
        let outcome = match parse_queued_op(
            &row.operation,
            row.item_id.as_deref(),
            row.payload.as_deref(),
        ) {
            Ok(op) => sink.push(&op).await.map_err(|e| PushFailure {
                // An unreachable server is not the row's fault: burning its
                // budget would abandon perfectly good rows just because the app
                // was opened offline a few times.
                transient: matches!(e, RepoError::Offline | RepoError::Network { .. }),
                reason: e.to_string(),
            }),
            // An unreadable or unsupported row can never succeed, so it does
            // burn attempts rather than being deleted outright — the panel shows
            // the reason until it is abandoned.
            Err(reason) => Err(PushFailure {
                reason,
                transient: false,
            }),
        };

        match outcome {
            Ok(()) => {
                mark_completed(db, row.id).await?;
                report.pushed += 1;
            }
            Err(failure) if failure.transient => {
                mark_deferred(db, row.id, &failure.reason).await?;
                debug!(
                    "[SyncQueue] Server unreachable, {} stays queued: {}",
                    row.operation, failure.reason
                );
                report.deferred += 1;
            }
            Err(failure) => {
                let attempts = row.retry_count + 1;
                let give_up = attempts >= MAX_SYNC_ATTEMPTS;
                mark_failed(db, row.id, attempts, give_up, &failure.reason).await?;
                if give_up {
                    warn!(
                        "[SyncQueue] Giving up on {} after {} attempts: {}",
                        row.operation, attempts, failure.reason
                    );
                    report.abandoned += 1;
                } else {
                    debug!(
                        "[SyncQueue] Deferring {} (attempt {}): {}",
                        row.operation, attempts, failure.reason
                    );
                    report.deferred += 1;
                }
            }
        }
    }

    report.remaining = remaining_count(db, user_id).await?;
    info!(
        "[SyncQueue] Drain finished: {} pushed, {} deferred, {} abandoned, {} remaining",
        report.pushed, report.deferred, report.abandoned, report.remaining
    );

    Ok(report)
}

async fn mark_completed(db: &Arc<RusqliteService>, id: i64) -> Result<(), String> {
    db.execute(Query::with_params(
        "UPDATE sync_queue \
         SET status = 'completed', processed_at = CURRENT_TIMESTAMP, error_message = NULL \
         WHERE id = ?",
        vec![QueryParam::Int64(id)],
    ))
    .await?;
    Ok(())
}

/// Put a row back in the queue untouched apart from its error note — used when
/// the server was simply unreachable.
async fn mark_deferred(db: &Arc<RusqliteService>, id: i64, reason: &str) -> Result<(), String> {
    db.execute(Query::with_params(
        "UPDATE sync_queue SET status = 'pending', error_message = ? WHERE id = ?",
        vec![
            QueryParam::String(reason.to_string()),
            QueryParam::Int64(id),
        ],
    ))
    .await?;
    Ok(())
}

async fn mark_failed(
    db: &Arc<RusqliteService>,
    id: i64,
    attempts: i32,
    give_up: bool,
    reason: &str,
) -> Result<(), String> {
    db.execute(Query::with_params(
        "UPDATE sync_queue \
         SET status = ?, retry_count = ?, error_message = ?, processed_at = CURRENT_TIMESTAMP \
         WHERE id = ?",
        vec![
            QueryParam::String(if give_up { "abandoned" } else { "failed" }.to_string()),
            QueryParam::Int(attempts),
            QueryParam::String(reason.to_string()),
            QueryParam::Int64(id),
        ],
    ))
    .await?;
    Ok(())
}

/// Queue a watch position that could not be reported to the server.
///
/// The stop-report path pushed straight to the server and, on failure, logged
/// and dropped the position — so closing a video while the server was
/// unreachable lost the resume point outright, even though the queue and its
/// drain (DR-131) were built and running. This is the missing producer.
///
/// The pending row for an item is *replaced* rather than appended to. Progress
/// is reported every 10s, so a server that stays down would otherwise grow one
/// row per tick, all of them superseded by the newest — the unbounded queue
/// DR-131 exists to prevent. Only `pending`/`failed` rows are superseded:
/// an `abandoned` row has been given up on and must not be revived, and a
/// `completed` one is history.
///
/// TRACES: UR-025 | DR-154 | UT-151
pub async fn enqueue_playback_stopped(
    db: &Arc<RusqliteService>,
    user_id: &str,
    item_id: &str,
    position_ticks: i64,
) -> Result<(), String> {
    let payload = format!(r#"{{"position_ticks": {}}}"#, position_ticks);

    // Supersede an already-queued position for this item, keeping its place in
    // the queue order (created_at) so a later item cannot overtake it.
    let updated = db
        .execute(Query::with_params(
            "UPDATE sync_queue \
             SET payload = ?, status = 'pending', error_message = NULL \
             WHERE user_id = ? AND item_id = ? AND operation = 'report_playback_stopped' \
               AND status IN ('pending', 'failed')",
            vec![
                QueryParam::String(payload.clone()),
                QueryParam::String(user_id.to_string()),
                QueryParam::String(item_id.to_string()),
            ],
        ))
        .await?;

    if updated == 0 {
        db.execute(Query::with_params(
            "INSERT INTO sync_queue (user_id, operation, item_id, payload, status, created_at) \
             VALUES (?, 'report_playback_stopped', ?, ?, 'pending', CURRENT_TIMESTAMP)",
            vec![
                QueryParam::String(user_id.to_string()),
                QueryParam::String(item_id.to_string()),
                QueryParam::String(payload),
            ],
        ))
        .await?;
    }

    debug!(
        "[SyncQueue] Queued unreported stop for {} at {} ticks",
        item_id, position_ticks
    );
    Ok(())
}

/// Drain on every offline→online transition.
///
/// TRACES: UR-025 | DR-131
pub fn spawn_sync_queue_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!("[SyncQueue] Drain skipped: {}", e);
            }
        });
    });
}

/// Resolve app state and drain. Shared by the reconnect hook and the manual
/// "Sync now" command.
pub async fn run_drain(app: &tauri::AppHandle) -> Result<DrainReport, 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(DrainReport::default());
        };
        let repo = manager.0.get(handle).ok_or("Repository not found")?;
        let user_id = repo.user_id().to_string();
        (repo, user_id)
    };

    let report = drain_sync_queue(&db_service, repo.as_ref(), &user_id).await?;

    if report.pushed > 0 || report.abandoned > 0 {
        if let Err(e) = app.emit(SYNC_QUEUE_CHANGED_EVENT, &report) {
            warn!("[SyncQueue] Failed to emit change event: {}", e);
        }
    }

    Ok(report)
}

/// Push the queue now, on the user's say-so, instead of waiting for a reconnect.
///
/// TRACES: UR-025 | DR-132
#[tauri::command]
#[specta::specta]
pub async fn sync_process_pending(app: tauri::AppHandle) -> Result<DrainReport, String> {
    run_drain(&app).await
}

#[cfg(test)]
mod tests {
    use super::*;
    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<QueuedOp>>,
        fail_with: Option<RepoError>,
    }

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

        /// The server is there and refuses the operation — the row's own fault.
        fn always_rejecting() -> Self {
            Self {
                calls: Mutex::new(Vec::new()),
                fail_with: Some(RepoError::Server {
                    message: "HTTP 400".to_string(),
                }),
            }
        }

        /// The server cannot be reached at all — nothing to do with the row.
        fn unreachable() -> Self {
            Self {
                calls: Mutex::new(Vec::new()),
                fail_with: Some(RepoError::Offline),
            }
        }

        fn calls(&self) -> Vec<QueuedOp> {
            self.calls.lock().unwrap().clone()
        }
    }

    #[async_trait]
    impl SyncSink for RecordingSink {
        async fn push(&self, op: &QueuedOp) -> Result<(), RepoError> {
            if let Some(err) = &self.fail_with {
                return Err(err.clone());
            }
            self.calls.lock().unwrap().push(op.clone());
            Ok(())
        }
    }

    fn test_db() -> Arc<RusqliteService> {
        let conn = Connection::open_in_memory().unwrap();
        conn.execute_batch(
            r#"
            CREATE TABLE sync_queue (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id TEXT NOT NULL,
                operation TEXT NOT NULL,
                item_id TEXT,
                payload TEXT,
                status TEXT DEFAULT 'pending',
                retry_count INTEGER DEFAULT 0,
                created_at TEXT DEFAULT CURRENT_TIMESTAMP,
                processed_at TEXT,
                error_message TEXT
            );
            "#,
        )
        .unwrap();
        Arc::new(RusqliteService::new(Arc::new(Mutex::new(conn))))
    }

    /// (user, operation, item_id, payload, status, retry_count, created_at)
    type Seed<'a> = (
        &'a str,
        &'a str,
        &'a str,
        Option<&'a str>,
        &'a str,
        i32,
        &'a str,
    );

    async fn seed(db: &Arc<RusqliteService>, rows: &[Seed<'_>]) {
        for (user, op, item, payload, status, retries, created) in rows {
            db.execute(Query::with_params(
                "INSERT INTO sync_queue (user_id, operation, item_id, payload, status, retry_count, created_at) \
                 VALUES (?, ?, ?, ?, ?, ?, ?)",
                vec![
                    QueryParam::String(user.to_string()),
                    QueryParam::String(op.to_string()),
                    QueryParam::String(item.to_string()),
                    payload
                        .map(|p| QueryParam::String(p.to_string()))
                        .unwrap_or(QueryParam::Null),
                    QueryParam::String(status.to_string()),
                    QueryParam::Int(*retries),
                    QueryParam::String(created.to_string()),
                ],
            ))
            .await
            .unwrap();
        }
    }

    async fn row_state(db: &Arc<RusqliteService>, item_id: &str) -> (String, i32) {
        db.query_one(
            Query::with_params(
                "SELECT status, COALESCE(retry_count, 0) FROM sync_queue WHERE item_id = ?",
                vec![QueryParam::String(item_id.to_string())],
            ),
            |row| Ok((row.get::<_, String>(0)?, row.get::<_, i32>(1)?)),
        )
        .await
        .unwrap()
    }

    /// UT-122 — the bug itself: rows queued while offline reach the server on
    /// reconnect and stop counting towards the offline banner's badge.
    ///
    /// TRACES: UR-025 | DR-131 | UT-122
    #[tokio::test]
    async fn test_drain_pushes_queued_operations_and_clears_the_queue() {
        let db = test_db();
        seed(
            &db,
            &[
                (
                    "u1",
                    "report_playback_start",
                    "ep1",
                    Some(r#"{"position_ticks": 100}"#),
                    "pending",
                    0,
                    "2026-08-01T10:00:00Z",
                ),
                (
                    "u1",
                    "report_playback_stopped",
                    "ep2",
                    Some(r#"{"position_ticks": 5000}"#),
                    "pending",
                    0,
                    "2026-08-01T10:01:00Z",
                ),
                (
                    "u1",
                    "mark_played",
                    "ep3",
                    None,
                    "pending",
                    0,
                    "2026-08-01T10:02:00Z",
                ),
            ],
        )
        .await;

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

        assert_eq!(
            sink.calls(),
            vec![
                QueuedOp::PlaybackStart {
                    item_id: "ep1".to_string(),
                    position_ticks: 100
                },
                QueuedOp::PlaybackStopped {
                    item_id: "ep2".to_string(),
                    position_ticks: 5000
                },
                QueuedOp::MarkPlayed {
                    item_id: "ep3".to_string()
                },
            ],
            "every queued operation pushes, oldest first"
        );
        assert_eq!(report.pushed, 3);
        assert_eq!(report.remaining, 0, "the badge must reach zero");
        assert_eq!(row_state(&db, "ep1").await.0, "completed");
    }

    /// A push that fails stays queued for the next reconnect rather than being
    /// dropped.
    ///
    /// TRACES: UR-025 | DR-131 | UT-122
    #[tokio::test]
    async fn test_drain_defers_failed_pushes() {
        let db = test_db();
        seed(
            &db,
            &[(
                "u1",
                "report_playback_stopped",
                "ep1",
                Some(r#"{"position_ticks": 42}"#),
                "pending",
                0,
                "2026-08-01T10:00:00Z",
            )],
        )
        .await;

        let report = drain_sync_queue(&db, &RecordingSink::always_rejecting(), "u1")
            .await
            .unwrap();

        assert_eq!(report.deferred, 1);
        assert_eq!(report.remaining, 1);
        assert_eq!(row_state(&db, "ep1").await, ("failed".to_string(), 1));
    }

    /// An unreachable server does not charge the row an attempt — otherwise
    /// opening the app offline a few times abandons perfectly good rows.
    ///
    /// TRACES: UR-025 | DR-131 | UT-122
    #[tokio::test]
    async fn test_unreachable_server_does_not_burn_the_retry_budget() {
        let db = test_db();
        seed(
            &db,
            &[(
                "u1",
                "mark_played",
                "ep1",
                None,
                "pending",
                MAX_SYNC_ATTEMPTS - 1,
                "2026-08-01T10:00:00Z",
            )],
        )
        .await;

        let report = drain_sync_queue(&db, &RecordingSink::unreachable(), "u1")
            .await
            .unwrap();

        assert_eq!(report.deferred, 1);
        assert_eq!(report.abandoned, 0);
        assert_eq!(
            row_state(&db, "ep1").await,
            ("pending".to_string(), MAX_SYNC_ATTEMPTS - 1),
            "still queued, with its budget intact"
        );
    }

    /// A row that can never succeed must eventually leave the queue, or the
    /// count climbs forever — which is the bug this module exists to fix.
    ///
    /// TRACES: UR-025 | DR-131 | UT-122
    #[tokio::test]
    async fn test_drain_abandons_a_row_after_max_attempts() {
        let db = test_db();
        seed(
            &db,
            &[(
                "u1",
                "report_playback_stopped",
                "doomed",
                Some(r#"{"position_ticks": 1}"#),
                "failed",
                MAX_SYNC_ATTEMPTS - 1,
                "2026-08-01T10:00:00Z",
            )],
        )
        .await;

        let report = drain_sync_queue(&db, &RecordingSink::always_rejecting(), "u1")
            .await
            .unwrap();

        assert_eq!(report.abandoned, 1);
        assert_eq!(report.remaining, 0, "an abandoned row stops being counted");
        assert_eq!(row_state(&db, "doomed").await.0, "abandoned");
    }

    /// An operation this build cannot push does not wedge the queue behind it.
    ///
    /// TRACES: UR-025 | DR-131 | UT-122
    #[tokio::test]
    async fn test_unsupported_operation_records_a_reason_and_lets_others_through() {
        let db = test_db();
        seed(
            &db,
            &[
                (
                    "u1",
                    "playlist_reorder_item",
                    "pl1",
                    None,
                    "pending",
                    0,
                    "2026-08-01T10:00:00Z",
                ),
                (
                    "u1",
                    "mark_played",
                    "ep1",
                    None,
                    "pending",
                    0,
                    "2026-08-01T10:01:00Z",
                ),
            ],
        )
        .await;

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

        assert_eq!(
            sink.calls(),
            vec![QueuedOp::MarkPlayed {
                item_id: "ep1".to_string()
            }],
            "the unsupported row must not block the ones behind it"
        );
        assert_eq!(report.pushed, 1);
        assert_eq!(report.deferred, 1);
        assert_eq!(row_state(&db, "pl1").await, ("failed".to_string(), 1));
    }

    /// Another user's queued changes are not pushed with this user's token.
    ///
    /// TRACES: UR-025 | DR-131 | UT-122
    #[tokio::test]
    async fn test_drain_only_touches_the_given_user() {
        let db = test_db();
        seed(
            &db,
            &[
                (
                    "u1",
                    "mark_played",
                    "mine",
                    None,
                    "pending",
                    0,
                    "2026-08-01T10:00:00Z",
                ),
                (
                    "u2",
                    "mark_played",
                    "theirs",
                    None,
                    "pending",
                    0,
                    "2026-08-01T10:00:00Z",
                ),
            ],
        )
        .await;

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

        assert_eq!(
            sink.calls(),
            vec![QueuedOp::MarkPlayed {
                item_id: "mine".to_string()
            }]
        );
        assert_eq!(row_state(&db, "theirs").await.0, "pending");
    }

    /// The bug DR-154 fixes: a stop-report that could not reach the server was
    /// logged and dropped, so the watch position was lost outright. It must
    /// land in the queue the drain already knows how to push.
    ///
    /// TRACES: UR-025 | DR-154 | UT-151
    #[tokio::test]
    async fn test_failed_stop_report_is_queued_rather_than_dropped() {
        let db = test_db();

        enqueue_playback_stopped(&db, "u1", "ep1", 5_000_000_000)
            .await
            .unwrap();

        // The very drain that already exists must be able to push it.
        let sink = RecordingSink::new();
        let report = drain_sync_queue(&db, &sink, "u1").await.unwrap();

        assert_eq!(
            sink.calls(),
            vec![QueuedOp::PlaybackStopped {
                item_id: "ep1".to_string(),
                position_ticks: 5_000_000_000,
            }]
        );
        assert_eq!(report.pushed, 1);
        assert_eq!(report.remaining, 0);
    }

    /// Progress is reported every 10s, and a server that stays unreachable
    /// would otherwise add a row per tick — an unbounded queue of positions
    /// that are all superseded by the newest one. The pending row for an item
    /// is replaced in place, so the queue holds the latest position only.
    ///
    /// TRACES: UR-025 | DR-154 | UT-151
    #[tokio::test]
    async fn test_requeueing_the_same_item_supersedes_the_earlier_position() {
        let db = test_db();

        enqueue_playback_stopped(&db, "u1", "ep1", 1_000)
            .await
            .unwrap();
        enqueue_playback_stopped(&db, "u1", "ep1", 2_000)
            .await
            .unwrap();
        enqueue_playback_stopped(&db, "u1", "ep1", 3_000)
            .await
            .unwrap();

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

        // One row, carrying the newest position — not three.
        assert_eq!(
            sink.calls(),
            vec![QueuedOp::PlaybackStopped {
                item_id: "ep1".to_string(),
                position_ticks: 3_000,
            }]
        );
    }

    /// Distinct items must not collide — superseding is per item, not global.
    ///
    /// TRACES: UR-025 | DR-154 | UT-151
    #[tokio::test]
    async fn test_requeueing_keeps_positions_for_different_items_apart() {
        let db = test_db();

        enqueue_playback_stopped(&db, "u1", "ep1", 1_000)
            .await
            .unwrap();
        enqueue_playback_stopped(&db, "u1", "ep2", 2_000)
            .await
            .unwrap();

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

        let mut calls = sink.calls();
        calls.sort_by_key(|op| match op {
            QueuedOp::PlaybackStopped { item_id, .. } => item_id.clone(),
            _ => String::new(),
        });
        assert_eq!(
            calls,
            vec![
                QueuedOp::PlaybackStopped {
                    item_id: "ep1".to_string(),
                    position_ticks: 1_000,
                },
                QueuedOp::PlaybackStopped {
                    item_id: "ep2".to_string(),
                    position_ticks: 2_000,
                },
            ]
        );
    }

    /// A row already abandoned (DR-131 gave up on it) must not be resurrected
    /// by a later report — that would restore the queue-that-only-grows this
    /// whole area exists to prevent. The new report is queued as its own row.
    ///
    /// TRACES: UR-025 | DR-154 | UT-151
    #[tokio::test]
    async fn test_requeueing_does_not_revive_an_abandoned_row() {
        let db = test_db();
        seed(
            &db,
            &[(
                "u1",
                "report_playback_stopped",
                "ep1",
                Some(r#"{"position_ticks": 111}"#),
                "abandoned",
                MAX_SYNC_ATTEMPTS,
                "2026-08-01T10:00:00Z",
            )],
        )
        .await;

        enqueue_playback_stopped(&db, "u1", "ep1", 999)
            .await
            .unwrap();

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

        // Only the fresh row is pushed; the abandoned one stays abandoned.
        assert_eq!(
            sink.calls(),
            vec![QueuedOp::PlaybackStopped {
                item_id: "ep1".to_string(),
                position_ticks: 999,
            }]
        );
    }

    /// Nothing queued means no server calls at all — a reconnect must not
    /// generate traffic just because it happened.
    ///
    /// TRACES: UR-025 | DR-131 | UT-122
    #[tokio::test]
    async fn test_drain_is_a_noop_when_the_queue_is_empty() {
        let db = test_db();
        let sink = RecordingSink::new();

        let report = drain_sync_queue(&db, &sink, "u1").await.unwrap();

        assert_eq!(report, DrainReport::default());
        assert!(sink.calls().is_empty());
    }

    /// Both payload dialects parse: `position_ticks` from the Rust reporter and
    /// camelCase `positionMs` from the frontend's queue helper.
    ///
    /// TRACES: UR-025 | DR-131 | UT-122
    #[test]
    fn test_parse_accepts_both_payload_dialects() {
        assert_eq!(
            parse_queued_op(
                "report_playback_stopped",
                Some("ep1"),
                Some(r#"{"position_ticks": 1234}"#)
            )
            .unwrap(),
            QueuedOp::PlaybackStopped {
                item_id: "ep1".to_string(),
                position_ticks: 1234
            }
        );

        assert_eq!(
            parse_queued_op("update_progress", Some("ep1"), Some(r#"{"positionMs": 5}"#)).unwrap(),
            QueuedOp::PlaybackStopped {
                item_id: "ep1".to_string(),
                position_ticks: 50_000
            },
            "milliseconds convert to ticks"
        );

        assert_eq!(
            parse_queued_op("mark_played", Some("ep1"), None).unwrap(),
            QueuedOp::MarkPlayed {
                item_id: "ep1".to_string()
            },
            "a payload-less operation is not an error"
        );

        assert!(parse_queued_op("mark_played", None, None).is_err());
        assert!(parse_queued_op("teleport", Some("ep1"), None).is_err());
    }

    /// Un-marking watched queues like marking watched does, so the toggle works
    /// in both directions while the server is unreachable rather than only one.
    ///
    /// TRACES: UR-073 | DR-158 | UT-154
    #[test]
    fn test_parse_accepts_mark_unplayed() {
        assert_eq!(
            parse_queued_op("mark_unplayed", Some("ep1"), None).unwrap(),
            QueuedOp::MarkUnplayed {
                item_id: "ep1".to_string()
            },
        );

        assert!(parse_queued_op("mark_unplayed", None, None).is_err());
    }

    /// The queued un-mark reaches the server as `clear_watch_history` — Jellyfin's
    /// mark-unplayed, which also zeroes the resume position, so a series returns
    /// to "never watched" rather than keeping a stale position.
    ///
    /// TRACES: UR-073 | DR-158 | UT-154
    #[tokio::test]
    async fn test_drain_pushes_mark_unplayed() {
        let db = test_db();
        seed(
            &db,
            &[(
                "u1",
                "mark_unplayed",
                "ep9",
                None,
                "pending",
                0,
                "2026-08-01T10:00:00Z",
            )],
        )
        .await;

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

        assert_eq!(
            sink.calls(),
            vec![QueuedOp::MarkUnplayed {
                item_id: "ep9".to_string()
            }],
        );
        assert_eq!(report.pushed, 1);
        assert_eq!(report.remaining, 0);
    }
}