feat(offline): play downloaded video, and drain the offline sync queue (0.4.6)

Bundles this session's work plus the concurrent search/offline/player changes.
Every gate passes on the combined tree: 885 frontend tests, 610 Rust tests,
clippy clean, boundary clean, trace coverage 86%.

Offline video playback — four separate defects, each of which alone stopped it:

  DR-133  A completed download's file_path is already absolute (the worker
          rewrites it on completion), but the player rooted it a second time and
          handed the webview /data/user/0/app//data/user/0/app/videos/x.mp4.
  DR-134  The asset protocol was never enabled: no protocol-asset feature and no
          assetProtocol config, so convertFileSrc produced URLs nothing answered.
          Also silently defeated the cached-thumbnail path, which fails soft to
          the server copy and hid it whenever the server was reachable.
  DR-137  Tauri's asset protocol answers a range-less request by reading the
          whole file into memory, and only advertises Accept-Ranges from inside
          its range branch, so the first request never learns ranges exist.
          Chromium gave up with PIPELINE_ERROR_READ after ~31s. Local media is
          now served by a loopback HTTP server: bounded 4 MiB chunks streamed
          from the file handle, every response length-delimited, and a range-less
          request answered with one chunk rather than the file. Confined by a
          per-session token and to the app data directory, because loopback is
          shared between apps on Android.
  DR-138  Release builds set usesCleartextTraffic=false, so Android rejected the
          request to that server before any I/O. A network-security-config
          exempts 127.0.0.1 only; a remote server must still be HTTPS.

Downloads:

  DR-135  download_item never records media_type and the reconnect resolver read
          that NULL as 'audio', so a movie queued from a media card had its URL
          resolved by get_audio_stream_url and completed as an audio-only
          transcode. The item's own type now decides.
  DR-136  Rows already downloaded that way are requeued on reconnect, since
          prevention alone leaves them reading "downloaded" and still unplayable.

Known limitation: a download taken at `original` quality is a byte copy of the
source, so it can be any container. One such file is an AVI holding XVID, which
the webview cannot play in any case — the media server serves it correctly and
Chromium refuses it. That needs either a transcoded download preset or the
native ExoPlayer surface work, and is not addressed here.

Also fixes two ID collisions between concurrent work: DR-143 defined twice
(search vs offline gate) and UT-131 defined twice (Episode Focus hero vs channel
cap). The search requirement is now DR-147 and the channel-cap test UT-141, with
their code references and matrix rows updated.

src-tauri/src/commands/sync_drain.rs

@@ -0,0 +1,838 @@
+//! 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,
+    },
+    /// 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_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::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(())
+}
+
+/// 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");
+    }
+
+    /// 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());
+    }
+}