clipbin-desktop/crates/clipbin-obs/src/recorder.rs

use std::{
    fmt,
    path::{Path, PathBuf},
    sync::{
        Arc, Mutex,
        mpsc::{self, Receiver},
    },
    time::{Duration, Instant},
};

use clipbin_core::{RecordingStatus, VideoCodec};

use crate::{
    RuntimeLayout,
    capture::{SettingValue, SourceSpec},
    encoders::registered_candidates,
    ffi::{self, Pipeline, PipelineSpec, Session, StartupSpec},
    runtime::PLUGINS,
    thread::ObsThread,
};

/// Segments are Matroska: a segment cut short by a crash stays readable, and
/// export remuxes to MP4 without re-encoding.
pub const SEGMENT_EXTENSION: &str = "mkv";

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct VideoFormat {
    pub width: u32,
    pub height: u32,
    pub frames_per_second: u32,
}

/// The single libobs instance. Dropping it shuts OBS down.
pub struct Engine {
    obs: Arc<ObsThread>,
    session: Option<Session>,
    format: VideoFormat,
    #[cfg(target_os = "linux")]
    _glib: ffi::GlibLoop,
}

impl Engine {
    pub fn start(layout: &RuntimeLayout, format: VideoFormat) -> Result<Self, EngineError> {
        let path = |path: PathBuf| path.to_string_lossy().into_owned();
        let spec = StartupSpec {
            libobs_data_dir: path(layout.libobs_data_dir()),
            plugins: PLUGINS
                .iter()
                .map(|plugin| {
                    (
                        path(layout.plugin_library(plugin)),
                        path(layout.plugin_data_dir(plugin)),
                    )
                })
                .collect(),
            width: format.width,
            height: format.height,
            frames_per_second: format.frames_per_second,
        };
        #[cfg(target_os = "linux")]
        let glib = ffi::GlibLoop::start();
        let obs = Arc::new(ObsThread::spawn());
        let session = obs
            .run(move || ffi::startup(&spec))
            .map_err(EngineError::Startup)?;
        Ok(Self {
            obs,
            session: Some(session),
            format,
            #[cfg(target_os = "linux")]
            _glib: glib,
        })
    }

    pub fn obs_version(&self) -> &str {
        &self.session().version
    }

    /// Bundled plugins that declined to initialise, such as NVENC without an
    /// NVIDIA GPU. Their encoders are simply not registered.
    pub fn plugins_failed(&self) -> &[String] {
        &self.session().plugins_failed
    }

    fn session(&self) -> &Session {
        self.session.as_ref().expect("the session lives until drop")
    }

    /// Registered video encoders as `(id, codec)` pairs, such as `("ffmpeg_vaapi_tex", "h264")`.
    pub fn registered_video_encoders(&self) -> Vec<(String, String)> {
        self.obs.run(ffi::registered_video_encoders)
    }

    /// Registered hardware encoders for `codec`, most preferred first.
    pub fn hardware_encoders(&self, codec: VideoCodec) -> Vec<String> {
        let registered: Vec<String> = self
            .registered_video_encoders()
            .into_iter()
            .map(|(id, _)| id)
            .collect();
        registered_candidates(codec, &registered)
            .into_iter()
            .map(str::to_owned)
            .collect()
    }

    pub fn start_recording(&self, config: RecordingConfig) -> Result<Recording, EngineError> {
        std::fs::create_dir_all(&config.segment_dir).map_err(EngineError::Io)?;
        let spec = PipelineSpec {
            canvas: (self.format.width, self.format.height),
            video_source: config.video_source,
            system_audio: config.system_audio,
            microphone: config.microphone,
            video_encoder: config.video_encoder,
            video_encoder_settings: vec![
                ("rate_control", SettingValue::String("CBR".into())),
                ("bitrate", SettingValue::Int(config.video_bitrate_kbps)),
                (
                    "keyint_sec",
                    SettingValue::Int(config.keyframe_interval_seconds),
                ),
            ],
            audio_bitrate_kbps: 160,
            segment_dir: forward_slashes(&config.segment_dir),
            segment_extension: SEGMENT_EXTENSION,
            segment_seconds: config.segment_seconds,
        };
        let (sender, events) = mpsc::channel();
        let current_segment = Arc::new(Mutex::new(None));
        let segment = Arc::clone(&current_segment);
        let pipeline = self
            .obs
            .run(move || Pipeline::build(&spec, sender, segment).map(Arc::new))
            .map_err(EngineError::Pipeline)?;
        let recording = Recording {
            obs: Arc::clone(&self.obs),
            pipeline: Some(pipeline),
            events,
            current_segment,
            written: Mutex::default(),
            segment_dir: config.segment_dir,
        };
        recording
            .with_pipeline(|pipeline| pipeline.start())
            .map_err(EngineError::Start)?;
        Ok(recording)
    }
}

impl Drop for Engine {
    fn drop(&mut self) {
        if let Some(session) = self.session.take() {
            self.obs.run(move || ffi::shutdown(session));
        }
    }
}

/// Paths in OBS settings use forward slashes on every platform.
fn forward_slashes(path: &Path) -> String {
    path.to_string_lossy().replace('\\', "/")
}

pub struct RecordingConfig {
    pub video_source: Option<SourceSpec>,
    pub system_audio: Option<SourceSpec>,
    pub microphone: Option<SourceSpec>,
    pub video_encoder: String,
    pub video_bitrate_kbps: i64,
    pub keyframe_interval_seconds: i64,
    /// Segments split at the first keyframe after this many seconds.
    pub segment_seconds: i64,
    pub segment_dir: PathBuf,
}

/// An active recording. Dropping it stops the output and releases its OBS
/// objects on the OBS thread. Drop recordings before their `Engine`.
pub struct Recording {
    obs: Arc<ObsThread>,
    pipeline: Option<Arc<Pipeline>>,
    events: Receiver<RecorderEvent>,
    current_segment: Arc<Mutex<Option<String>>>,
    written: Mutex<Written>,
    segment_dir: PathBuf,
}

/// Bytes on disk across this recording's segments.
#[derive(Default)]
struct Written {
    current: Option<String>,
    finished_bytes: u64,
}

impl Recording {
    fn with_pipeline<T: Send + 'static>(
        &self,
        operation: impl FnOnce(&Pipeline) -> T + Send + 'static,
    ) -> T {
        let pipeline = Arc::clone(
            self.pipeline
                .as_ref()
                .expect("the pipeline lives until drop"),
        );
        self.obs.run(move || operation(&pipeline))
    }

    pub fn events(&self) -> &Receiver<RecorderEvent> {
        &self.events
    }

    /// OBS's counters plus what reached disk. OBS counts bytes handed to its
    /// muxer process, which keeps rising even if that process cannot write, so
    /// only growth on disk proves footage is saved.
    pub fn health(&self) -> Health {
        let mut health = self.with_pipeline(Pipeline::health);
        health.disk_available_bytes = fs4::available_space(&self.segment_dir).ok();
        let current = self
            .current_segment
            .lock()
            .ok()
            .and_then(|path| path.clone());
        let size = |path: &str| std::fs::metadata(path).map_or(0, |metadata| metadata.len());
        if let Ok(mut written) = self.written.lock() {
            if written.current != current {
                // The previous segment is complete; count its final size.
                if let Some(finished) = written.current.take() {
                    written.finished_bytes += size(&finished);
                }
                written.current = current;
            }
            health.written_bytes =
                written.finished_bytes + written.current.as_deref().map_or(0, size);
        }
        health
    }

    pub fn restore_token(&self) -> Option<String> {
        self.with_pipeline(Pipeline::restore_token)
    }

    /// Requests a stop; `OutputStopped` follows once the last segment is finalised.
    pub fn stop(&self) {
        self.with_pipeline(Pipeline::stop)
    }
}

impl Drop for Recording {
    fn drop(&mut self) {
        if let Some(pipeline) = self.pipeline.take() {
            // OBS objects must be released on the OBS thread.
            self.obs.run(move || drop(pipeline));
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RecorderEvent {
    /// OBS reports the output as started. Not yet proof that footage is written.
    OutputStarted,
    /// The muxer opened a new segment file; the previous one is complete.
    SegmentStarted { path: String },
    OutputStopped {
        code: i64,
        last_error: Option<String>,
    },
}

/// Counters for honest recording status: OBS accepting a start request is not
/// enough, frames must reach disk and the source must have a picture.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Health {
    pub source_width: u32,
    pub source_height: u32,
    /// PipeWire capture's portal and stream state. `None` for other sources,
    /// which report loss through their size, and before the portal answers.
    pub pipewire: Option<PipeWireState>,
    pub output_active: bool,
    pub total_frames: u64,
    pub dropped_frames: u64,
    pub total_bytes: u64,
    pub lagged_frames: u64,
    /// Bytes on disk across all of this recording's segments.
    pub written_bytes: u64,
    /// Free space on the recording volume, if it could be read.
    pub disk_available_bytes: Option<u64>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PipeWireState {
    /// A screen-share session was requested from the portal.
    Requesting = 1,
    /// The portal is showing its picker or restoring an earlier selection.
    Selecting = 2,
    Streaming = 3,
    /// The stream exists but is not delivering frames.
    Stopped = 4,
    /// The portal request was denied, cancelled by the user, or failed.
    Refused = 5,
}

/// Derives the status shown to the user from health samples and output events.
///
/// Recording is reported only while the source has a picture and encoded data
/// keeps reaching the muxer. A stall after recording began is a failure, not a
/// silent pause.
#[derive(Debug)]
pub struct StatusTracker {
    status: RecordingStatus,
    min_free_bytes: u64,
    stall_limit: Duration,
    last_written: u64,
    last_progress: Instant,
    requesting_since: Option<Instant>,
}

impl StatusTracker {
    /// Free space below which recording stops, so the last segment can still
    /// be finalised rather than failing mid-write.
    pub const DEFAULT_MIN_FREE_BYTES: u64 = 1 << 30;

    /// How long the portal may take to reach its picker or restore a selection.
    pub const PORTAL_RESPONSE_LIMIT: Duration = Duration::from_secs(10);

    /// Matroska output can stay buffered until its segment closes, so footage
    /// may appear on disk only once per segment. A stall is declared after two
    /// segment lengths without growth.
    pub fn new(now: Instant, min_free_bytes: u64, segment_length: Duration) -> Self {
        Self {
            status: RecordingStatus::Starting,
            min_free_bytes,
            stall_limit: (segment_length * 2 + Duration::from_secs(1)).max(Duration::from_secs(3)),
            last_written: 0,
            last_progress: now,
            requesting_since: None,
        }
    }

    /// How long footage may stop reaching disk before recording counts as failed.
    pub fn stall_limit(&self) -> Duration {
        self.stall_limit
    }

    pub fn status(&self) -> &RecordingStatus {
        &self.status
    }

    pub fn sample(&mut self, health: &Health, now: Instant) -> &RecordingStatus {
        if matches!(self.status, RecordingStatus::Failed { .. }) {
            return &self.status;
        }
        let progressed = health.written_bytes > self.last_written;
        if progressed {
            self.last_written = health.written_bytes;
            self.last_progress = now;
        }
        // Before the picker appears the portal needs no user input, so a
        // request stuck there has stalled (seen with Hyprland's portal).
        if health.pipewire == Some(PipeWireState::Requesting) {
            let since = *self.requesting_since.get_or_insert(now);
            if now.duration_since(since) >= Self::PORTAL_RESPONSE_LIMIT {
                self.status = RecordingStatus::Failed {
                    reason:
                        "the screen-share portal did not respond; select a capture source again"
                            .into(),
                };
                return &self.status;
            }
        } else {
            self.requesting_since = None;
        }
        if health.pipewire == Some(PipeWireState::Refused) {
            self.status = RecordingStatus::Failed {
                reason: "screen capture was not allowed: the source selection was denied, cancelled or failed"
                    .into(),
            };
            return &self.status;
        }
        let has_picture = health.source_width > 0
            && health.source_height > 0
            && matches!(health.pipewire, None | Some(PipeWireState::Streaming));
        let stalled = now.duration_since(self.last_progress) >= self.stall_limit;
        if let Some(available) = health.disk_available_bytes
            && available < self.min_free_bytes
        {
            self.status = RecordingStatus::Failed {
                reason: format!(
                    "not enough disk space: {} MB free, {} MB needed",
                    available / 1_000_000,
                    self.min_free_bytes / 1_000_000
                ),
            };
            return &self.status;
        }

        self.status = match (&self.status, has_picture, progressed || !stalled) {
            (_, true, true)
                if health.output_active && health.total_frames > 0 && self.last_written > 0 =>
            {
                RecordingStatus::Recording
            }
            (RecordingStatus::Recording, false, _) => RecordingStatus::Failed {
                reason: "the capture source stopped providing a picture".into(),
            },
            (RecordingStatus::Recording, true, false) => RecordingStatus::Failed {
                reason: format!(
                    "no footage was written for {} seconds",
                    self.stall_limit.as_secs()
                ),
            },
            (_, false, _) => RecordingStatus::WaitingForCapture,
            (status, _, _) => status.clone(),
        };
        &self.status
    }

    /// A failure stays visible after the output stops cleanly; it clears only
    /// when a new recording starts.
    pub fn stopped(&mut self, reason: StopReason, last_error: Option<&str>) -> &RecordingStatus {
        self.status = match reason {
            StopReason::Success if matches!(self.status, RecordingStatus::Failed { .. }) => {
                return &self.status;
            }
            StopReason::Success => RecordingStatus::Idle,
            reason => RecordingStatus::Failed {
                reason: match last_error {
                    Some(error) => format!("{reason:?}: {error}"),
                    None => format!("{reason:?}"),
                },
            },
        };
        &self.status
    }
}

/// Why an output stopped, from `obs-output.h`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StopReason {
    Success,
    BadPath,
    Unsupported,
    NoSpace,
    EncodeError,
    Disconnected,
    Error,
    Other(i64),
}

impl StopReason {
    pub fn from_code(code: i64) -> Self {
        match code as i32 {
            code if code == libobs::OBS_OUTPUT_SUCCESS as i32 => Self::Success,
            libobs::OBS_OUTPUT_BAD_PATH => Self::BadPath,
            libobs::OBS_OUTPUT_UNSUPPORTED => Self::Unsupported,
            libobs::OBS_OUTPUT_NO_SPACE => Self::NoSpace,
            libobs::OBS_OUTPUT_ENCODE_ERROR => Self::EncodeError,
            libobs::OBS_OUTPUT_DISCONNECTED => Self::Disconnected,
            libobs::OBS_OUTPUT_ERROR => Self::Error,
            _ => Self::Other(code),
        }
    }
}

#[derive(Debug)]
pub enum EngineError {
    Startup(String),
    Io(std::io::Error),
    Pipeline(String),
    Start(String),
}

impl fmt::Display for EngineError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Startup(error) => write!(formatter, "OBS did not start: {error}"),
            Self::Io(error) => write!(formatter, "{error}"),
            Self::Pipeline(error) => write!(formatter, "could not set up recording: {error}"),
            Self::Start(error) => write!(formatter, "recording did not start: {error}"),
        }
    }
}

impl std::error::Error for EngineError {}

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

    const SEGMENT: Duration = Duration::from_secs(2);

    fn healthy(bytes: u64) -> Health {
        Health {
            source_width: 1920,
            source_height: 1080,
            pipewire: None,
            output_active: true,
            total_frames: bytes / 1000,
            dropped_frames: 0,
            total_bytes: bytes,
            lagged_frames: 0,
            written_bytes: bytes,
            disk_available_bytes: Some(u64::MAX),
        }
    }

    #[test]
    fn low_disk_space_fails_before_the_disk_is_full() {
        let start = Instant::now();
        let mut tracker = StatusTracker::new(start, 1000, SEGMENT);
        assert_eq!(
            tracker.sample(&healthy(5000), start),
            &RecordingStatus::Recording
        );
        let low = Health {
            disk_available_bytes: Some(999),
            ..healthy(9000)
        };
        assert!(
            matches!(tracker.sample(&low, start), RecordingStatus::Failed { reason } if reason.contains("disk space"))
        );
    }

    #[test]
    fn footage_must_reach_disk_not_just_the_muxer() {
        let start = Instant::now();
        let mut tracker = StatusTracker::new(start, 0, SEGMENT);
        let unwritten = Health {
            written_bytes: 0,
            ..healthy(5000)
        };
        assert_ne!(
            tracker.sample(&unwritten, start),
            &RecordingStatus::Recording
        );
        assert_eq!(
            tracker.sample(&healthy(5000), start),
            &RecordingStatus::Recording
        );
        // OBS keeps counting, but the file on disk stopped growing.
        let stuck = Health {
            total_bytes: 90_000,
            ..healthy(5000)
        };
        let later = start + tracker.stall_limit();
        assert!(matches!(
            tracker.sample(&stuck, later),
            RecordingStatus::Failed { .. }
        ));
    }

    #[test]
    fn recording_needs_a_picture_and_written_footage() {
        let start = Instant::now();
        let mut tracker = StatusTracker::new(start, 0, SEGMENT);
        let no_picture = Health {
            source_width: 0,
            source_height: 0,
            ..healthy(5000)
        };
        assert_eq!(
            tracker.sample(&no_picture, start),
            &RecordingStatus::WaitingForCapture
        );
        assert_eq!(
            tracker.sample(&healthy(0), start),
            &RecordingStatus::WaitingForCapture
        );
        assert_eq!(
            tracker.sample(&healthy(5000), start),
            &RecordingStatus::Recording
        );
    }

    #[test]
    fn a_stall_or_lost_source_fails_instead_of_pausing() {
        let start = Instant::now();
        let mut tracker = StatusTracker::new(start, 0, SEGMENT);
        tracker.sample(&healthy(5000), start);
        let later = start + tracker.stall_limit();
        assert!(matches!(
            tracker.sample(&healthy(5000), later),
            RecordingStatus::Failed { .. }
        ));

        let mut tracker = StatusTracker::new(start, 0, SEGMENT);
        tracker.sample(&healthy(5000), start);
        let lost = Health {
            source_width: 0,
            ..healthy(6000)
        };
        assert!(matches!(
            tracker.sample(&lost, start),
            RecordingStatus::Failed { .. }
        ));
        // Failure persists until a new recording starts.
        assert!(matches!(
            tracker.sample(&healthy(9000), start),
            RecordingStatus::Failed { .. }
        ));
    }

    #[test]
    fn a_stopped_pipewire_stream_is_a_lost_source_despite_its_size() {
        let start = Instant::now();
        let mut tracker = StatusTracker::new(start, 0, SEGMENT);
        let streaming = Health {
            pipewire: Some(PipeWireState::Streaming),
            ..healthy(5000)
        };
        assert_eq!(
            tracker.sample(&streaming, start),
            &RecordingStatus::Recording
        );
        let paused = Health {
            pipewire: Some(PipeWireState::Stopped),
            ..healthy(9000)
        };
        assert!(matches!(
            tracker.sample(&paused, start),
            RecordingStatus::Failed { .. }
        ));
    }

    #[test]
    fn a_refused_portal_request_fails_while_waiting_for_capture() {
        let start = Instant::now();
        let mut tracker = StatusTracker::new(start, 0, SEGMENT);
        let waiting = Health {
            source_width: 0,
            source_height: 0,
            ..healthy(5000)
        };
        assert_eq!(
            tracker.sample(&waiting, start),
            &RecordingStatus::WaitingForCapture
        );
        let refused = Health {
            pipewire: Some(PipeWireState::Refused),
            ..waiting
        };
        assert!(matches!(
            tracker.sample(&refused, start),
            RecordingStatus::Failed { reason } if reason.contains("not allowed")
        ));
    }

    #[test]
    fn a_portal_that_never_reaches_its_picker_fails() {
        let start = Instant::now();
        let mut tracker = StatusTracker::new(start, 0, SEGMENT);
        let requesting = Health {
            source_width: 0,
            source_height: 0,
            pipewire: Some(PipeWireState::Requesting),
            ..healthy(5000)
        };
        assert_eq!(
            tracker.sample(&requesting, start),
            &RecordingStatus::WaitingForCapture
        );
        // The user may take as long as they like in the picker.
        let selecting = Health {
            pipewire: Some(PipeWireState::Selecting),
            ..requesting.clone()
        };
        let much_later = start + StatusTracker::PORTAL_RESPONSE_LIMIT * 3;
        assert_eq!(
            tracker.sample(&selecting, much_later),
            &RecordingStatus::WaitingForCapture
        );
        let mut tracker = StatusTracker::new(start, 0, SEGMENT);
        tracker.sample(&requesting, start);
        assert!(matches!(
            tracker.sample(&requesting, start + StatusTracker::PORTAL_RESPONSE_LIMIT),
            RecordingStatus::Failed { reason } if reason.contains("did not respond")
        ));
    }

    #[test]
    fn output_errors_fail_and_clean_stops_go_idle() {
        let mut tracker = StatusTracker::new(Instant::now(), 0, SEGMENT);
        assert!(matches!(
            tracker.stopped(StopReason::NoSpace, Some("disk full")),
            RecordingStatus::Failed { reason } if reason.contains("disk full")
        ));
        let mut tracker = StatusTracker::new(Instant::now(), 0, SEGMENT);
        assert_eq!(
            tracker.stopped(StopReason::Success, None),
            &RecordingStatus::Idle
        );
    }

    #[test]
    fn stopping_after_a_failure_keeps_the_failure() {
        let start = Instant::now();
        let mut tracker = StatusTracker::new(start, 0, SEGMENT);
        tracker.sample(&healthy(5000), start);
        let lost = Health {
            source_width: 0,
            ..healthy(6000)
        };
        tracker.sample(&lost, start);
        assert!(matches!(
            tracker.stopped(StopReason::Success, None),
            RecordingStatus::Failed { .. }
        ));
    }

    #[test]
    fn stop_codes_map_to_reasons() {
        assert_eq!(StopReason::from_code(0), StopReason::Success);
        assert_eq!(
            StopReason::from_code(libobs::OBS_OUTPUT_NO_SPACE.into()),
            StopReason::NoSpace
        );
        assert_eq!(StopReason::from_code(-99), StopReason::Other(-99));
    }
}