Proxy storage: flows table, bodies in blobs.sqlite, settings

apps/yaak-proxy/components/ExchangesTable.tsx

@@ -1,4 +1,4 @@
-import type { HttpExchange, ProxyHeader } from "@yaakapp-internal/proxy-lib";
+import type { Flow } from "@yaakapp-internal/proxy-lib";
 import {
   Table,
   TableBody,
@@ -11,7 +11,7 @@ import {
 import classNames from "classnames";
 
 interface Props {
-  exchanges: HttpExchange[];
+  exchanges: Flow[];
   style?: React.CSSProperties;
   className?: string;
 }
@@ -40,10 +40,10 @@ export function ExchangesTable({ exchanges, style, className }: Props) {
                 {ex.url}
               </TruncatedWideTableCell>
               <TableCell>
-                <StatusBadge status={ex.resStatus} error={ex.error} />
+                <StatusBadge status={ex.status} error={ex.error} />
               </TableCell>
               <TableCell className="text-text-subtle text-xs">
-                {getContentType(ex.resHeaders)}
+                {getContentType(ex.contentType)}
               </TableCell>
             </TableRow>
           ))}
@@ -69,8 +69,7 @@ function StatusBadge({ status, error }: { status: number | null; error: string |
   return <span className={classNames("text-xs font-mono", color)}>{status}</span>;
 }
 
-function getContentType(headers: ProxyHeader[]): string {
-  const ct = headers.find((h) => h.name.toLowerCase() === "content-type")?.value;
+function getContentType(ct: string | null): string {
   if (ct == null) return "—";
   // Strip parameters (e.g. "; charset=utf-8")
   return ct.split(";")[0]?.trim() ?? ct;

apps/yaak-proxy/components/Sidebar.tsx

@@ -1,10 +1,10 @@
-import type { HttpExchange } from "@yaakapp-internal/proxy-lib";
+import type { Flow } from "@yaakapp-internal/proxy-lib";
 import type { TreeNode } from "@yaakapp-internal/ui";
 import { selectedIdsFamily, Tree } from "@yaakapp-internal/ui";
 import { atom, useAtomValue } from "jotai";
 import { atomFamily } from "jotai-family";
 import { useCallback } from "react";
-import { httpExchangesAtom } from "../lib/store";
+import { flowsAtom } from "../lib/store";
 
 /** A node in the sidebar tree — either a domain or a path segment. */
 export type SidebarItem = {
@@ -18,13 +18,13 @@ const collapsedAtom = atomFamily((_treeId: string) => atom<Record<string, boolea
 export const SIDEBAR_TREE_ID = "proxy-sidebar";
 
 const sidebarTreeAtom = atom<TreeNode<SidebarItem>>((get) => {
-  const exchanges = get(httpExchangesAtom);
+  const exchanges = get(flowsAtom);
   return buildTree(exchanges);
 });
 
 /** Exchanges filtered by the currently selected sidebar node(s). */
 export const filteredExchangesAtom = atom((get) => {
-  const exchanges = get(httpExchangesAtom);
+  const exchanges = get(flowsAtom);
   const tree = get(sidebarTreeAtom);
   const selectedIds = get(selectedIdsFamily(SIDEBAR_TREE_ID));
 
@@ -72,7 +72,7 @@ function collectNodes(node: TreeNode<SidebarItem>, map: Map<string, SidebarItem>
  *         /123
  *       /orders
  */
-function buildTree(exchanges: HttpExchange[]): TreeNode<SidebarItem> {
+function buildTree(exchanges: Flow[]): TreeNode<SidebarItem> {
   const root: SidebarItem = { id: "root", label: "All Traffic", exchangeIds: [] };
   const rootNode: TreeNode<SidebarItem> = {
     item: root,

apps/yaak-proxy/lib/store.ts

@@ -1,11 +1,11 @@
 import { createModelStore } from "@yaakapp-internal/model-store";
-import type { HttpExchange } from "@yaakapp-internal/proxy-lib";
+import type { Flow } from "@yaakapp-internal/proxy-lib";
 
 type ProxyModels = {
-  http_exchange: HttpExchange;
+  flow: Flow;
 };
 
 export const { dataAtom, applyChange, replaceAll, listAtom, orderedListAtom } =
-  createModelStore<ProxyModels>(["http_exchange"]);
+  createModelStore<ProxyModels>(["flow"]);
 
-export const httpExchangesAtom = orderedListAtom("http_exchange", "createdAt", "desc");
+export const flowsAtom = orderedListAtom("flow", "startedAt", "desc");

apps/yaak-proxy/main.tsx

@@ -14,17 +14,15 @@ const jotaiStore = createStore();
 
 // Load initial models from the database
 void rpc("list_models", {}).then((res) => {
-  jotaiStore.set(dataAtom, (prev) => replaceAll(prev, "http_exchange", res.httpExchanges));
+  jotaiStore.set(dataAtom, (prev) => replaceAll(prev, "flow", res.flows));
 });
 
 // Register hotkeys from action metadata
 void initHotkeys();
 
 // Subscribe to model change events from the backend
 void listen("model_write", (payload) => {
-  jotaiStore.set(dataAtom, (prev) =>
-    applyChange(prev, "http_exchange", payload.model, payload.change),
-  );
+  jotaiStore.set(dataAtom, (prev) => applyChange(prev, "flow", payload.model, payload.change));
 });
 
 const motionFeatures = () => import("framer-motion").then((mod) => mod.domAnimation);

crates-cli/yaak-cli/Cargo.toml

@@ -45,6 +45,7 @@ webbrowser = "1"
 zip = "7"
 yaak = { workspace = true }
 yaak-api = { workspace = true }
+yaak-common = { workspace = true }
 yaak-core = { workspace = true }
 yaak-crypto = { workspace = true }
 yaak-http = { workspace = true }

crates-cli/yaak-cli/src/main.rs

@@ -10,7 +10,7 @@ mod version_check;
 use clap::{CommandFactory, FromArgMatches};
 use cli::{AGENT_HINTS, Cli, Commands, PluginCommands, RequestCommands};
 use context::{CliContext, CliExecutionContext};
-use std::path::PathBuf;
+use yaak_common::data_dir::resolve_data_dir;
 use yaak_models::queries::any_request::AnyRequest;
 
 /// Built at runtime so root `--help` can report whether the installed agent skill is
@@ -286,46 +286,3 @@ fn resolve_cookie_jar_id(
         .map(|jar| jar.id);
     Ok(default_cookie_jar)
 }
-
-fn resolve_data_dir(app_id: &str) -> PathBuf {
-    if let Some(dir) = wsl_data_dir(app_id) {
-        return dir;
-    }
-    dirs::data_dir().expect("Could not determine data directory").join(app_id)
-}
-
-/// Detect WSL and resolve the Windows AppData\Roaming path for the Yaak data directory.
-fn wsl_data_dir(app_id: &str) -> Option<PathBuf> {
-    if !cfg!(target_os = "linux") {
-        return None;
-    }
-
-    let proc_version = std::fs::read_to_string("/proc/version").ok()?;
-    let is_wsl = proc_version.to_lowercase().contains("microsoft");
-    if !is_wsl {
-        return None;
-    }
-
-    // We're in WSL, so try to resolve the Yaak app's data directory in Windows
-
-    // Get the Windows %APPDATA% path via cmd.exe
-    let appdata_output =
-        std::process::Command::new("cmd.exe").args(["/C", "echo", "%APPDATA%"]).output().ok()?;
-
-    let win_path = String::from_utf8(appdata_output.stdout).ok()?.trim().to_string();
-    if win_path.is_empty() || win_path == "%APPDATA%" {
-        return None;
-    }
-
-    // Convert Windows path to WSL path using wslpath (handles custom mount points)
-    let wslpath_output = std::process::Command::new("wslpath").arg(&win_path).output().ok()?;
-
-    let wsl_appdata = String::from_utf8(wslpath_output.stdout).ok()?.trim().to_string();
-    if wsl_appdata.is_empty() {
-        return None;
-    }
-
-    let wsl_path = PathBuf::from(wsl_appdata).join(app_id);
-
-    if wsl_path.exists() { Some(wsl_path) } else { None }
-}

crates-proxy/yaak-proxy-lib/Cargo.toml

@@ -15,7 +15,20 @@ rusqlite = { version = "0.38", features = ["bundled", "chrono"] }
 sea-query = { version = "1.0", features = ["with-chrono", "attr"] }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
+tokio = { workspace = true, features = ["io-util"] }
 ts-rs = { workspace = true, features = ["chrono-impl"] }
 yaak-database = { workspace = true }
+yaak-http = { workspace = true }
+yaak-models = { workspace = true }
 yaak-proxy = { workspace = true }
 yaak-rpc = { workspace = true }
+
+[dev-dependencies]
+bytes = "1"
+flate2 = "1"
+http-body-util = "0.1"
+hyper = { version = "1", features = ["http1", "server"] }
+hyper-util = { version = "0.1", features = ["tokio"] }
+reqwest = { workspace = true }
+tempfile = "3"
+tokio = { workspace = true, features = ["macros", "net", "rt-multi-thread", "time"] }

crates-proxy/yaak-proxy-lib/migrations/00000002_flows.sql

@@ -0,0 +1,27 @@
+DROP TABLE http_exchanges;
+
+CREATE TABLE flows
+(
+    id                    TEXT     NOT NULL PRIMARY KEY,
+    created_at            DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
+    updated_at            DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
+    started_at            DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
+    method                TEXT     NOT NULL DEFAULT '',
+    url                   TEXT     NOT NULL DEFAULT '',
+    client_http_version   TEXT     NOT NULL DEFAULT '',
+    upstream_http_version TEXT,
+    remote_addr           TEXT,
+    status                INTEGER,
+    content_type          TEXT,
+    request_headers       TEXT     NOT NULL DEFAULT '[]',
+    response_headers      TEXT     NOT NULL DEFAULT '[]',
+    request_size          INTEGER  NOT NULL DEFAULT 0,
+    response_size         INTEGER  NOT NULL DEFAULT 0,
+    request_truncated     BOOLEAN  NOT NULL DEFAULT FALSE,
+    response_truncated    BOOLEAN  NOT NULL DEFAULT FALSE,
+    elapsed_headers_ms    INTEGER,
+    elapsed_ms            INTEGER,
+    error                 TEXT
+);
+
+CREATE INDEX idx_flows_started_at ON flows (started_at);

crates-proxy/yaak-proxy-lib/src/bodies.rs

@@ -0,0 +1,37 @@
+use crate::models::{Flow, FlowBody, find_header};
+use std::io::Cursor;
+use tokio::io::AsyncReadExt;
+use yaak_http::decompress::{ContentEncoding, streaming_decoder};
+use yaak_models::blob_manager::BlobManager;
+
+/// A flow's body as stored: the bytes on the wire, up to the capture limit.
+pub fn read_raw_body(
+    blobs: &BlobManager,
+    flow_id: &str,
+    body: FlowBody,
+) -> Result<Vec<u8>, String> {
+    let chunks =
+        blobs.connect().get_chunks(&Flow::body_id(flow_id, body)).map_err(|e| e.to_string())?;
+    Ok(chunks.into_iter().flat_map(|c| c.data).collect())
+}
+
+/// A flow's body decoded with its Content-Encoding. A truncated body decodes as far as its
+/// stored bytes go.
+pub async fn read_body(
+    blobs: &BlobManager,
+    flow: &Flow,
+    body: FlowBody,
+) -> Result<Vec<u8>, String> {
+    let raw = read_raw_body(blobs, &flow.id, body)?;
+    let encoding = ContentEncoding::from_header(find_header(flow.headers(body), "content-encoding"));
+    if encoding == ContentEncoding::Identity {
+        return Ok(raw);
+    }
+
+    let mut decoded = Vec::new();
+    match streaming_decoder(Cursor::new(raw), encoding).read_to_end(&mut decoded).await {
+        Ok(_) => Ok(decoded),
+        Err(_) if flow.truncated(body) => Ok(decoded),
+        Err(e) => Err(format!("Failed to decode {encoding:?} body: {e}")),
+    }
+}