oximedia_proxy/

transcode_queue.rs

//! Proxy transcode queue management.
//!
//! Provides a priority-based queue for scheduling and tracking proxy transcode
//! jobs, along with batch request support and queue statistics.

#![allow(dead_code)]

use serde::{Deserialize, Serialize};
use std::collections::VecDeque;

/// Specification for a proxy transcode output.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProxySpec {
    /// Target resolution (width, height).
    pub resolution: (u32, u32),
    /// Codec identifier (e.g. "h264", "prores_proxy").
    pub codec: String,
    /// Target bitrate in kilobits per second.
    pub bitrate_kbps: u32,
}

impl ProxySpec {
    /// Create a new proxy spec.
    #[must_use]
    pub fn new(resolution: (u32, u32), codec: impl Into<String>, bitrate_kbps: u32) -> Self {
        Self {
            resolution,
            codec: codec.into(),
            bitrate_kbps,
        }
    }

    /// Standard H.264 HD proxy.
    #[must_use]
    pub fn h264_hd() -> Self {
        Self::new((1920, 1080), "h264", 8_000)
    }

    /// Standard ProRes Proxy.
    #[must_use]
    pub fn prores_proxy() -> Self {
        Self::new((1920, 1080), "prores_proxy", 45_000)
    }
}

/// A proxy transcode request submitted to the queue.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProxyRequest {
    /// Unique request identifier.
    pub id: String,
    /// Path to the source media file.
    pub source_path: String,
    /// Desired proxy specification.
    pub proxy_spec: ProxySpec,
    /// Priority (0 = lowest, 255 = highest).
    pub priority: u8,
    /// Submission timestamp in milliseconds since epoch.
    pub submitted_at_ms: u64,
}

impl ProxyRequest {
    /// Create a new proxy request.
    #[must_use]
    pub fn new(
        id: impl Into<String>,
        source_path: impl Into<String>,
        proxy_spec: ProxySpec,
        priority: u8,
        submitted_at_ms: u64,
    ) -> Self {
        Self {
            id: id.into(),
            source_path: source_path.into(),
            proxy_spec,
            priority,
            submitted_at_ms,
        }
    }
}

/// Status of a proxy transcode job.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum JobStatus {
    /// Waiting in the queue.
    Queued,
    /// Currently being transcoded.
    Running,
    /// Successfully completed.
    Completed,
    /// Failed due to an error.
    Failed,
    /// Cancelled before execution.
    Cancelled,
}

/// A proxy transcode job with its current status and timing information.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProxyTranscodeJob {
    /// The original request.
    pub request: ProxyRequest,
    /// Current job status.
    pub status: JobStatus,
    /// When the job started, in milliseconds since epoch.
    pub started_at_ms: Option<u64>,
    /// When the job completed (or failed), in milliseconds since epoch.
    pub completed_at_ms: Option<u64>,
    /// Path to the output proxy file (set on completion).
    pub output_path: Option<String>,
    /// Error message if the job failed.
    pub error: Option<String>,
}

impl ProxyTranscodeJob {
    /// Create a new job in the `Queued` state.
    #[must_use]
    pub fn new(request: ProxyRequest) -> Self {
        Self {
            request,
            status: JobStatus::Queued,
            started_at_ms: None,
            completed_at_ms: None,
            output_path: None,
            error: None,
        }
    }

    /// Duration the job waited in the queue before starting, in milliseconds.
    #[must_use]
    pub fn wait_duration_ms(&self) -> Option<u64> {
        self.started_at_ms
            .map(|start| start.saturating_sub(self.request.submitted_at_ms))
    }

    /// Total processing duration, in milliseconds.
    #[must_use]
    pub fn processing_duration_ms(&self) -> Option<u64> {
        match (self.started_at_ms, self.completed_at_ms) {
            (Some(start), Some(end)) => Some(end.saturating_sub(start)),
            _ => None,
        }
    }
}

/// Priority-based proxy transcode queue.
///
/// Higher-priority jobs are dispatched first. Within the same priority,
/// earlier submission time wins (FIFO).
#[derive(Debug, Default)]
pub struct ProxyTranscodeQueue {
    /// All jobs indexed by their ID.
    jobs: std::collections::HashMap<String, ProxyTranscodeJob>,
    /// Queue ordering: sorted by (priority desc, submitted_at_ms asc).
    order: VecDeque<String>,
}

impl ProxyTranscodeQueue {
    /// Create an empty transcode queue.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Submit a new request and return the job ID.
    pub fn submit(&mut self, request: ProxyRequest) -> String {
        let id = request.id.clone();
        let job = ProxyTranscodeJob::new(request);
        // Insert into order queue maintaining priority order
        let priority = job.request.priority;
        let submitted = job.request.submitted_at_ms;
        // Find insertion point: higher priority first, then earlier submission
        let pos = self
            .order
            .iter()
            .position(|existing_id| {
                if let Some(j) = self.jobs.get(existing_id) {
                    let ep = j.request.priority;
                    let es = j.request.submitted_at_ms;
                    // Insert before this entry if we have higher priority,
                    // or equal priority and earlier submission
                    ep < priority || (ep == priority && es > submitted)
                } else {
                    false
                }
            })
            .unwrap_or(self.order.len());
        self.order.insert(pos, id.clone());
        self.jobs.insert(id.clone(), job);
        id
    }

    /// Get the next queued job (highest priority, earliest submitted).
    #[must_use]
    pub fn next_job(&mut self) -> Option<&mut ProxyTranscodeJob> {
        // Find the first job in order that is still Queued
        let next_id = self
            .order
            .iter()
            .find(|id| {
                self.jobs
                    .get(*id)
                    .map(|j| j.status == JobStatus::Queued)
                    .unwrap_or(false)
            })
            .cloned();
        next_id.and_then(|id| self.jobs.get_mut(&id))
    }

    /// Mark a job as started at the given timestamp.
    pub fn start_job(&mut self, id: &str, started_at_ms: u64) {
        if let Some(job) = self.jobs.get_mut(id) {
            if job.status == JobStatus::Queued {
                job.status = JobStatus::Running;
                job.started_at_ms = Some(started_at_ms);
            }
        }
    }

    /// Mark a job as successfully completed.
    pub fn complete_job(&mut self, id: &str, output: &str) {
        if let Some(job) = self.jobs.get_mut(id) {
            job.status = JobStatus::Completed;
            job.output_path = Some(output.to_string());
            // Use a synthetic completion time if not already set
            if job.completed_at_ms.is_none() {
                job.completed_at_ms = job.started_at_ms.map(|s| s + 1);
            }
        }
    }

    /// Mark a job as failed with an error message.
    pub fn fail_job(&mut self, id: &str, error: &str) {
        if let Some(job) = self.jobs.get_mut(id) {
            job.status = JobStatus::Failed;
            job.error = Some(error.to_string());
        }
    }

    /// Cancel a queued job.
    pub fn cancel_job(&mut self, id: &str) {
        if let Some(job) = self.jobs.get_mut(id) {
            if job.status == JobStatus::Queued {
                job.status = JobStatus::Cancelled;
            }
        }
    }

    /// Get a job by ID (immutable).
    #[must_use]
    pub fn get(&self, id: &str) -> Option<&ProxyTranscodeJob> {
        self.jobs.get(id)
    }

    /// Total number of jobs (all statuses).
    #[must_use]
    pub fn len(&self) -> usize {
        self.jobs.len()
    }

    /// Whether the queue has any jobs at all.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.jobs.is_empty()
    }

    /// Iterate over all jobs.
    pub fn iter(&self) -> impl Iterator<Item = &ProxyTranscodeJob> {
        self.jobs.values()
    }
}

/// Aggregated statistics for a `ProxyTranscodeQueue`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QueueStats {
    /// Number of jobs currently queued (waiting).
    pub pending: u32,
    /// Number of jobs currently running.
    pub running: u32,
    /// Total jobs completed successfully.
    pub completed: u64,
    /// Total jobs that failed.
    pub failed: u64,
    /// Average wait time from submission to start, in milliseconds.
    pub avg_wait_ms: f64,
}

impl QueueStats {
    /// Compute statistics from the current state of the queue.
    #[must_use]
    pub fn compute(queue: &ProxyTranscodeQueue) -> Self {
        let mut pending = 0u32;
        let mut running = 0u32;
        let mut completed = 0u64;
        let mut failed = 0u64;
        let mut total_wait_ms = 0u64;
        let mut wait_count = 0u32;

        for job in queue.iter() {
            match job.status {
                JobStatus::Queued => pending += 1,
                JobStatus::Running => running += 1,
                JobStatus::Completed => completed += 1,
                JobStatus::Failed => failed += 1,
                JobStatus::Cancelled => {}
            }
            if let Some(wait) = job.wait_duration_ms() {
                total_wait_ms += wait;
                wait_count += 1;
            }
        }

        let avg_wait_ms = if wait_count == 0 {
            0.0
        } else {
            total_wait_ms as f64 / wait_count as f64
        };

        Self {
            pending,
            running,
            completed,
            failed,
            avg_wait_ms,
        }
    }
}

/// A batch proxy request: transcode many sources with the same spec.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProxyBatchRequest {
    /// Source file paths to transcode.
    pub source_paths: Vec<String>,
    /// Proxy specification to apply to all sources.
    pub spec: ProxySpec,
    /// Maximum number of concurrent transcode jobs.
    pub concurrent_limit: u32,
}

impl ProxyBatchRequest {
    /// Create a new batch request.
    #[must_use]
    pub fn new(source_paths: Vec<String>, spec: ProxySpec, concurrent_limit: u32) -> Self {
        Self {
            source_paths,
            spec,
            concurrent_limit,
        }
    }

    /// Estimate the total time in minutes to transcode `items` files at `fps` frames-per-second.
    ///
    /// Assumes each source file has ~1 minute of content at `fps` fps, and that
    /// the transcode runs at 2× real-time per concurrent slot.
    #[must_use]
    pub fn estimate_duration_mins(items: usize, fps: f32) -> f32 {
        if fps <= 0.0 || items == 0 {
            return 0.0;
        }
        // Simple model: each item takes 0.5 real-time minutes (2× speed),
        // batched by concurrent_limit which defaults to 1 here for the pure fn.
        let per_item_mins = 1.0 / 2.0; // 1 minute of source → 0.5 min at 2× speed
                                       // fps affects CPU cost (higher fps = proportionally more work)
        let fps_factor = fps / 25.0; // normalise to 25 fps
        items as f32 * per_item_mins * fps_factor
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn make_request(id: &str, priority: u8, submitted_at_ms: u64) -> ProxyRequest {
        ProxyRequest::new(
            id,
            format!("/source/{id}.mov"),
            ProxySpec::h264_hd(),
            priority,
            submitted_at_ms,
        )
    }

    #[test]
    fn test_proxy_spec_new() {
        let spec = ProxySpec::new((1280, 720), "h264", 5_000);
        assert_eq!(spec.resolution, (1280, 720));
        assert_eq!(spec.codec, "h264");
        assert_eq!(spec.bitrate_kbps, 5_000);
    }

    #[test]
    fn test_proxy_spec_presets() {
        let hd = ProxySpec::h264_hd();
        assert_eq!(hd.codec, "h264");
        assert_eq!(hd.resolution, (1920, 1080));

        let prores = ProxySpec::prores_proxy();
        assert_eq!(prores.codec, "prores_proxy");
    }

    #[test]
    fn test_submit_returns_id() {
        let mut queue = ProxyTranscodeQueue::new();
        let req = make_request("job_001", 100, 1000);
        let id = queue.submit(req);
        assert_eq!(id, "job_001");
        assert_eq!(queue.len(), 1);
    }

    #[test]
    fn test_priority_ordering() {
        let mut queue = ProxyTranscodeQueue::new();
        queue.submit(make_request("low", 10, 1000));
        queue.submit(make_request("high", 200, 2000));
        queue.submit(make_request("mid", 100, 1500));

        // next_job should return "high" (priority 200)
        let next = queue.next_job().expect("should succeed in test");
        assert_eq!(next.request.id, "high");
    }

    #[test]
    fn test_complete_job() {
        let mut queue = ProxyTranscodeQueue::new();
        queue.submit(make_request("j1", 50, 1000));
        queue.start_job("j1", 1100);
        queue.complete_job("j1", "/proxy/j1.mp4");
        let job = queue.get("j1").expect("should succeed in test");
        assert_eq!(job.status, JobStatus::Completed);
        assert_eq!(job.output_path.as_deref(), Some("/proxy/j1.mp4"));
    }

    #[test]
    fn test_fail_job() {
        let mut queue = ProxyTranscodeQueue::new();
        queue.submit(make_request("j2", 50, 1000));
        queue.start_job("j2", 1100);
        queue.fail_job("j2", "codec error");
        let job = queue.get("j2").expect("should succeed in test");
        assert_eq!(job.status, JobStatus::Failed);
        assert_eq!(job.error.as_deref(), Some("codec error"));
    }

    #[test]
    fn test_cancel_job() {
        let mut queue = ProxyTranscodeQueue::new();
        queue.submit(make_request("j3", 50, 1000));
        queue.cancel_job("j3");
        let job = queue.get("j3").expect("should succeed in test");
        assert_eq!(job.status, JobStatus::Cancelled);
    }

    #[test]
    fn test_queue_stats() {
        let mut queue = ProxyTranscodeQueue::new();
        queue.submit(make_request("a", 10, 0));
        queue.submit(make_request("b", 10, 0));
        queue.submit(make_request("c", 10, 0));
        queue.start_job("a", 100);
        queue.complete_job("a", "/out/a.mp4");
        queue.start_job("b", 100);
        queue.fail_job("b", "err");

        let stats = QueueStats::compute(&queue);
        assert_eq!(stats.pending, 1);
        assert_eq!(stats.completed, 1);
        assert_eq!(stats.failed, 1);
    }

    #[test]
    fn test_queue_is_empty() {
        let queue = ProxyTranscodeQueue::new();
        assert!(queue.is_empty());
    }

    #[test]
    fn test_wait_duration_ms() {
        let mut queue = ProxyTranscodeQueue::new();
        queue.submit(make_request("w", 10, 1000));
        queue.start_job("w", 2000);
        let job = queue.get("w").expect("should succeed in test");
        assert_eq!(job.wait_duration_ms(), Some(1000));
    }

    #[test]
    fn test_batch_estimate_duration_zero_fps() {
        assert!((ProxyBatchRequest::estimate_duration_mins(10, 0.0) - 0.0).abs() < f32::EPSILON);
    }

    #[test]
    fn test_batch_estimate_duration_zero_items() {
        assert!((ProxyBatchRequest::estimate_duration_mins(0, 25.0) - 0.0).abs() < f32::EPSILON);
    }

    #[test]
    fn test_batch_estimate_duration_positive() {
        let mins = ProxyBatchRequest::estimate_duration_mins(10, 25.0);
        assert!(mins > 0.0);
    }
}