diff --git a/apps/desktop-gpui/src/auth.rs b/apps/desktop-gpui/src/auth.rs index cfc44f06190..40a69b74a2d 100644 --- a/apps/desktop-gpui/src/auth.rs +++ b/apps/desktop-gpui/src/auth.rs @@ -402,19 +402,7 @@ pub async fn authed_request_with( path: &str, body: Option, ) -> Result { - let (server, token) = if let Ok(context) = UPLOAD_REQUEST.try_with(Clone::clone) { - let token = context.credentials()?; - (context.server_url, token) - } else { - let auth = store::auth_snapshot(); - ( - normalize_server_url(&auth.server_url), - auth.token.ok_or(AuthApiError::InvalidAuthentication)?, - ) - }; - let url = format!("{server}{path}"); - let mut request = client.request(method, url); - request = apply_desktop_headers(request).bearer_auth(token); + let mut request = authed_request_builder(client, method, path)?; if let Some(body) = body { request = request .header("Content-Type", "application/json") @@ -433,6 +421,25 @@ pub async fn authed_request_with( Ok(response) } +pub(crate) fn authed_request_builder( + client: &reqwest::Client, + method: reqwest::Method, + path: &str, +) -> Result { + let (server, token) = if let Ok(context) = UPLOAD_REQUEST.try_with(Clone::clone) { + let token = context.credentials()?; + (context.server_url, token) + } else { + let auth = store::auth_snapshot(); + ( + normalize_server_url(&auth.server_url), + auth.token.ok_or(AuthApiError::InvalidAuthentication)?, + ) + }; + let url = format!("{server}{path}"); + Ok(apply_desktop_headers(client.request(method, url)).bearer_auth(token)) +} + pub fn process_auth_data(params: AuthParams) -> Result<(), String> { let (secret, user_id) = match params { AuthParams::ApiKey { api_key, user_id } => (json!({ "api_key": api_key }), user_id), diff --git a/apps/desktop-gpui/src/main.rs b/apps/desktop-gpui/src/main.rs index 61d7e8d88e1..e75dc70ae4f 100644 --- a/apps/desktop-gpui/src/main.rs +++ b/apps/desktop-gpui/src/main.rs @@ -62,6 +62,7 @@ mod main_window; mod media_access; mod menus; mod mode_select_window; +mod network_health; mod onboarding_audio; mod onboarding_doodles; mod onboarding_window; @@ -401,6 +402,7 @@ fn main() { platform::follow_system_reduce_motion(cx); let session = RecordingSession::init(cx); + crate::network_health::init(cx); crate::sign_in::SignIn::init(cx); crate::feeds::Feeds::init(cx); crate::media_access::MediaAccess::init(cx); diff --git a/apps/desktop-gpui/src/main_window.rs b/apps/desktop-gpui/src/main_window.rs index 28f6daac7bc..8e7fa7e9fe6 100644 --- a/apps/desktop-gpui/src/main_window.rs +++ b/apps/desktop-gpui/src/main_window.rs @@ -3810,6 +3810,27 @@ impl MainWindow { ) }; + let upload_health = crate::network_health::indicator().map(|health| { + let color = if health.failed { + theme.red_11 + } else if health.limited { + theme.amber_11 + } else { + theme.gray_11 + }; + div() + .id("upload-health") + .tab_index(0) + .flex_shrink_0() + .text_size(px(9.)) + .text_color(color) + .cursor_pointer() + .child(health.label) + .tooltip(move |_, cx| ui::Tooltip::new(&theme, health.tooltip.clone()).view(cx)) + .tooltip_show_delay(ui::TOOLTIP_SHOW_DELAY) + .on_click(cx.listener(|_, _, _, cx| crate::network_health::request(cx, true))) + }); + let actions = div() .flex() .flex_1() @@ -3817,6 +3838,7 @@ impl MainWindow { .gap(px(4.)) .mx(px(8.)) .min_w_0() + .children(upload_health) .child( icon_button("help", "icons/circle-help.svg", 16.).on_click(cx.listener( |_, _, _window, cx| { diff --git a/apps/desktop-gpui/src/network_health.rs b/apps/desktop-gpui/src/network_health.rs new file mode 100644 index 00000000000..303cefb1631 --- /dev/null +++ b/apps/desktop-gpui/src/network_health.rs @@ -0,0 +1,724 @@ +mod policy; + +use std::{ + sync::{Mutex, MutexGuard, OnceLock}, + time::{Duration, SystemTime}, +}; + +use futures_util::future::{AbortHandle, AbortRegistration, Abortable, select}; +use gpui::App; +use policy::{AUTOMATIC_PROBE_INTERVAL, Admission, Measurement, Sample, Scope}; +use serde::Deserialize; + +use crate::{auth, session::RecordingSession, store}; + +#[derive(Clone, Copy, Default)] +enum Status { + #[default] + Unknown, + Checking, + Ready, + Failed, + Unsupported, +} + +struct Running { + ticket: u64, + abort: AbortHandle, + done: flume::Receiver<()>, +} + +#[derive(Default)] +struct Controller { + admission: Admission, + scope: Option, + measurement: Option, + status: Status, + running: Option, + attempted_at: Option, + applied_width: Option, +} + +fn controller() -> MutexGuard<'static, Controller> { + static STATE: OnceLock> = OnceLock::new(); + STATE + .get_or_init(|| Mutex::new(Controller::default())) + .lock() + .unwrap_or_else(|error| error.into_inner()) +} + +fn wake_channel() -> &'static (flume::Sender<()>, flume::Receiver<()>) { + static WAKE: OnceLock<(flume::Sender<()>, flume::Receiver<()>)> = OnceLock::new(); + WAKE.get_or_init(|| flume::bounded(1)) +} + +fn notify_wake() { + let _ = wake_channel().0.try_send(()); +} + +impl Controller { + fn admit_recording(&mut self) { + self.applied_width = None; + self.status = Status::Unknown; + self.admission.admit_recording(); + if let Some(running) = &self.running { + running.abort.abort(); + } + } + + fn admit_transfer(&mut self) { + self.status = Status::Unknown; + self.admission.admit_transfer(); + if let Some(running) = &self.running { + running.abort.abort(); + } + } + + fn release_transfer(&mut self) { + self.admission.release_transfer(); + } + + fn begin_request( + &mut self, + scope: Option, + recording_in_flight: bool, + manual: bool, + now: SystemTime, + ) -> Option<(Scope, u64, AbortRegistration, Completion)> { + if self.scope != scope { + self.admission.invalidate(); + if let Some(running) = &self.running { + running.abort.abort(); + } + self.scope = scope.clone(); + self.measurement = None; + self.attempted_at = None; + self.status = Status::Unknown; + } + let scope = scope?; + let interval = if manual { + Duration::from_secs(60) + } else { + AUTOMATIC_PROBE_INTERVAL + }; + if self.running.is_some() + || recording_in_flight + || self + .attempted_at + .is_some_and(|at| now.duration_since(at).is_ok_and(|age| age < interval)) + { + return None; + } + let ticket = self.admission.begin_probe()?; + let (abort, registration) = AbortHandle::new_pair(); + let (done, receiver) = flume::bounded(1); + self.running = Some(Running { + ticket, + abort, + done: receiver, + }); + self.status = Status::Checking; + self.attempted_at = Some(now); + Some((scope, ticket, registration, Completion(done))) + } + + fn finish_request( + &mut self, + ticket: u64, + expected_scope: &Scope, + current_scope: Option<&Scope>, + result: Result, + now: SystemTime, + ) -> bool { + if !self + .running + .as_ref() + .is_some_and(|running| running.ticket == ticket) + { + return false; + } + self.running = None; + if !self.admission.accepts(ticket) || current_scope != Some(expected_scope) { + self.status = Status::Unknown; + self.attempted_at = self + .measurement + .as_ref() + .map(|measurement| measurement.checked_at); + } else { + match result { + Ok(measurement) => { + self.attempted_at = Some(measurement.checked_at); + self.measurement = Some(measurement); + self.status = Status::Ready; + } + Err(ProbeError::Unsupported) => { + self.measurement = None; + self.status = Status::Unsupported; + } + Err(ProbeError::Failed) => { + if self + .measurement + .as_ref() + .is_none_or(|measurement| !measurement.fresh_for(expected_scope, now)) + { + self.measurement = None; + } + self.status = Status::Failed; + } + } + } + true + } +} + +fn current_scope() -> Option { + store::auth_snapshot().token?; + Some(Scope { + server: auth::server_url(), + owner: store::store_section("auth") + .get("user_id")? + .as_str()? + .to_string(), + organization: store::store_section(store::RECORDING_SETTINGS) + .get("organizationId") + .and_then(serde_json::Value::as_str) + .filter(|id| !id.is_empty()) + .map(str::to_string), + }) +} + +pub(crate) fn invalidate() { + let mut state = controller(); + state.admission.invalidate(); + if let Some(running) = &state.running { + running.abort.abort(); + } + state.measurement = None; + state.attempted_at = None; + state.status = Status::Unknown; + drop(state); + notify_wake(); +} + +pub(crate) fn recording_admitted() { + controller().admit_recording(); +} + +pub(crate) fn recording_released() { + let mut state = controller(); + state.admission.release_recording(); + state.applied_width = None; + drop(state); + notify_wake(); +} + +pub(crate) struct TransferLease; + +impl Drop for TransferLease { + fn drop(&mut self) { + let mut state = controller(); + state.release_transfer(); + drop(state); + notify_wake(); + } +} + +pub(crate) fn acquire_transfer() -> TransferLease { + controller().admit_transfer(); + TransferLease +} + +#[cfg(test)] +pub(crate) fn probe_admitted_for_test() -> bool { + controller().admission.begin_probe().is_some() +} + +pub(crate) async fn wait_for_probe() { + let done = controller() + .running + .as_ref() + .map(|running| running.done.clone()); + if let Some(done) = done { + let _ = done.recv_async().await; + } +} + +pub(crate) fn max_width(configured: u32, organization: Option<&str>) -> u32 { + let Some(scope) = current_scope() else { + return configured; + }; + let mut state = controller(); + let width = state + .measurement + .as_ref() + .and_then(|measurement| measurement.max_width_for(&scope, organization, SystemTime::now())) + .map_or(configured, |cap| configured.min(cap)); + state.applied_width = (width < configured).then_some(width); + width +} + +pub(crate) struct Indicator { + pub label: String, + pub tooltip: String, + pub failed: bool, + pub limited: bool, +} + +pub(crate) fn indicator() -> Option { + let scope = current_scope()?; + let state = controller(); + let measurement = state + .measurement + .as_ref() + .filter(|measurement| { + !matches!(state.status, Status::Failed) + && measurement.fresh_for(&scope, SystemTime::now()) + }); + let (label, tooltip, failed, limited) = if let Some(width) = state.applied_width { + ( + format!("Instant limited: {}p", width * 9 / 16), + "This recording uses a lower resolution because repeated upload samples were slow. Your saved setting is unchanged; retest after recording.".into(), + false, + true, + ) + } else if state.scope.as_ref() == Some(&scope) && matches!(state.status, Status::Checking) { + ( + "Checking upload".into(), + "Checking upload availability and transfer speed.".into(), + false, + false, + ) + } else if let Some(measurement) = measurement { + let speed = measurement + .mbps + .map_or_else(|| "Upload OK".to_string(), |rate| format!("{rate:.1} Mb/s")); + let configured = if store::auth_snapshot().is_upgraded() { + store::GeneralSettings::load().instant_mode_max_resolution + } else { + cap_recording::FREE_INSTANT_MODE_MAX_RESOLUTION + }; + let cap = measurement.max_width.filter(|cap| *cap < configured); + match cap { + Some(width) => ( + format!("{speed} · {}p", width * 9 / 16), + format!( + "Repeated slow uploads: Instant recordings are limited to {}p. Your saved resolution is unchanged. Click to retest while idle.", + width * 9 / 16 + ), + false, + true, + ), + None => ( + speed, + "Upload destination is writable. Click to retest while idle. Tests are paused during recording.".into(), + false, + false, + ), + } + } else { + match if state.scope.as_ref() == Some(&scope) { + state.status + } else { + Status::Unknown + } { + Status::Checking => ( + "Checking upload".into(), + "Checking upload availability and transfer speed.".into(), + false, + false, + ), + Status::Failed => ( + "Upload check failed".into(), + "The upload check failed. Check your connection and storage settings, retry while idle, or contact support. Recording is still available.".into(), + true, + false, + ), + Status::Unsupported => ( + "Upload check unavailable".into(), + "This server does not support upload diagnostics. Your recording resolution is unchanged.".into(), + false, + false, + ), + _ => ( + "Check upload".into(), + "Click to check the upload destination and speed while idle.".into(), + false, + false, + ), + } + }; + Some(Indicator { + label, + tooltip, + failed, + limited, + }) +} + +struct Completion(flume::Sender<()>); + +impl Drop for Completion { + fn drop(&mut self) { + let _ = self.0.send(()); + } +} + +pub(crate) fn request(cx: &mut App, manual: bool) { + let scope = current_scope(); + let recording_in_flight = RecordingSession::recording_in_flight(cx); + let Some((scope, ticket, registration, completion)) = + controller().begin_request(scope, recording_in_flight, manual, SystemTime::now()) + else { + return; + }; + let expected_scope = scope.clone(); + let task = gpui_tokio::Tokio::spawn(cx, async move { + let _completion = completion; + Abortable::new(probe(scope, ticket), registration).await + }); + cx.spawn(async move |cx| { + let result = match task.await { + Ok(Ok(result)) => result, + _ => Err(ProbeError::Failed), + }; + cx.update(|cx| { + let active_scope = current_scope(); + let updated = controller().finish_request( + ticket, + &expected_scope, + active_scope.as_ref(), + result, + SystemTime::now(), + ); + if updated { + cx.refresh_windows(); + } + }); + }) + .detach(); + cx.refresh_windows(); +} + +pub(crate) fn init(cx: &mut App) { + let wake = wake_channel().1.clone(); + cx.spawn(async move |cx| { + loop { + cx.update(|cx| request(cx, false)); + let timer = cx.background_executor().timer(Duration::from_secs(15)); + let notified = wake.recv_async(); + let _ = select(std::pin::pin!(timer), std::pin::pin!(notified)).await; + } + }) + .detach(); +} + +#[derive(Debug)] +enum ProbeError { + Failed, + Unsupported, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct ProbeResponse { + bytes: usize, + body_read_ms: f64, + organization_id: String, +} + +fn probe_admitted(ticket: u64, scope: &Scope) -> bool { + current_scope().as_ref() == Some(scope) && controller().admission.accepts(ticket) +} + +async fn probe(scope: Scope, ticket: u64) -> Result { + let client = auth::api_client().map_err(|_| ProbeError::Failed)?; + let context = auth::UploadRequestContext::for_upload(&scope.server, Some(&scope.owner)) + .map_err(|_| ProbeError::Failed)?; + context + .run(async { + if !probe_admitted(ticket, &scope) { + return Err(ProbeError::Failed); + } + let warm = auth::authed_request_with( + &client, + reqwest::Method::GET, + "/api/desktop/upload-health", + None, + ) + .await + .map_err(|_| ProbeError::Failed)?; + if warm.status() == reqwest::StatusCode::NOT_FOUND { + return Err(ProbeError::Unsupported); + } + if !warm.status().is_success() { + return Err(ProbeError::Failed); + } + warm.bytes().await.map_err(|_| ProbeError::Failed)?; + let mut payload = vec![0_u8; 2 * 1024 * 1024]; + let mut random = 0x3ab1_5fc9_u32; + for byte in &mut payload { + random ^= random << 13; + random ^= random >> 17; + random ^= random << 5; + *byte = random as u8; + } + let payload = bytes::Bytes::from(payload); + let mut samples = [Sample { + bytes: 0, + body_read_ms: 0., + }; 2]; + let mut organization: Option = None; + for (index, sample) in samples.iter_mut().enumerate() { + if !probe_admitted(ticket, &scope) { + return Err(ProbeError::Failed); + } + let path = if index == 0 { + "/api/desktop/upload-health?checkStorage=true" + } else { + "/api/desktop/upload-health" + }; + let mut request = + auth::authed_request_builder(&client, reqwest::Method::POST, path) + .map_err(|_| ProbeError::Failed)?; + if let Some(id) = organization.as_ref().or(scope.organization.as_ref()) { + request = request.query(&[("orgId", id)]); + } + let response = request + .header("Content-Type", "application/octet-stream") + .timeout(Duration::from_secs(30)) + .body(payload.clone()) + .send() + .await + .map_err(|_| ProbeError::Failed)?; + if !response.status().is_success() { + return Err(ProbeError::Failed); + } + let response = response + .json::() + .await + .map_err(|_| ProbeError::Failed)?; + if response.bytes != payload.len() + || response.organization_id.is_empty() + || organization + .as_ref() + .is_some_and(|id| *id != response.organization_id) + { + return Err(ProbeError::Failed); + } + organization = Some(response.organization_id); + *sample = Sample { + bytes: response.bytes, + body_read_ms: response.body_read_ms, + }; + } + Ok(Measurement::from_samples( + scope.clone(), + organization.ok_or(ProbeError::Failed)?, + samples, + SystemTime::now(), + )) + }) + .await +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }; + + struct ActiveTransfer(Arc); + + impl Drop for ActiveTransfer { + fn drop(&mut self) { + self.0.store(false, Ordering::SeqCst); + } + } + + #[tokio::test] + async fn recording_waits_for_an_in_flight_transfer_to_be_dropped() { + let active = Arc::new(AtomicBool::new(false)); + let mut state = Controller::default(); + let ticket = state.admission.begin_probe().unwrap(); + let (abort, registration) = AbortHandle::new_pair(); + let (done, receiver) = flume::bounded(1); + let (started, ready) = flume::bounded(1); + state.running = Some(Running { + ticket, + abort, + done: receiver.clone(), + }); + let completion = Completion(done); + let transfer_active = active.clone(); + let task = tokio::spawn(async move { + let _completion = completion; + Abortable::new( + async move { + transfer_active.store(true, Ordering::SeqCst); + let _transfer = ActiveTransfer(transfer_active); + started.send(()).unwrap(); + std::future::pending::<()>().await; + }, + registration, + ) + .await + }); + ready.recv_async().await.unwrap(); + assert!(active.load(Ordering::SeqCst)); + state.admit_recording(); + tokio::time::timeout(Duration::from_secs(1), receiver.recv_async()) + .await + .unwrap() + .unwrap(); + assert!(!active.load(Ordering::SeqCst)); + assert!(!state.admission.accepts(ticket)); + assert_eq!(state.admission.begin_probe(), None); + assert!(task.await.unwrap().is_err()); + } + + #[test] + fn dropping_an_unpolled_probe_still_releases_recording_startup() { + let (done, receiver) = flume::bounded(1); + let completion = Completion(done); + let task = async move { + let _completion = completion; + std::future::pending::<()>().await; + }; + drop(task); + assert_eq!(receiver.try_recv(), Ok(())); + } + + fn test_scope() -> Scope { + Scope { + server: "https://cap.example".into(), + owner: "owner".into(), + organization: Some("org-1".into()), + } + } + + fn slow_measurement(scope: Scope, checked_at: SystemTime) -> Measurement { + let sample = Sample { + bytes: 2 * 1024 * 1024, + body_read_ms: 2. * 1024. * 1024. * 8. / (3. * 1000.), + }; + Measurement::from_samples(scope, "org-1".into(), [sample, sample], checked_at) + } + + #[tokio::test] + async fn background_transfer_aborts_an_in_flight_probe_and_blocks_new_ones() { + let active = Arc::new(AtomicBool::new(false)); + let mut state = Controller::default(); + let scope = test_scope(); + let t0 = SystemTime::now(); + let (_, ticket, registration, completion) = state + .begin_request(Some(scope.clone()), false, false, t0) + .unwrap(); + let receiver = state.running.as_ref().unwrap().done.clone(); + let (started, ready) = flume::bounded(1); + let transfer_active = active.clone(); + let task = tokio::spawn(async move { + let _completion = completion; + Abortable::new( + async move { + transfer_active.store(true, Ordering::SeqCst); + let _transfer = ActiveTransfer(transfer_active); + started.send(()).unwrap(); + std::future::pending::<()>().await; + }, + registration, + ) + .await + }); + ready.recv_async().await.unwrap(); + assert!(active.load(Ordering::SeqCst)); + state.admit_transfer(); + tokio::time::timeout(Duration::from_secs(1), receiver.recv_async()) + .await + .unwrap() + .unwrap(); + assert!(!active.load(Ordering::SeqCst)); + assert!(task.await.unwrap().is_err()); + assert!( + state.finish_request( + ticket, + &scope, + Some(&scope), + Err(ProbeError::Failed), + t0 + Duration::from_secs(1), + ) + ); + assert_eq!(state.attempted_at, None); + assert!( + state + .begin_request( + Some(scope.clone()), + false, + true, + t0 + Duration::from_secs(120), + ) + .is_none() + ); + state.release_transfer(); + assert!( + state + .begin_request(Some(scope), false, false, t0 + Duration::from_secs(2)) + .is_some() + ); + } + + #[test] + fn automatic_checks_refresh_before_measurement_ttl_expires_and_preserve_fresh_caps() { + let mut state = Controller::default(); + let scope = test_scope(); + let t0 = SystemTime::now(); + let (_, first_ticket, _, _completion) = state + .begin_request(Some(scope.clone()), false, false, t0) + .unwrap(); + let first_checked_at = t0 + Duration::from_secs(5); + assert!(state.finish_request( + first_ticket, + &scope, + Some(&scope), + Ok(slow_measurement(scope.clone(), first_checked_at)), + first_checked_at, + )); + assert!( + state + .begin_request( + Some(scope.clone()), + false, + false, + first_checked_at + AUTOMATIC_PROBE_INTERVAL - Duration::from_secs(1), + ) + .is_none() + ); + let refresh_start = first_checked_at + AUTOMATIC_PROBE_INTERVAL; + assert_eq!( + state + .measurement + .as_ref() + .and_then(|m| m.max_width_for(&scope, Some("org-1"), refresh_start)), + Some(1280) + ); + let (_, second_ticket, _, _completion) = state + .begin_request(Some(scope.clone()), false, false, refresh_start) + .unwrap(); + let refresh_failed_at = refresh_start + Duration::from_secs(5); + assert!(state.finish_request( + second_ticket, + &scope, + Some(&scope), + Err(ProbeError::Failed), + refresh_failed_at, + )); + assert_eq!( + state + .measurement + .as_ref() + .and_then(|m| m.max_width_for(&scope, Some("org-1"), refresh_failed_at)), + Some(1280) + ); + } +} diff --git a/apps/desktop-gpui/src/network_health/policy.rs b/apps/desktop-gpui/src/network_health/policy.rs new file mode 100644 index 00000000000..dcb6bde075b --- /dev/null +++ b/apps/desktop-gpui/src/network_health/policy.rs @@ -0,0 +1,322 @@ +use std::time::{Duration, SystemTime}; + +pub const MEASUREMENT_TTL: Duration = Duration::from_secs(15 * 60); +pub const AUTOMATIC_PROBE_INTERVAL: Duration = Duration::from_secs(10 * 60); + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Scope { + pub server: String, + pub owner: String, + pub organization: Option, +} + +#[derive(Clone, Copy, Debug)] +pub struct Sample { + pub bytes: usize, + pub body_read_ms: f64, +} + +impl Sample { + fn mbps(self) -> Option { + (self.bytes >= 1024 * 1024 + && self.bytes <= 8 * 1024 * 1024 + && self.body_read_ms.is_finite() + && self.body_read_ms > 0.) + .then(|| self.bytes as f64 * 8. / (self.body_read_ms * 1000.)) + } +} + +#[derive(Clone, Debug)] +pub struct Measurement { + pub scope: Scope, + pub organization: String, + pub checked_at: SystemTime, + pub mbps: Option, + pub max_width: Option, +} + +impl Measurement { + pub fn from_samples( + scope: Scope, + organization: String, + samples: [Sample; 2], + now: SystemTime, + ) -> Self { + let rates = samples[0].mbps().zip(samples[1].mbps()); + let mbps = rates.map(|(first, second)| first.max(second)); + let max_width = rates.and_then(|(first, second)| { + if samples.iter().any(|sample| sample.body_read_ms < 500.) + || first.max(second) > first.min(second) * 2. + { + return None; + } + match first.max(second) { + rate if rate < 5. => Some(1280), + rate if rate < 10. => Some(1920), + rate if rate < 20. => Some(2560), + _ => None, + } + }); + Self { + scope, + organization, + checked_at: now, + mbps, + max_width, + } + } + + pub fn fresh_for(&self, scope: &Scope, now: SystemTime) -> bool { + self.scope == *scope + && now + .duration_since(self.checked_at) + .is_ok_and(|age| age < MEASUREMENT_TTL) + } + + pub fn max_width_for( + &self, + scope: &Scope, + organization: Option<&str>, + now: SystemTime, + ) -> Option { + (self.fresh_for(scope, now) && organization == Some(self.organization.as_str())) + .then_some(self.max_width) + .flatten() + } +} + +#[derive(Default)] +pub struct Admission { + recording: bool, + transfers: usize, + generation: u64, +} + +impl Admission { + pub fn begin_probe(&mut self) -> Option { + if self.recording || self.transfers > 0 { + return None; + } + self.invalidate(); + Some(self.generation) + } + + pub fn accepts(&self, generation: u64) -> bool { + !self.recording && self.transfers == 0 && self.generation == generation + } + + pub fn admit_recording(&mut self) { + self.recording = true; + self.invalidate(); + } + + pub fn release_recording(&mut self) { + self.recording = false; + } + + pub fn admit_transfer(&mut self) { + self.transfers = self.transfers.saturating_add(1); + self.invalidate(); + } + + pub fn release_transfer(&mut self) { + self.transfers = self.transfers.saturating_sub(1); + } + + pub fn invalidate(&mut self) { + self.generation = self.generation.wrapping_add(1); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn scope() -> Scope { + Scope { + server: "https://cap.example".into(), + owner: "owner".into(), + organization: None, + } + } + + fn measure(first_mbps: f64, second_mbps: f64) -> Measurement { + let sample = |mbps| Sample { + bytes: 2 * 1024 * 1024, + body_read_ms: 2. * 1024. * 1024. * 8. / (mbps * 1000.), + }; + Measurement::from_samples( + scope(), + "org-1".into(), + [sample(first_mbps), sample(second_mbps)], + SystemTime::now(), + ) + } + + #[test] + fn only_repeatably_slow_samples_lower_resolution() { + assert_eq!(measure(3., 4.).max_width, Some(1280)); + assert_eq!(measure(6., 7.).max_width, Some(1920)); + assert_eq!(measure(12., 16.).max_width, Some(2560)); + assert_eq!(measure(20., 20.).max_width, None); + } + + #[test] + fn a_fast_or_inconsistent_sample_preserves_quality() { + assert_eq!(measure(3., 100.).max_width, None); + assert_eq!(measure(3., 9.).max_width, None); + assert_eq!(measure(100., 100.).max_width, None); + } + + #[test] + fn invalid_or_buffered_samples_never_apply_a_cap() { + for duration in [0., -1., f64::NAN, f64::INFINITY] { + let sample = Sample { + bytes: 2 * 1024 * 1024, + body_read_ms: duration, + }; + let result = + Measurement::from_samples(scope(), "org-1".into(), [sample; 2], SystemTime::now()); + assert_eq!(result.max_width, None); + assert_eq!(result.mbps, None); + } + let sample = Sample { + bytes: 100, + body_read_ms: 1000., + }; + assert_eq!( + Measurement::from_samples(scope(), "org-1".into(), [sample; 2], SystemTime::now()) + .max_width, + None + ); + } + + #[test] + fn threshold_boundaries_use_the_less_restrictive_tier() { + assert_eq!(measure(5., 5.).max_width, Some(1920)); + assert_eq!(measure(10., 10.).max_width, Some(2560)); + assert_eq!(measure(20., 20.).max_width, None); + } + + #[test] + fn stale_account_and_server_measurements_are_ignored() { + let result = measure(3., 3.); + assert!(result.fresh_for(&scope(), result.checked_at)); + assert!(!result.fresh_for(&scope(), result.checked_at + MEASUREMENT_TTL)); + let mut other = scope(); + other.owner = "other".into(); + assert!(!result.fresh_for(&other, result.checked_at)); + other = scope(); + other.server = "https://other.example".into(); + assert!(!result.fresh_for(&other, result.checked_at)); + } + + #[test] + fn automatic_refresh_interval_keeps_slow_connection_limits_fresh() { + assert!(AUTOMATIC_PROBE_INTERVAL < MEASUREMENT_TTL); + let first = measure(3., 3.); + let refreshed_at = first.checked_at + AUTOMATIC_PROBE_INTERVAL; + assert_eq!( + first.max_width_for(&scope(), Some("org-1"), refreshed_at), + Some(1280) + ); + let second = Measurement::from_samples( + scope(), + "org-1".into(), + [ + Sample { + bytes: 2 * 1024 * 1024, + body_read_ms: 2. * 1024. * 1024. * 8. / (3. * 1000.), + }, + Sample { + bytes: 2 * 1024 * 1024, + body_read_ms: 2. * 1024. * 1024. * 8. / (3. * 1000.), + }, + ], + refreshed_at + Duration::from_secs(30), + ); + for elapsed_mins in 0..=24 { + let now = first.checked_at + Duration::from_secs(elapsed_mins * 60); + let active = if now >= second.checked_at { + &second + } else { + &first + }; + assert_eq!( + active.max_width_for(&scope(), Some("org-1"), now), + Some(1280) + ); + } + } + + #[test] + fn recording_admission_rejects_in_flight_and_new_probe_results() { + let mut admission = Admission::default(); + let ticket = admission.begin_probe().unwrap(); + assert!(admission.accepts(ticket)); + admission.admit_recording(); + assert!(!admission.accepts(ticket)); + assert_eq!(admission.begin_probe(), None); + admission.release_recording(); + assert!(!admission.accepts(ticket)); + let next = admission.begin_probe().unwrap(); + assert!(admission.accepts(next)); + assert!(!admission.accepts(ticket)); + } + + #[test] + fn background_transfers_reject_in_flight_and_new_probe_results() { + let mut admission = Admission::default(); + let ticket = admission.begin_probe().unwrap(); + assert!(admission.accepts(ticket)); + admission.admit_transfer(); + admission.admit_transfer(); + assert!(!admission.accepts(ticket)); + assert_eq!(admission.begin_probe(), None); + admission.release_transfer(); + assert_eq!(admission.begin_probe(), None); + admission.release_transfer(); + assert!(!admission.accepts(ticket)); + let next = admission.begin_probe().unwrap(); + assert!(admission.accepts(next)); + } + + #[test] + fn storage_selection_and_actual_recording_destination_bound_the_cap() { + let result = measure(3., 3.); + assert_eq!( + result.max_width_for(&scope(), Some("org-1"), result.checked_at), + Some(1280) + ); + assert_eq!( + result.max_width_for(&scope(), Some("org-2"), result.checked_at), + None + ); + assert_eq!( + result.max_width_for(&scope(), None, result.checked_at), + None + ); + let mut other = scope(); + other.organization = Some("different-selection".into()); + assert!(!result.fresh_for(&other, result.checked_at)); + } + + #[test] + fn invalidation_rejects_late_results_without_entering_recording() { + let mut admission = Admission::default(); + let ticket = admission.begin_probe().unwrap(); + admission.invalidate(); + assert!(!admission.accepts(ticket)); + assert!(admission.begin_probe().is_some()); + } + + #[test] + fn moving_the_wall_clock_backwards_preserves_the_users_resolution() { + let result = measure(3., 3.); + let earlier = result + .checked_at + .checked_sub(Duration::from_secs(1)) + .unwrap(); + assert!(!result.fresh_for(&scope(), earlier)); + } +} diff --git a/apps/desktop-gpui/src/recording.rs b/apps/desktop-gpui/src/recording.rs index 1b7e1cabd45..a3708fb6dad 100644 --- a/apps/desktop-gpui/src/recording.rs +++ b/apps/desktop-gpui/src/recording.rs @@ -2284,6 +2284,7 @@ async fn start_internal( config: StartConfig, #[cfg(target_os = "linux")] attempt: Option, ) -> anyhow::Result { + crate::network_health::wait_for_probe().await; #[cfg(target_os = "linux")] validate_linux_camera_request( config.mode, @@ -2310,8 +2311,8 @@ async fn start_internal( .filter(|organization| !organization.is_empty()) .map(str::to_string); - let pre_created_video = if config.mode == RecordingMode::Instant { - let video = crate::upload::prepare_instant_upload( + let (pre_created_video, upload_organization_id) = if config.mode == RecordingMode::Instant { + let (video, organization) = crate::upload::prepare_instant_upload( matches!(config.target, ScreenCaptureTarget::CameraOnly), project_name, organization_id, @@ -2323,15 +2324,16 @@ async fn start_internal( if let Some(attempt) = &attempt { *attempt.0.video.lock().unwrap() = Some((video.id.clone(), project_dir.clone())); } - Some(video) + (Some(video), organization) } else { - None + (None, None) }; let result = start_attempt_with_upload( config, project_dir, pre_created_video.clone(), + upload_organization_id, #[cfg(target_os = "linux")] attempt.clone(), ) @@ -2380,6 +2382,7 @@ async fn start_attempt_with_upload( config: StartConfig, project_dir: PathBuf, pre_created_video: Option, + upload_organization_id: Option, #[cfg(target_os = "linux")] attempt: Option, ) -> anyhow::Result { #[cfg(target_os = "linux")] @@ -2535,6 +2538,11 @@ async fn start_attempt_with_upload( } else { cap_recording::FREE_INSTANT_MODE_MAX_RESOLUTION }; + let instant_max_resolution = if config.mode == RecordingMode::Instant { + crate::network_health::max_width(instant_max_resolution, upload_organization_id.as_deref()) + } else { + instant_max_resolution + }; #[cfg(target_os = "macos")] let excluded_windows = { diff --git a/apps/desktop-gpui/src/session.rs b/apps/desktop-gpui/src/session.rs index c32419e0b42..c4470572dba 100644 --- a/apps/desktop-gpui/src/session.rs +++ b/apps/desktop-gpui/src/session.rs @@ -493,8 +493,8 @@ struct SessionGlobal(Entity); impl Global for SessionGlobal {} impl RecordingSession { - pub fn init(cx: &mut App) -> Entity { - let session = cx.new(|_| Self { + fn new() -> Self { + Self { phase: Phase::Idle, active: None, error: None, @@ -538,7 +538,11 @@ impl RecordingSession { clean_control: CleanControlState::default(), #[cfg(target_os = "linux")] show_controls_after_pause: false, - }); + } + } + + pub fn init(cx: &mut App) -> Entity { + let session = cx.new(|_| Self::new()); cx.set_global(SessionGlobal(session.clone())); session } @@ -776,6 +780,7 @@ impl RecordingSession { self.clean_control.invalidate(); self.clean_control.uncertain = false; } + crate::network_health::recording_admitted(); self.phase = Phase::Starting; self.error = None; self.storage_warning = false; @@ -935,6 +940,8 @@ impl RecordingSession { } else { false }; + #[cfg(not(target_os = "linux"))] + let startup_joined = true; this.update(cx, |this, cx| { if this.recording_generation != generation || this.phase != Phase::Starting { if let Ok(Ok(active)) = result { @@ -955,39 +962,11 @@ impl RecordingSession { } Ok(Err(error)) => { tracing::error!("recording failed to start: {error:#}"); - this.error = Some(format!("{error:#}")); - this.abort_start_cue(); - this.countdown_remaining = None; - #[cfg(target_os = "linux")] - { - this.phase = if startup_joined { - Phase::Idle - } else { - Phase::Starting - }; - } - #[cfg(not(target_os = "linux"))] - { - this.phase = Phase::Idle; - } + this.fail_start(format!("{error:#}"), startup_joined); } Err(join_error) => { tracing::error!("recording start task died: {join_error}"); - this.error = Some("Recording task failed.".into()); - this.abort_start_cue(); - this.countdown_remaining = None; - #[cfg(target_os = "linux")] - { - this.phase = if startup_joined { - Phase::Idle - } else { - Phase::Starting - }; - } - #[cfg(not(target_os = "linux"))] - { - this.phase = Phase::Idle; - } + this.fail_start("Recording task failed.".into(), startup_joined); } } #[cfg(target_os = "linux")] @@ -1012,6 +991,20 @@ impl RecordingSession { .detach(); } + fn fail_start(&mut self, error: String, cleanup_joined: bool) { + self.error = Some(error); + self.abort_start_cue(); + self.countdown_remaining = None; + self.phase = if cleanup_joined { + Phase::Idle + } else { + Phase::Starting + }; + if cleanup_joined { + crate::network_health::recording_released(); + } + } + fn activate_primed(&mut self, cx: &mut Context) { if let Some(active) = self.primed.take() { self.activate(active, cx); @@ -1874,6 +1867,7 @@ impl RecordingSession { { self.failure_monitor = None; } + crate::network_health::recording_admitted(); self.phase = Phase::Starting; self.started_at = None; cx.notify(); @@ -2126,6 +2120,7 @@ impl RecordingSession { self.storage_monitor = None; self.countdown_remaining = None; self.phase = Phase::Idle; + crate::network_health::recording_released(); self.mic_muted = false; self.storage_warning = false; self.stopped_elapsed = None; @@ -2993,4 +2988,28 @@ mod studio_editor_presentation_tests { state.closed(2); assert_eq!(state, StudioEditorPresentation::Dismissed); } + + #[test] + fn failed_starts_release_upload_checks_only_after_capture_cleanup() { + let mut session = RecordingSession::new(); + for error in [ + "Missing device", + "Upload setup failed", + "Recording task failed.", + ] { + crate::network_health::recording_admitted(); + session.phase = Phase::Starting; + assert!(!crate::network_health::probe_admitted_for_test()); + session.fail_start(error.into(), true); + assert_eq!(session.phase, Phase::Idle); + assert_eq!(session.error.as_deref(), Some(error)); + assert!(crate::network_health::probe_admitted_for_test()); + } + crate::network_health::recording_admitted(); + session.fail_start("Cleanup unconfirmed".into(), false); + assert_eq!(session.phase, Phase::Starting); + assert!(!crate::network_health::probe_admitted_for_test()); + session.fail_start("Cleanup joined".into(), true); + assert!(crate::network_health::probe_admitted_for_test()); + } } diff --git a/apps/desktop-gpui/src/settings_pages.rs b/apps/desktop-gpui/src/settings_pages.rs index ae760ec0c4b..f06e85b221d 100644 --- a/apps/desktop-gpui/src/settings_pages.rs +++ b/apps/desktop-gpui/src/settings_pages.rs @@ -4510,6 +4510,7 @@ impl SettingsWindow { Ok((200, _)) => { this.s3_fetch(window, cx); this.info_dialog(tr::settings::integrations::s3::saved(), window, cx); + crate::network_health::invalidate(); } _ => { this.pages.s3.error = @@ -4549,6 +4550,7 @@ impl SettingsWindow { Ok((200, _)) => { this.s3_fetch(window, cx); this.info_dialog(tr::settings::integrations::s3::deleted(), window, cx); + crate::network_health::invalidate(); } _ => { this.pages.s3.error = @@ -4741,6 +4743,7 @@ impl SettingsWindow { this.pages.storage = Some(parsed); } if connected { + crate::network_health::invalidate(); this.pages.gdrive.waiting = false; this.gdrive_fetch_s3_presence(window, cx); } @@ -4856,7 +4859,10 @@ impl SettingsWindow { |this, result, window, cx| { this.pages.gdrive.set_active_pending = false; match result { - Ok((200, _)) => this.gdrive_refresh(window, cx), + Ok((200, _)) => { + crate::network_health::invalidate(); + this.gdrive_refresh(window, cx); + } _ => { this.pages.gdrive.error = Some( tr::settings::integrations::gdrive::set_active_failed().to_string(), @@ -4903,6 +4909,7 @@ impl SettingsWindow { window, cx, ); + crate::network_health::invalidate(); } _ => { this.pages.gdrive.error = Some( diff --git a/apps/desktop-gpui/src/settings_window.rs b/apps/desktop-gpui/src/settings_window.rs index 6cced2d66ce..cdf3285a9ac 100644 --- a/apps/desktop-gpui/src/settings_window.rs +++ b/apps/desktop-gpui/src/settings_window.rs @@ -3514,6 +3514,22 @@ impl SettingsWindow { ); let mut rows = vec![resolution.into_any_element()]; + if let Some(health) = crate::network_health::indicator() { + rows.push( + div() + .p(px(12.)) + .text_size(px(12.)) + .text_color(Hsla::from(if health.failed { + theme.red_11 + } else if health.limited { + theme.amber_11 + } else { + theme.gray_11 + })) + .child(health.label) + .into_any_element(), + ); + } if let Some(InstantQualityNotice::SaveFailed) = self.instant_quality_notice { rows.push( div() diff --git a/apps/desktop-gpui/src/upload.rs b/apps/desktop-gpui/src/upload.rs index ce2d1cd9f13..b40af3f6b9c 100644 --- a/apps/desktop-gpui/src/upload.rs +++ b/apps/desktop-gpui/src/upload.rs @@ -286,7 +286,7 @@ pub async fn prepare_instant_upload( camera_only: bool, project_name: String, organization_id: Option, -) -> Result { +) -> Result<(VideoUploadInfo, Option), String> { if store_auth_missing() { return Err("Please sign in to use instant recording".to_string()); } @@ -315,11 +315,14 @@ pub async fn prepare_instant_upload( let link = created.link(); let config = created.config; - Ok(VideoUploadInfo { - id: config.id.clone(), - link, - config, - }) + Ok(( + VideoUploadInfo { + id: config.id.clone(), + link, + config, + }, + created.organization_id, + )) } pub(crate) fn start_instant_upload( @@ -394,6 +397,7 @@ fn start_instant_upload_with_ownership( .map_err(|error| format!("Failed to start instant upload: {error}"))?, ); + let transfer = crate::network_health::acquire_transfer(); let upload_video = video.clone(); let upload_cancel = cancel.clone(); let upload_metadata_lock = metadata_lock.clone(); @@ -403,6 +407,7 @@ fn start_instant_upload_with_ownership( context, upload_reads, async move { + let _transfer = transfer; run_segment_upload( upload_video, project_path, @@ -590,6 +595,7 @@ async fn run_segment_upload( metadata_lock: Arc>, authorization: Option, ) -> Result<(), String> { + let _transfer = crate::network_health::acquire_transfer(); let result = upload_segments( &LiveSegmentTransport::default(), &video.id, @@ -1378,6 +1384,7 @@ async fn upload_exported_video_inner( cancel: Arc, defer_completion: bool, ) -> Result<(UploadResult, Option), String> { + let _transfer = crate::network_health::acquire_transfer(); check_export_cancelled(&cancel).map_err(|error| error.to_string())?; if store_auth_missing() { let _ = crate::store::set_auth(None); @@ -1638,6 +1645,7 @@ async fn create_or_get_video_with_mode( struct CreatedVideo { config: S3UploadMeta, share_url: Option, + organization_id: Option, } impl CreatedVideo { @@ -1654,6 +1662,8 @@ fn parse_created_video(body: &str) -> Result { id: String, #[serde(default)] share_url: Option, + #[serde(default, rename = "organizationId")] + organization_id: Option, } let response = serde_json::from_str::(body)?; @@ -1663,6 +1673,7 @@ fn parse_created_video(body: &str) -> Result { Ok(CreatedVideo { config: S3UploadMeta { id: response.id }, share_url, + organization_id: response.organization_id, }) } @@ -2735,6 +2746,19 @@ fn urlencoding(value: &str) -> String { mod tests { use super::*; + #[test] + fn created_video_accepts_legacy_and_organization_bound_responses() { + for organization in [None, Some("org-recording")] { + let mut response = json!({ "id": "video-recording" }); + if let Some(id) = organization { + response["organizationId"] = json!(id); + } + let created = parse_created_video(&response.to_string()).unwrap(); + assert_eq!(created.config.id, "video-recording"); + assert_eq!(created.organization_id.as_deref(), organization); + } + } + #[test] fn created_video_uses_a_custom_domain_share_link() { let created = parse_created_video( diff --git a/apps/desktop-gpui/src/upload/queue.rs b/apps/desktop-gpui/src/upload/queue.rs index 3879a94b74f..fa328f38268 100644 --- a/apps/desktop-gpui/src/upload/queue.rs +++ b/apps/desktop-gpui/src/upload/queue.rs @@ -698,9 +698,11 @@ impl Manager { request_context: upload.request_context.clone(), reads: upload.reads.clone(), }; + let transfer = crate::network_health::acquire_transfer(); let manager = self.clone(); let path = canonical.clone(); let task = tokio::spawn(async move { + let _transfer = transfer; manager.run_job(path, owned, state, cancelled).await; }); jobs.insert( @@ -723,6 +725,7 @@ impl Manager { mut state: UploadState, mut cancelled: tokio::sync::watch::Receiver, ) { + let _transfer = crate::network_health::acquire_transfer(); let context = upload.request_context.clone(); let reads = upload.reads.clone(); let result = std::panic::AssertUnwindSafe(with_upload_context(context, reads, async { diff --git a/apps/web/__tests__/unit/desktop-video-create.test.ts b/apps/web/__tests__/unit/desktop-video-create.test.ts index 5e06c118988..996ac5ac4a5 100644 --- a/apps/web/__tests__/unit/desktop-video-create.test.ts +++ b/apps/web/__tests__/unit/desktop-video-create.test.ts @@ -578,6 +578,7 @@ describe("GET /create", () => { const body = await response.json(); expect(body.id).toBeTruthy(); + expect(body.organizationId).toBe(orgValues?.id); }); it("does not provision when the user already belongs to an org", async () => { @@ -711,6 +712,7 @@ describe("GET /create", () => { ); expect(response.status).toBe(200); + expect((await response.json()).organizationId).toBe("org-1"); expect(mockDb.transaction).toHaveBeenCalledTimes(1); expect(insertedValues(schema.videos)).toMatchObject({ @@ -738,6 +740,7 @@ describe("GET /create", () => { const response = await app.request("https://cap.test/create?orgId=org-2"); expect(response.status).toBe(200); + expect((await response.json()).organizationId).toBe("org-2"); expect(mockDb.transaction).toHaveBeenCalledTimes(1); expect(insertedValues(schema.videos)).toMatchObject({ diff --git a/apps/web/__tests__/unit/google-drive-storage.test.ts b/apps/web/__tests__/unit/google-drive-storage.test.ts index 6b5b5cd3bf6..6e37c28d72f 100644 --- a/apps/web/__tests__/unit/google-drive-storage.test.ts +++ b/apps/web/__tests__/unit/google-drive-storage.test.ts @@ -2,6 +2,7 @@ import { createHash } from "node:crypto"; import { Organisation, Storage, User, Video } from "@cap/web-domain"; import { Cause, Effect, Exit, Layer, Option } from "effect"; import { afterEach, describe, expect, it, vi } from "vitest"; +import { checkUploadStorage } from "@/lib/upload-health"; vi.mock("@cap/env", () => ({ serverEnv: () => ({ @@ -1846,6 +1847,70 @@ const getStorageAccess = (harness: RepoHarness) => { ); }; +describe("Google Drive upload-health objects", () => { + it.each(["video", "userVideo"] as const)( + "probes %s storage without video IDs or recording folders", + async (layout) => { + const harness = makeRepoHarness(); + vi.spyOn(harness.repo, "getGoogleDriveConfig").mockReturnValue( + Effect.succeed(makeConfig(layout)), + ); + const reserve = harness.repo.reserveObject; + vi.spyOn(harness.repo, "reserveObject").mockImplementation((input) => { + if (input.videoId && input.videoId.length > 15) + return Effect.die(new Error("videoId exceeds database width")); + return reserve(input); + }); + let uploaded = false; + const fetchMock = vi.fn( + async (input: string | URL | Request, init?: RequestInit) => { + const url = String(input); + if (url.includes("/files/generateIds")) + return jsonResponse({ ids: ["probe-file"] }); + if (url.includes("uploadType=resumable") && init?.method === "POST") { + expect(requestBody(init)).toMatchObject({ + parents: ["root-folder"], + appProperties: { + capObjectKey: `${videoOwnerId}/.cap-upload-health/probe`, + }, + }); + return response(200, { Location: "https://upload.test/probe" }); + } + if (url === "https://upload.test/probe" && init?.method === "PUT") { + uploaded = true; + return response(200); + } + if (url.includes("/files/probe-file") && init?.method === "DELETE") + return response(204); + throw new Error( + `Unexpected Drive probe request: ${init?.method} ${url}`, + ); + }, + ); + vi.stubGlobal("fetch", fetchMock); + const access = await getStorageAccess(harness); + await Effect.runPromise( + checkUploadStorage({ + userId: videoOwnerId, + bytes: new Uint8Array([1, 2, 3]), + uuid: () => "probe", + put: (key, bytes) => access.putProbeObject(key, bytes), + remove: (key) => access.deleteObject(key), + onCleanupFailure: () => { + throw new Error("Cleanup failed"); + }, + }), + ); + expect(uploaded).toBe(true); + expect(harness.reservations).toHaveLength(1); + expect(harness.reservations[0]?.videoId).toBeNull(); + expect(harness.reservations[0]?.metadata?.videoId).toBeUndefined(); + expect(harness.objects.size).toBe(0); + expect(fetchMock).toHaveBeenCalledTimes(4); + }, + ); +}); + const getStorageObjectExit = async (harness: RepoHarness) => { const access = await getStorageAccess(harness); return Effect.runPromiseExit(access.getObject(resultKey)); diff --git a/apps/web/__tests__/unit/upload-health-route.test.ts b/apps/web/__tests__/unit/upload-health-route.test.ts new file mode 100644 index 00000000000..0c91e0042d8 --- /dev/null +++ b/apps/web/__tests__/unit/upload-health-route.test.ts @@ -0,0 +1,333 @@ +import { HttpApiBuilder, HttpServer } from "@effect/platform"; +import { Context, Effect, Layer } from "effect"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const state = vi.hoisted(() => ({ + writes: [] as string[], + deletes: [] as string[], + storageAvailable: true, + cleanupAvailable: true, + organizationAllowed: true, + cookieValid: false, + apiKeyValid: true, + rateChecks: 0, + selectedOrganization: "", + rateLimited: false, +})); + +vi.mock("@cap/web-backend", async () => { + const { Context } = await import("effect"); + return { Storage: Context.GenericTag("UploadHealthTestStorage") }; +}); + +vi.mock("@/lib/rate-limit", () => ({ + isRateLimited: async () => { + state.rateChecks++; + return state.rateLimited; + }, + RATE_LIMIT_IDS: { DESKTOP_UPLOAD_HEALTH: "rl_desktop_upload_health" }, +})); + +vi.mock("@/lib/desktop-recording-organization", async (importOriginal) => { + const { Organisation } = await import("@cap/web-domain"); + const original = + await importOriginal< + typeof import("@/lib/desktop-recording-organization") + >(); + return { + ...original, + getDesktopRecordingOrganization: async ( + _user: string, + requested?: string, + ) => + state.organizationAllowed + ? original.selectDesktopRecordingOrganization( + [{ id: "org-default" }, { id: "org-selected" }].map(({ id }) => ({ + id: Organisation.OrganisationId.make(id), + })), + requested ? Organisation.OrganisationId.make(requested) : undefined, + Organisation.OrganisationId.make("org-default"), + ) + : undefined, + }; +}); + +vi.mock("@cap/env", () => ({ + serverEnv: () => ({ WEB_URL: "https://cap.test" }), +})); +vi.mock("@cap/database/auth/auth-options", () => ({ + getServerSession: async () => + state.cookieValid ? { user: { id: "test-user" } } : null, +})); + +vi.mock("@/lib/server", async () => { + const storage = Context.GenericTag("UploadHealthTestStorage"); + const { HttpAuthMiddlewareLive } = await import( + "../../../../packages/web-backend/src/Auth" + ); + const { Database } = await import( + "../../../../packages/web-backend/src/Database" + ); + const user = { + id: "test-user", + email: "dev@example.test", + activeOrganizationId: "org-dashboard", + image: null, + }; + const database = { + use: (run: (db: unknown) => Promise) => + Effect.promise(() => + run({ + select: () => ({ + from: () => ({ + leftJoin: () => ({ + where: async () => (state.apiKeyValid ? [{ users: user }] : []), + }), + where: async () => [user], + }), + }), + }), + ), + }; + const authentication = HttpAuthMiddlewareLive.pipe( + Layer.provide( + Layer.succeed( + Database, + database as unknown as Context.Tag.Service, + ), + ), + ); + + const access = { + putProbeObject: (key: string) => + Effect.suspend(() => { + state.writes.push(key); + return state.storageAvailable + ? Effect.void + : Effect.fail("write failed"); + }), + deleteObject: (key: string) => + Effect.suspend(() => { + state.deletes.push(key); + return state.cleanupAvailable + ? Effect.void + : Effect.fail("delete failed"); + }), + }; + return { + apiToHandler: (live: Parameters[0]) => + HttpApiBuilder.toWebHandler( + live.pipe( + Layer.provide(authentication), + Layer.provide( + Layer.succeed(storage, { + getWritableAccessForUser: ( + _user: string, + organization: string, + ) => { + state.selectedOrganization = organization; + return Effect.succeed({ access }); + }, + }), + ), + Layer.provide(HttpServer.layerContext), + ), + ).handler, + }; +}); + +import { GET, POST } from "@/app/api/desktop/upload-health/route"; + +function request( + checkStorage = true, + body = new Uint8Array(1024), + token = "12345678-1234-1234-1234-123456789abc", +) { + return new Request( + `http://localhost/api/desktop/upload-health${checkStorage ? "?checkStorage=true" : ""}`, + { + method: "POST", + headers: { authorization: `Bearer ${token}` }, + body, + }, + ); +} + +let clock = 0; +beforeEach(() => { + clock += 60_000; + vi.spyOn(Date, "now").mockReturnValue(clock); + state.writes = []; + state.deletes = []; + state.storageAvailable = true; + state.cleanupAvailable = true; + state.organizationAllowed = true; + state.rateLimited = false; + state.cookieValid = false; + state.apiKeyValid = true; + state.rateChecks = 0; + state.selectedOrganization = ""; +}); + +describe("authenticated upload-health API", () => { + it("rejects unauthenticated probes without touching storage", async () => { + const response = await POST(request(true, new Uint8Array(1), "wrong")); + expect(response.status).toBe(401); + expect(state.writes).toEqual([]); + }); + + it("warms the connection without writing or deleting an object", async () => { + const response = await GET( + new Request("http://localhost/api/desktop/upload-health", { + headers: { + authorization: "Bearer 12345678-1234-1234-1234-123456789abc", + }, + }), + ); + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ ok: true }); + expect(state.writes).toEqual([]); + }); + + it("checks the selected storage under the user's prefix and returns body timing", async () => { + const response = await POST(request()); + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ + bytes: 1024, + bodyReadMs: 0, + organizationId: "org-default", + }); + expect(state.writes[0]).toMatch(/^test-user\/\.cap-upload-health\//); + expect(state.deletes).toEqual(state.writes); + }); + + it("measures subsequent samples without another storage write", async () => { + const response = await POST(request(false)); + expect(response.status).toBe(200); + expect(state.writes).toEqual([]); + }); + + it("reports no available organization without touching storage", async () => { + state.organizationAllowed = false; + const response = await POST(request()); + expect(response.status).toBe(503); + expect(state.writes).toEqual([]); + }); + + it("returns unavailable when the chosen storage cannot accept a write", async () => { + state.storageAvailable = false; + const response = await POST(request()); + expect(response.status).toBe(503); + }); + + it("rejects empty and oversized streamed bodies without a storage write", async () => { + expect((await POST(request(true, new Uint8Array()))).status).toBe(400); + expect( + (await POST(request(true, new Uint8Array(8 * 1024 * 1024 + 1)))).status, + ).toBe(413); + expect(state.writes).toEqual([]); + }); + + it("rejects an oversized Content-Length before reading or writing", async () => { + const probe = request(); + probe.headers.set("content-length", String(8 * 1024 * 1024 + 1)); + expect((await POST(probe)).status).toBe(413); + expect(state.writes).toEqual([]); + }); + + it("keeps upload availability successful when object cleanup fails", async () => { + state.cleanupAvailable = false; + vi.spyOn(console, "warn").mockImplementation(() => {}); + expect((await POST(request())).status).toBe(200); + expect(console.warn).toHaveBeenCalledOnce(); + }); + + it("rate-limits repeated authenticated probes", async () => { + for (let index = 0; index < 4; index++) { + expect((await POST(request(false))).status).toBe(200); + } + expect((await POST(request())).status).toBe(429); + expect(state.writes).toEqual([]); + }); + + it("rejects a distributed rate-limit decision before writing", async () => { + state.rateLimited = true; + expect((await POST(request())).status).toBe(429); + expect(state.writes).toEqual([]); + }); +}); + +describe("upload probe credential boundary", () => { + const cookieProbe = (headers: Record = {}) => { + state.cookieValid = true; + return new Request( + "https://cap.test/api/desktop/upload-health?checkStorage=true", + { + method: "POST", + headers: { + cookie: "next-auth.session-token=valid", + "content-type": "text/plain", + ...headers, + }, + body: "probe", + }, + ); + }; + it.each>([ + {}, + { origin: "https://attacker.test" }, + { origin: "null" }, + { "sec-fetch-site": "cross-site", origin: "https://attacker.test" }, + { "sec-fetch-site": "same-site", origin: "https://sub.cap.test" }, + { "sec-fetch-site": "none", origin: "https://cap.test" }, + { authorization: "Bearer invalid", origin: "https://attacker.test" }, + { authorization: "Bearer", origin: "https://attacker.test" }, + ])( + "rejects ambient cookie POSTs before rate limiting/storage: %j", + async (headers) => { + expect([401, 403]).toContain((await POST(cookieProbe(headers))).status); + expect(state.rateChecks).toBe(0); + expect(state.writes).toEqual([]); + expect(state.deletes).toEqual([]); + }, + ); + it.each>([ + { "sec-fetch-site": "same-origin" }, + { origin: "https://cap.test" }, + ])("allows legitimate same-origin cookies: %j", async (headers) => { + expect((await POST(cookieProbe(headers))).status).toBe(200); + expect(state.writes).toHaveLength(1); + }); + it("does not fall back to a cookie for an invalid API key", async () => { + state.apiKeyValid = false; + expect( + ( + await POST( + cookieProbe({ + authorization: "Bearer 00000000-0000-0000-0000-000000000000", + origin: "https://attacker.test", + }), + ) + ).status, + ).toBe(401); + expect(state.writes).toEqual([]); + }); + it("checks the saved desktop organization instead of the dashboard organization", async () => { + const probe = request(); + const url = new URL(probe.url); + url.searchParams.set("orgId", "org-selected"); + const response = await POST(new Request(url, probe)); + expect(response.status).toBe(200); + expect(state.selectedOrganization).toBe("org-selected"); + expect(await response.json()).toMatchObject({ + organizationId: "org-selected", + }); + }); + it("uses the default organization for a stale desktop selection", async () => { + const probe = request(); + const url = new URL(probe.url); + url.searchParams.set("orgId", "departed-org"); + expect((await POST(new Request(url, probe))).status).toBe(200); + expect(state.selectedOrganization).toBe("org-default"); + }); +}); diff --git a/apps/web/__tests__/unit/upload-health.test.ts b/apps/web/__tests__/unit/upload-health.test.ts new file mode 100644 index 00000000000..c6912eb93b6 --- /dev/null +++ b/apps/web/__tests__/unit/upload-health.test.ts @@ -0,0 +1,175 @@ +import { Data, Effect } from "effect"; +import { describe, expect, it, vi } from "vitest"; +import { + checkUploadStorage, + MAX_UPLOAD_PROBE_BYTES, + UploadProbeBody, + UploadProbeLimiter, + UploadProbeTooLarge, +} from "@/lib/upload-health"; + +class TestStorageError extends Data.TaggedError("TestStorageError")<{ + message: string; +}> {} + +describe("upload measurement", () => { + it("times body transfer separately from cold-start and storage latency", () => { + let now = 12_000; + const probe = new UploadProbeBody(() => now); + now += 4000; + probe.append(new Uint8Array(1024)); + now += 800; + probe.append(new Uint8Array(1024)); + now += 30_000; + expect(probe.result()).toEqual({ bytes: 2048, bodyReadMs: 800 }); + }); + + it("reports buffered or empty bodies as inconclusive timing", () => { + const probe = new UploadProbeBody(() => 10); + expect(probe.result()).toEqual({ bytes: 0, bodyReadMs: 0 }); + probe.append(new Uint8Array(1024)); + expect(probe.result()).toEqual({ bytes: 1024, bodyReadMs: 0 }); + }); + + it("bounds streamed bodies even when Content-Length is absent or false", () => { + const probe = new UploadProbeBody(); + probe.append(new Uint8Array(MAX_UPLOAD_PROBE_BYTES)); + expect(() => probe.append(new Uint8Array(1))).toThrow(UploadProbeTooLarge); + expect(probe.result().bytes).toBe(MAX_UPLOAD_PROBE_BYTES); + expect(probe.sample().byteLength).toBe(1024); + }); + + it("does not retain the upload or an alias of its storage sample", () => { + const probe = new UploadProbeBody(); + const bytes = new Uint8Array(2048).fill(7); + probe.append(bytes); + bytes.fill(0); + expect(probe.sample()).toEqual(new Uint8Array(1024).fill(7)); + }); +}); + +describe("selected-storage health check", () => { + const fixture = () => ({ + userId: "owner", + bytes: new Uint8Array([1, 2, 3]), + put: vi.fn((): Effect.Effect => Effect.void), + remove: vi.fn((): Effect.Effect => Effect.void), + onCleanupFailure: vi.fn(), + uuid: () => "test-id", + }); + + it("writes under the authenticated user's prefix and cleans up", async () => { + const deps = fixture(); + await Effect.runPromise(checkUploadStorage(deps)); + expect(deps.put).toHaveBeenCalledWith( + "owner/.cap-upload-health/test-id", + deps.bytes, + ); + expect(deps.remove).toHaveBeenCalledWith( + "owner/.cap-upload-health/test-id", + ); + }); + + it("keeps a successful write healthy when cleanup fails", async () => { + const deps = fixture(); + deps.remove.mockReturnValue( + Effect.fail(new TestStorageError({ message: "delete denied" })), + ); + await expect( + Effect.runPromise(checkUploadStorage(deps)), + ).resolves.toBeUndefined(); + expect(deps.onCleanupFailure).toHaveBeenCalledOnce(); + }); + + it("reports an unavailable storage destination and still attempts cleanup", async () => { + const deps = fixture(); + deps.put.mockReturnValue( + Effect.fail(new TestStorageError({ message: "write failed" })), + ); + await expect(Effect.runPromise(checkUploadStorage(deps))).rejects.toThrow( + "write failed", + ); + expect(deps.remove).toHaveBeenCalledOnce(); + }); +}); + +describe("storage effect execution", () => { + it("executes cleanup after the write effect fails", async () => { + const events: string[] = []; + await expect( + Effect.runPromise( + checkUploadStorage({ + userId: "owner", + bytes: new Uint8Array([1]), + put: () => + Effect.sync(() => events.push("write")).pipe( + Effect.andThen( + Effect.fail(new TestStorageError({ message: "failed" })), + ), + ), + remove: () => + Effect.sync(() => { + events.push("delete"); + }), + onCleanupFailure: () => {}, + }), + ), + ).rejects.toThrow("failed"); + expect(events).toEqual(["write", "delete"]); + }); + + it("aborts in-flight promise writes before running cleanup on timeout", async () => { + vi.useFakeTimers(); + try { + const events: string[] = []; + const promise = Effect.runPromise( + checkUploadStorage({ + userId: "owner", + bytes: new Uint8Array([1]), + put: () => + Effect.tryPromise({ + try: (signal) => + new Promise((_resolve, reject) => { + events.push("write"); + signal.addEventListener("abort", () => { + events.push("abort"); + reject(signal.reason); + }); + }), + catch: (cause) => + new TestStorageError({ message: String(cause) }), + }), + remove: () => + Effect.sync(() => { + events.push("delete"); + }), + onCleanupFailure: () => {}, + }), + ); + const assertion = expect(promise).rejects.toThrow(); + await vi.advanceTimersByTimeAsync(10_000); + await assertion; + expect(events).toEqual(["write", "abort", "delete"]); + } finally { + vi.useRealTimers(); + } + }); +}); + +describe("upload probe rate limit", () => { + it("allows two samples and a retry, then rejects repeated checks", () => { + const limiter = new UploadProbeLimiter(); + for (let i = 0; i < 4; i++) expect(limiter.admit("owner", 0)).toBe(true); + expect(limiter.admit("owner", 59_999)).toBe(false); + expect(limiter.admit("other", 59_999)).toBe(true); + expect(limiter.admit("owner", 60_000)).toBe(true); + }); + + it("bounds its per-process state and expires old entries", () => { + const limiter = new UploadProbeLimiter(); + for (let i = 0; i < 1024; i++) + expect(limiter.admit(String(i), 0)).toBe(true); + expect(limiter.admit("extra", 0)).toBe(false); + expect(limiter.admit("extra", 60_000)).toBe(true); + }); +}); diff --git a/apps/web/app/api/desktop/[...route]/video.ts b/apps/web/app/api/desktop/[...route]/video.ts index e4cdf55fa4f..8a850585a3c 100644 --- a/apps/web/app/api/desktop/[...route]/video.ts +++ b/apps/web/app/api/desktop/[...route]/video.ts @@ -22,6 +22,10 @@ import { Hono } from "hono"; import { after } from "next/server"; import { z } from "zod"; import { readInstantCleanupReport } from "@/lib/desktop-instant-cleanup"; +import { + mergeUserOrganizationSelections, + selectDesktopRecordingOrganization, +} from "@/lib/desktop-recording-organization"; import { getDesktopShareUrl } from "@/lib/desktop-share-url"; import { invalidateGoogleDriveStorageQuotaCache } from "@/lib/google-drive-storage-quota"; import { maybeStartLiveTranscription } from "@/lib/live-transcribe"; @@ -38,28 +42,6 @@ import { withAuth } from "../../utils"; export const app = new Hono().use(withAuth); -type UserOrganizationSelection = { - id: Organisation.OrganisationId; - name: string; - createdAt: Date; -}; - -function mergeUserOrganizationSelections( - ...rowSets: UserOrganizationSelection[][] -) { - const organizationsById = new Map(); - - for (const rows of rowSets) { - for (const row of rows) { - organizationsById.set(row.id, row); - } - } - - return Array.from(organizationsById.values()) - .sort((a, b) => a.createdAt.getTime() - b.createdAt.getTime()) - .map(({ createdAt, ...organization }) => organization); -} - app.get("/new-id", (c) => c.json({ id: nanoId() })); app.get( @@ -243,35 +225,22 @@ app.get( userOrganizations.push({ id: organizationId, name: "My Organization" }); } - const userOrgIds = userOrganizations.map((org) => org.id); - - let videoOrgId: Organisation.OrganisationId; - // Desktop persists orgId in settings and keeps sending it after the - // user leaves the org, so an unknown orgId falls back to the - // default/first org below instead of hard-failing every upload. - if (orgId && userOrgIds.includes(orgId)) { - videoOrgId = orgId; - } else if (user.defaultOrgId) { - // User's defaultOrgId is no longer valid, switch to first available org - if (!userOrgIds.includes(user.defaultOrgId)) { - if (!userOrganizations[0]) - return c.json({ error: "no_valid_org" }, { status: 403 }); - - videoOrgId = userOrganizations[0].id; - - // Update user's defaultOrgId to the new valid org - await db() - .update(users) - .set({ - defaultOrgId: videoOrgId, - }) - .where(eq(users.id, user.id)); - } else videoOrgId = user.defaultOrgId; - } else { - // No orgId provided and no defaultOrgId, use first available org - if (!userOrganizations[0]) - return c.json({ error: "no_valid_org" }, { status: 403 }); - videoOrgId = userOrganizations[0].id; + const videoOrgId = selectDesktopRecordingOrganization( + userOrganizations, + orgId, + user.defaultOrgId, + ); + if (!videoOrgId) + return c.json({ error: "no_valid_org" }, { status: 403 }); + if ( + orgId !== videoOrgId && + user.defaultOrgId && + !userOrganizations.some((org) => org.id === user.defaultOrgId) + ) { + await db() + .update(users) + .set({ defaultOrgId: videoOrgId }) + .where(eq(users.id, user.id)); } const idToUse = Video.VideoId.make( @@ -395,6 +364,7 @@ app.get( return c.json({ id: idToUse, + organizationId: videoOrgId, share_url: await getDesktopShareUrl(videoOrgId, idToUse), // All deprecated user_id: user.id, diff --git a/apps/web/app/api/desktop/upload-health/route.ts b/apps/web/app/api/desktop/upload-health/route.ts new file mode 100644 index 00000000000..725c997c74b --- /dev/null +++ b/apps/web/app/api/desktop/upload-health/route.ts @@ -0,0 +1,166 @@ +import { serverEnv } from "@cap/env"; +import { Storage } from "@cap/web-backend"; +import { CurrentUser, HttpAuthMiddleware, Organisation } from "@cap/web-domain"; +import { + HttpApi, + HttpApiBuilder, + HttpApiEndpoint, + HttpApiError, + HttpApiGroup, + HttpApiSchema, + HttpServerRequest, +} from "@effect/platform"; +import { Effect, Layer, Schema, Stream } from "effect"; +import { getDesktopRecordingOrganization } from "@/lib/desktop-recording-organization"; +import { isRateLimited, RATE_LIMIT_IDS } from "@/lib/rate-limit"; +import { apiToHandler } from "@/lib/server"; +import { + checkUploadStorage, + MAX_UPLOAD_PROBE_BYTES, + UploadProbeBody, + UploadProbeLimiter, + UploadProbeTooLarge, +} from "@/lib/upload-health"; + +class ProbeTooLarge extends HttpApiSchema.EmptyError()({ + tag: "ProbeTooLarge", + status: 413, +}) {} + +class ProbeRateLimited extends HttpApiSchema.EmptyError()({ + tag: "ProbeRateLimited", + status: 429, +}) {} + +class Api extends HttpApi.make("DesktopUploadHealth").add( + HttpApiGroup.make("root") + .add( + HttpApiEndpoint.get("warm")`/api/desktop/upload-health`.addSuccess( + Schema.Struct({ ok: Schema.Literal(true) }), + ), + ) + .add( + HttpApiEndpoint.post("probe")`/api/desktop/upload-health` + .setUrlParams( + Schema.Struct({ + checkStorage: Schema.optional(Schema.Literal("true")), + orgId: Schema.optional(Organisation.OrganisationId), + }), + ) + .addSuccess( + Schema.Struct({ + bytes: Schema.Number, + bodyReadMs: Schema.Number, + organizationId: Organisation.OrganisationId, + }), + ) + .addError(ProbeTooLarge) + .addError(ProbeRateLimited) + .addError(HttpApiError.BadRequest) + .addError(HttpApiError.Forbidden) + .addError(HttpApiError.ServiceUnavailable) + .addError(HttpApiError.RequestTimeout), + ) + .middleware(HttpAuthMiddleware), +) {} + +const limiter = new UploadProbeLimiter(); + +const ApiLive = HttpApiBuilder.api(Api).pipe( + Layer.provide( + HttpApiBuilder.group(Api, "root", (handlers) => + handlers + .handle("warm", () => Effect.succeed({ ok: true as const })) + .handle("probe", ({ urlParams }) => + Effect.gen(function* () { + const user = yield* CurrentUser; + const request = yield* HttpServerRequest.HttpServerRequest; + const apiKey = request.headers.authorization?.split(" ")[1]; + if (apiKey?.length !== 36) { + const site = request.headers["sec-fetch-site"]; + if ( + site !== undefined + ? site !== "same-origin" + : request.headers.origin !== + new URL(serverEnv().WEB_URL).origin + ) { + return yield* new HttpApiError.Forbidden(); + } + } + if ( + !limiter.admit(user.id) || + (yield* Effect.promise(() => + isRateLimited(RATE_LIMIT_IDS.DESKTOP_UPLOAD_HEALTH, { + key: user.id, + headers: new Headers(request.headers), + }), + )) + ) { + return yield* new ProbeRateLimited(); + } + if ( + Number(request.headers["content-length"]) > MAX_UPLOAD_PROBE_BYTES + ) { + return yield* new ProbeTooLarge(); + } + const probe = new UploadProbeBody(); + yield* request.stream.pipe( + Stream.runForEach((chunk) => + Effect.try({ + try: () => probe.append(chunk), + catch: (error) => + error instanceof UploadProbeTooLarge + ? new ProbeTooLarge() + : new HttpApiError.BadRequest(), + }), + ), + Effect.catchTag( + "RequestError", + () => new HttpApiError.BadRequest(), + ), + ); + const result = probe.result(); + if (!result.bytes) return yield* new HttpApiError.BadRequest(); + const organizationId = yield* Effect.tryPromise({ + try: () => + getDesktopRecordingOrganization(user.id, urlParams.orgId), + catch: () => new HttpApiError.ServiceUnavailable(), + }); + if (!organizationId) + return yield* new HttpApiError.ServiceUnavailable(); + if (urlParams.checkStorage) { + const storage = yield* Storage; + const { access } = yield* storage + .getWritableAccessForUser(user.id, organizationId) + .pipe( + Effect.mapError(() => new HttpApiError.ServiceUnavailable()), + ); + yield* checkUploadStorage({ + userId: user.id, + bytes: probe.sample(), + put: (key, bytes) => + access.putProbeObject(key, Buffer.from(bytes)), + remove: (key) => access.deleteObject(key), + onCleanupFailure: () => + console.warn("Upload-health probe cleanup failed"), + }).pipe( + Effect.mapError(() => new HttpApiError.ServiceUnavailable()), + ); + } + return { ...result, organizationId }; + }).pipe( + Effect.timeoutFail({ + duration: "25 seconds", + onTimeout: () => new HttpApiError.RequestTimeout(), + }), + ), + ), + ), + ), +); + +const handler = apiToHandler(ApiLive); +export const GET = handler; +export const POST = handler; +export const OPTIONS = handler; +export const maxDuration = 30; diff --git a/apps/web/lib/desktop-recording-organization.ts b/apps/web/lib/desktop-recording-organization.ts new file mode 100644 index 00000000000..25690027988 --- /dev/null +++ b/apps/web/lib/desktop-recording-organization.ts @@ -0,0 +1,87 @@ +import { db } from "@cap/database"; +import { + organizationMembers, + organizations, + users, +} from "@cap/database/schema"; +import type { Organisation, User } from "@cap/web-domain"; +import { and, eq, isNull } from "drizzle-orm"; + +type UserOrganizationSelection = { + id: Organisation.OrganisationId; + name: string; + createdAt: Date; +}; + +export function mergeUserOrganizationSelections( + ...rowSets: UserOrganizationSelection[][] +) { + const byId = new Map(); + for (const rows of rowSets) { + for (const row of rows) byId.set(row.id, row); + } + return Array.from(byId.values()) + .sort((a, b) => a.createdAt.getTime() - b.createdAt.getTime()) + .map(({ createdAt, ...organization }) => organization); +} + +export function selectDesktopRecordingOrganization( + available: Array<{ id: Organisation.OrganisationId }>, + requested?: Organisation.OrganisationId, + defaultId?: Organisation.OrganisationId | null, +) { + return ( + [requested, defaultId].find( + (id): id is Organisation.OrganisationId => + !!id && available.some((organization) => organization.id === id), + ) ?? available[0]?.id + ); +} + +export async function getDesktopRecordingOrganization( + userId: User.UserId, + requested?: Organisation.OrganisationId, +) { + const [[user], owned, member] = await Promise.all([ + db() + .select({ defaultOrgId: users.defaultOrgId }) + .from(users) + .where(eq(users.id, userId)) + .limit(1), + db() + .select({ + id: organizations.id, + name: organizations.name, + createdAt: organizations.createdAt, + }) + .from(organizations) + .where( + and( + eq(organizations.ownerId, userId), + isNull(organizations.tombstoneAt), + ), + ), + db() + .select({ + id: organizations.id, + name: organizations.name, + createdAt: organizations.createdAt, + }) + .from(organizationMembers) + .innerJoin( + organizations, + eq(organizations.id, organizationMembers.organizationId), + ) + .where( + and( + eq(organizationMembers.userId, userId), + isNull(organizations.tombstoneAt), + ), + ), + ]); + return selectDesktopRecordingOrganization( + mergeUserOrganizationSelections(owned, member), + requested, + user?.defaultOrgId, + ); +} diff --git a/apps/web/lib/rate-limit.ts b/apps/web/lib/rate-limit.ts index d0a72451e8d..84e1842574e 100644 --- a/apps/web/lib/rate-limit.ts +++ b/apps/web/lib/rate-limit.ts @@ -164,6 +164,8 @@ export const RATE_LIMIT_IDS = { GUEST_CHECKOUT: "rl_guest_checkout", /** Unauthed desktop log → Discord forwarding (spam). Suggested: 10 / 1m per IP. */ DESKTOP_LOGS: "rl_desktop_logs", + /** Authenticated desktop upload health probes. Suggested: 4 / 1m per key (user). */ + DESKTOP_UPLOAD_HEALTH: "rl_desktop_upload_health", /** Mobile share password attempts (brute-force guard). Suggested: 20 / 10m per key (user). */ MOBILE_VIDEO_PASSWORD: "rl_mobile_video_password", /** Unauthed docs Ask AI questions (Anthropic cost). Suggested: 10 / 1m per IP. */ @@ -206,6 +208,7 @@ const FALLBACK_RATE_LIMITS: Record = { [RATE_LIMIT_IDS.ANALYTICS_TRACK]: { limit: 120, windowMs: MINUTE_MS }, [RATE_LIMIT_IDS.GUEST_CHECKOUT]: { limit: 20, windowMs: 10 * MINUTE_MS }, [RATE_LIMIT_IDS.DESKTOP_LOGS]: { limit: 10, windowMs: MINUTE_MS }, + [RATE_LIMIT_IDS.DESKTOP_UPLOAD_HEALTH]: { limit: 4, windowMs: MINUTE_MS }, [RATE_LIMIT_IDS.MOBILE_VIDEO_PASSWORD]: { limit: 20, windowMs: 10 * MINUTE_MS, diff --git a/apps/web/lib/upload-health.ts b/apps/web/lib/upload-health.ts new file mode 100644 index 00000000000..8377e36a0ce --- /dev/null +++ b/apps/web/lib/upload-health.ts @@ -0,0 +1,86 @@ +import { Effect } from "effect"; + +export const MAX_UPLOAD_PROBE_BYTES = 8 * 1024 * 1024; + +export class UploadProbeTooLarge extends Error {} + +export class UploadProbeBody { + private startedAt: number | undefined; + private finishedAt: number | undefined; + private bytes = 0; + private storageSample = new Uint8Array(); + + constructor(private readonly now: () => number = () => performance.now()) {} + + append(chunk: Uint8Array) { + if (this.bytes + chunk.byteLength > MAX_UPLOAD_PROBE_BYTES) { + throw new UploadProbeTooLarge(); + } + if (!chunk.byteLength) return; + const now = this.now(); + this.startedAt ??= now; + this.finishedAt = now; + if (!this.bytes) this.storageSample = chunk.slice(0, 1024); + this.bytes += chunk.byteLength; + } + + result() { + return { + bytes: this.bytes, + bodyReadMs: Math.max(0, (this.finishedAt ?? 0) - (this.startedAt ?? 0)), + }; + } + + sample() { + return this.storageSample; + } +} + +export function checkUploadStorage({ + userId, + bytes, + put, + remove, + onCleanupFailure, + uuid = () => crypto.randomUUID(), +}: { + userId: string; + bytes: Uint8Array; + put: (key: string, bytes: Uint8Array) => Effect.Effect; + remove: (key: string) => Effect.Effect; + onCleanupFailure: (error: unknown) => void; + uuid?: () => string; +}) { + const key = `${userId}/.cap-upload-health/${uuid()}`; + return put(key, bytes).pipe( + Effect.timeout("10 seconds"), + Effect.ensuring( + remove(key).pipe( + Effect.timeout("3 seconds"), + Effect.catchAll((error) => Effect.sync(() => onCleanupFailure(error))), + ), + ), + ); +} + +export class UploadProbeLimiter { + private readonly windows = new Map< + string, + { until: number; count: number } + >(); + + admit(userId: string, now = Date.now()) { + for (const [key, window] of this.windows) { + if (window.until <= now) this.windows.delete(key); + } + const window = this.windows.get(userId); + if (window) { + if (window.count >= 4) return false; + window.count++; + return true; + } + if (this.windows.size >= 1024) return false; + this.windows.set(userId, { until: now + 60_000, count: 1 }); + return true; + } +} diff --git a/packages/web-backend/src/S3Buckets/S3BucketAccess.ts b/packages/web-backend/src/S3Buckets/S3BucketAccess.ts index ea28e837d8c..55592606341 100644 --- a/packages/web-backend/src/S3Buckets/S3BucketAccess.ts +++ b/packages/web-backend/src/S3Buckets/S3BucketAccess.ts @@ -324,39 +324,44 @@ export const createS3BucketAccess = Effect.gen(function* () { ifNoneMatch?: string; }, ) => - wrapS3Promise( - provider.getInternal.pipe( - Effect.flatMap((client) => - Effect.gen(function* () { - let _body: S3.PutObjectCommandInput["Body"]; - - if (typeof body === "string" || body instanceof Uint8Array) { - _body = body; - } else if (body instanceof ArrayBuffer) { - _body = new Uint8Array(body); - } else { - _body = body.pipe( - Stream.toReadableStreamRuntime(yield* Effect.runtime()), - (s) => - Readable.fromWeb(s as unknown as NodeReadableWebStream), - ); - } + provider.getInternal.pipe( + Effect.flatMap((client) => + Effect.gen(function* () { + let _body: S3.PutObjectCommandInput["Body"]; - return client.send( - new S3.PutObjectCommand({ - Bucket: provider.bucket, - Key: key, - Body: _body, - ContentType: fields?.contentType, - ContentLength: fields?.contentLength, - IfMatch: fields?.ifMatch, - IfNoneMatch: fields?.ifNoneMatch, - }), + if (typeof body === "string" || body instanceof Uint8Array) { + _body = body; + } else if (body instanceof ArrayBuffer) { + _body = new Uint8Array(body); + } else { + _body = body.pipe( + Stream.toReadableStreamRuntime(yield* Effect.runtime()), + (s) => + Readable.fromWeb(s as unknown as NodeReadableWebStream), ); - }), - ), + } + + return yield* Effect.tryPromise({ + try: (signal) => + client.send( + new S3.PutObjectCommand({ + Bucket: provider.bucket, + Key: key, + Body: _body, + ContentType: fields?.contentType, + ContentLength: fields?.contentLength, + IfMatch: fields?.ifMatch, + IfNoneMatch: fields?.ifNoneMatch, + }), + { abortSignal: signal }, + ), + catch: (cause) => new S3Error({ cause }), + }).pipe(Effect.tapError(Effect.logError)); + }), + ), + Effect.catchTag("UnknownException", (cause) => + Effect.fail(new S3Error({ cause })), ), - ).pipe( Effect.withSpan("S3BucketAccess.putObject", { attributes: { key } }), ), /** Copy an object within the same bucket */ diff --git a/packages/web-backend/src/Storage/GoogleDrive.ts b/packages/web-backend/src/Storage/GoogleDrive.ts index 03cfebbc8ee..a832e10abcd 100644 --- a/packages/web-backend/src/Storage/GoogleDrive.ts +++ b/packages/web-backend/src/Storage/GoogleDrive.ts @@ -92,6 +92,7 @@ export type CreateGoogleDriveUploadInput = { contentType: string; contentLength?: number; videoTitle?: string; + skipRecordingFolders?: boolean; }; const normalizeContentType = (contentType?: string | null) => @@ -492,10 +493,10 @@ const sendDriveRequest = ( init?: RequestInit, ) => Effect.tryPromise({ - try: () => { + try: (signal) => { const headers = new Headers(init?.headers); headers.set("Authorization", `Bearer ${accessToken}`); - return fetch(url, { ...init, headers }); + return fetch(url, { ...init, headers, signal: init?.signal ?? signal }); }, catch: (cause) => new Storage.StorageError({ cause }), }); @@ -1100,6 +1101,7 @@ const getGoogleDriveUploadParentId = ( tokenStore?: GoogleDriveTokenStore, ) => Effect.gen(function* () { + if (input.skipRecordingFolders) return config.folderId; const folderParts = getDriveFolderParts(input.key, config); let parentId = config.folderId; const pathParts: string[] = []; diff --git a/packages/web-backend/src/Storage/index.ts b/packages/web-backend/src/Storage/index.ts index aa6c128fe0a..d86aac1ecc0 100644 --- a/packages/web-backend/src/Storage/index.ts +++ b/packages/web-backend/src/Storage/index.ts @@ -470,6 +470,10 @@ const makeS3Access = (s3: S3BucketAccess) => ({ body: Parameters[1], fields?: Parameters[2], ) => mapStorageError(s3.putObject(key, body, fields)).pipe(Effect.asVoid), + putProbeObject: (key: string, body: Uint8Array) => + mapStorageError( + s3.putObject(key, body, { contentType: "application/octet-stream" }), + ).pipe(Effect.asVoid), putObjectWithETag: ( key: string, body: string, @@ -830,6 +834,7 @@ const makeGoogleDriveAccess = ({ key: string, body: string | Uint8Array | ArrayBuffer, fields?: { contentType?: string; contentLength?: number }, + probe = false, ) => Effect.gen(function* () { const contentType = fields?.contentType ?? "application/octet-stream"; @@ -844,10 +849,13 @@ const makeGoogleDriveAccess = ({ { integrationId, ownerId, - videoId: parseVideoIdFromObjectKey(key).pipe( - Option.map((id) => id as Video.VideoId), - Option.getOrNull, - ), + videoId: probe + ? null + : parseVideoIdFromObjectKey(key).pipe( + Option.map((id) => id as Video.VideoId), + Option.getOrNull, + ), + skipRecordingFolders: probe, key, contentType, contentLength, @@ -857,11 +865,12 @@ const makeGoogleDriveAccess = ({ const uploadBody = body instanceof Uint8Array ? new Uint8Array(body).buffer : body; const response = yield* Effect.tryPromise({ - try: () => + try: (signal) => fetch(uploadUrl, { method: "PUT", headers: getGoogleDriveUploadHeaders(contentType, contentLength), body: uploadBody, + signal, }), catch: (cause) => new StorageDomain.StorageError({ cause }), }); @@ -985,6 +994,8 @@ const makeGoogleDriveAccess = ({ body: string | Uint8Array | ArrayBuffer, fields?: { contentType?: string; contentLength?: number }, ) => putObjectWithChecksum(key, body, fields).pipe(Effect.asVoid), + putProbeObject: (key: string, body: Uint8Array) => + putObjectWithChecksum(key, body, undefined, true).pipe(Effect.asVoid), putObjectWithChecksum, copyObject, copyObjectForRecording,