jellytau_lib/playback_reporting/

reporter.rs

//! 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);
    }
}