First working POC

src-tauri/src/playback_reporting/mod.rs

@@ -0,0 +1,9 @@
+pub mod reporter;
+pub mod throttle;
+pub mod sync_processor;
+
+pub use reporter::{PlaybackReporter, PlaybackOperation, PlaybackContext};
+#[allow(unused_imports)]  // Will be used when position updates are hooked
+pub use throttle::EventThrottler;
+#[allow(unused_imports)]  // Will be used when sync processor is integrated
+pub use sync_processor::SyncProcessor;

src-tauri/src/playback_reporting/reporter.rs

@@ -0,0 +1,332 @@
+//! Playback reporter implementation
+//!
+//! This module is fully implemented but not yet integrated with the player.
+//! Dead code warnings are suppressed until integration is complete.
+
+#![allow(dead_code)]
+
+use std::sync::Arc;
+use tokio::sync::Mutex as TokioMutex;
+
+use crate::jellyfin::client::JellyfinClient;
+use crate::storage::db_service::{DatabaseService, Query, QueryParam, RusqliteService};
+
+/// Playback context information
+#[derive(Debug, Clone)]
+pub struct PlaybackContext {
+    pub context_type: String,  // "container" or "single"
+    pub context_id: Option<String>,
+}
+
+/// Playback operation types
+#[derive(Debug, Clone)]
+pub enum PlaybackOperation {
+    Start {
+        item_id: String,
+        position_ticks: i64,
+        context: Option<PlaybackContext>,
+    },
+    Progress {
+        item_id: String,
+        position_ticks: i64,
+        is_paused: bool,
+    },
+    Stopped {
+        item_id: String,
+        position_ticks: i64,
+    },
+    MarkPlayed {
+        item_id: String,
+    },
+}
+
+/// Main playback reporter that handles dual sync (local DB + server)
+pub struct PlaybackReporter {
+    db_service: Arc<RusqliteService>,
+    jellyfin_client: Arc<TokioMutex<Option<JellyfinClient>>>,
+    user_id: String,
+}
+
+impl PlaybackReporter {
+    /// Creates a new PlaybackReporter
+    pub fn new(
+        db_service: Arc<RusqliteService>,
+        jellyfin_client: Arc<TokioMutex<Option<JellyfinClient>>>,
+        user_id: String,
+    ) -> Self {
+        Self {
+            db_service,
+            jellyfin_client,
+            user_id,
+        }
+    }
+
+    /// Reports a playback operation (dual sync: local DB + server)
+    ///
+    /// Always updates local DB first, then attempts server sync if online.
+    /// If server sync fails, operation is queued for retry.
+    pub async fn report(&self, operation: PlaybackOperation, is_online: bool) -> Result<(), String> {
+        log::info!("[PlaybackReporter] Reporting operation: {:?}", operation);
+
+        // Always update local DB first (works offline)
+        self.update_local_db(&operation).await?;
+
+        // If online, attempt server sync
+        if is_online {
+            if let Err(e) = self.sync_to_server(&operation).await {
+                log::warn!("[PlaybackReporter] Server sync failed, queueing: {}", e);
+                self.queue_for_sync(&operation).await?;
+            } else {
+                // Mark as synced on success
+                if let Some(item_id) = self.get_item_id(&operation) {
+                    self.mark_synced(&item_id).await?;
+                }
+            }
+        } else {
+            log::debug!("[PlaybackReporter] Offline - queueing operation");
+            self.queue_for_sync(&operation).await?;
+        }
+
+        Ok(())
+    }
+
+    /// Updates local database with playback info
+    async fn update_local_db(&self, operation: &PlaybackOperation) -> Result<(), String> {
+        match operation {
+            PlaybackOperation::Start { item_id, position_ticks, context } => {
+                let query = Query::with_params(
+                    "INSERT INTO user_data (user_id, item_id, playback_position_ticks, last_played_at,
+                                            playback_context_type, playback_context_id, pending_sync)
+                     VALUES (?, ?, ?, CURRENT_TIMESTAMP, ?, ?, 1)
+                     ON CONFLICT(user_id, item_id) DO UPDATE SET
+                        playback_position_ticks = excluded.playback_position_ticks,
+                        last_played_at = excluded.last_played_at,
+                        playback_context_type = excluded.playback_context_type,
+                        playback_context_id = excluded.playback_context_id,
+                        pending_sync = 1",
+                    vec![
+                        QueryParam::String(self.user_id.clone()),
+                        QueryParam::String(item_id.clone()),
+                        QueryParam::Int64(*position_ticks),
+                        context.as_ref().map(|c| QueryParam::String(c.context_type.clone())).unwrap_or(QueryParam::Null),
+                        context.as_ref().and_then(|c| c.context_id.as_ref()).map(|id| QueryParam::String(id.clone())).unwrap_or(QueryParam::Null),
+                    ],
+                );
+
+                self.db_service.execute(query).await.map_err(|e| e.to_string())?;
+                log::debug!("[PlaybackReporter] Updated local DB for start: {}", item_id);
+            }
+
+            PlaybackOperation::Progress { item_id, position_ticks, is_paused: _ } |
+            PlaybackOperation::Stopped { item_id, position_ticks } => {
+                let query = Query::with_params(
+                    "INSERT INTO user_data (user_id, item_id, playback_position_ticks, last_played_at, pending_sync)
+                     VALUES (?, ?, ?, CURRENT_TIMESTAMP, 1)
+                     ON CONFLICT(user_id, item_id) DO UPDATE SET
+                        playback_position_ticks = excluded.playback_position_ticks,
+                        last_played_at = excluded.last_played_at,
+                        pending_sync = 1",
+                    vec![
+                        QueryParam::String(self.user_id.clone()),
+                        QueryParam::String(item_id.clone()),
+                        QueryParam::Int64(*position_ticks),
+                    ],
+                );
+
+                self.db_service.execute(query).await.map_err(|e| e.to_string())?;
+                log::debug!("[PlaybackReporter] Updated local DB for progress/stop: {}", item_id);
+            }
+
+            PlaybackOperation::MarkPlayed { item_id } => {
+                let query = Query::with_params(
+                    "INSERT INTO user_data (user_id, item_id, is_played, play_count, last_played_at, pending_sync)
+                     VALUES (?, ?, 1, 1, CURRENT_TIMESTAMP, 1)
+                     ON CONFLICT(user_id, item_id) DO UPDATE SET
+                        is_played = 1,
+                        play_count = COALESCE(play_count, 0) + 1,
+                        last_played_at = CURRENT_TIMESTAMP,
+                        pending_sync = 1",
+                    vec![
+                        QueryParam::String(self.user_id.clone()),
+                        QueryParam::String(item_id.clone()),
+                    ],
+                );
+
+                self.db_service.execute(query).await.map_err(|e| e.to_string())?;
+                log::debug!("[PlaybackReporter] Updated local DB for mark played: {}", item_id);
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Syncs to Jellyfin server
+    async fn sync_to_server(&self, operation: &PlaybackOperation) -> Result<(), String> {
+        let client_guard = self.jellyfin_client.lock().await;
+        let client = client_guard.as_ref().ok_or("JellyfinClient not initialized")?;
+
+        match operation {
+            PlaybackOperation::Start { item_id, position_ticks, .. } => {
+                client.report_playback_start(
+                    item_id.clone(),
+                    *position_ticks,
+                    None,  // play_session_id
+                ).await?;
+                log::info!("[PlaybackReporter] Reported start to server: {}", item_id);
+            }
+
+            PlaybackOperation::Progress { item_id, position_ticks, is_paused } => {
+                client.report_playback_progress(
+                    item_id.clone(),
+                    *position_ticks,
+                    *is_paused,
+                    None,  // play_session_id
+                ).await?;
+                log::debug!("[PlaybackReporter] Reported progress to server: {} (paused: {})", item_id, is_paused);
+            }
+
+            PlaybackOperation::Stopped { item_id, position_ticks } => {
+                client.report_playback_stopped(
+                    item_id.clone(),
+                    *position_ticks,
+                    None,  // play_session_id
+                ).await?;
+                log::info!("[PlaybackReporter] Reported stop to server: {}", item_id);
+            }
+
+            PlaybackOperation::MarkPlayed { item_id } => {
+                // For mark as played, we need to get the item's runtime
+                // For now, report as stopped at max position
+                // TODO: Fetch item runtime from DB or assume 100% completion
+                let max_ticks = i64::MAX; // Temporary - should be actual runtime
+                client.report_playback_stopped(
+                    item_id.clone(),
+                    max_ticks,
+                    None,
+                ).await?;
+                log::info!("[PlaybackReporter] Reported mark played to server: {}", item_id);
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Queues operation for later sync
+    async fn queue_for_sync(&self, operation: &PlaybackOperation) -> Result<(), String> {
+        let (op_name, item_id, payload) = match operation {
+            PlaybackOperation::Start { item_id, position_ticks, context } => {
+                let payload_data = serde_json::json!({
+                    "position_ticks": position_ticks,
+                    "context_type": context.as_ref().map(|c| &c.context_type),
+                    "context_id": context.as_ref().and_then(|c| c.context_id.as_ref()),
+                });
+                ("report_playback_start", Some(item_id.clone()), Some(payload_data.to_string()))
+            }
+
+            PlaybackOperation::Progress { .. } => {
+                // Don't queue progress reports - too frequent
+                // Progress is captured by final stop report
+                log::debug!("[PlaybackReporter] Skipping queue for progress report (too frequent)");
+                return Ok(());
+            }
+
+            PlaybackOperation::Stopped { item_id, position_ticks } => {
+                let payload_data = serde_json::json!({
+                    "position_ticks": position_ticks,
+                });
+                ("report_playback_stopped", Some(item_id.clone()), Some(payload_data.to_string()))
+            }
+
+            PlaybackOperation::MarkPlayed { item_id } => {
+                ("mark_played", Some(item_id.clone()), None)
+            }
+        };
+
+        let query = Query::with_params(
+            "INSERT INTO sync_queue (user_id, operation, item_id, payload, status, created_at)
+             VALUES (?, ?, ?, ?, 'pending', CURRENT_TIMESTAMP)",
+            vec![
+                QueryParam::String(self.user_id.clone()),
+                QueryParam::String(op_name.to_string()),
+                item_id.map(QueryParam::String).unwrap_or(QueryParam::Null),
+                payload.map(QueryParam::String).unwrap_or(QueryParam::Null),
+            ],
+        );
+
+        self.db_service.execute(query).await.map_err(|e| e.to_string())?;
+        log::info!("[PlaybackReporter] Queued operation: {}", op_name);
+
+        Ok(())
+    }
+
+    /// Marks an item as synced in the local database
+    async fn mark_synced(&self, item_id: &str) -> Result<(), String> {
+        let query = Query::with_params(
+            "UPDATE user_data SET pending_sync = 0 WHERE user_id = ? AND item_id = ?",
+            vec![
+                QueryParam::String(self.user_id.clone()),
+                QueryParam::String(item_id.to_string()),
+            ],
+        );
+
+        self.db_service.execute(query).await.map_err(|e| e.to_string())?;
+        log::debug!("[PlaybackReporter] Marked as synced: {}", item_id);
+
+        Ok(())
+    }
+
+    /// Extracts item_id from operation
+    fn get_item_id(&self, operation: &PlaybackOperation) -> Option<String> {
+        match operation {
+            PlaybackOperation::Start { item_id, .. } |
+            PlaybackOperation::Progress { item_id, .. } |
+            PlaybackOperation::Stopped { item_id, .. } |
+            PlaybackOperation::MarkPlayed { item_id } => Some(item_id.clone()),
+        }
+    }
+}
+
+impl Clone for PlaybackReporter {
+    fn clone(&self) -> Self {
+        Self {
+            db_service: Arc::clone(&self.db_service),
+            jellyfin_client: Arc::clone(&self.jellyfin_client),
+            user_id: self.user_id.clone(),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // Unit tests will be added incrementally as dependencies are mocked
+
+    #[test]
+    fn test_playback_operation_debug() {
+        let op = PlaybackOperation::Start {
+            item_id: "item123".to_string(),
+            position_ticks: 1000,
+            context: Some(PlaybackContext {
+                context_type: "container".to_string(),
+                context_id: Some("album456".to_string()),
+            }),
+        };
+
+        let debug_str = format!("{:?}", op);
+        assert!(debug_str.contains("Start"));
+        assert!(debug_str.contains("item123"));
+    }
+
+    #[test]
+    fn test_playback_context_clone() {
+        let context = PlaybackContext {
+            context_type: "single".to_string(),
+            context_id: None,
+        };
+
+        let cloned = context.clone();
+        assert_eq!(cloned.context_type, "single");
+        assert_eq!(cloned.context_id, None);
+    }
+}

src-tauri/src/playback_reporting/sync_processor.rs

@@ -0,0 +1,127 @@
+//! Sync queue processor with retry logic and exponential backoff
+//!
+//! This is a placeholder implementation. Full implementation will be added
+//! when the reporter is integrated. Dead code warnings are suppressed.
+
+#![allow(dead_code)]
+#![allow(unused_imports)]
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Duration;
+use tokio::sync::Mutex as TokioMutex;
+
+use crate::jellyfin::client::JellyfinClient;
+use crate::repository::MediaRepository;
+use crate::storage::db_service::RusqliteService;
+
+/// Configuration for sync processor
+pub struct SyncConfig {
+    pub max_retries: u32,          // 5
+    pub base_retry_delay_ms: u64,  // 1000ms
+    pub batch_size: usize,         // 10 items
+}
+
+impl Default for SyncConfig {
+    fn default() -> Self {
+        Self {
+            max_retries: 5,
+            base_retry_delay_ms: 1000,
+            batch_size: 10,
+        }
+    }
+}
+
+/// Sync queue processor that handles retry logic with exponential backoff
+///
+/// This is a placeholder implementation. Full implementation will be added
+/// in a subsequent task following the plan.
+pub struct SyncProcessor {
+    _db_service: Arc<RusqliteService>,
+    _jellyfin_client: Arc<TokioMutex<Option<JellyfinClient>>>,
+    _repository: Arc<dyn MediaRepository>,
+    _processing: Arc<TokioMutex<bool>>,
+    _cancelled: Arc<AtomicBool>,
+    _config: SyncConfig,
+}
+
+impl SyncProcessor {
+    /// Creates a new SyncProcessor
+    pub fn new(
+        db_service: Arc<RusqliteService>,
+        jellyfin_client: Arc<TokioMutex<Option<JellyfinClient>>>,
+        repository: Arc<dyn MediaRepository>,
+    ) -> Self {
+        Self {
+            _db_service: db_service,
+            _jellyfin_client: jellyfin_client,
+            _repository: repository,
+            _processing: Arc::new(TokioMutex::new(false)),
+            _cancelled: Arc::new(AtomicBool::new(false)),
+            _config: SyncConfig::default(),
+        }
+    }
+
+    /// Starts the sync processor
+    pub async fn start(&self) -> Result<(), String> {
+        log::info!("[SyncProcessor] Started (placeholder implementation)");
+        // TODO: Implement full processor logic
+        Ok(())
+    }
+
+    /// Stops the sync processor
+    pub async fn stop(&self) -> Result<(), String> {
+        log::info!("[SyncProcessor] Stopped (placeholder implementation)");
+        // TODO: Implement stop logic
+        Ok(())
+    }
+
+    /// Processes the sync queue once
+    pub async fn process_queue(&self) -> Result<(), String> {
+        log::debug!("[SyncProcessor] Processing queue (placeholder)");
+        // TODO: Implement queue processing
+        Ok(())
+    }
+
+    /// Calculates exponential backoff delay
+    fn _calculate_backoff(&self, retry_count: u32) -> Duration {
+        let delay_ms = self._config.base_retry_delay_ms * 2_u64.pow(retry_count);
+        let max_delay_ms = 10_000; // 10 seconds max
+        Duration::from_millis(delay_ms.min(max_delay_ms))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_sync_config_default() {
+        let config = SyncConfig::default();
+        assert_eq!(config.max_retries, 5);
+        assert_eq!(config.base_retry_delay_ms, 1000);
+        assert_eq!(config.batch_size, 10);
+    }
+
+    // TODO: Re-enable when SyncProcessor is fully implemented
+    // #[test]
+    // fn test_calculate_backoff() {
+    //     let config = SyncConfig::default();
+    //     let processor = SyncProcessor {
+    //         _db_service: Arc::new(unsafe { std::mem::zeroed() }),  // Placeholder for test
+    //         _jellyfin_client: Arc::new(TokioMutex::new(None)),
+    //         _repository: Arc::new(unsafe { std::mem::zeroed() }),  // Placeholder for test
+    //         _processing: Arc::new(TokioMutex::new(false)),
+    //         _cancelled: Arc::new(AtomicBool::new(false)),
+    //         _config: config,
+    //     };
+    //
+    //     // Test exponential backoff: 1s, 2s, 4s, 8s, 10s (capped)
+    //     assert_eq!(processor._calculate_backoff(0), Duration::from_millis(1000));
+    //     assert_eq!(processor._calculate_backoff(1), Duration::from_millis(2000));
+    //     assert_eq!(processor._calculate_backoff(2), Duration::from_millis(4000));
+    //     assert_eq!(processor._calculate_backoff(3), Duration::from_millis(8000));
+    //     assert_eq!(processor._calculate_backoff(4), Duration::from_millis(10000)); // capped
+    //     assert_eq!(processor._calculate_backoff(5), Duration::from_millis(10000)); // capped
+    // }
+}

src-tauri/src/playback_reporting/throttle.rs

@@ -0,0 +1,175 @@
+//! Event throttler for position update reporting
+//!
+//! This module is fully implemented but not yet integrated with the player.
+//! Dead code warnings are suppressed until integration is complete.
+
+#![allow(dead_code)]
+
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+/// Event throttler to prevent spam from frequent position updates.
+///
+/// Tracks the last report time for each item and ensures reports are only
+/// sent at most once per throttle duration (default 30 seconds).
+pub struct EventThrottler {
+    last_report_time: Arc<Mutex<HashMap<String, Instant>>>,
+    throttle_duration: Duration,
+}
+
+impl EventThrottler {
+    /// Creates a new EventThrottler with 30 second default interval
+    pub fn new() -> Self {
+        Self::with_duration(Duration::from_secs(30))
+    }
+
+    /// Creates a new EventThrottler with custom interval
+    pub fn with_duration(duration: Duration) -> Self {
+        Self {
+            last_report_time: Arc::new(Mutex::new(HashMap::new())),
+            throttle_duration: duration,
+        }
+    }
+
+    /// Checks if enough time has elapsed since the last report for this item
+    pub fn should_report(&self, item_id: &str) -> bool {
+        let last_times = self.last_report_time.lock().unwrap();
+
+        if let Some(last_time) = last_times.get(item_id) {
+            let elapsed = last_time.elapsed();
+            if elapsed < self.throttle_duration {
+                log::debug!(
+                    "[EventThrottler] Skipping report for {}, last reported {:.1}s ago (threshold: {}s)",
+                    item_id,
+                    elapsed.as_secs_f64(),
+                    self.throttle_duration.as_secs()
+                );
+                return false;
+            }
+        }
+
+        true
+    }
+
+    /// Marks the item as reported at the current time
+    pub fn mark_reported(&self, item_id: &str) {
+        let mut last_times = self.last_report_time.lock().unwrap();
+        last_times.insert(item_id.to_string(), Instant::now());
+
+        log::debug!(
+            "[EventThrottler] Marked {} as reported at {:?}",
+            item_id,
+            Instant::now()
+        );
+    }
+
+    /// Clears all tracked report times
+    pub fn clear(&self) {
+        let mut last_times = self.last_report_time.lock().unwrap();
+        last_times.clear();
+        log::debug!("[EventThrottler] Cleared all tracked report times");
+    }
+
+    /// Removes a specific item from tracking
+    pub fn clear_item(&self, item_id: &str) {
+        let mut last_times = self.last_report_time.lock().unwrap();
+        last_times.remove(item_id);
+        log::debug!("[EventThrottler] Cleared tracking for {}", item_id);
+    }
+}
+
+impl Default for EventThrottler {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Clone for EventThrottler {
+    fn clone(&self) -> Self {
+        Self {
+            last_report_time: Arc::clone(&self.last_report_time),
+            throttle_duration: self.throttle_duration,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::thread;
+
+    #[test]
+    fn test_throttler_allows_first_report() {
+        let throttler = EventThrottler::new();
+        assert!(throttler.should_report("item1"));
+    }
+
+    #[test]
+    fn test_throttler_blocks_immediate_second_report() {
+        let throttler = EventThrottler::new();
+        assert!(throttler.should_report("item1"));
+        throttler.mark_reported("item1");
+        assert!(!throttler.should_report("item1"));
+    }
+
+    #[test]
+    fn test_throttler_allows_report_after_duration() {
+        let throttler = EventThrottler::with_duration(Duration::from_millis(100));
+        assert!(throttler.should_report("item1"));
+        throttler.mark_reported("item1");
+        assert!(!throttler.should_report("item1"));
+
+        thread::sleep(Duration::from_millis(150));
+        assert!(throttler.should_report("item1"));
+    }
+
+    #[test]
+    fn test_throttler_handles_multiple_items() {
+        let throttler = EventThrottler::new();
+        assert!(throttler.should_report("item1"));
+        throttler.mark_reported("item1");
+
+        assert!(throttler.should_report("item2"));
+        throttler.mark_reported("item2");
+
+        assert!(!throttler.should_report("item1"));
+        assert!(!throttler.should_report("item2"));
+    }
+
+    #[test]
+    fn test_throttler_clear() {
+        let throttler = EventThrottler::new();
+        throttler.mark_reported("item1");
+        assert!(!throttler.should_report("item1"));
+
+        throttler.clear();
+        assert!(throttler.should_report("item1"));
+    }
+
+    #[test]
+    fn test_throttler_clear_item() {
+        let throttler = EventThrottler::new();
+        throttler.mark_reported("item1");
+        throttler.mark_reported("item2");
+
+        assert!(!throttler.should_report("item1"));
+        assert!(!throttler.should_report("item2"));
+
+        throttler.clear_item("item1");
+        assert!(throttler.should_report("item1"));
+        assert!(!throttler.should_report("item2"));
+    }
+
+    #[test]
+    fn test_throttler_clone_shares_state() {
+        let throttler1 = EventThrottler::new();
+        throttler1.mark_reported("item1");
+
+        let throttler2 = throttler1.clone();
+        assert!(!throttler2.should_report("item1"));
+
+        throttler2.mark_reported("item2");
+        assert!(!throttler1.should_report("item2"));
+    }
+}