diff --git a/crates/yaak/Cargo.toml b/crates/yaak/Cargo.toml index d24bc012..b0af1b57 100644 --- a/crates/yaak/Cargo.toml +++ b/crates/yaak/Cargo.toml @@ -21,3 +21,4 @@ yaak-tls = { workspace = true } [dev-dependencies] tempfile = "3" +tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } diff --git a/crates/yaak/src/send.rs b/crates/yaak/src/send.rs index 7f00fbfe..3c8b52e7 100644 --- a/crates/yaak/src/send.rs +++ b/crates/yaak/src/send.rs @@ -24,9 +24,9 @@ use yaak_http::types::{ }; use yaak_models::blob_manager::{BlobManager, BodyChunk}; use yaak_models::models::{ - ClientCertificate, CookieJar, DnsOverride, Environment, HttpRequest, HttpResponse, + ClientCertificate, Cookie, CookieJar, DnsOverride, Environment, HttpRequest, HttpResponse, HttpResponseEvent, HttpResponseHeader, HttpResponseState, ProxySetting, ProxySettingAuth, - ResolvedSetting, + ResolvedHttpRequestSettings, ResolvedSetting, }; use yaak_models::query_manager::QueryManager; use yaak_models::util::{UpdateSource, generate_prefixed_id}; @@ -170,8 +170,7 @@ impl PrepareSendableRequest for PluginPrepareSendableRequest { struct ConnectionManagerSendRequestExecutor<'a> { connection_manager: &'a HttpConnectionManager, plugin_context_id: String, - query_manager: QueryManager, - request: HttpRequest, + runtime_config: HttpSendRuntimeConfig, cancelled_rx: Option>, } @@ -183,18 +182,17 @@ impl SendRequestExecutor for ConnectionManagerSendRequestExecutor<'_> { event_tx: mpsc::Sender, cookie_behavior: CookieBehavior, ) -> yaak_http::error::Result { - let runtime_config = resolve_http_send_runtime_config(&self.query_manager, &self.request) - .map_err(|e| yaak_http::error::Error::RequestError(e.to_string()))?; + let runtime_config = &self.runtime_config; let client_certificate = find_client_certificate(&sendable_request.url, &runtime_config.client_certificates); let cached_client = self .connection_manager .get_client(&HttpConnectionOptions { id: self.plugin_context_id.clone(), - validate_certificates: runtime_config.validate_certificates, - proxy: runtime_config.proxy, + validate_certificates: runtime_config.settings.validate_certificates.value, + proxy: runtime_config.proxy.clone(), client_certificate, - dns_overrides: runtime_config.dns_overrides, + dns_overrides: runtime_config.dns_overrides.clone(), }) .await?; @@ -238,20 +236,37 @@ pub struct SendHttpRequestByIdParams<'a, T: TemplateCallback> { pub executor: &'a dyn SendRequestExecutor, } -pub struct SendHttpRequestParams<'a, T: TemplateCallback> { +/// Everything a send needs that would otherwise be read from the database. +/// +/// Callers backed by a database build this with [`resolve_send_inputs`]. A stateless caller +/// constructs it directly, which is what lets [`send_http_request`] run with no database at all. +pub struct HttpSendInputs { + /// The request with inherited authentication and headers already applied. + pub request: HttpRequest, + pub auth_context_id: String, + pub environment_chain: Vec, + pub runtime_config: HttpSendRuntimeConfig, + /// Cookies the send starts with. The store is shared, so reading it back after the send + /// returns (or fails) yields the cookies the transaction collected. + pub cookie_store: Option, +} + +/// Where a send writes its response. Without it, the send keeps everything in memory: no +/// response body file, no model writes, and no timeline rows. +pub struct ResponseStorage<'a> { pub query_manager: &'a QueryManager, pub blob_manager: &'a BlobManager, - pub request: HttpRequest, - pub environment_id: Option<&'a str>, - pub template_callback: &'a T, - pub send_options: Option, pub update_source: UpdateSource, - pub cookie_jar_id: Option, pub response_dir: &'a Path, +} + +pub struct SendHttpRequestParams<'a, T: TemplateCallback> { + pub inputs: HttpSendInputs, + pub template_callback: &'a T, + pub storage: Option>, pub emit_events_to: Option>, pub emit_response_body_chunks_to: Option>>, pub cancelled_rx: Option>, - pub auth_context_id: Option, pub existing_response: Option, pub prepare_sendable_request: Option<&'a dyn PrepareSendableRequest>, pub executor: &'a dyn SendRequestExecutor, @@ -296,43 +311,71 @@ pub struct SendHttpRequestResult { pub rendered_request: HttpRequest, pub response: HttpResponse, pub response_body: Vec, + /// The cookies held by the jar after the send, for callers that persist one. + pub cookies: Option>, } +#[derive(Clone)] pub struct HttpSendRuntimeConfig { - pub send_options: SendableHttpRequestOptions, - pub validate_certificates: bool, + pub settings: ResolvedHttpRequestSettings, pub proxy: HttpConnectionProxySetting, pub dns_overrides: Vec, pub client_certificates: Vec, } -pub fn resolve_http_send_runtime_config( - query_manager: &QueryManager, - request: &HttpRequest, -) -> Result { - let db = query_manager.connect(); - let workspace = - db.get_workspace(&request.workspace_id).map_err(SendHttpRequestError::LoadWorkspace)?; - let resolved_settings = db - .resolve_settings_for_http_request(request) - .map_err(SendHttpRequestError::ResolveRequestInheritance)?; - let settings = db.get_settings(); - - Ok(HttpSendRuntimeConfig { - send_options: SendableHttpRequestOptions { - follow_redirects: resolved_settings.follow_redirects.value, - timeout: if resolved_settings.request_timeout.value > 0 { - Some(std::time::Duration::from_millis( - resolved_settings.request_timeout.value.unsigned_abs() as u64, +impl HttpSendRuntimeConfig { + pub fn send_options(&self) -> SendableHttpRequestOptions { + SendableHttpRequestOptions { + follow_redirects: self.settings.follow_redirects.value, + timeout: if self.settings.request_timeout.value > 0 { + Some(Duration::from_millis( + self.settings.request_timeout.value.unsigned_abs() as u64 )) } else { None }, + } + } +} + +/// Resolve every database-backed input a send needs, in one pass. +pub fn resolve_send_inputs( + query_manager: &QueryManager, + request: &HttpRequest, + environment_id: Option<&str>, + cookies: Option>, +) -> Result { + let db = query_manager.connect(); + + let environment_chain = db + .resolve_environments(&request.workspace_id, request.folder_id.as_deref(), environment_id) + .map_err(SendHttpRequestError::ResolveEnvironments)?; + + let (authentication_type, authentication, auth_context_id) = db + .resolve_auth_for_http_request(request) + .map_err(SendHttpRequestError::ResolveRequestInheritance)?; + let headers = db + .resolve_headers_for_http_request(request) + .map_err(SendHttpRequestError::ResolveRequestInheritance)?; + + let workspace = + db.get_workspace(&request.workspace_id).map_err(SendHttpRequestError::LoadWorkspace)?; + let settings = db.get_settings(); + let resolved_settings = db + .resolve_settings_for_http_request(request) + .map_err(SendHttpRequestError::ResolveRequestInheritance)?; + + Ok(HttpSendInputs { + request: HttpRequest { authentication_type, authentication, headers, ..request.clone() }, + auth_context_id, + environment_chain, + runtime_config: HttpSendRuntimeConfig { + settings: resolved_settings, + proxy: proxy_setting_from_settings(settings.proxy), + dns_overrides: workspace.setting_dns_overrides, + client_certificates: settings.client_certificates, }, - validate_certificates: resolved_settings.validate_certificates.value, - proxy: proxy_setting_from_settings(settings.proxy), - dns_overrides: workspace.setting_dns_overrides, - client_certificates: settings.client_certificates, + cookie_store: cookies.map(CookieStore::from_cookies), }) } @@ -368,6 +411,14 @@ pub async fn send_http_request_by_id_with_plugins( pub async fn send_http_request_with_plugins( params: SendHttpRequestWithPluginsParams<'_>, ) -> Result { + let mut cookie_jar = load_cookie_jar(params.query_manager, params.cookie_jar_id.as_deref())?; + let inputs = resolve_send_inputs( + params.query_manager, + ¶ms.request, + params.environment_id, + cookie_jar.as_ref().map(|jar| jar.cookies.clone()), + )?; + let template_callback = PluginTemplateCallback::new( params.plugin_manager.clone(), params.encryption_manager.clone(), @@ -382,30 +433,37 @@ pub async fn send_http_request_with_plugins( let executor = ConnectionManagerSendRequestExecutor { connection_manager: params.connection_manager, plugin_context_id: params.plugin_context.id.clone(), - query_manager: params.query_manager.clone(), - request: params.request.clone(), + runtime_config: inputs.runtime_config.clone(), cancelled_rx: params.cancelled_rx.clone(), }; + let cookie_store = inputs.cookie_store.clone(); - send_http_request(SendHttpRequestParams { - query_manager: params.query_manager, - blob_manager: params.blob_manager, - request: params.request, - environment_id: params.environment_id, + let result = send_http_request(SendHttpRequestParams { + inputs, template_callback: &template_callback, - send_options: None, - update_source: params.update_source, - cookie_jar_id: params.cookie_jar_id, - response_dir: params.response_dir, + storage: Some(ResponseStorage { + query_manager: params.query_manager, + blob_manager: params.blob_manager, + update_source: params.update_source, + response_dir: params.response_dir, + }), emit_events_to: params.emit_events_to, emit_response_body_chunks_to: params.emit_response_body_chunks_to, cancelled_rx: params.cancelled_rx, - auth_context_id: None, existing_response: params.existing_response, prepare_sendable_request: Some(&auth_hook), executor: &executor, }) - .await + .await; + + persist_cookies_after_send( + params.query_manager, + cookie_jar.as_mut(), + cookie_store.as_ref(), + &result, + )?; + + result } pub async fn send_http_request_by_id( @@ -416,50 +474,56 @@ pub async fn send_http_request_by_id( .connect() .get_http_request(params.request_id) .map_err(SendHttpRequestError::LoadRequest)?; - let (request, auth_context_id) = resolve_inherited_request(params.query_manager, &request)?; + let mut cookie_jar = load_cookie_jar(params.query_manager, params.cookie_jar_id.as_deref())?; + let inputs = resolve_send_inputs( + params.query_manager, + &request, + params.environment_id, + cookie_jar.as_ref().map(|jar| jar.cookies.clone()), + )?; + let cookie_store = inputs.cookie_store.clone(); - send_http_request(SendHttpRequestParams { - query_manager: params.query_manager, - blob_manager: params.blob_manager, - request, - environment_id: params.environment_id, + let result = send_http_request(SendHttpRequestParams { + inputs, template_callback: params.template_callback, - send_options: None, - update_source: params.update_source, - cookie_jar_id: params.cookie_jar_id, - response_dir: params.response_dir, + storage: Some(ResponseStorage { + query_manager: params.query_manager, + blob_manager: params.blob_manager, + update_source: params.update_source, + response_dir: params.response_dir, + }), emit_events_to: params.emit_events_to, emit_response_body_chunks_to: params.emit_response_body_chunks_to, cancelled_rx: params.cancelled_rx, existing_response: None, prepare_sendable_request: params.prepare_sendable_request, executor: params.executor, - auth_context_id: Some(auth_context_id), }) - .await + .await; + + persist_cookies_after_send( + params.query_manager, + cookie_jar.as_mut(), + cookie_store.as_ref(), + &result, + )?; + + result } pub async fn send_http_request( params: SendHttpRequestParams<'_, T>, ) -> Result { - let environment_chain = - resolve_environment_chain(params.query_manager, ¶ms.request, params.environment_id)?; - let (resolved_request, auth_context_id) = - if let Some(auth_context_id) = params.auth_context_id.clone() { - (params.request.clone(), auth_context_id) - } else { - resolve_inherited_request(params.query_manager, ¶ms.request)? - }; - let runtime_config = resolve_http_send_runtime_config(params.query_manager, ¶ms.request)?; - let send_options = params.send_options.unwrap_or(runtime_config.send_options); - let resolved_settings = params - .query_manager - .connect() - .resolve_settings_for_http_request(¶ms.request) - .map_err(SendHttpRequestError::ResolveRequestInheritance)?; - let mut cookie_jar = load_cookie_jar(params.query_manager, params.cookie_jar_id.as_deref())?; - let cookie_store = - cookie_jar.as_ref().map(|jar| CookieStore::from_cookies(jar.cookies.clone())); + let HttpSendInputs { + request, + auth_context_id, + environment_chain, + runtime_config, + cookie_store, + } = params.inputs; + let storage = params.storage; + let send_options = runtime_config.send_options(); + let resolved_settings = &runtime_config.settings; let cookie_behavior = CookieBehavior { store: cookie_store, send_cookies: resolved_settings.send_cookies.value, @@ -467,7 +531,7 @@ pub async fn send_http_request( }; let rendered_request = render_http_request( - &resolved_request, + &request, environment_chain, params.template_callback, &RenderOptions::throw(), @@ -491,8 +555,8 @@ pub async fn send_http_request( Some(SendableBody::Stream { .. }) | None => None, }; let mut response = params.existing_response.unwrap_or_default(); - response.request_id = params.request.id.clone(); - response.workspace_id = params.request.workspace_id.clone(); + response.request_id = request.id.clone(); + response.workspace_id = request.workspace_id.clone(); response.request_content_length = request_content_length; response.request_headers = sendable_request .headers @@ -513,12 +577,15 @@ pub async fn send_http_request( response.elapsed = 0; response.elapsed_headers = 0; response.elapsed_dns = 0; - let persist_response = !response.request_id.is_empty(); - if persist_response { - response = params + // Responses with no request behind them are ephemeral: they belong to whoever called this + // function and never reach the model store. + let store = storage.as_ref().filter(|_| !response.request_id.is_empty()); + let persist_response = store.is_some(); + if let Some(store) = store { + response = store .query_manager .connect() - .upsert_http_response(&response, ¶ms.update_source, params.blob_manager) + .upsert_http_response(&response, &store.update_source, store.blob_manager) .map_err(SendHttpRequestError::PersistResponse)?; } else if response.id.is_empty() { response.id = generate_prefixed_id("rs"); @@ -527,14 +594,12 @@ pub async fn send_http_request( let request_body_id = format!("{}.request", response.id); let mut request_body_capture_task = None; let mut request_body_capture_error = None; - if persist_response { + if let Some(store) = store { match sendable_request.body.as_mut() { Some(SendableBody::Bytes(bytes)) => { - if let Err(err) = persist_request_body_bytes( - params.blob_manager, - &request_body_id, - bytes.as_ref(), - ) { + if let Err(err) = + persist_request_body_bytes(store.blob_manager, &request_body_id, bytes.as_ref()) + { request_body_capture_error = Some(err); } } @@ -543,7 +608,7 @@ pub async fn send_http_request( let inner = std::mem::replace(data, Box::pin(tokio::io::empty())); let tee_reader = TeeReader::new(inner, tx); *data = Box::pin(tee_reader); - let blob_manager = params.blob_manager.clone(); + let blob_manager = store.blob_manager.clone(); let body_id = request_body_id.clone(); request_body_capture_task = Some(tokio::spawn(async move { persist_request_body_stream(blob_manager, body_id, rx).await @@ -555,10 +620,9 @@ pub async fn send_http_request( let (event_tx, mut event_rx) = mpsc::channel::(HTTP_EVENT_CHANNEL_CAPACITY); - let event_query_manager = params.query_manager.clone(); + let event_store = store.map(|store| (store.query_manager.clone(), store.update_source.clone())); let event_response_id = response.id.clone(); - let event_workspace_id = params.request.workspace_id.clone(); - let event_update_source = params.update_source.clone(); + let event_workspace_id = request.workspace_id.clone(); let emit_events_to = params.emit_events_to.clone(); let dns_elapsed = Arc::new(AtomicI32::new(0)); let event_dns_elapsed = dns_elapsed.clone(); @@ -568,15 +632,14 @@ pub async fn send_http_request( event_dns_elapsed.store(u64_to_i32(*duration), Ordering::Relaxed); } - if persist_response { + if let Some((query_manager, update_source)) = event_store.as_ref() { let db_event = HttpResponseEvent::new( &event_response_id, &event_workspace_id, event.clone().into(), ); - if let Err(err) = event_query_manager - .connect() - .upsert_http_response_event(&db_event, &event_update_source) + if let Err(err) = + query_manager.connect().upsert_http_response_event(&db_event, update_source) { warn!("Failed to persist HTTP response event: {}", err); } @@ -595,7 +658,7 @@ pub async fn send_http_request( send_setting_event( &event_tx, "validate_certificates", - runtime_config.validate_certificates.to_string(), + resolved_settings.validate_certificates.value.to_string(), &resolved_settings.validate_certificates, ); send_setting_event( @@ -627,16 +690,9 @@ pub async fn send_http_request( match executor.send(sendable_request, event_tx, cookie_behavior.clone()).await { Ok(response) => response, Err(err) => { - persist_cookie_jar( - params.query_manager, - cookie_jar.as_mut(), - cookie_behavior.store.as_ref(), - )?; - if persist_response { + if let Some(store) = store { let _ = persist_response_error( - params.query_manager, - params.blob_manager, - ¶ms.update_source, + store, &response, started_at, err.to_string(), @@ -654,14 +710,19 @@ pub async fn send_http_request( }; let headers_elapsed = duration_to_i32(started_at.elapsed()); - std::fs::create_dir_all(params.response_dir).map_err(|source| { - SendHttpRequestError::CreateResponseDirectory { - path: params.response_dir.to_path_buf(), - source, + let body_path = match storage.as_ref() { + Some(storage) => { + std::fs::create_dir_all(storage.response_dir).map_err(|source| { + SendHttpRequestError::CreateResponseDirectory { + path: storage.response_dir.to_path_buf(), + source, + } + })?; + Some(storage.response_dir.join(&response.id)) } - })?; - let body_path = params.response_dir.join(&response.id); - let response_body_path = body_path.to_string_lossy().to_string(); + None => None, + }; + let response_body_path = body_path.as_ref().map(|p| p.to_string_lossy().to_string()); let connected_response = HttpResponse { state: HttpResponseState::Connected, elapsed_headers: headers_elapsed, @@ -671,7 +732,7 @@ pub async fn send_http_request( remote_addr: http_response.remote_addr.clone(), version: http_response.version.clone(), elapsed_dns: dns_elapsed.load(Ordering::Relaxed), - body_path: Some(response_body_path.clone()), + body_path: response_body_path.clone(), content_length: http_response.content_length.map(u64_to_i32), headers: http_response .headers @@ -685,20 +746,26 @@ pub async fn send_http_request( .collect(), ..response }; - if persist_response { - response = params + if let Some(store) = store { + response = store .query_manager .connect() - .upsert_http_response(&connected_response, ¶ms.update_source, params.blob_manager) + .upsert_http_response(&connected_response, &store.update_source, store.blob_manager) .map_err(SendHttpRequestError::PersistResponse)?; } else { response = connected_response; } - let mut file = - File::options().create(true).truncate(true).write(true).open(&body_path).await.map_err( - |source| SendHttpRequestError::WriteResponseBody { path: body_path.clone(), source }, - )?; + let mut body_file = match body_path { + Some(path) => { + let file = + File::options().create(true).truncate(true).write(true).open(&path).await.map_err( + |source| SendHttpRequestError::WriteResponseBody { path: path.clone(), source }, + )?; + Some((file, path)) + } + None => None, + }; let mut body_stream = http_response.into_body_stream().map_err(SendHttpRequestError::ReadResponseBody)?; let mut response_body = Vec::new(); @@ -737,9 +804,11 @@ pub async fn send_http_request( Ok(n) => { written_bytes += n; let chunk = &read_buf[..n]; - file.write_all(chunk).await.map_err(|source| { - SendHttpRequestError::WriteResponseBody { path: body_path.clone(), source } - })?; + if let Some((file, path)) = body_file.as_mut() { + file.write_all(chunk).await.map_err(|source| { + SendHttpRequestError::WriteResponseBody { path: path.clone(), source } + })?; + } if let Some(tx) = params.emit_response_body_chunks_to.as_ref() { let _ = tx.send(chunk.to_vec()); } else if collect_response_body { @@ -757,14 +826,14 @@ pub async fn send_http_request( elapsed_dns: dns_elapsed.load(Ordering::Relaxed), ..response.clone() }; - if persist_response { - response = params + if let Some(store) = store { + response = store .query_manager .connect() .upsert_http_response( &progress_response, - ¶ms.update_source, - params.blob_manager, + &store.update_source, + store.blob_manager, ) .map_err(SendHttpRequestError::PersistResponse)?; } else { @@ -782,10 +851,12 @@ pub async fn send_http_request( } } - file.flush().await.map_err(|source| SendHttpRequestError::WriteResponseBody { - path: body_path.clone(), - source, - })?; + if let Some((file, path)) = body_file.as_mut() { + file.flush().await.map_err(|source| SendHttpRequestError::WriteResponseBody { + path: path.clone(), + source, + })?; + } drop(body_stream); if let Some(err) = request_body_capture_error.take() { @@ -796,22 +867,15 @@ pub async fn send_http_request( } if let Some(err) = body_read_error { - if persist_response { + if let Some(store) = store { let _ = persist_response_error( - params.query_manager, - params.blob_manager, - ¶ms.update_source, + store, &response, started_at, err.to_string(), request_started_url, ); } - persist_cookie_jar( - params.query_manager, - cookie_jar.as_mut(), - cookie_behavior.store.as_ref(), - )?; if let Some(task) = request_body_capture_task.take() { match task.await { Ok(Ok(_)) => {} @@ -827,7 +891,7 @@ pub async fn send_http_request( let compressed_length = http_response.content_length.unwrap_or(written_bytes as u64); let final_response = HttpResponse { - body_path: Some(response_body_path), + body_path: response_body_path, content_length: Some(usize_to_i32(written_bytes)), content_length_compressed: Some(u64_to_i32(compressed_length)), elapsed: duration_to_i32(started_at.elapsed()), @@ -836,18 +900,16 @@ pub async fn send_http_request( state: HttpResponseState::Closed, ..response }; - if persist_response { - response = params + if let Some(store) = store { + response = store .query_manager .connect() - .upsert_http_response(&final_response, ¶ms.update_source, params.blob_manager) + .upsert_http_response(&final_response, &store.update_source, store.blob_manager) .map_err(SendHttpRequestError::PersistResponse)?; } else { response = final_response; } - persist_cookie_jar(params.query_manager, cookie_jar.as_mut(), cookie_behavior.store.as_ref())?; - // Request-body history can be much larger than the response. It should not keep the // response in a loading state after the network/response-body work has completed. if let Some(task) = request_body_capture_task.take() { @@ -876,11 +938,11 @@ pub async fn send_http_request( } } - if update_response && persist_response { - response = params + if update_response && let Some(store) = store { + response = store .query_manager .connect() - .upsert_http_response(&response, ¶ms.update_source, params.blob_manager) + .upsert_http_response(&response, &store.update_source, store.blob_manager) .map_err(SendHttpRequestError::PersistResponse)?; } } @@ -891,7 +953,12 @@ pub async fn send_http_request( warn!("Failed to join response event task: {}", join_err); } - Ok(SendHttpRequestResult { rendered_request, response, response_body }) + Ok(SendHttpRequestResult { + rendered_request, + response, + response_body, + cookies: cookie_behavior.store.as_ref().map(|store| store.get_all_cookies()), + }) } fn persist_request_body_bytes( @@ -953,37 +1020,7 @@ fn append_error_message(existing_error: Option, message: String) -> Stri } } -fn resolve_environment_chain( - query_manager: &QueryManager, - request: &HttpRequest, - environment_id: Option<&str>, -) -> Result> { - let db = query_manager.connect(); - db.resolve_environments(&request.workspace_id, request.folder_id.as_deref(), environment_id) - .map_err(SendHttpRequestError::ResolveEnvironments) -} - -fn resolve_inherited_request( - query_manager: &QueryManager, - request: &HttpRequest, -) -> Result<(HttpRequest, String)> { - let db = query_manager.connect(); - let (authentication_type, authentication, auth_context_id) = db - .resolve_auth_for_http_request(request) - .map_err(SendHttpRequestError::ResolveRequestInheritance)?; - let resolved_headers = db - .resolve_headers_for_http_request(request) - .map_err(SendHttpRequestError::ResolveRequestInheritance)?; - - let mut request = request.clone(); - request.authentication_type = authentication_type; - request.authentication = authentication; - request.headers = resolved_headers; - - Ok((request, auth_context_id)) -} - -fn load_cookie_jar( +pub fn load_cookie_jar( query_manager: &QueryManager, cookie_jar_id: Option<&str>, ) -> Result> { @@ -998,6 +1035,30 @@ fn load_cookie_jar( .map_err(SendHttpRequestError::LoadCookieJar) } +/// Write the cookies a send collected back to its jar. +/// +/// The store is shared with the HTTP transaction, so it holds cookies picked up along the way — +/// worth keeping even when the send later failed partway through a redirect chain. A failure +/// before the transaction started leaves the jar alone. +pub fn persist_cookies_after_send( + query_manager: &QueryManager, + cookie_jar: Option<&mut CookieJar>, + cookie_store: Option<&CookieStore>, + result: &Result, +) -> Result<()> { + let transaction_ran = result.as_ref().err().is_none_or(|err| { + matches!( + err, + SendHttpRequestError::SendRequest(_) | SendHttpRequestError::ReadResponseBody(_) + ) + }); + if !transaction_ran { + return Ok(()); + } + + persist_cookie_jar(query_manager, cookie_jar, cookie_store) +} + fn persist_cookie_jar( query_manager: &QueryManager, cookie_jar: Option<&mut CookieJar>, @@ -1118,16 +1179,15 @@ pub async fn apply_plugin_authentication( } fn persist_response_error( - query_manager: &QueryManager, - blob_manager: &BlobManager, - update_source: &UpdateSource, + store: &ResponseStorage, response: &HttpResponse, started_at: Instant, error: String, fallback_url: String, ) -> Result { let elapsed = duration_to_i32(started_at.elapsed()); - query_manager + store + .query_manager .connect() .upsert_http_response( &HttpResponse { @@ -1142,8 +1202,8 @@ fn persist_response_error( url: if response.url.is_empty() { fallback_url } else { response.url.clone() }, ..response.clone() }, - update_source, - blob_manager, + &store.update_source, + store.blob_manager, ) .map_err(SendHttpRequestError::PersistResponse) } @@ -1173,3 +1233,134 @@ fn u64_to_i32(value: u64) -> i32 { fn u128_to_i32(value: u128) -> i32 { if value > i32::MAX as u128 { i32::MAX } else { value as i32 } } + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashMap; + use std::pin::Pin; + use tokio::io::AsyncRead; + use yaak_http::decompress::ContentEncoding; + + struct NoopTemplateCallback; + + impl TemplateCallback for NoopTemplateCallback { + async fn run( + &self, + _fn_name: &str, + _args: HashMap, + ) -> yaak_templates::error::Result { + Ok(String::new()) + } + + fn transform_arg( + &self, + _fn_name: &str, + _arg_name: &str, + arg_value: &str, + ) -> yaak_templates::error::Result { + Ok(arg_value.to_string()) + } + } + + struct StubExecutor { + body: &'static [u8], + } + + #[async_trait] + impl SendRequestExecutor for StubExecutor { + async fn send( + &self, + sendable_request: SendableHttpRequest, + event_tx: mpsc::Sender, + _cookie_behavior: CookieBehavior, + ) -> yaak_http::error::Result { + let _ = event_tx.try_send(SenderHttpResponseEvent::HeaderDown( + "content-type".to_string(), + "text/plain".to_string(), + )); + let body: Pin> = + Box::pin(std::io::Cursor::new(self.body.to_vec())); + Ok(yaak_http::sender::HttpResponse::new( + 200, + Some("OK".to_string()), + vec![("content-type".to_string(), "text/plain".to_string())], + Vec::new(), + Some(self.body.len() as u64), + sendable_request.url.clone(), + None, + Some("HTTP/1.1".to_string()), + body, + ContentEncoding::Identity, + )) + } + } + + /// The hosted sender runs with no query manager, blob manager, or response directory. Nothing + /// here touches storage, so the caller's only view of the response is what gets streamed out. + #[tokio::test] + async fn sends_without_a_database() { + let (event_tx, mut event_rx) = mpsc::channel(HTTP_EVENT_CHANNEL_CAPACITY); + let (chunk_tx, mut chunk_rx) = mpsc::unbounded_channel(); + let executor = StubExecutor { body: b"hello world" }; + + let result = send_http_request(SendHttpRequestParams { + inputs: HttpSendInputs { + request: HttpRequest { + workspace_id: "wk_test".to_string(), + url: "http://localhost/test".to_string(), + ..Default::default() + }, + auth_context_id: String::new(), + environment_chain: Vec::new(), + runtime_config: HttpSendRuntimeConfig { + settings: ResolvedHttpRequestSettings::default(), + proxy: HttpConnectionProxySetting::System, + dns_overrides: Vec::new(), + client_certificates: Vec::new(), + }, + cookie_store: Some(CookieStore::new()), + }, + template_callback: &NoopTemplateCallback, + storage: None, + emit_events_to: Some(event_tx), + emit_response_body_chunks_to: Some(chunk_tx), + cancelled_rx: None, + existing_response: None, + prepare_sendable_request: None, + executor: &executor, + }) + .await + .expect("send should succeed without a database"); + + assert_eq!(result.response.status, 200); + assert!(matches!(result.response.state, HttpResponseState::Closed)); + assert_eq!(result.response.content_length, Some(11)); + assert!(result.cookies.is_some()); + + // Nothing was written, so the response carries no body file and only ever lived in memory. + assert_eq!(result.response.body_path, None); + assert!(!result.response.id.is_empty()); + + let mut body = Vec::new(); + while let Some(chunk) = chunk_rx.recv().await { + body.extend_from_slice(&chunk); + } + assert_eq!(body, b"hello world"); + + let mut events = Vec::new(); + while let Some(event) = event_rx.recv().await { + events.push(event); + } + assert!( + events.iter().any( + |e| matches!(e, SenderHttpResponseEvent::Setting { name, .. } if name == "redirects") + ), + "expected timeline settings events, got {events:?}" + ); + assert!( + events.iter().any(|e| matches!(e, SenderHttpResponseEvent::HeaderDown(..))), + "expected timeline events from the executor, got {events:?}" + ); + } +}