Overhaul recording pipeline (#1119)

* hoooly

* windows

* fixes

* bruh

* windows works

* studio mode actor

* system audio mostly

* clippy

* fix capture

* hook up error handling for tasks and done_rx

* desktop instant recording

* windows system audio

* more windows

* use flume for video tx

* native encoder working

* windows working properly

* formatting

* Implement completion future for studio recording

* cleanup

* fix

* fix microphone capture on Windows

* extract ffmpeg encoder logic into common base

* audio works

* cargo fix

* start moving windows capturer to dedicated thread

* fix windows muxer

* some more logging

* fix recording-cli

* clippy

* fix log

---------

Co-authored-by: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com>

Author: Brendonovich

Cargo.lock

@@ -75,7 +75,7 @@ impl RecordStart {
             .await
             .unwrap();
 
-        actor.0.stop().await.unwrap();
+        actor.stop().await.unwrap();
 
         Ok(())
     }

apps/cli/src/record.rs

@@ -1,7 +1,7 @@
 use anyhow::{Context, anyhow};
-use cap_recording::feeds::{
-    self,
-    camera::{CameraFeed, RawCameraFrame},
+use cap_recording::{
+    FFmpegVideoFrame,
+    feeds::{self, camera::CameraFeed},
 };
 use ffmpeg::{
     format::{self, Pixel},
@@ -465,7 +465,7 @@ impl Renderer {
         window: WebviewWindow,
         default_state: CameraPreviewState,
         mut reconfigure: broadcast::Receiver<ReconfigureEvent>,
-        camera_rx: flume::Receiver<RawCameraFrame>,
+        camera_rx: flume::Receiver<FFmpegVideoFrame>,
     ) {
         let mut resampler_frame = Cached::default();
         let Ok(mut scaler) = scaling::Context::get(
@@ -496,7 +496,7 @@ impl Renderer {
         } {
             match event {
                 Ok(frame) => {
-                    let aspect_ratio = frame.frame.width() as f32 / frame.frame.height() as f32;
+                    let aspect_ratio = frame.inner.width() as f32 / frame.inner.height() as f32;
                     self.sync_ratio_uniform_and_resize_window_to_it(&window, &state, aspect_ratio);
 
                     if let Ok(surface) = self.surface.get_current_texture().map_err(|err| {
@@ -509,16 +509,16 @@ impl Renderer {
                             .get_or_init((output_width, output_height), frame::Video::empty);
 
                         scaler.cached(
-                            frame.frame.format(),
-                            frame.frame.width(),
-                            frame.frame.height(),
+                            frame.inner.format(),
+                            frame.inner.width(),
+                            frame.inner.height(),
                             format::Pixel::RGBA,
                             output_width,
                             output_height,
                             ffmpeg::software::scaling::flag::Flags::FAST_BILINEAR,
                         );
 
-                        if let Err(err) = scaler.run(&frame.frame, resampler_frame) {
+                        if let Err(err) = scaler.run(&frame.inner, resampler_frame) {
                             error!("Error rescaling frame with ffmpeg: {err:?}");
                             continue;
                         }

apps/desktop/src-tauri/src/camera.rs

@@ -1,19 +1,19 @@
-use cap_recording::feeds::camera::RawCameraFrame;
+use cap_recording::FFmpegVideoFrame;
 use flume::Sender;
 use tokio_util::sync::CancellationToken;
 
 use crate::frame_ws::{WSFrame, create_frame_ws};
 
-pub async fn create_camera_preview_ws() -> (Sender<RawCameraFrame>, u16, CancellationToken) {
-    let (camera_tx, mut _camera_rx) = flume::bounded::<RawCameraFrame>(4);
+pub async fn create_camera_preview_ws() -> (Sender<FFmpegVideoFrame>, u16, CancellationToken) {
+    let (camera_tx, mut _camera_rx) = flume::bounded::<FFmpegVideoFrame>(4);
     let (_camera_tx, camera_rx) = flume::bounded::<WSFrame>(4);
     std::thread::spawn(move || {
         use ffmpeg::format::Pixel;
 
         let mut converter: Option<(Pixel, ffmpeg::software::scaling::Context)> = None;
 
         while let Ok(raw_frame) = _camera_rx.recv() {
-            let mut frame = raw_frame.frame;
+            let mut frame = raw_frame.inner;
 
             if frame.format() != Pixel::RGBA || frame.width() > 1280 || frame.height() > 720 {
                 let converter = match &mut converter {

apps/desktop/src-tauri/src/camera_legacy.rs

@@ -1,4 +1,6 @@
-use cap_recording::{RecordingMode, feeds::camera::DeviceOrModelID, sources::ScreenCaptureTarget};
+use cap_recording::{
+    RecordingMode, feeds::camera::DeviceOrModelID, sources::screen_capture::ScreenCaptureTarget,
+};
 use serde::{Deserialize, Serialize};
 use std::path::{Path, PathBuf};
 use tauri::{AppHandle, Manager, Url};

apps/desktop/src-tauri/src/deeplink_actions.rs

@@ -40,7 +40,7 @@ use cap_recording::{
         camera::{CameraFeed, DeviceOrModelID},
         microphone::{self, MicrophoneFeed},
     },
-    sources::ScreenCaptureTarget,
+    sources::screen_capture::ScreenCaptureTarget,
 };
 use cap_rendering::{ProjectRecordingsMeta, RenderedFrame};
 use clipboard_rs::common::RustImage;

apps/desktop/src-tauri/src/lib.rs

@@ -9,7 +9,10 @@ use cap_recording::{
     RecordingError, RecordingMode,
     feeds::{camera, microphone},
     instant_recording,
-    sources::{CaptureDisplay, CaptureWindow, ScreenCaptureTarget, screen_capture},
+    sources::{
+        screen_capture,
+        screen_capture::{CaptureDisplay, CaptureWindow, ScreenCaptureTarget},
+    },
     studio_recording,
 };
 use cap_rendering::ProjectRecordingsMeta;
@@ -79,17 +82,19 @@ impl InProgressRecording {
     }
 
     pub async fn pause(&self) -> Result<(), RecordingError> {
-        match self {
-            Self::Instant { handle, .. } => handle.pause().await,
-            Self::Studio { handle, .. } => handle.pause().await,
-        }
+        todo!()
+        // match self {
+        //     Self::Instant { handle, .. } => handle.pause().await,
+        //     Self::Studio { handle, .. } => handle.pause().await,
+        // }
     }
 
     pub async fn resume(&self) -> Result<(), String> {
-        match self {
-            Self::Instant { handle, .. } => handle.resume().await.map_err(|e| e.to_string()),
-            Self::Studio { handle, .. } => handle.resume().await.map_err(|e| e.to_string()),
-        }
+        todo!()
+        // match self {
+        //     Self::Instant { handle, .. } => handle.resume().await.map_err(|e| e.to_string()),
+        //     Self::Studio { handle, .. } => handle.resume().await.map_err(|e| e.to_string()),
+        // }
     }
 
     pub fn recording_dir(&self) -> &PathBuf {
@@ -99,7 +104,7 @@ impl InProgressRecording {
         }
     }
 
-    pub async fn stop(self) -> Result<CompletedRecording, RecordingError> {
+    pub async fn stop(self) -> anyhow::Result<CompletedRecording> {
         Ok(match self {
             Self::Instant {
                 handle,
@@ -124,13 +129,21 @@ impl InProgressRecording {
         })
     }
 
-    pub async fn cancel(self) -> Result<(), RecordingError> {
+    pub fn done_fut(&self) -> cap_recording::DoneFut {
         match self {
-            Self::Instant { handle, .. } => handle.cancel().await,
-            Self::Studio { handle, .. } => handle.cancel().await,
+            Self::Instant { handle, .. } => handle.done_fut(),
+            Self::Studio { handle, .. } => handle.done_fut(),
         }
     }
 
+    pub async fn cancel(self) -> Result<(), RecordingError> {
+        todo!()
+        // match self {
+        //     Self::Instant { handle, .. } => handle.cancel().await,
+        //     Self::Studio { handle, .. } => handle.cancel().await,
+        // }
+    }
+
     pub fn mode(&self) -> RecordingMode {
         match self {
             Self::Instant { .. } => RecordingMode::Instant,
@@ -147,7 +160,7 @@ pub enum CompletedRecording {
         video_upload_info: VideoUploadInfo,
     },
     Studio {
-        recording: studio_recording::CompletedStudioRecording,
+        recording: studio_recording::CompletedRecording,
         target_name: String,
     },
 }
@@ -406,14 +419,13 @@ pub async fn start_recording(
                     Err(SendError::HandlerError(camera::LockFeedError::NoInput)) => None,
                     Err(e) => return Err(e.to_string()),
                 };
-
                 #[cfg(target_os = "macos")]
                 let shareable_content = crate::platform::get_shareable_content()
                     .await
                     .map_err(|e| format!("GetShareableContent: {e}"))?
                     .ok_or_else(|| format!("GetShareableContent/NotAvailable"))?;
 
-                let (actor, actor_done_rx) = match inputs.mode {
+                let actor = match inputs.mode {
                     RecordingMode::Studio => {
                         let mut builder = studio_recording::Actor::builder(
                             recording_dir.clone(),
@@ -434,7 +446,7 @@ pub async fn start_recording(
                             builder = builder.with_mic_feed(mic_feed);
                         }
 
-                        let (handle, actor_done_rx) = builder
+                        let handle = builder
                             .build(
                                 #[cfg(target_os = "macos")]
                                 shareable_content,
@@ -445,15 +457,12 @@ pub async fn start_recording(
                                 e.to_string()
                             })?;
 
-                        (
-                            InProgressRecording::Studio {
-                                handle,
-                                target_name,
-                                inputs,
-                                recording_dir: recording_dir.clone(),
-                            },
-                            actor_done_rx,
-                        )
+                        InProgressRecording::Studio {
+                            handle,
+                            target_name,
+                            inputs,
+                            recording_dir: recording_dir.clone(),
+                        }
                     }
                     RecordingMode::Instant => {
                         let Some(video_upload_info) = video_upload_info.clone() else {
@@ -470,7 +479,7 @@ pub async fn start_recording(
                             builder = builder.with_mic_feed(mic_feed);
                         }
 
-                        let (handle, actor_done_rx) = builder
+                        let handle = builder
                             .build(
                                 #[cfg(target_os = "macos")]
                                 shareable_content,
@@ -481,32 +490,31 @@ pub async fn start_recording(
                                 e.to_string()
                             })?;
 
-                        (
-                            InProgressRecording::Instant {
-                                handle,
-                                progressive_upload,
-                                video_upload_info,
-                                target_name,
-                                inputs,
-                                recording_dir: recording_dir.clone(),
-                            },
-                            actor_done_rx,
-                        )
+                        InProgressRecording::Instant {
+                            handle,
+                            progressive_upload,
+                            video_upload_info,
+                            target_name,
+                            inputs,
+                            recording_dir: recording_dir.clone(),
+                        }
                     }
                 };
 
+                let done_fut = actor.done_fut();
+
                 state.set_current_recording(actor);
 
-                Ok::<_, String>(actor_done_rx)
+                Ok::<_, String>(done_fut)
             }
         })
         .await
         .map_err(|e| format!("Failed to spawn recording actor: {e}"))?
     }
     .await;
 
-    let actor_done_rx = match spawn_actor_res {
-        Ok(rx) => rx,
+    let actor_done_fut = match spawn_actor_res {
+        Ok(fut) => fut,
         Err(e) => {
             let _ = RecordingEvent::Failed { error: e.clone() }.emit(&app);
 
@@ -537,22 +545,25 @@ pub async fn start_recording(
         let state_mtx = Arc::clone(&state_mtx);
         async move {
             fail!("recording::wait_actor_done");
-            let res = actor_done_rx.await;
+            let res = actor_done_fut.await;
             info!("recording wait actor done: {:?}", &res);
             match res {
-                Ok(Ok(_)) => {
+                Ok(()) => {
                     let _ = finish_upload_tx.send(());
                     let _ = RecordingEvent::Stopped.emit(&app);
                 }
-                Ok(Err(e)) => {
+                Err(e) => {
                     let mut state = state_mtx.write().await;
 
-                    let _ = RecordingEvent::Failed { error: e.clone() }.emit(&app);
+                    let _ = RecordingEvent::Failed {
+                        error: e.to_string(),
+                    }
+                    .emit(&app);
 
                     let mut dialog = MessageDialogBuilder::new(
                         app.dialog().clone(),
                         "An error occurred".to_string(),
-                        e,
+                        e.to_string(),
                     )
                     .kind(tauri_plugin_dialog::MessageDialogKind::Error);
 
@@ -565,10 +576,6 @@ pub async fn start_recording(
                     // this clears the current recording for us
                     handle_recording_end(app, None, &mut state).await.ok();
                 }
-                // Actor hasn't errored, it's just finished
-                v => {
-                    info!("recording actor ended: {v:?}");
-                }
             }
         }
     });
@@ -628,7 +635,7 @@ pub async fn restart_recording(app: AppHandle, state: MutableState<'_, App>) ->
 
     let inputs = recording.inputs().clone();
 
-    let _ = recording.cancel().await;
+    // let _ = recording.cancel().await;
 
     tokio::time::sleep(Duration::from_millis(1000)).await;
 
@@ -658,7 +665,7 @@ pub async fn delete_recording(app: AppHandle, state: MutableState<'_, App>) -> R
         CurrentRecordingChanged.emit(&app).ok();
         RecordingStopped {}.emit(&app).ok();
 
-        let _ = recording.cancel().await;
+        // let _ = recording.cancel().await;
 
         std::fs::remove_dir_all(&recording_dir).ok();
 
@@ -1105,7 +1112,7 @@ fn generate_zoom_segments_from_clicks_impl(
 /// Generates zoom segments based on mouse click events during recording.
 /// Used during the recording completion process.
 pub fn generate_zoom_segments_from_clicks(
-    recording: &studio_recording::CompletedStudioRecording,
+    recording: &studio_recording::CompletedRecording,
     recordings: &ProjectRecordingsMeta,
 ) -> Vec<ZoomSegment> {
     // Build a temporary RecordingMeta so we can use the common implementation
@@ -1162,7 +1169,7 @@ pub fn generate_zoom_segments_for_project(
 
 fn project_config_from_recording(
     app: &AppHandle,
-    completed_recording: &studio_recording::CompletedStudioRecording,
+    completed_recording: &studio_recording::CompletedRecording,
     recordings: &ProjectRecordingsMeta,
     default_config: Option<ProjectConfiguration>,
 ) -> ProjectConfiguration {

apps/desktop/src-tauri/src/recording.rs

@@ -1,4 +1,6 @@
-use cap_recording::{RecordingMode, feeds::camera::DeviceOrModelID, sources::ScreenCaptureTarget};
+use cap_recording::{
+    RecordingMode, feeds::camera::DeviceOrModelID, sources::screen_capture::ScreenCaptureTarget,
+};
 use serde_json::json;
 use tauri::{AppHandle, Wry};
 use tauri_plugin_store::StoreExt;

apps/desktop/src-tauri/src/recording_settings.rs

@@ -1,4 +1,4 @@
-use cap_recording::sources::{list_displays, list_windows};
+use cap_recording::sources::screen_capture::{list_displays, list_windows};
 use serde::{Deserialize, Serialize};
 use specta::Type;
 use tracing::*;