starred/yaak-mountain-loop
1
0
Fork 0
mirror of https://github.com/mountain-loop/yaak.git synced 2026-08-25 12:54:09 +02:00
Code Issues 8 Packages Projects Releases 10 Wiki Activity

Add a plugin API for reading HTTP response bodies

Browse Source 
Plugins read bodies by response id through ctx.httpResponse.body()
instead of opening HttpResponse.bodyPath themselves. The accessors are
named after fetch's, minus the single-use semantics, since the bytes are
durable and re-reading should work.

Underneath is a chunked pull over the existing plugin protocol, so the
host can move bodies off the filesystem without plugins noticing.
text() now decodes with the response's charset rather than assuming
UTF-8, and the buffering accessors refuse past 32 MiB and point at
chunks().
This commit is contained in:
Gregory Schier
2026-08-16 09:20:29 -07:00
parent 6a02cbe525
commit 96c8a95094
18 changed files with 915 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/yaak-plugins/bindings/gen_events.ts
Generated
+42 -1
View File
File diff suppressed because one or more lines are too long
crates/yaak-plugins/src/events.rs
+62
View File
@@ -171,6 +171,12 @@ pub enum InternalEventPayload {
FindHttpResponsesRequest(FindHttpResponsesRequest),
FindHttpResponsesResponse(FindHttpResponsesResponse),
GetHttpResponseBodyInfoRequest(GetHttpResponseBodyInfoRequest),
GetHttpResponseBodyInfoResponse(GetHttpResponseBodyInfoResponse),
ReadHttpResponseBodyChunkRequest(ReadHttpResponseBodyChunkRequest),
ReadHttpResponseBodyChunkResponse(ReadHttpResponseBodyChunkResponse),
ListHttpRequestsRequest(ListHttpRequestsRequest),
ListHttpRequestsResponse(ListHttpRequestsResponse),
ListFoldersRequest(ListFoldersRequest),
@@ -1413,6 +1419,62 @@ pub struct FindHttpResponsesResponse {
pub http_responses: Vec<HttpResponse>,
}
/// Ask what a response's body is, before deciding whether to pull it.
///
/// Bodies are addressed by response id and never by path, so where the host
/// keeps the bytes is its own business.
#[derive(Debug, Clone, Default, Serialize, Deserialize, TS)]
#[serde(default, rename_all = "camelCase")]
#[ts(export, export_to = "gen_events.ts")]
pub struct GetHttpResponseBodyInfoRequest {
pub response_id: String,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize, TS)]
#[serde(default, rename_all = "camelCase")]
#[ts(export, export_to = "gen_events.ts")]
pub struct GetHttpResponseBodyInfoResponse {
/// How many bytes are actually stored, which is not necessarily what the
/// `Content-Length` header claimed. Zero when the response has no body.
#[ts(type = "number")]
pub content_length: u64,
/// The response's `Content-Type` header, verbatim, so the reader can pick a
/// charset.
#[ts(optional = nullable)]
pub content_type: Option<String>,
}
/// Pull one window of a response body.
///
/// Reads are idempotent: the bytes live in durable storage, so the same window
/// can be asked for as many times as the plugin likes.
#[derive(Debug, Clone, Default, Serialize, Deserialize, TS)]
#[serde(default, rename_all = "camelCase")]
#[ts(export, export_to = "gen_events.ts")]
pub struct ReadHttpResponseBodyChunkRequest {
pub response_id: String,
#[ts(type = "number")]
pub offset: u64,
#[ts(type = "number")]
pub length: u64,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize, TS)]
#[serde(default, rename_all = "camelCase")]
#[ts(export, export_to = "gen_events.ts")]
pub struct ReadHttpResponseBodyChunkResponse {
/// Base64, because the desktop transport is a WebSocket that only sends
/// text frames today. A host that can carry binary sends the bytes as they
/// are and fills this in from them.
pub data: String,
/// Bytes decoded from `data`. Short of the requested length means the body
/// ended here.
#[ts(type = "number")]
pub length: u64,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize, TS)]
#[serde(default, rename_all = "camelCase")]
#[ts(export, export_to = "gen_events.ts")]
crates/yaak/Cargo.toml
+1
View File
@@ -6,6 +6,7 @@ publish = false
[dependencies]
async-trait = "0.1"
base64 = "0.22.1" # For carrying body chunks over a text-only plugin transport
log = { workspace = true }
md5 = "0.8.0"
serde_json = { workspace = true }
crates/yaak/src/lib.rs
+1
View File
@@ -3,6 +3,7 @@ pub mod export;
pub mod import;
pub mod plugin_events;
pub mod render;
pub mod response_body;
pub mod send;
pub use error::Error;
crates/yaak/src/plugin_events.rs
+172 -16
View File
@@ -1,3 +1,6 @@
use crate::response_body::ResponseBodyStore;
use base64::Engine;
use base64::prelude::BASE64_STANDARD;
use yaak_models::models::AnyModel;
use yaak_models::query_manager::QueryManager;
use yaak_models::util::UpdateSource;
@@ -5,12 +8,14 @@ use yaak_plugins::events::{
CloseWindowRequest, CopyTextRequest, DeleteKeyValueRequest, DeleteKeyValueResponse,
DeleteModelRequest, DeleteModelResponse, ErrorResponse, FindHttpResponsesRequest,
FindHttpResponsesResponse, GetCookieValueRequest, GetHttpRequestByIdRequest,
GetHttpRequestByIdResponse, GetKeyValueRequest, GetKeyValueResponse, InternalEventPayload,
ListCookieNamesRequest, ListFoldersRequest, ListFoldersResponse, ListHttpRequestsRequest,
ListHttpRequestsResponse, ListOpenWorkspacesRequest, OpenExternalUrlRequest, OpenWindowRequest,
PromptFormRequest, PromptTextRequest, ReloadResponse, RenderGrpcRequestRequest,
RenderHttpRequestRequest, SendHttpRequestRequest, SetKeyValueRequest, ShowToastRequest,
TemplateRenderRequest, UpsertModelRequest, UpsertModelResponse, WindowInfoRequest,
GetHttpRequestByIdResponse, GetHttpResponseBodyInfoRequest, GetHttpResponseBodyInfoResponse,
GetKeyValueRequest, GetKeyValueResponse, InternalEventPayload, ListCookieNamesRequest,
ListFoldersRequest, ListFoldersResponse, ListHttpRequestsRequest, ListHttpRequestsResponse,
ListOpenWorkspacesRequest, OpenExternalUrlRequest, OpenWindowRequest, PromptFormRequest,
PromptTextRequest, ReadHttpResponseBodyChunkRequest, ReadHttpResponseBodyChunkResponse,
ReloadResponse, RenderGrpcRequestRequest, RenderHttpRequestRequest, SendHttpRequestRequest,
SetKeyValueRequest, ShowToastRequest, TemplateRenderRequest, UpsertModelRequest,
UpsertModelResponse, WindowInfoRequest,
};
pub struct SharedPluginEventContext<'a> {
@@ -40,6 +45,8 @@ pub enum SharedRequest<'a> {
ListFolders(&'a ListFoldersRequest),
ListHttpRequests(&'a ListHttpRequestsRequest),
FindHttpResponses(&'a FindHttpResponsesRequest),
GetHttpResponseBodyInfo(&'a GetHttpResponseBodyInfoRequest),
ReadHttpResponseBodyChunk(&'a ReadHttpResponseBodyChunkRequest),
UpsertModel(&'a UpsertModelRequest),
DeleteModel(&'a DeleteModelRequest),
}
@@ -136,6 +143,12 @@ impl<'a> From<&'a InternalEventPayload> for GroupedPluginRequest<'a> {
InternalEventPayload::FindHttpResponsesRequest(req) => {
GroupedPluginRequest::Shared(SharedRequest::FindHttpResponses(req))
}
InternalEventPayload::GetHttpResponseBodyInfoRequest(req) => {
GroupedPluginRequest::Shared(SharedRequest::GetHttpResponseBodyInfo(req))
}
InternalEventPayload::ReadHttpResponseBodyChunkRequest(req) => {
GroupedPluginRequest::Shared(SharedRequest::ReadHttpResponseBodyChunk(req))
}
InternalEventPayload::UpsertModelRequest(req) => {
GroupedPluginRequest::Shared(SharedRequest::UpsertModel(req))
}
@@ -182,13 +195,17 @@ impl<'a> From<&'a InternalEventPayload> for GroupedPluginRequest<'a> {
pub fn handle_shared_plugin_event<'a>(
query_manager: &QueryManager,
body_store: &dyn ResponseBodyStore,
payload: &'a InternalEventPayload,
context: SharedPluginEventContext<'_>,
) -> GroupedPluginEvent<'a> {
match GroupedPluginRequest::from(payload) {
GroupedPluginRequest::Shared(req) => {
GroupedPluginEvent::Handled(Some(build_shared_reply(query_manager, req, context)))
}
GroupedPluginRequest::Shared(req) => GroupedPluginEvent::Handled(Some(build_shared_reply(
query_manager,
body_store,
req,
context,
))),
GroupedPluginRequest::Host(req) => GroupedPluginEvent::ToHandle(req),
GroupedPluginRequest::Ignore => GroupedPluginEvent::Handled(None),
}
@@ -196,6 +213,7 @@ pub fn handle_shared_plugin_event<'a>(
fn build_shared_reply(
query_manager: &QueryManager,
body_store: &dyn ResponseBodyStore,
request: SharedRequest<'_>,
context: SharedPluginEventContext<'_>,
) -> InternalEventPayload {
@@ -283,6 +301,30 @@ fn build_shared_reply(
http_responses,
})
}
SharedRequest::GetHttpResponseBodyInfo(req) => match body_store.info(&req.response_id) {
Ok(info) => InternalEventPayload::GetHttpResponseBodyInfoResponse(
GetHttpResponseBodyInfoResponse {
content_length: info.content_length,
content_type: info.content_type,
},
),
Err(err) => InternalEventPayload::ErrorResponse(ErrorResponse {
error: format!("Failed to read body of response {}: {err}", req.response_id),
}),
},
SharedRequest::ReadHttpResponseBodyChunk(req) => {
match body_store.read_chunk(&req.response_id, req.offset, req.length) {
Ok(bytes) => InternalEventPayload::ReadHttpResponseBodyChunkResponse(
ReadHttpResponseBodyChunkResponse {
length: bytes.len() as u64,
data: BASE64_STANDARD.encode(bytes),
},
),
Err(err) => InternalEventPayload::ErrorResponse(ErrorResponse {
error: format!("Failed to read body of response {}: {err}", req.response_id),
}),
}
}
SharedRequest::UpsertModel(req) => {
use AnyModel::*;
@@ -437,10 +479,26 @@ fn build_shared_reply(
#[cfg(test)]
mod tests {
use super::*;
use crate::response_body::{FileResponseBodyStore, ResponseBodyInfo};
use std::cell::RefCell;
use tempfile::TempDir;
use yaak_models::models::{AnyModel, Folder, HttpRequest, Workspace};
use yaak_models::util::UpdateSource;
/// The real dispatch, with the store the desktop and CLI hand it.
fn dispatch<'a>(
query_manager: &QueryManager,
payload: &'a InternalEventPayload,
context: SharedPluginEventContext<'_>,
) -> GroupedPluginEvent<'a> {
handle_shared_plugin_event(
query_manager,
&FileResponseBodyStore::new(query_manager),
payload,
context,
)
}
fn seed_query_manager() -> (QueryManager, TempDir) {
let temp_dir = TempDir::new().expect("Failed to create temp dir");
let db_path = temp_dir.path().join("db.sqlite");
@@ -498,7 +556,7 @@ mod tests {
let payload = InternalEventPayload::ListHttpRequestsRequest(
yaak_plugins::events::ListHttpRequestsRequest { folder_id: None },
);
let result = handle_shared_plugin_event(
let result = dispatch(
&query_manager,
&payload,
SharedPluginEventContext { plugin_name: "@yaak/test", workspace_id: None },
@@ -517,7 +575,7 @@ mod tests {
let by_workspace_payload = InternalEventPayload::ListHttpRequestsRequest(
yaak_plugins::events::ListHttpRequestsRequest { folder_id: None },
);
let by_workspace = handle_shared_plugin_event(
let by_workspace = dispatch(
&query_manager,
&by_workspace_payload,
SharedPluginEventContext { plugin_name: "@yaak/test", workspace_id: Some("wk_test") },
@@ -536,7 +594,7 @@ mod tests {
folder_id: Some("fl_test".to_string()),
},
);
let by_folder = handle_shared_plugin_event(
let by_folder = dispatch(
&query_manager,
&by_folder_payload,
SharedPluginEventContext { plugin_name: "@yaak/test", workspace_id: None },
@@ -559,7 +617,7 @@ mod tests {
limit: Some(1),
});
let result = handle_shared_plugin_event(
let result = dispatch(
&query_manager,
&payload,
SharedPluginEventContext { plugin_name: "@yaak/test", workspace_id: Some("wk_test") },
@@ -575,6 +633,104 @@ mod tests {
}
}
/// A store that answers from memory, standing in for whatever holds the
/// bytes — the point being that the dispatch below never learns which.
struct FakeBodyStore {
body: Vec<u8>,
reads: RefCell<Vec<(u64, u64)>>,
}
impl ResponseBodyStore for FakeBodyStore {
fn info(&self, _response_id: &str) -> crate::error::Result<ResponseBodyInfo> {
Ok(ResponseBodyInfo {
content_length: self.body.len() as u64,
content_type: Some("text/plain; charset=utf-8".to_string()),
})
}
fn read_chunk(
&self,
_response_id: &str,
offset: u64,
length: u64,
) -> crate::error::Result<Vec<u8>> {
self.reads.borrow_mut().push((offset, length));
let start = (offset as usize).min(self.body.len());
let end = (start + length as usize).min(self.body.len());
Ok(self.body[start..end].to_vec())
}
}
#[test]
fn response_body_is_read_by_id_through_the_store() {
let (query_manager, _temp_dir) = seed_query_manager();
let store = FakeBodyStore { body: b"hello".to_vec(), reads: RefCell::new(Vec::new()) };
let info_payload = InternalEventPayload::GetHttpResponseBodyInfoRequest(
GetHttpResponseBodyInfoRequest { response_id: "rs_test".to_string() },
);
let info = handle_shared_plugin_event(
&query_manager,
&store,
&info_payload,
SharedPluginEventContext { plugin_name: "@yaak/test", workspace_id: None },
);
match info {
GroupedPluginEvent::Handled(Some(
InternalEventPayload::GetHttpResponseBodyInfoResponse(resp),
)) => {
assert_eq!(resp.content_length, 5);
assert_eq!(resp.content_type.as_deref(), Some("text/plain; charset=utf-8"));
}
other => panic!("unexpected body info result: {other:?}"),
}
let chunk_payload = InternalEventPayload::ReadHttpResponseBodyChunkRequest(
ReadHttpResponseBodyChunkRequest {
response_id: "rs_test".to_string(),
offset: 1,
length: 3,
},
);
let chunk = handle_shared_plugin_event(
&query_manager,
&store,
&chunk_payload,
SharedPluginEventContext { plugin_name: "@yaak/test", workspace_id: None },
);
match chunk {
GroupedPluginEvent::Handled(Some(
InternalEventPayload::ReadHttpResponseBodyChunkResponse(resp),
)) => {
assert_eq!(resp.length, 3);
assert_eq!(BASE64_STANDARD.decode(resp.data).unwrap(), b"ell");
}
other => panic!("unexpected body chunk result: {other:?}"),
}
assert_eq!(*store.reads.borrow(), vec![(1, 3)]);
}
#[test]
fn an_unreadable_response_body_becomes_an_error_reply() {
let (query_manager, _temp_dir) = seed_query_manager();
let payload = InternalEventPayload::GetHttpResponseBodyInfoRequest(
GetHttpResponseBodyInfoRequest { response_id: "rs_never_persisted".to_string() },
);
let result = dispatch(
&query_manager,
&payload,
SharedPluginEventContext { plugin_name: "@yaak/test", workspace_id: None },
);
match result {
GroupedPluginEvent::Handled(Some(InternalEventPayload::ErrorResponse(resp))) => {
assert!(resp.error.contains("rs_never_persisted"), "unhelpful error: {}", resp.error)
}
other => panic!("unexpected missing-response result: {other:?}"),
}
}
#[test]
fn upsert_and_delete_model_are_shared_handled() {
let (query_manager, _temp_dir) = seed_query_manager();
@@ -590,7 +746,7 @@ mod tests {
}),
});
let upsert_result = handle_shared_plugin_event(
let upsert_result = dispatch(
&query_manager,
&upsert_payload,
SharedPluginEventContext { plugin_name: "@yaak/test", workspace_id: Some("wk_test") },
@@ -609,7 +765,7 @@ mod tests {
model: "http_request".to_string(),
id: "rq_test".to_string(),
});
let delete_result = handle_shared_plugin_event(
let delete_result = dispatch(
&query_manager,
&delete_payload,
SharedPluginEventContext { plugin_name: "@yaak/test", workspace_id: Some("wk_test") },
@@ -631,7 +787,7 @@ mod tests {
let payload = InternalEventPayload::WindowInfoRequest(WindowInfoRequest {
label: "main".to_string(),
});
let result = handle_shared_plugin_event(
let result = dispatch(
&query_manager,
&payload,
SharedPluginEventContext { plugin_name: "@yaak/test", workspace_id: None },
crates/yaak/src/response_body.rs
+197
View File
@@ -0,0 +1,197 @@
//! Reading response bodies back out, by response id.
//!
//! Plugins only ever name a response. Where its bytes actually live — files the
//! engine wrote under `<data dir>/responses/<id>` today, blob rows later — is
//! behind [`ResponseBodyStore`], so moving the bytes is a change to this file
//! and nothing a plugin can see.
use crate::error::Result;
use std::fs::File;
use std::io::{Read, Seek, SeekFrom};
use yaak_models::query_manager::QueryManager;
/// The most bytes one read will hand back, however much was asked for.
///
/// A chunk is buffered whole and, on the desktop transport, base64'd into a
/// single WebSocket frame, so an unbounded request is a way to make the host
/// allocate on a plugin's say-so.
pub const MAX_CHUNK_BYTES: u64 = 8 * 1024 * 1024;
/// What a stored body is, without reading any of it.
#[derive(Debug, Clone, Default)]
pub struct ResponseBodyInfo {
/// Bytes actually stored, which is not necessarily what `Content-Length`
/// claimed. Zero when the response has no body.
pub content_length: u64,
/// The response's `Content-Type` header, verbatim.
pub content_type: Option<String>,
}
/// Somewhere response bodies can be read from, a window at a time.
///
/// Reads are repeatable — the bytes are durable, so nothing is consumed by
/// looking at it.
pub trait ResponseBodyStore {
fn info(&self, response_id: &str) -> Result<ResponseBodyInfo>;
/// Bytes `[offset, offset + length)`, clamped to what is there. A short
/// read means the body ended.
fn read_chunk(&self, response_id: &str, offset: u64, length: u64) -> Result<Vec<u8>>;
}
/// The desktop and CLI store: the database says where the file is, and the
/// filesystem holds it.
pub struct FileResponseBodyStore<'a> {
query_manager: &'a QueryManager,
}
impl<'a> FileResponseBodyStore<'a> {
pub fn new(query_manager: &'a QueryManager) -> Self {
Self { query_manager }
}
/// The file backing a response, or `None` when the response stored no body.
///
/// A response that was never persisted (GraphQL introspection and other
/// ephemeral sends) has no row here at all, so it fails as "not found"
/// rather than reading as an empty body.
fn body_path(&self, response_id: &str) -> Result<Option<String>> {
Ok(self.query_manager.connect().get_http_response(response_id)?.body_path)
}
}
impl ResponseBodyStore for FileResponseBodyStore<'_> {
fn info(&self, response_id: &str) -> Result<ResponseBodyInfo> {
let response = self.query_manager.connect().get_http_response(response_id)?;
let content_type = response
.headers
.iter()
.find(|h| h.name.eq_ignore_ascii_case("content-type"))
.map(|h| h.value.clone());
let content_length = match response.body_path {
Some(path) => std::fs::metadata(path)?.len(),
None => 0,
};
Ok(ResponseBodyInfo { content_length, content_type })
}
fn read_chunk(&self, response_id: &str, offset: u64, length: u64) -> Result<Vec<u8>> {
let Some(path) = self.body_path(response_id)? else {
return Ok(Vec::new());
};
let length = length.min(MAX_CHUNK_BYTES);
if length == 0 {
return Ok(Vec::new());
}
let mut file = File::open(path)?;
file.seek(SeekFrom::Start(offset))?;
let mut buf = Vec::new();
file.take(length).read_to_end(&mut buf)?;
Ok(buf)
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::Write;
use tempfile::TempDir;
use yaak_models::models::{HttpRequest, HttpResponse, HttpResponseHeader, Workspace};
use yaak_models::util::UpdateSource;
fn seed(body: Option<&[u8]>) -> (QueryManager, TempDir, String) {
let temp_dir = TempDir::new().unwrap();
let (query_manager, blob_manager, _rx) = yaak_models::init_standalone(
&temp_dir.path().join("db.sqlite"),
&temp_dir.path().join("blobs.sqlite"),
)
.unwrap();
query_manager
.connect()
.upsert_workspace(
&Workspace { id: "wk_test".to_string(), ..Default::default() },
&UpdateSource::Sync,
)
.unwrap();
query_manager
.connect()
.upsert_http_request(
&HttpRequest {
id: "rq_test".to_string(),
workspace_id: "wk_test".to_string(),
..Default::default()
},
&UpdateSource::Sync,
)
.unwrap();
let body_path = body.map(|bytes| {
let path = temp_dir.path().join("body");
let mut f = std::fs::File::create(&path).unwrap();
f.write_all(bytes).unwrap();
path.to_string_lossy().to_string()
});
let response = query_manager
.connect()
.upsert_http_response(
&HttpResponse {
workspace_id: "wk_test".to_string(),
request_id: "rq_test".to_string(),
body_path,
headers: vec![HttpResponseHeader {
name: "Content-Type".to_string(),
value: "application/json; charset=utf-8".to_string(),
}],
..Default::default()
},
&UpdateSource::Sync,
&blob_manager,
)
.unwrap();
let id = response.id.clone();
(query_manager, temp_dir, id)
}
#[test]
fn info_reports_stored_size_and_content_type() {
let (qm, _tmp, id) = seed(Some(b"hello world"));
let info = FileResponseBodyStore::new(&qm).info(&id).unwrap();
assert_eq!(info.content_length, 11);
assert_eq!(info.content_type.as_deref(), Some("application/json; charset=utf-8"));
}
#[test]
fn chunks_cover_the_body_and_stop_short_at_the_end() {
let (qm, _tmp, id) = seed(Some(b"hello world"));
let store = FileResponseBodyStore::new(&qm);
assert_eq!(store.read_chunk(&id, 0, 5).unwrap(), b"hello");
assert_eq!(store.read_chunk(&id, 6, 100).unwrap(), b"world");
assert!(store.read_chunk(&id, 11, 100).unwrap().is_empty());
// Reading the same window twice gives the same bytes; nothing is consumed.
assert_eq!(store.read_chunk(&id, 0, 5).unwrap(), b"hello");
}
#[test]
fn a_response_with_no_body_is_empty_not_an_error() {
let (qm, _tmp, id) = seed(None);
let store = FileResponseBodyStore::new(&qm);
assert_eq!(store.info(&id).unwrap().content_length, 0);
assert!(store.read_chunk(&id, 0, 100).unwrap().is_empty());
}
#[test]
fn an_unknown_response_fails() {
let (qm, _tmp, _id) = seed(Some(b"hi"));
assert!(FileResponseBodyStore::new(&qm).info("rs_nope").is_err());
}
}