CLI command architecture and DB-backed model update syncing (#397)

crates/yaak-http/src/sender.rs

@@ -74,15 +74,31 @@ impl Display for HttpResponseEvent {
                 };
                 write!(f, "* Redirect {} -> {} ({})", status, url, behavior_str)
             }
-            HttpResponseEvent::SendUrl { method, scheme, username, password, host, port, path, query, fragment } => {
+            HttpResponseEvent::SendUrl {
+                method,
+                scheme,
+                username,
+                password,
+                host,
+                port,
+                path,
+                query,
+                fragment,
+            } => {
                 let auth_str = if username.is_empty() && password.is_empty() {
                     String::new()
                 } else {
                     format!("{}:{}@", username, password)
                 };
-                let query_str = if query.is_empty() { String::new() } else { format!("?{}", query) };
-                let fragment_str = if fragment.is_empty() { String::new() } else { format!("#{}", fragment) };
-                write!(f, "> {} {}://{}{}:{}{}{}{}", method, scheme, auth_str, host, port, path, query_str, fragment_str)
+                let query_str =
+                    if query.is_empty() { String::new() } else { format!("?{}", query) };
+                let fragment_str =
+                    if fragment.is_empty() { String::new() } else { format!("#{}", fragment) };
+                write!(
+                    f,
+                    "> {} {}://{}{}:{}{}{}{}",
+                    method, scheme, auth_str, host, port, path, query_str, fragment_str
+                )
             }
             HttpResponseEvent::ReceiveUrl { version, status } => {
                 write!(f, "< {} {}", version_to_str(version), status)
@@ -122,7 +138,17 @@ impl From<HttpResponseEvent> for yaak_models::models::HttpResponseEventData {
                     RedirectBehavior::DropBody => "drop_body".to_string(),
                 },
             },
-            HttpResponseEvent::SendUrl { method, scheme, username, password, host, port, path, query, fragment } => {
+            HttpResponseEvent::SendUrl {
+                method,
+                scheme,
+                username,
+                password,
+                host,
+                port,
+                path,
+                query,
+                fragment,
+            } => {
                 D::SendUrl { method, scheme, username, password, host, port, path, query, fragment }
             }
             HttpResponseEvent::ReceiveUrl { version, status } => {
@@ -546,7 +572,10 @@ impl<S> SizedBody<S> {
 
 impl<S> HttpBody for SizedBody<S>
 where
-    S: futures_util::Stream<Item = std::result::Result<Bytes, std::io::Error>> + Send + Unpin + 'static,
+    S: futures_util::Stream<Item = std::result::Result<Bytes, std::io::Error>>
+        + Send
+        + Unpin
+        + 'static,
 {
     type Data = Bytes;
     type Error = std::io::Error;

crates/yaak-http/src/types.rs

@@ -37,10 +37,9 @@ impl From<SendableBodyWithMeta> for SendableBody {
     fn from(value: SendableBodyWithMeta) -> Self {
         match value {
             SendableBodyWithMeta::Bytes(b) => SendableBody::Bytes(b),
-            SendableBodyWithMeta::Stream { data, content_length } => SendableBody::Stream {
-                data,
-                content_length: content_length.map(|l| l as u64),
-            },
+            SendableBodyWithMeta::Stream { data, content_length } => {
+                SendableBody::Stream { data, content_length: content_length.map(|l| l as u64) }
+            }
         }
     }
 }

crates/yaak-models/migrations/20260216000000_model-changes.sql

@@ -0,0 +1,12 @@
+CREATE TABLE model_changes
+(
+    id            INTEGER PRIMARY KEY AUTOINCREMENT,
+    model         TEXT                                                    NOT NULL,
+    model_id      TEXT                                                    NOT NULL,
+    change        TEXT                                                    NOT NULL,
+    update_source TEXT                                                    NOT NULL,
+    payload       TEXT                                                    NOT NULL,
+    created_at    DATETIME DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')) NOT NULL
+);
+
+CREATE INDEX idx_model_changes_created_at ON model_changes (created_at);

crates/yaak-models/src/db_context.rs

@@ -3,8 +3,7 @@ use crate::error::Error::ModelNotFound;
 use crate::error::Result;
 use crate::models::{AnyModel, UpsertModelInfo};
 use crate::util::{ModelChangeEvent, ModelPayload, UpdateSource};
-use log::error;
-use rusqlite::OptionalExtension;
+use rusqlite::{OptionalExtension, params};
 use sea_query::{
     Asterisk, Expr, Func, IntoColumnRef, IntoIden, IntoTableRef, OnConflict, Query, SimpleExpr,
     SqliteQueryBuilder,
@@ -14,7 +13,7 @@ use std::fmt::Debug;
 use std::sync::mpsc;
 
 pub struct DbContext<'a> {
-    pub(crate) events_tx: mpsc::Sender<ModelPayload>,
+    pub(crate) _events_tx: mpsc::Sender<ModelPayload>,
     pub(crate) conn: ConnectionOrTx<'a>,
 }
 
@@ -180,9 +179,8 @@ impl<'a> DbContext<'a> {
             change: ModelChangeEvent::Upsert { created },
         };
 
-        if let Err(e) = self.events_tx.send(payload.clone()) {
-            error!("Failed to send model change {source:?}: {e:?}");
-        }
+        self.record_model_change(&payload)?;
+        let _ = self._events_tx.send(payload);
 
         Ok(m)
     }
@@ -203,9 +201,31 @@ impl<'a> DbContext<'a> {
             change: ModelChangeEvent::Delete,
         };
 
-        if let Err(e) = self.events_tx.send(payload) {
-            error!("Failed to send model change {source:?}: {e:?}");
-        }
+        self.record_model_change(&payload)?;
+        let _ = self._events_tx.send(payload);
+
         Ok(m.clone())
     }
+
+    fn record_model_change(&self, payload: &ModelPayload) -> Result<()> {
+        let payload_json = serde_json::to_string(payload)?;
+        let source_json = serde_json::to_string(&payload.update_source)?;
+        let change_json = serde_json::to_string(&payload.change)?;
+
+        self.conn.resolve().execute(
+            r#"
+                INSERT INTO model_changes (model, model_id, change, update_source, payload)
+                VALUES (?1, ?2, ?3, ?4, ?5)
+            "#,
+            params![
+                payload.model.model(),
+                payload.model.id(),
+                change_json,
+                source_json,
+                payload_json,
+            ],
+        )?;
+
+        Ok(())
+    }
 }

crates/yaak-models/src/models.rs

@@ -2347,6 +2347,15 @@ macro_rules! define_any_model {
                     )*
                 }
             }
+
+            #[inline]
+            pub fn model(&self) -> &str {
+                match self {
+                    $(
+                        AnyModel::$type(inner) => &inner.model,
+                    )*
+                }
+            }
         }
 
         $(
@@ -2400,30 +2409,29 @@ impl<'de> Deserialize<'de> for AnyModel {
     {
         let value = Value::deserialize(deserializer)?;
         let model = value.as_object().unwrap();
+        use AnyModel::*;
         use serde_json::from_value as fv;
 
         let model = match model.get("model") {
-            Some(m) if m == "cookie_jar" => AnyModel::CookieJar(fv(value).unwrap()),
-            Some(m) if m == "environment" => AnyModel::Environment(fv(value).unwrap()),
-            Some(m) if m == "folder" => AnyModel::Folder(fv(value).unwrap()),
-            Some(m) if m == "graphql_introspection" => {
-                AnyModel::GraphQlIntrospection(fv(value).unwrap())
-            }
-            Some(m) if m == "grpc_connection" => AnyModel::GrpcConnection(fv(value).unwrap()),
-            Some(m) if m == "grpc_event" => AnyModel::GrpcEvent(fv(value).unwrap()),
-            Some(m) if m == "grpc_request" => AnyModel::GrpcRequest(fv(value).unwrap()),
-            Some(m) if m == "http_request" => AnyModel::HttpRequest(fv(value).unwrap()),
-            Some(m) if m == "http_response" => AnyModel::HttpResponse(fv(value).unwrap()),
-            Some(m) if m == "key_value" => AnyModel::KeyValue(fv(value).unwrap()),
-            Some(m) if m == "plugin" => AnyModel::Plugin(fv(value).unwrap()),
-            Some(m) if m == "settings" => AnyModel::Settings(fv(value).unwrap()),
-            Some(m) if m == "websocket_connection" => {
-                AnyModel::WebsocketConnection(fv(value).unwrap())
-            }
-            Some(m) if m == "websocket_event" => AnyModel::WebsocketEvent(fv(value).unwrap()),
-            Some(m) if m == "websocket_request" => AnyModel::WebsocketRequest(fv(value).unwrap()),
-            Some(m) if m == "workspace" => AnyModel::Workspace(fv(value).unwrap()),
-            Some(m) if m == "workspace_meta" => AnyModel::WorkspaceMeta(fv(value).unwrap()),
+            Some(m) if m == "cookie_jar" => CookieJar(fv(value).unwrap()),
+            Some(m) if m == "environment" => Environment(fv(value).unwrap()),
+            Some(m) if m == "folder" => Folder(fv(value).unwrap()),
+            Some(m) if m == "graphql_introspection" => GraphQlIntrospection(fv(value).unwrap()),
+            Some(m) if m == "grpc_connection" => GrpcConnection(fv(value).unwrap()),
+            Some(m) if m == "grpc_event" => GrpcEvent(fv(value).unwrap()),
+            Some(m) if m == "grpc_request" => GrpcRequest(fv(value).unwrap()),
+            Some(m) if m == "http_request" => HttpRequest(fv(value).unwrap()),
+            Some(m) if m == "http_response" => HttpResponse(fv(value).unwrap()),
+            Some(m) if m == "http_response_event" => HttpResponseEvent(fv(value).unwrap()),
+            Some(m) if m == "key_value" => KeyValue(fv(value).unwrap()),
+            Some(m) if m == "plugin" => Plugin(fv(value).unwrap()),
+            Some(m) if m == "settings" => Settings(fv(value).unwrap()),
+            Some(m) if m == "sync_state" => SyncState(fv(value).unwrap()),
+            Some(m) if m == "websocket_connection" => WebsocketConnection(fv(value).unwrap()),
+            Some(m) if m == "websocket_event" => WebsocketEvent(fv(value).unwrap()),
+            Some(m) if m == "websocket_request" => WebsocketRequest(fv(value).unwrap()),
+            Some(m) if m == "workspace" => Workspace(fv(value).unwrap()),
+            Some(m) if m == "workspace_meta" => WorkspaceMeta(fv(value).unwrap()),
             Some(m) => {
                 return Err(serde::de::Error::custom(format!(
                     "Failed to deserialize AnyModel {}",

crates/yaak-models/src/queries/mod.rs

@@ -11,6 +11,7 @@ mod http_requests;
 mod http_response_events;
 mod http_responses;
 mod key_values;
+mod model_changes;
 mod plugin_key_values;
 mod plugins;
 mod settings;
@@ -20,6 +21,7 @@ mod websocket_events;
 mod websocket_requests;
 mod workspace_metas;
 pub mod workspaces;
+pub use model_changes::PersistedModelChange;
 
 const MAX_HISTORY_ITEMS: usize = 20;
 

crates/yaak-models/src/queries/model_changes.rs

@@ -0,0 +1,289 @@
+use crate::db_context::DbContext;
+use crate::error::Result;
+use crate::util::ModelPayload;
+use rusqlite::params;
+use rusqlite::types::Type;
+
+#[derive(Debug, Clone)]
+pub struct PersistedModelChange {
+    pub id: i64,
+    pub created_at: String,
+    pub payload: ModelPayload,
+}
+
+impl<'a> DbContext<'a> {
+    pub fn list_model_changes_after(
+        &self,
+        after_id: i64,
+        limit: usize,
+    ) -> Result<Vec<PersistedModelChange>> {
+        let mut stmt = self.conn.prepare(
+            r#"
+                SELECT id, created_at, payload
+                FROM model_changes
+                WHERE id > ?1
+                ORDER BY id ASC
+                LIMIT ?2
+            "#,
+        )?;
+
+        let items = stmt.query_map(params![after_id, limit as i64], |row| {
+            let id: i64 = row.get(0)?;
+            let created_at: String = row.get(1)?;
+            let payload_raw: String = row.get(2)?;
+            let payload = serde_json::from_str::<ModelPayload>(&payload_raw).map_err(|e| {
+                rusqlite::Error::FromSqlConversionFailure(2, Type::Text, Box::new(e))
+            })?;
+            Ok(PersistedModelChange { id, created_at, payload })
+        })?;
+
+        Ok(items.collect::<std::result::Result<Vec<_>, rusqlite::Error>>()?)
+    }
+
+    pub fn list_model_changes_since(
+        &self,
+        since_created_at: &str,
+        since_id: i64,
+        limit: usize,
+    ) -> Result<Vec<PersistedModelChange>> {
+        let mut stmt = self.conn.prepare(
+            r#"
+                SELECT id, created_at, payload
+                FROM model_changes
+                WHERE created_at > ?1
+                   OR (created_at = ?1 AND id > ?2)
+                ORDER BY created_at ASC, id ASC
+                LIMIT ?3
+            "#,
+        )?;
+
+        let items = stmt.query_map(params![since_created_at, since_id, limit as i64], |row| {
+            let id: i64 = row.get(0)?;
+            let created_at: String = row.get(1)?;
+            let payload_raw: String = row.get(2)?;
+            let payload = serde_json::from_str::<ModelPayload>(&payload_raw).map_err(|e| {
+                rusqlite::Error::FromSqlConversionFailure(2, Type::Text, Box::new(e))
+            })?;
+            Ok(PersistedModelChange { id, created_at, payload })
+        })?;
+
+        Ok(items.collect::<std::result::Result<Vec<_>, rusqlite::Error>>()?)
+    }
+
+    pub fn prune_model_changes_older_than_days(&self, days: i64) -> Result<usize> {
+        let offset = format!("-{days} days");
+        Ok(self.conn.resolve().execute(
+            r#"
+                DELETE FROM model_changes
+                WHERE created_at < STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW', ?1)
+            "#,
+            params![offset],
+        )?)
+    }
+
+    pub fn prune_model_changes_older_than_hours(&self, hours: i64) -> Result<usize> {
+        let offset = format!("-{hours} hours");
+        Ok(self.conn.resolve().execute(
+            r#"
+                DELETE FROM model_changes
+                WHERE created_at < STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW', ?1)
+            "#,
+            params![offset],
+        )?)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::init_in_memory;
+    use crate::models::Workspace;
+    use crate::util::{ModelChangeEvent, UpdateSource};
+    use serde_json::json;
+
+    #[test]
+    fn records_model_changes_for_upsert_and_delete() {
+        let (query_manager, _blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
+        let db = query_manager.connect();
+
+        let workspace = db
+            .upsert_workspace(
+                &Workspace {
+                    name: "Changes Test".to_string(),
+                    setting_follow_redirects: true,
+                    setting_validate_certificates: true,
+                    ..Default::default()
+                },
+                &UpdateSource::Sync,
+            )
+            .expect("Failed to upsert workspace");
+
+        let created_changes = db.list_model_changes_after(0, 10).expect("Failed to list changes");
+        assert_eq!(created_changes.len(), 1);
+        assert_eq!(created_changes[0].payload.model.id(), workspace.id);
+        assert_eq!(created_changes[0].payload.model.model(), "workspace");
+        assert!(matches!(
+            created_changes[0].payload.change,
+            ModelChangeEvent::Upsert { created: true }
+        ));
+        assert!(matches!(created_changes[0].payload.update_source, UpdateSource::Sync));
+
+        db.delete_workspace_by_id(&workspace.id, &UpdateSource::Sync)
+            .expect("Failed to delete workspace");
+
+        let all_changes = db.list_model_changes_after(0, 10).expect("Failed to list changes");
+        assert_eq!(all_changes.len(), 2);
+        assert!(matches!(all_changes[1].payload.change, ModelChangeEvent::Delete));
+        assert!(all_changes[1].id > all_changes[0].id);
+
+        let changes_after_first = db
+            .list_model_changes_after(all_changes[0].id, 10)
+            .expect("Failed to list changes after cursor");
+        assert_eq!(changes_after_first.len(), 1);
+        assert!(matches!(changes_after_first[0].payload.change, ModelChangeEvent::Delete));
+    }
+
+    #[test]
+    fn prunes_old_model_changes() {
+        let (query_manager, _blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
+        let db = query_manager.connect();
+
+        db.upsert_workspace(
+            &Workspace {
+                name: "Prune Test".to_string(),
+                setting_follow_redirects: true,
+                setting_validate_certificates: true,
+                ..Default::default()
+            },
+            &UpdateSource::Sync,
+        )
+        .expect("Failed to upsert workspace");
+
+        let changes = db.list_model_changes_after(0, 10).expect("Failed to list changes");
+        assert_eq!(changes.len(), 1);
+
+        db.conn
+            .resolve()
+            .execute(
+                "UPDATE model_changes SET created_at = '2000-01-01 00:00:00.000' WHERE id = ?1",
+                params![changes[0].id],
+            )
+            .expect("Failed to age model change row");
+
+        let pruned =
+            db.prune_model_changes_older_than_days(30).expect("Failed to prune model changes");
+        assert_eq!(pruned, 1);
+        assert!(db.list_model_changes_after(0, 10).expect("Failed to list changes").is_empty());
+    }
+
+    #[test]
+    fn list_model_changes_since_uses_timestamp_with_id_tiebreaker() {
+        let (query_manager, _blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
+        let db = query_manager.connect();
+
+        let workspace = db
+            .upsert_workspace(
+                &Workspace {
+                    name: "Cursor Test".to_string(),
+                    setting_follow_redirects: true,
+                    setting_validate_certificates: true,
+                    ..Default::default()
+                },
+                &UpdateSource::Sync,
+            )
+            .expect("Failed to upsert workspace");
+        db.delete_workspace_by_id(&workspace.id, &UpdateSource::Sync)
+            .expect("Failed to delete workspace");
+
+        let all = db.list_model_changes_after(0, 10).expect("Failed to list changes");
+        assert_eq!(all.len(), 2);
+
+        let fixed_ts = "2026-02-16 00:00:00.000";
+        db.conn
+            .resolve()
+            .execute("UPDATE model_changes SET created_at = ?1", params![fixed_ts])
+            .expect("Failed to normalize timestamps");
+
+        let after_first =
+            db.list_model_changes_since(fixed_ts, all[0].id, 10).expect("Failed to query cursor");
+        assert_eq!(after_first.len(), 1);
+        assert_eq!(after_first[0].id, all[1].id);
+    }
+
+    #[test]
+    fn prunes_old_model_changes_by_hours() {
+        let (query_manager, _blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
+        let db = query_manager.connect();
+
+        db.upsert_workspace(
+            &Workspace {
+                name: "Prune Hour Test".to_string(),
+                setting_follow_redirects: true,
+                setting_validate_certificates: true,
+                ..Default::default()
+            },
+            &UpdateSource::Sync,
+        )
+        .expect("Failed to upsert workspace");
+
+        let changes = db.list_model_changes_after(0, 10).expect("Failed to list changes");
+        assert_eq!(changes.len(), 1);
+
+        db.conn
+            .resolve()
+            .execute(
+                "UPDATE model_changes SET created_at = STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW', '-2 hours') WHERE id = ?1",
+                params![changes[0].id],
+            )
+            .expect("Failed to age model change row");
+
+        let pruned =
+            db.prune_model_changes_older_than_hours(1).expect("Failed to prune model changes");
+        assert_eq!(pruned, 1);
+    }
+
+    #[test]
+    fn list_model_changes_deserializes_http_response_event_payload() {
+        let (query_manager, _blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
+        let db = query_manager.connect();
+
+        let payload = json!({
+            "model": {
+                "model": "http_response_event",
+                "id": "re_test",
+                "createdAt": "2026-02-16T21:01:34.809162",
+                "updatedAt": "2026-02-16T21:01:34.809163",
+                "workspaceId": "wk_test",
+                "responseId": "rs_test",
+                "event": {
+                    "type": "info",
+                    "message": "hello"
+                }
+            },
+            "updateSource": { "type": "sync" },
+            "change": { "type": "upsert", "created": false }
+        });
+
+        db.conn
+            .resolve()
+            .execute(
+                r#"
+                INSERT INTO model_changes (model, model_id, change, update_source, payload)
+                VALUES (?1, ?2, ?3, ?4, ?5)
+                "#,
+                params![
+                    "http_response_event",
+                    "re_test",
+                    r#"{"type":"upsert","created":false}"#,
+                    r#"{"type":"sync"}"#,
+                    payload.to_string(),
+                ],
+            )
+            .expect("Failed to insert model change row");
+
+        let changes = db.list_model_changes_after(0, 10).expect("Failed to list changes");
+        assert_eq!(changes.len(), 1);
+        assert_eq!(changes[0].payload.model.model(), "http_response_event");
+        assert_eq!(changes[0].payload.model.id(), "re_test");
+    }
+}

crates/yaak-models/src/query_manager.rs

@@ -25,7 +25,7 @@ impl QueryManager {
             .expect("Failed to gain lock on DB")
             .get()
             .expect("Failed to get a new DB connection from the pool");
-        DbContext { events_tx: self.events_tx.clone(), conn: ConnectionOrTx::Connection(conn) }
+        DbContext { _events_tx: self.events_tx.clone(), conn: ConnectionOrTx::Connection(conn) }
     }
 
     pub fn with_conn<F, T>(&self, func: F) -> T
@@ -39,8 +39,10 @@ impl QueryManager {
             .get()
             .expect("Failed to get new DB connection from the pool");
 
-        let db_context =
-            DbContext { events_tx: self.events_tx.clone(), conn: ConnectionOrTx::Connection(conn) };
+        let db_context = DbContext {
+            _events_tx: self.events_tx.clone(),
+            conn: ConnectionOrTx::Connection(conn),
+        };
 
         func(&db_context)
     }
@@ -62,8 +64,10 @@ impl QueryManager {
             .transaction_with_behavior(TransactionBehavior::Immediate)
             .expect("Failed to start DB transaction");
 
-        let db_context =
-            DbContext { events_tx: self.events_tx.clone(), conn: ConnectionOrTx::Transaction(&tx) };
+        let db_context = DbContext {
+            _events_tx: self.events_tx.clone(),
+            conn: ConnectionOrTx::Transaction(&tx),
+        };
 
         match func(&db_context) {
             Ok(val) => {

crates/yaak-plugins/src/nodejs.rs

@@ -68,7 +68,9 @@ pub async fn start_nodejs_plugin_runtime(
     // Handle kill signal
     let mut kill_rx = kill_rx.clone();
     tokio::spawn(async move {
-        kill_rx.wait_for(|b| *b == true).await.expect("Kill channel errored");
+        if kill_rx.wait_for(|b| *b == true).await.is_err() {
+            warn!("Kill channel closed before explicit shutdown; terminating plugin runtime");
+        }
         info!("Killing plugin runtime");
         if let Err(e) = child.kill().await {
             warn!("Failed to kill plugin runtime: {e}");

crates/yaak/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "yaak"
+version = "0.1.0"
+edition = "2024"
+publish = false
+
+[dependencies]
+async-trait = "0.1"
+log = { workspace = true }
+md5 = "0.8.0"
+serde_json = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true, features = ["sync", "rt"] }
+yaak-http = { workspace = true }
+yaak-crypto = { workspace = true }
+yaak-models = { workspace = true }
+yaak-plugins = { workspace = true }
+yaak-templates = { workspace = true }
+yaak-tls = { workspace = true }

crates/yaak/src/error.rs

@@ -0,0 +1,9 @@
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum Error {
+    #[error(transparent)]
+    Send(#[from] crate::send::SendHttpRequestError),
+}
+
+pub type Result<T> = std::result::Result<T, Error>;

crates/yaak/src/lib.rs

@@ -0,0 +1,6 @@
+pub mod error;
+pub mod render;
+pub mod send;
+
+pub use error::Error;
+pub type Result<T> = error::Result<T>;

crates/yaak/src/render.rs

@@ -0,0 +1,157 @@
+use log::info;
+use serde_json::Value;
+use std::collections::BTreeMap;
+use yaak_http::path_placeholders::apply_path_placeholders;
+use yaak_models::models::{Environment, HttpRequest, HttpRequestHeader, HttpUrlParameter};
+use yaak_models::render::make_vars_hashmap;
+use yaak_templates::{RenderOptions, TemplateCallback, parse_and_render, render_json_value_raw};
+
+pub async fn render_http_request<T: TemplateCallback>(
+    request: &HttpRequest,
+    environment_chain: Vec<Environment>,
+    callback: &T,
+    options: &RenderOptions,
+) -> yaak_templates::error::Result<HttpRequest> {
+    let vars = &make_vars_hashmap(environment_chain);
+
+    let mut url_parameters = Vec::new();
+    for parameter in request.url_parameters.clone() {
+        if !parameter.enabled {
+            continue;
+        }
+
+        url_parameters.push(HttpUrlParameter {
+            enabled: parameter.enabled,
+            name: parse_and_render(parameter.name.as_str(), vars, callback, options).await?,
+            value: parse_and_render(parameter.value.as_str(), vars, callback, options).await?,
+            id: parameter.id,
+        })
+    }
+
+    let mut headers = Vec::new();
+    for header in request.headers.clone() {
+        if !header.enabled {
+            continue;
+        }
+
+        headers.push(HttpRequestHeader {
+            enabled: header.enabled,
+            name: parse_and_render(header.name.as_str(), vars, callback, options).await?,
+            value: parse_and_render(header.value.as_str(), vars, callback, options).await?,
+            id: header.id,
+        })
+    }
+
+    let mut body = BTreeMap::new();
+    for (key, value) in request.body.clone() {
+        let value = if key == "form" { strip_disabled_form_entries(value) } else { value };
+        body.insert(key, render_json_value_raw(value, vars, callback, options).await?);
+    }
+
+    let authentication = {
+        let mut disabled = false;
+        let mut auth = BTreeMap::new();
+
+        match request.authentication.get("disabled") {
+            Some(Value::Bool(true)) => {
+                disabled = true;
+            }
+            Some(Value::String(template)) => {
+                disabled = parse_and_render(template.as_str(), vars, callback, options)
+                    .await
+                    .unwrap_or_default()
+                    .is_empty();
+                info!(
+                    "Rendering authentication.disabled as a template: {disabled} from \"{template}\""
+                );
+            }
+            _ => {}
+        }
+
+        if disabled {
+            auth.insert("disabled".to_string(), Value::Bool(true));
+        } else {
+            for (key, value) in request.authentication.clone() {
+                if key == "disabled" {
+                    auth.insert(key, Value::Bool(false));
+                } else {
+                    auth.insert(key, render_json_value_raw(value, vars, callback, options).await?);
+                }
+            }
+        }
+
+        auth
+    };
+
+    let url = parse_and_render(request.url.clone().as_str(), vars, callback, options).await?;
+    let (url, url_parameters) = apply_path_placeholders(&url, &url_parameters);
+
+    Ok(HttpRequest { url, url_parameters, headers, body, authentication, ..request.to_owned() })
+}
+
+fn strip_disabled_form_entries(v: Value) -> Value {
+    match v {
+        Value::Array(items) => Value::Array(
+            items
+                .into_iter()
+                .filter(|item| item.get("enabled").and_then(|e| e.as_bool()).unwrap_or(true))
+                .collect(),
+        ),
+        v => v,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    #[test]
+    fn test_strip_disabled_form_entries() {
+        let input = json!([
+            {"enabled": true, "name": "foo", "value": "bar"},
+            {"enabled": false, "name": "disabled", "value": "gone"},
+            {"enabled": true, "name": "baz", "value": "qux"},
+        ]);
+        let result = strip_disabled_form_entries(input);
+        assert_eq!(
+            result,
+            json!([
+                {"enabled": true, "name": "foo", "value": "bar"},
+                {"enabled": true, "name": "baz", "value": "qux"},
+            ])
+        );
+    }
+
+    #[test]
+    fn test_strip_disabled_form_entries_all_disabled() {
+        let input = json!([
+            {"enabled": false, "name": "a", "value": "b"},
+            {"enabled": false, "name": "c", "value": "d"},
+        ]);
+        let result = strip_disabled_form_entries(input);
+        assert_eq!(result, json!([]));
+    }
+
+    #[test]
+    fn test_strip_disabled_form_entries_missing_enabled_defaults_to_kept() {
+        let input = json!([
+            {"name": "no_enabled_field", "value": "kept"},
+            {"enabled": false, "name": "disabled", "value": "gone"},
+        ]);
+        let result = strip_disabled_form_entries(input);
+        assert_eq!(
+            result,
+            json!([
+                {"name": "no_enabled_field", "value": "kept"},
+            ])
+        );
+    }
+
+    #[test]
+    fn test_strip_disabled_form_entries_non_array_passthrough() {
+        let input = json!("just a string");
+        let result = strip_disabled_form_entries(input.clone());
+        assert_eq!(result, input);
+    }
+}

crates/yaak/src/send.rs

@@ -0,0 +1,813 @@
+use crate::render::render_http_request;
+use async_trait::async_trait;
+use log::warn;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::time::Instant;
+use thiserror::Error;
+use tokio::sync::mpsc;
+use tokio::sync::watch;
+use yaak_crypto::manager::EncryptionManager;
+use yaak_http::client::{
+    HttpConnectionOptions, HttpConnectionProxySetting, HttpConnectionProxySettingAuth,
+};
+use yaak_http::cookies::CookieStore;
+use yaak_http::manager::HttpConnectionManager;
+use yaak_http::sender::{HttpResponseEvent as SenderHttpResponseEvent, ReqwestSender};
+use yaak_http::transaction::HttpTransaction;
+use yaak_http::types::{
+    SendableBody, SendableHttpRequest, SendableHttpRequestOptions, append_query_params,
+};
+use yaak_models::blob_manager::BlobManager;
+use yaak_models::models::{
+    ClientCertificate, CookieJar, DnsOverride, Environment, HttpRequest, HttpResponse,
+    HttpResponseEvent, HttpResponseHeader, HttpResponseState, ProxySetting, ProxySettingAuth,
+};
+use yaak_models::query_manager::QueryManager;
+use yaak_models::util::UpdateSource;
+use yaak_plugins::events::{
+    CallHttpAuthenticationRequest, HttpHeader, PluginContext, RenderPurpose,
+};
+use yaak_plugins::manager::PluginManager;
+use yaak_plugins::template_callback::PluginTemplateCallback;
+use yaak_templates::{RenderOptions, TemplateCallback};
+use yaak_tls::find_client_certificate;
+
+const HTTP_EVENT_CHANNEL_CAPACITY: usize = 100;
+
+#[derive(Debug, Error)]
+pub enum SendHttpRequestError {
+    #[error("Failed to load request: {0}")]
+    LoadRequest(#[source] yaak_models::error::Error),
+
+    #[error("Failed to load workspace: {0}")]
+    LoadWorkspace(#[source] yaak_models::error::Error),
+
+    #[error("Failed to resolve environments: {0}")]
+    ResolveEnvironments(#[source] yaak_models::error::Error),
+
+    #[error("Failed to resolve inherited request settings: {0}")]
+    ResolveRequestInheritance(#[source] yaak_models::error::Error),
+
+    #[error("Failed to load cookie jar: {0}")]
+    LoadCookieJar(#[source] yaak_models::error::Error),
+
+    #[error("Failed to persist cookie jar: {0}")]
+    PersistCookieJar(#[source] yaak_models::error::Error),
+
+    #[error("Failed to render request templates: {0}")]
+    RenderRequest(#[source] yaak_templates::error::Error),
+
+    #[error("Failed to prepare request before send: {0}")]
+    PrepareSendableRequest(String),
+
+    #[error("Failed to persist response metadata: {0}")]
+    PersistResponse(#[source] yaak_models::error::Error),
+
+    #[error("Failed to create HTTP client: {0}")]
+    CreateHttpClient(#[source] yaak_http::error::Error),
+
+    #[error("Failed to build sendable request: {0}")]
+    BuildSendableRequest(#[source] yaak_http::error::Error),
+
+    #[error("Failed to send request: {0}")]
+    SendRequest(#[source] yaak_http::error::Error),
+
+    #[error("Failed to read response body: {0}")]
+    ReadResponseBody(#[source] yaak_http::error::Error),
+
+    #[error("Failed to create response directory {path:?}: {source}")]
+    CreateResponseDirectory {
+        path: PathBuf,
+        #[source]
+        source: std::io::Error,
+    },
+
+    #[error("Failed to write response body to {path:?}: {source}")]
+    WriteResponseBody {
+        path: PathBuf,
+        #[source]
+        source: std::io::Error,
+    },
+}
+
+pub type Result<T> = std::result::Result<T, SendHttpRequestError>;
+
+#[async_trait]
+pub trait PrepareSendableRequest: Send + Sync {
+    async fn prepare_sendable_request(
+        &self,
+        rendered_request: &HttpRequest,
+        auth_context_id: &str,
+        sendable_request: &mut SendableHttpRequest,
+    ) -> std::result::Result<(), String>;
+}
+
+#[async_trait]
+pub trait SendRequestExecutor: Send + Sync {
+    async fn send(
+        &self,
+        sendable_request: SendableHttpRequest,
+        event_tx: mpsc::Sender<SenderHttpResponseEvent>,
+        cookie_store: Option<CookieStore>,
+    ) -> yaak_http::error::Result<yaak_http::sender::HttpResponse>;
+}
+
+struct DefaultSendRequestExecutor;
+
+#[async_trait]
+impl SendRequestExecutor for DefaultSendRequestExecutor {
+    async fn send(
+        &self,
+        sendable_request: SendableHttpRequest,
+        event_tx: mpsc::Sender<SenderHttpResponseEvent>,
+        cookie_store: Option<CookieStore>,
+    ) -> yaak_http::error::Result<yaak_http::sender::HttpResponse> {
+        let sender = ReqwestSender::new()?;
+        let transaction = match cookie_store {
+            Some(store) => HttpTransaction::with_cookie_store(sender, store),
+            None => HttpTransaction::new(sender),
+        };
+        let (_cancel_tx, cancel_rx) = watch::channel(false);
+        transaction.execute_with_cancellation(sendable_request, cancel_rx, event_tx).await
+    }
+}
+
+struct PluginPrepareSendableRequest {
+    plugin_manager: Arc<PluginManager>,
+    plugin_context: PluginContext,
+    cancelled_rx: Option<watch::Receiver<bool>>,
+}
+
+#[async_trait]
+impl PrepareSendableRequest for PluginPrepareSendableRequest {
+    async fn prepare_sendable_request(
+        &self,
+        rendered_request: &HttpRequest,
+        auth_context_id: &str,
+        sendable_request: &mut SendableHttpRequest,
+    ) -> std::result::Result<(), String> {
+        if let Some(cancelled_rx) = &self.cancelled_rx {
+            let mut cancelled_rx = cancelled_rx.clone();
+            tokio::select! {
+                result = apply_plugin_authentication(
+                    sendable_request,
+                    rendered_request,
+                    auth_context_id,
+                    &self.plugin_manager,
+                    &self.plugin_context,
+                ) => result,
+                _ = cancelled_rx.changed() => Err("Request canceled".to_string()),
+            }
+        } else {
+            apply_plugin_authentication(
+                sendable_request,
+                rendered_request,
+                auth_context_id,
+                &self.plugin_manager,
+                &self.plugin_context,
+            )
+            .await
+        }
+    }
+}
+
+struct ConnectionManagerSendRequestExecutor<'a> {
+    connection_manager: &'a HttpConnectionManager,
+    plugin_context_id: String,
+    query_manager: QueryManager,
+    workspace_id: String,
+    cancelled_rx: Option<watch::Receiver<bool>>,
+}
+
+#[async_trait]
+impl SendRequestExecutor for ConnectionManagerSendRequestExecutor<'_> {
+    async fn send(
+        &self,
+        sendable_request: SendableHttpRequest,
+        event_tx: mpsc::Sender<SenderHttpResponseEvent>,
+        cookie_store: Option<CookieStore>,
+    ) -> yaak_http::error::Result<yaak_http::sender::HttpResponse> {
+        let runtime_config =
+            resolve_http_send_runtime_config(&self.query_manager, &self.workspace_id)
+                .map_err(|e| yaak_http::error::Error::RequestError(e.to_string()))?;
+        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,
+                client_certificate,
+                dns_overrides: runtime_config.dns_overrides,
+            })
+            .await?;
+
+        cached_client.resolver.set_event_sender(Some(event_tx.clone())).await;
+
+        let sender = ReqwestSender::with_client(cached_client.client);
+        let transaction = match cookie_store {
+            Some(cs) => HttpTransaction::with_cookie_store(sender, cs),
+            None => HttpTransaction::new(sender),
+        };
+
+        let result = if let Some(cancelled_rx) = self.cancelled_rx.clone() {
+            transaction.execute_with_cancellation(sendable_request, cancelled_rx, event_tx).await
+        } else {
+            let (_cancel_tx, cancel_rx) = watch::channel(false);
+            transaction.execute_with_cancellation(sendable_request, cancel_rx, event_tx).await
+        };
+        cached_client.resolver.set_event_sender(None).await;
+        result
+    }
+}
+
+pub struct SendHttpRequestByIdParams<'a, T: TemplateCallback> {
+    pub query_manager: &'a QueryManager,
+    pub blob_manager: &'a BlobManager,
+    pub request_id: &'a str,
+    pub environment_id: Option<&'a str>,
+    pub template_callback: &'a T,
+    pub update_source: UpdateSource,
+    pub cookie_jar_id: Option<String>,
+    pub response_dir: &'a Path,
+    pub emit_events_to: Option<mpsc::Sender<SenderHttpResponseEvent>>,
+    pub prepare_sendable_request: Option<&'a dyn PrepareSendableRequest>,
+    pub executor: Option<&'a dyn SendRequestExecutor>,
+}
+
+pub struct SendHttpRequestParams<'a, T: TemplateCallback> {
+    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<SendableHttpRequestOptions>,
+    pub update_source: UpdateSource,
+    pub cookie_jar_id: Option<String>,
+    pub response_dir: &'a Path,
+    pub emit_events_to: Option<mpsc::Sender<SenderHttpResponseEvent>>,
+    pub auth_context_id: Option<String>,
+    pub existing_response: Option<HttpResponse>,
+    pub prepare_sendable_request: Option<&'a dyn PrepareSendableRequest>,
+    pub executor: Option<&'a dyn SendRequestExecutor>,
+}
+
+pub struct SendHttpRequestWithPluginsParams<'a> {
+    pub query_manager: &'a QueryManager,
+    pub blob_manager: &'a BlobManager,
+    pub request: HttpRequest,
+    pub environment_id: Option<&'a str>,
+    pub update_source: UpdateSource,
+    pub cookie_jar_id: Option<String>,
+    pub response_dir: &'a Path,
+    pub emit_events_to: Option<mpsc::Sender<SenderHttpResponseEvent>>,
+    pub existing_response: Option<HttpResponse>,
+    pub plugin_manager: Arc<PluginManager>,
+    pub encryption_manager: Arc<EncryptionManager>,
+    pub plugin_context: &'a PluginContext,
+    pub cancelled_rx: Option<watch::Receiver<bool>>,
+    pub connection_manager: Option<&'a HttpConnectionManager>,
+}
+
+pub struct SendHttpRequestByIdWithPluginsParams<'a> {
+    pub query_manager: &'a QueryManager,
+    pub blob_manager: &'a BlobManager,
+    pub request_id: &'a str,
+    pub environment_id: Option<&'a str>,
+    pub update_source: UpdateSource,
+    pub cookie_jar_id: Option<String>,
+    pub response_dir: &'a Path,
+    pub emit_events_to: Option<mpsc::Sender<SenderHttpResponseEvent>>,
+    pub plugin_manager: Arc<PluginManager>,
+    pub encryption_manager: Arc<EncryptionManager>,
+    pub plugin_context: &'a PluginContext,
+    pub cancelled_rx: Option<watch::Receiver<bool>>,
+    pub connection_manager: Option<&'a HttpConnectionManager>,
+}
+
+pub struct SendHttpRequestResult {
+    pub rendered_request: HttpRequest,
+    pub response: HttpResponse,
+    pub response_body: Vec<u8>,
+}
+
+pub struct HttpSendRuntimeConfig {
+    pub send_options: SendableHttpRequestOptions,
+    pub validate_certificates: bool,
+    pub proxy: HttpConnectionProxySetting,
+    pub dns_overrides: Vec<DnsOverride>,
+    pub client_certificates: Vec<ClientCertificate>,
+}
+
+pub fn resolve_http_send_runtime_config(
+    query_manager: &QueryManager,
+    workspace_id: &str,
+) -> Result<HttpSendRuntimeConfig> {
+    let db = query_manager.connect();
+    let workspace = db.get_workspace(workspace_id).map_err(SendHttpRequestError::LoadWorkspace)?;
+    let settings = db.get_settings();
+
+    Ok(HttpSendRuntimeConfig {
+        send_options: SendableHttpRequestOptions {
+            follow_redirects: workspace.setting_follow_redirects,
+            timeout: if workspace.setting_request_timeout > 0 {
+                Some(std::time::Duration::from_millis(
+                    workspace.setting_request_timeout.unsigned_abs() as u64,
+                ))
+            } else {
+                None
+            },
+        },
+        validate_certificates: workspace.setting_validate_certificates,
+        proxy: proxy_setting_from_settings(settings.proxy),
+        dns_overrides: workspace.setting_dns_overrides,
+        client_certificates: settings.client_certificates,
+    })
+}
+
+pub async fn send_http_request_by_id_with_plugins(
+    params: SendHttpRequestByIdWithPluginsParams<'_>,
+) -> Result<SendHttpRequestResult> {
+    let request = params
+        .query_manager
+        .connect()
+        .get_http_request(params.request_id)
+        .map_err(SendHttpRequestError::LoadRequest)?;
+
+    send_http_request_with_plugins(SendHttpRequestWithPluginsParams {
+        query_manager: params.query_manager,
+        blob_manager: params.blob_manager,
+        request,
+        environment_id: params.environment_id,
+        update_source: params.update_source,
+        cookie_jar_id: params.cookie_jar_id,
+        response_dir: params.response_dir,
+        emit_events_to: params.emit_events_to,
+        existing_response: None,
+        plugin_manager: params.plugin_manager,
+        encryption_manager: params.encryption_manager,
+        plugin_context: params.plugin_context,
+        cancelled_rx: params.cancelled_rx,
+        connection_manager: params.connection_manager,
+    })
+    .await
+}
+
+pub async fn send_http_request_with_plugins(
+    params: SendHttpRequestWithPluginsParams<'_>,
+) -> Result<SendHttpRequestResult> {
+    let template_callback = PluginTemplateCallback::new(
+        params.plugin_manager.clone(),
+        params.encryption_manager.clone(),
+        params.plugin_context,
+        RenderPurpose::Send,
+    );
+    let auth_hook = PluginPrepareSendableRequest {
+        plugin_manager: params.plugin_manager,
+        plugin_context: params.plugin_context.clone(),
+        cancelled_rx: params.cancelled_rx.clone(),
+    };
+    let executor =
+        params.connection_manager.map(|connection_manager| ConnectionManagerSendRequestExecutor {
+            connection_manager,
+            plugin_context_id: params.plugin_context.id.clone(),
+            query_manager: params.query_manager.clone(),
+            workspace_id: params.request.workspace_id.clone(),
+            cancelled_rx: params.cancelled_rx.clone(),
+        });
+
+    send_http_request(SendHttpRequestParams {
+        query_manager: params.query_manager,
+        blob_manager: params.blob_manager,
+        request: params.request,
+        environment_id: params.environment_id,
+        template_callback: &template_callback,
+        send_options: None,
+        update_source: params.update_source,
+        cookie_jar_id: params.cookie_jar_id,
+        response_dir: params.response_dir,
+        emit_events_to: params.emit_events_to,
+        auth_context_id: None,
+        existing_response: params.existing_response,
+        prepare_sendable_request: Some(&auth_hook),
+        executor: executor.as_ref().map(|e| e as &dyn SendRequestExecutor),
+    })
+    .await
+}
+
+pub async fn send_http_request_by_id<T: TemplateCallback>(
+    params: SendHttpRequestByIdParams<'_, T>,
+) -> Result<SendHttpRequestResult> {
+    let request = params
+        .query_manager
+        .connect()
+        .get_http_request(params.request_id)
+        .map_err(SendHttpRequestError::LoadRequest)?;
+    let (request, auth_context_id) = resolve_inherited_request(params.query_manager, &request)?;
+
+    send_http_request(SendHttpRequestParams {
+        query_manager: params.query_manager,
+        blob_manager: params.blob_manager,
+        request,
+        environment_id: params.environment_id,
+        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,
+        emit_events_to: params.emit_events_to,
+        existing_response: None,
+        prepare_sendable_request: params.prepare_sendable_request,
+        executor: params.executor,
+        auth_context_id: Some(auth_context_id),
+    })
+    .await
+}
+
+pub async fn send_http_request<T: TemplateCallback>(
+    params: SendHttpRequestParams<'_, T>,
+) -> Result<SendHttpRequestResult> {
+    let environment_chain =
+        resolve_environment_chain(params.query_manager, &params.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, &params.request)?
+        };
+    let runtime_config =
+        resolve_http_send_runtime_config(params.query_manager, &params.request.workspace_id)?;
+    let send_options = params.send_options.unwrap_or(runtime_config.send_options);
+    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 rendered_request = render_http_request(
+        &resolved_request,
+        environment_chain,
+        params.template_callback,
+        &RenderOptions::throw(),
+    )
+    .await
+    .map_err(SendHttpRequestError::RenderRequest)?;
+
+    let mut sendable_request =
+        SendableHttpRequest::from_http_request(&rendered_request, send_options)
+            .await
+            .map_err(SendHttpRequestError::BuildSendableRequest)?;
+
+    if let Some(hook) = params.prepare_sendable_request {
+        hook.prepare_sendable_request(&rendered_request, &auth_context_id, &mut sendable_request)
+            .await
+            .map_err(SendHttpRequestError::PrepareSendableRequest)?;
+    }
+
+    let request_content_length = sendable_body_length(sendable_request.body.as_ref());
+    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_content_length = request_content_length;
+    response.request_headers = sendable_request
+        .headers
+        .iter()
+        .map(|(name, value)| HttpResponseHeader { name: name.clone(), value: value.clone() })
+        .collect();
+    response.url = sendable_request.url.clone();
+    response.state = HttpResponseState::Initialized;
+    response.error = None;
+    response.content_length = None;
+    response.content_length_compressed = None;
+    response.body_path = None;
+    response.status = 0;
+    response.status_reason = None;
+    response.headers = Vec::new();
+    response.remote_addr = None;
+    response.version = None;
+    response.elapsed = 0;
+    response.elapsed_headers = 0;
+    response.elapsed_dns = 0;
+    response = params
+        .query_manager
+        .connect()
+        .upsert_http_response(&response, &params.update_source, params.blob_manager)
+        .map_err(SendHttpRequestError::PersistResponse)?;
+
+    let (event_tx, mut event_rx) =
+        mpsc::channel::<SenderHttpResponseEvent>(HTTP_EVENT_CHANNEL_CAPACITY);
+    let event_query_manager = params.query_manager.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 emit_events_to = params.emit_events_to.clone();
+    let event_handle = tokio::spawn(async move {
+        while let Some(event) = event_rx.recv().await {
+            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)
+            {
+                warn!("Failed to persist HTTP response event: {}", err);
+            }
+
+            if let Some(tx) = emit_events_to.as_ref() {
+                let _ = tx.try_send(event);
+            }
+        }
+    });
+
+    let default_executor = DefaultSendRequestExecutor;
+    let executor = params.executor.unwrap_or(&default_executor);
+    let started_at = Instant::now();
+    let request_started_url = sendable_request.url.clone();
+
+    let http_response = match executor.send(sendable_request, event_tx, cookie_store.clone()).await
+    {
+        Ok(response) => response,
+        Err(err) => {
+            persist_cookie_jar(params.query_manager, cookie_jar.as_mut(), cookie_store.as_ref())?;
+            let _ = persist_response_error(
+                params.query_manager,
+                params.blob_manager,
+                &params.update_source,
+                &response,
+                started_at,
+                err.to_string(),
+                request_started_url,
+            );
+            if let Err(join_err) = event_handle.await {
+                warn!("Failed to join response event task: {}", join_err);
+            }
+            return Err(SendHttpRequestError::SendRequest(err));
+        }
+    };
+
+    let headers_elapsed = duration_to_i32(started_at.elapsed());
+    response = params
+        .query_manager
+        .connect()
+        .upsert_http_response(
+            &HttpResponse {
+                state: HttpResponseState::Connected,
+                elapsed_headers: headers_elapsed,
+                status: i32::from(http_response.status),
+                status_reason: http_response.status_reason.clone(),
+                url: http_response.url.clone(),
+                remote_addr: http_response.remote_addr.clone(),
+                version: http_response.version.clone(),
+                headers: http_response
+                    .headers
+                    .iter()
+                    .map(|(name, value)| HttpResponseHeader {
+                        name: name.clone(),
+                        value: value.clone(),
+                    })
+                    .collect(),
+                request_headers: http_response
+                    .request_headers
+                    .iter()
+                    .map(|(name, value)| HttpResponseHeader {
+                        name: name.clone(),
+                        value: value.clone(),
+                    })
+                    .collect(),
+                ..response
+            },
+            &params.update_source,
+            params.blob_manager,
+        )
+        .map_err(SendHttpRequestError::PersistResponse)?;
+
+    let (response_body, body_stats) =
+        http_response.bytes().await.map_err(SendHttpRequestError::ReadResponseBody)?;
+
+    std::fs::create_dir_all(params.response_dir).map_err(|source| {
+        SendHttpRequestError::CreateResponseDirectory {
+            path: params.response_dir.to_path_buf(),
+            source,
+        }
+    })?;
+
+    let body_path = params.response_dir.join(&response.id);
+    std::fs::write(&body_path, &response_body).map_err(|source| {
+        SendHttpRequestError::WriteResponseBody { path: body_path.clone(), source }
+    })?;
+
+    response = params
+        .query_manager
+        .connect()
+        .upsert_http_response(
+            &HttpResponse {
+                body_path: Some(body_path.to_string_lossy().to_string()),
+                content_length: Some(usize_to_i32(response_body.len())),
+                content_length_compressed: Some(u64_to_i32(body_stats.size_compressed)),
+                elapsed: duration_to_i32(started_at.elapsed()),
+                elapsed_headers: headers_elapsed,
+                state: HttpResponseState::Closed,
+                ..response
+            },
+            &params.update_source,
+            params.blob_manager,
+        )
+        .map_err(SendHttpRequestError::PersistResponse)?;
+
+    if let Err(join_err) = event_handle.await {
+        warn!("Failed to join response event task: {}", join_err);
+    }
+    persist_cookie_jar(params.query_manager, cookie_jar.as_mut(), cookie_store.as_ref())?;
+
+    Ok(SendHttpRequestResult { rendered_request, response, response_body })
+}
+
+fn resolve_environment_chain(
+    query_manager: &QueryManager,
+    request: &HttpRequest,
+    environment_id: Option<&str>,
+) -> Result<Vec<Environment>> {
+    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(
+    query_manager: &QueryManager,
+    cookie_jar_id: Option<&str>,
+) -> Result<Option<CookieJar>> {
+    let Some(cookie_jar_id) = cookie_jar_id else {
+        return Ok(None);
+    };
+
+    query_manager
+        .connect()
+        .get_cookie_jar(cookie_jar_id)
+        .map(Some)
+        .map_err(SendHttpRequestError::LoadCookieJar)
+}
+
+fn persist_cookie_jar(
+    query_manager: &QueryManager,
+    cookie_jar: Option<&mut CookieJar>,
+    cookie_store: Option<&CookieStore>,
+) -> Result<()> {
+    match (cookie_jar, cookie_store) {
+        (Some(cookie_jar), Some(cookie_store)) => {
+            cookie_jar.cookies = cookie_store.get_all_cookies();
+            query_manager
+                .connect()
+                .upsert_cookie_jar(cookie_jar, &UpdateSource::Background)
+                .map_err(SendHttpRequestError::PersistCookieJar)?;
+            Ok(())
+        }
+        _ => Ok(()),
+    }
+}
+
+fn proxy_setting_from_settings(proxy: Option<ProxySetting>) -> HttpConnectionProxySetting {
+    match proxy {
+        None => HttpConnectionProxySetting::System,
+        Some(ProxySetting::Disabled) => HttpConnectionProxySetting::Disabled,
+        Some(ProxySetting::Enabled { http, https, auth, bypass, disabled }) => {
+            if disabled {
+                HttpConnectionProxySetting::System
+            } else {
+                HttpConnectionProxySetting::Enabled {
+                    http,
+                    https,
+                    bypass,
+                    auth: auth.map(|ProxySettingAuth { user, password }| {
+                        HttpConnectionProxySettingAuth { user, password }
+                    }),
+                }
+            }
+        }
+    }
+}
+
+pub async fn apply_plugin_authentication(
+    sendable_request: &mut SendableHttpRequest,
+    request: &HttpRequest,
+    auth_context_id: &str,
+    plugin_manager: &PluginManager,
+    plugin_context: &PluginContext,
+) -> std::result::Result<(), String> {
+    match &request.authentication_type {
+        None => {}
+        Some(authentication_type) if authentication_type == "none" => {}
+        Some(authentication_type) => {
+            let req = CallHttpAuthenticationRequest {
+                context_id: format!("{:x}", md5::compute(auth_context_id)),
+                values: serde_json::from_value(
+                    serde_json::to_value(&request.authentication)
+                        .map_err(|e| format!("Failed to serialize auth values: {e}"))?,
+                )
+                .map_err(|e| format!("Failed to parse auth values: {e}"))?,
+                url: sendable_request.url.clone(),
+                method: sendable_request.method.clone(),
+                headers: sendable_request
+                    .headers
+                    .iter()
+                    .map(|(name, value)| HttpHeader {
+                        name: name.to_string(),
+                        value: value.to_string(),
+                    })
+                    .collect(),
+            };
+            let plugin_result = plugin_manager
+                .call_http_authentication(plugin_context, authentication_type, req)
+                .await
+                .map_err(|e| format!("Failed to apply authentication plugin: {e}"))?;
+
+            for header in plugin_result.set_headers.unwrap_or_default() {
+                sendable_request.insert_header((header.name, header.value));
+            }
+
+            if let Some(params) = plugin_result.set_query_parameters {
+                let params = params.into_iter().map(|p| (p.name, p.value)).collect::<Vec<_>>();
+                sendable_request.url = append_query_params(&sendable_request.url, params);
+            }
+        }
+    }
+    Ok(())
+}
+
+fn persist_response_error(
+    query_manager: &QueryManager,
+    blob_manager: &BlobManager,
+    update_source: &UpdateSource,
+    response: &HttpResponse,
+    started_at: Instant,
+    error: String,
+    fallback_url: String,
+) -> Result<HttpResponse> {
+    let elapsed = duration_to_i32(started_at.elapsed());
+    query_manager
+        .connect()
+        .upsert_http_response(
+            &HttpResponse {
+                state: HttpResponseState::Closed,
+                elapsed,
+                elapsed_headers: if response.elapsed_headers == 0 {
+                    elapsed
+                } else {
+                    response.elapsed_headers
+                },
+                error: Some(error),
+                url: if response.url.is_empty() { fallback_url } else { response.url.clone() },
+                ..response.clone()
+            },
+            update_source,
+            blob_manager,
+        )
+        .map_err(SendHttpRequestError::PersistResponse)
+}
+
+fn sendable_body_length(body: Option<&SendableBody>) -> Option<i32> {
+    match body {
+        Some(SendableBody::Bytes(bytes)) => Some(usize_to_i32(bytes.len())),
+        Some(SendableBody::Stream { content_length: Some(length), .. }) => {
+            Some(u64_to_i32(*length))
+        }
+        _ => None,
+    }
+}
+
+fn duration_to_i32(duration: std::time::Duration) -> i32 {
+    u128_to_i32(duration.as_millis())
+}
+
+fn usize_to_i32(value: usize) -> i32 {
+    if value > i32::MAX as usize { i32::MAX } else { value as i32 }
+}
+
+fn u64_to_i32(value: u64) -> i32 {
+    if value > i32::MAX as u64 { i32::MAX } else { value as i32 }
+}
+
+fn u128_to_i32(value: u128) -> i32 {
+    if value > i32::MAX as u128 { i32::MAX } else { value as i32 }
+}