diff --git a/Cargo.lock b/Cargo.lock index 551f9cf7..aa591d97 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -613,11 +613,15 @@ checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f" dependencies = [ "async-trait", "axum-core", + "axum-macros", + "base64 0.22.1", "bytes", "futures-util", "http", "http-body", "http-body-util", + "hyper", + "hyper-util", "itoa", "matchit", "memchr", @@ -626,10 +630,17 @@ dependencies = [ "pin-project-lite", "rustversion", "serde", + "serde_json", + "serde_path_to_error", + "serde_urlencoded", + "sha1", "sync_wrapper", + "tokio", + "tokio-tungstenite 0.24.0", "tower 0.5.2", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -650,6 +661,18 @@ dependencies = [ "sync_wrapper", "tower-layer", "tower-service", + "tracing", +] + +[[package]] +name = "axum-macros" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57d123550fa8d071b7255cb0cc04dc302baa6c8c4a79f55701552684d8399bce" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.101", ] [[package]] @@ -3233,6 +3256,12 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "21dec9db110f5f872ed9699c3ecf50cf16f423502706ba5c72462e28d3157573" +[[package]] +name = "http-range-header" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9171a2ea8a68358193d15dd5d70c1c10a2afc3e7e4c5bc92bc9f025cebd7359c" + [[package]] name = "httparse" version = "1.10.1" @@ -9080,6 +9109,18 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-tungstenite" +version = "0.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edc5f74e248dc973e0dbb7b74c7e0d6fcc301c694ff50049504004ef4d0cdcd9" +dependencies = [ + "futures-util", + "log 0.4.29", + "tokio", + "tungstenite 0.24.0", +] + [[package]] name = "tokio-tungstenite" version = "0.26.2" @@ -9093,7 +9134,7 @@ dependencies = [ "rustls-pki-types", "tokio", "tokio-rustls", - "tungstenite", + "tungstenite 0.26.2", ] [[package]] @@ -9287,6 +9328,7 @@ dependencies = [ "tokio", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -9297,13 +9339,23 @@ checksum = "68d6fdd9f81c2819c9a8b0e0cd91660e7746a8e6ea2ba7c6b2b057985f6bcb51" dependencies = [ "bitflags 2.11.0", "bytes", + "futures-core", "futures-util", "http", "http-body", + "http-body-util", + "http-range-header", + "httpdate", + "mime", + "mime_guess", + "percent-encoding", "pin-project-lite", + "tokio", + "tokio-util", "tower 0.5.2", "tower-layer", "tower-service", + "tracing", "url", ] @@ -9325,6 +9377,7 @@ version = "0.1.41" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "784e0ac535deb450455cbfa28a6f0df145ea1bb7ae51b821cf5e7927fdcfbdd0" dependencies = [ + "log 0.4.29", "pin-project-lite", "tracing-attributes", "tracing-core", @@ -9471,6 +9524,24 @@ dependencies = [ "termcolor", ] +[[package]] +name = "tungstenite" +version = "0.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18e5b8366ee7a95b16d32197d0b2604b43a0be89dc5fac9f8e96ccafbaedda8a" +dependencies = [ + "byteorder", + "bytes", + "data-encoding", + "http", + "httparse", + "log 0.4.29", + "rand 0.8.5", + "sha1", + "thiserror 1.0.69", + "utf-8", +] + [[package]] name = "tungstenite" version = "0.26.2" @@ -11016,7 +11087,7 @@ dependencies = [ "thiserror 2.0.17", "tokio", "tokio-stream", - "tokio-tungstenite", + "tokio-tungstenite 0.26.2", "tokio-util", "ts-rs", "url", @@ -11317,7 +11388,7 @@ dependencies = [ "sha2", "thiserror 2.0.17", "tokio", - "tokio-tungstenite", + "tokio-tungstenite 0.26.2", "ts-rs", "yaak-common", "yaak-crypto", @@ -11372,6 +11443,40 @@ dependencies = [ "ts-rs", ] +[[package]] +name = "yaak-server" +version = "0.1.0" +dependencies = [ + "axum", + "charset", + "chrono", + "clap", + "dirs", + "env_logger", + "eventsource-client", + "futures", + "include_dir", + "log 0.4.29", + "mime_guess", + "pretty_graphql", + "rand 0.8.5", + "serde", + "serde_json", + "serde_urlencoded", + "tokio", + "tower-http", + "yaak", + "yaak-common", + "yaak-core", + "yaak-crypto", + "yaak-http", + "yaak-models", + "yaak-plugins", + "yaak-rpc", + "yaak-sse", + "yaak-templates", +] + [[package]] name = "yaak-sse" version = "0.1.0" @@ -11472,7 +11577,7 @@ dependencies = [ "serde_json", "thiserror 2.0.17", "tokio", - "tokio-tungstenite", + "tokio-tungstenite 0.26.2", "url", "yaak-http", "yaak-models", diff --git a/Cargo.toml b/Cargo.toml index 81df1687..7e44de65 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,6 +25,8 @@ members = [ "crates-proxy/yaak-proxy-lib", # CLI crates "crates-cli/yaak-cli", + # Headless server crates + "crates-server/yaak-server", # Tauri-specific crates "crates-tauri/yaak-app-client", "crates-tauri/yaak-app-proxy", diff --git a/crates-server/yaak-server/Cargo.toml b/crates-server/yaak-server/Cargo.toml new file mode 100644 index 00000000..7fecf7cb --- /dev/null +++ b/crates-server/yaak-server/Cargo.toml @@ -0,0 +1,47 @@ +[package] +name = "yaak-server" +version = "0.1.0" +edition = "2024" +publish = false + +[[bin]] +name = "yaak-bridge" +path = "src/main.rs" + +[dependencies] +axum = { version = "0.7", features = ["ws", "macros"] } +charset = "0.1" +chrono = { workspace = true } +clap = { version = "4", features = ["derive", "env"] } +dirs = "6" +env_logger = "0.11" +eventsource-client = { git = "https://github.com/yaakapp/rust-eventsource-client", version = "0.14.0" } +futures = "0.3" +include_dir = "0.7" +log = { workspace = true } +mime_guess = "2" +pretty_graphql = "0.2" +rand = "0.8" +serde = { workspace = true } +serde_json = { workspace = true } +serde_urlencoded = "0.7" +tokio = { workspace = true, features = [ + "rt-multi-thread", + "macros", + "io-util", + "net", + "signal", + "time", + "sync", +] } +tower-http = { version = "0.6", features = ["cors", "fs", "trace"] } +yaak = { workspace = true } +yaak-common = { workspace = true } +yaak-core = { workspace = true } +yaak-crypto = { workspace = true } +yaak-http = { workspace = true } +yaak-models = { workspace = true } +yaak-plugins = { workspace = true } +yaak-rpc = { workspace = true } +yaak-sse = { workspace = true } +yaak-templates = { workspace = true } diff --git a/crates-server/yaak-server/README.md b/crates-server/yaak-server/README.md new file mode 100644 index 00000000..74d43dd4 --- /dev/null +++ b/crates-server/yaak-server/README.md @@ -0,0 +1,94 @@ +# Yaak Bridge + +A headless binary that runs the real Yaak engine for a browser tab. + +The tab is the unmodified Yaak UI. Everything a page cannot do — send an HTTP +request and see every response header, follow redirects, keep a cookie jar, run +the plugin runtime, read a response body off disk — happens in this process, +reached over local HTTP and a WebSocket. + +This is the reason a browser Yaak can be credible at all. An in-page `fetch` +sender only ever sees the CORS-safelisted response headers: measured against +httpbin, a server that sent 8 headers yielded 2. Through the bridge the same +request yields all 8, plus the redirect chain, `Set-Cookie`, connection timings +and client certificates. + +## Running it + +Start the bridge: + +```bash +cargo run -p yaak-server -- --port 9444 +``` + +It binds `127.0.0.1` only and prints a bearer token that every route requires. + +Then point a frontend at it. In dev, run Vite separately and tell it where the +bridge is: + +```bash +YAAK_CLIENT_DEV_PORT=1472 VITE_YAAK_BRIDGE_URL=http://127.0.0.1:9444 npm run dev --workspace apps/yaak-client +``` + +Open `http://localhost:1472/?bridgeToken=`. The token is consumed from +the query, kept for the session, and stripped from the address bar. Without one +you get a small connect form. + +To serve the built frontend from the bridge itself instead, so there is only one +process: + +```bash +npm run build --workspace apps/yaak-client +cargo run -p yaak-server -- --web-dir dist/apps/yaak-client +``` + +## Shape + +| Route | What it carries | +| --- | --- | +| `POST /rpc` | The yaak-rpc envelope, the same one Tauri's `invoke` wraps on the desktop | +| `GET /events` | WebSocket. Server to client: `model_writes`, `stream_{id}`, toasts, plugin events. Client to server: the tab's location, and replies to prompts | +| `GET /responses/:id/body` | Response bodies, with Range support. Replaces reading `bodyPath` off disk | +| `GET /bridge/info` | Capabilities and the implemented command list | + +Auth is a bearer token in the `Authorization` header, or a `token` query +parameter for the two requests the browser issues itself (the WebSocket, and +``-style body loads). It is dev-grade and deliberately minimal: OTP +pairing and request encryption replace it, and `require_token` in `http.rs` is +where they go. + +## Relationship to the other hosts + +The engine crates under `crates/` are Tauri-free, and `crates-cli/yaak-cli` +already proved they run headless. This crate is structurally the CLI's +`CliContext` with an event hub attached — same `init_standalone` database, same +`PluginManager` over the same Node sidecar. + +Two things are ported deliberately rather than invented: + +- **Model writes** (`model_writes.rs`) keep the desktop's two paths: an + in-memory channel for writes this process made, and a poll of the + `model_changes` table so external writers — the CLI, the desktop app open on + the same database — show up live in the browser. +- **Plugin host requests** (`plugin_events.rs`) let `yaak::plugin_events` + answer everything that is only a database question, exactly as the CLI and the + desktop do. Only the host-specific arms differ, and where the CLI answers a + prompt from a TTY, the bridge round-trips it to the tab the way the desktop + round-trips it to a window. + +## Known gaps + +- **Settings is unreachable.** The desktop opens it via `cmd_new_child_window`. + A tab is one window, `multiWindow` is false, and this task did not add in-page + routing for it. +- **One tab at a time.** Model writes broadcast correctly to every connected + tab, so two tabs stay in sync for reads. What breaks is the session: the + tab's reported URL lives in a single slot, so with two tabs in different + workspaces a plugin's template render resolves against whichever attached + last. Prompts also broadcast, so a dialog raised by one tab appears in both. +- **No local files.** There is no file dialog, so request bodies from disk, + export, and save-response are unsupported. `cmd_import_data` is registered and + works, but only for a path typed by hand on the bridge's machine. +- **Command subset.** Roughly 40 of the desktop's 107 commands are implemented. + The rest return a structured "not supported on this host" error naming the + command; `UNSUPPORTED_COMMANDS` in `rpc/mod.rs` lists them. diff --git a/crates-server/yaak-server/src/events.rs b/crates-server/yaak-server/src/events.rs new file mode 100644 index 00000000..47f62e75 --- /dev/null +++ b/crates-server/yaak-server/src/events.rs @@ -0,0 +1,117 @@ +//! The events channel: everything the browser tab would have received as a +//! Tauri window event. +//! +//! Two directions ride the same WebSocket. Server to client is a broadcast, so +//! `model_writes`, `stream_{id}` messages, toasts and plugin events all reach +//! the tab through one pipe. Client to server exists because some plugin host +//! requests are questions — a prompt round-trips through the UI and comes back +//! keyed by the originating event's id, exactly as the desktop app's +//! `call_frontend` does with window events. + +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use tokio::sync::{broadcast, mpsc}; + +/// One frame in either direction: a name and a JSON payload. +/// +/// Deliberately the same shape both ways, and the same shape as the desktop's +/// event payloads, so `platform.listen` on the browser side hands the payload +/// to callers unwrapped. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct EventFrame { + pub event: String, + #[serde(default)] + pub payload: serde_json::Value, +} + +#[derive(Clone)] +pub struct EventHub { + outbound: broadcast::Sender, + /// Listeners waiting on a named event from the client, keyed by event name. + inbound: Arc>>>>, +} + +/// A subscription to one named client-sent event. Deregisters on drop, so a +/// prompt that is never answered doesn't leak a listener for the process's life. +pub struct InboundSubscription { + event: String, + rx: mpsc::UnboundedReceiver, + inbound: Arc>>>>, +} + +impl InboundSubscription { + pub async fn recv(&mut self) -> Option { + self.rx.recv().await + } +} + +impl Drop for InboundSubscription { + fn drop(&mut self) { + let mut inbound = match self.inbound.lock() { + Ok(inbound) => inbound, + Err(poisoned) => poisoned.into_inner(), + }; + if let Some(senders) = inbound.get_mut(&self.event) { + senders.retain(|tx| !tx.is_closed()); + if senders.is_empty() { + inbound.remove(&self.event); + } + } + } +} + +impl EventHub { + pub fn new() -> Self { + // Bounded: a tab that stops reading gets dropped frames rather than + // growing the server's memory without limit. Model writes are the + // high-volume case (imports, bulk deletes) and they arrive in batches. + let (outbound, _) = broadcast::channel(1024); + Self { outbound, inbound: Arc::new(Mutex::new(HashMap::new())) } + } + + /// Send an event to every connected tab. Fails silently when none is + /// connected, which is the normal state before a browser attaches. + pub fn emit(&self, event: impl Into, payload: &T) { + let payload = match serde_json::to_value(payload) { + Ok(payload) => payload, + Err(e) => { + log::warn!("Failed to serialize event payload: {e}"); + return; + } + }; + let _ = self.outbound.send(EventFrame { event: event.into(), payload }); + } + + pub fn subscribe(&self) -> broadcast::Receiver { + self.outbound.subscribe() + } + + /// Listen for a named event sent *by* the client. + pub fn subscribe_inbound(&self, event: impl Into) -> InboundSubscription { + let event = event.into(); + let (tx, rx) = mpsc::unbounded_channel(); + let mut inbound = match self.inbound.lock() { + Ok(inbound) => inbound, + Err(poisoned) => poisoned.into_inner(), + }; + inbound.entry(event.clone()).or_default().push(tx); + drop(inbound); + InboundSubscription { event, rx, inbound: Arc::clone(&self.inbound) } + } + + /// Route a frame that arrived from a tab to whoever is waiting on it. + pub fn dispatch_inbound(&self, frame: EventFrame) { + let mut inbound = match self.inbound.lock() { + Ok(inbound) => inbound, + Err(poisoned) => poisoned.into_inner(), + }; + let Some(senders) = inbound.get_mut(&frame.event) else { + return; + }; + senders.retain(|tx| tx.send(frame.payload.clone()).is_ok()); + if senders.is_empty() { + inbound.remove(&frame.event); + } + } +} diff --git a/crates-server/yaak-server/src/http.rs b/crates-server/yaak-server/src/http.rs new file mode 100644 index 00000000..865cce12 --- /dev/null +++ b/crates-server/yaak-server/src/http.rs @@ -0,0 +1,355 @@ +//! The front door: one HTTP surface for the browser tab. +//! +//! Three routes carry everything. `POST /rpc` is the yaak-rpc envelope, byte for +//! byte what the desktop puts inside Tauri's `invoke`. `GET /events` is the +//! WebSocket that replaces window events, in both directions. And +//! `GET /responses/:id/body` replaces reading `bodyPath` off disk, which a tab +//! cannot do. + +use crate::events::EventFrame; +use crate::rpc::BridgeCtx; +use crate::session::SessionContext; +use crate::state::BridgeState; +use axum::body::Body; +use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade}; +use axum::extract::{Path, Query, Request, State}; +use axum::http::{HeaderMap, StatusCode, header}; +use axum::middleware::Next; +use axum::response::{IntoResponse, Response}; +use axum::routing::{get, post}; +use axum::{Json, Router}; +use serde::{Deserialize, Serialize}; +use std::sync::Arc; +use tokio::io::{AsyncReadExt, AsyncSeekExt}; +use tower_http::cors::CorsLayer; +use yaak_rpc::{RpcRequest, RpcResponse, RpcRouter}; + +#[derive(Clone)] +pub struct AppState { + pub state: Arc, + pub router: Arc>, +} + +pub fn build_app(state: Arc, router: Arc>) -> Router { + let app_state = AppState { state: state.clone(), router }; + + let api = Router::new() + .route("/bridge/info", get(bridge_info)) + .route("/rpc", post(rpc_handler)) + .route("/events", get(events_handler)) + .route("/responses/:id/body", get(response_body)) + .layer(axum::middleware::from_fn_with_state(state.clone(), require_token)) + // The dev setup serves the frontend from Vite on another port, so the + // tab's origin is not the bridge's. Credentials never ride on cookies + // here — the token is explicit — so a permissive CORS layer is safe and + // is bounded by the token check that runs before it. + .layer(CorsLayer::permissive()) + .with_state(app_state); + + match std::env::var("YAAK_BRIDGE_WEB_DIR").ok() { + // Serving the built frontend makes the bridge a single process to run. + // `index.html` is the fallback because the router owns the paths. + Some(dir) => api.fallback_service( + tower_http::services::ServeDir::new(&dir) + .fallback(tower_http::services::ServeFile::new(format!("{dir}/index.html"))), + ), + None => api, + } +} + +// -- Auth -- + +#[derive(Debug, Deserialize)] +struct TokenQuery { + token: Option, +} + +/// Dev-grade bearer check on every route. +/// +/// The header is the normal path. The query parameter exists because two of +/// these are opened by the browser itself — the WebSocket and the `` +/// pointing at a response body — and neither lets the page set headers. +/// +/// This is the seam where OTP pairing and per-session keys go. It is not one +/// today: the token is a process-lifetime shared secret, and anything that can +/// read the tab's URL can read it. +async fn require_token( + State(state): State>, + request: Request, + next: Next, +) -> Response { + let from_header = request + .headers() + .get(header::AUTHORIZATION) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.strip_prefix("Bearer ")) + .map(|v| v.to_string()); + + let from_query = request + .uri() + .query() + .and_then(|q| serde_urlencoded::from_str::(q).ok()) + .and_then(|q| q.token); + + let presented = from_header.or(from_query); + + match presented { + Some(token) if constant_time_eq(&token, &state.token) => next.run(request).await, + _ => (StatusCode::UNAUTHORIZED, "Invalid or missing bridge token").into_response(), + } +} + +/// Compares without returning early on the first differing byte, so a caller +/// can't learn the token one character at a time. +fn constant_time_eq(a: &str, b: &str) -> bool { + if a.len() != b.len() { + return false; + } + a.bytes().zip(b.bytes()).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0 +} + +// -- Routes -- + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct BridgeInfo { + name: String, + version: String, + capabilities: crate::state::BridgeCapabilities, + /// Commands this build implements. The browser host uses it to fail fast + /// with a clear message instead of waiting for a round trip. + commands: Vec, +} + +async fn bridge_info(State(app): State) -> Json { + Json(BridgeInfo { + name: "Yaak Bridge".to_string(), + version: env!("CARGO_PKG_VERSION").to_string(), + capabilities: app.state.capabilities.clone(), + commands: crate::rpc::implemented_commands(&app.router), + }) +} + +/// One envelope in, one out. Errors are carried inside the envelope, not as an +/// HTTP status, so the browser host can reject the caller's promise with the +/// backend's own message. +async fn rpc_handler( + State(app): State, + Json(req): Json, +) -> Json { + let ctx = BridgeCtx { state: app.state.clone(), session: app.state.session.get() }; + log::debug!("RPC {}", req.cmd); + let response = app.router.handle(req, &ctx).await; + if let RpcResponse::Error { error, .. } = &response { + log::warn!("RPC failed: {error}"); + } + Json(response) +} + +async fn events_handler(State(app): State, ws: WebSocketUpgrade) -> Response { + ws.on_upgrade(move |socket| handle_events_socket(socket, app)) +} + +/// The tab's first frame reports who and where it is; everything after that is +/// a reply to something the server asked. +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct AttachPayload { + label: String, + url: String, +} + +async fn handle_events_socket(socket: WebSocket, app: AppState) { + use futures::{SinkExt, StreamExt}; + + let (mut sink, mut stream) = socket.split(); + let mut outbound = app.state.events.subscribe(); + + // Server to client. + let send_task = tokio::spawn(async move { + loop { + match outbound.recv().await { + Ok(frame) => { + let Ok(text) = serde_json::to_string(&frame) else { + continue; + }; + if sink.send(Message::Text(text)).await.is_err() { + break; + } + } + // A tab that fell behind has missed writes, and the model store + // would be silently stale. Close instead, so a reconnect + // re-reads the workspace from scratch. + Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { + log::warn!("Events client lagged by {n} frames; closing so it resyncs"); + break; + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + } + }); + + // Client to server. + let state = app.state.clone(); + let recv_task = tokio::spawn(async move { + while let Some(Ok(message)) = stream.next().await { + let Message::Text(text) = message else { + continue; + }; + let Ok(frame) = serde_json::from_str::(&text) else { + log::warn!("Ignoring malformed event frame from browser"); + continue; + }; + + // `bridge_attach` is the browser telling us what the desktop would + // have read off the window: its label and its current URL. + if frame.event == "bridge_attach" { + match serde_json::from_value::(frame.payload.clone()) { + Ok(attach) => { + log::info!("Browser attached: {} at {}", attach.label, attach.url); + state.session.set(SessionContext { + label: attach.label, + url: attach.url, + }); + } + Err(e) => log::warn!("Bad bridge_attach payload: {e}"), + } + continue; + } + + state.events.dispatch_inbound(frame); + } + }); + + tokio::select! { + _ = send_task => {}, + _ = recv_task => {}, + } +} + +#[derive(Debug, Deserialize)] +struct BodyQuery { + /// Present so the shared token extractor doesn't reject the request; the + /// value itself is checked in the middleware. + #[allow(dead_code)] + token: Option, +} + +/// Stream a response body, with Range support. +/// +/// Keyed by response id rather than by path: the tab hands back a `bodyPath` +/// the backend gave it, and resolving that through the database means this +/// route can only ever serve a file the engine wrote, not an arbitrary path a +/// page asked for. Range matters because the video and audio viewers seek. +async fn response_body( + State(app): State, + Path(id): Path, + Query(_q): Query, + headers: HeaderMap, +) -> Response { + let response = match app.state.db().get_http_response(&id) { + Ok(response) => response, + Err(_) => return (StatusCode::NOT_FOUND, "No such response").into_response(), + }; + + let Some(body_path) = response.body_path else { + return (StatusCode::NOT_FOUND, "Response has no body").into_response(); + }; + + let mut file = match tokio::fs::File::open(&body_path).await { + Ok(file) => file, + Err(e) => return (StatusCode::NOT_FOUND, format!("Body unavailable: {e}")).into_response(), + }; + + let total = match file.metadata().await { + Ok(meta) => meta.len(), + Err(e) => { + return (StatusCode::INTERNAL_SERVER_ERROR, format!("Body unreadable: {e}")) + .into_response(); + } + }; + + let content_type = response + .headers + .iter() + .find(|h| h.name.eq_ignore_ascii_case("content-type")) + .map(|h| h.value.clone()) + .unwrap_or_else(|| "application/octet-stream".to_string()); + + let range = headers.get(header::RANGE).and_then(|v| v.to_str().ok()).and_then(parse_range); + + let (start, end, status) = match range { + Some((start, end)) => { + let end = end.unwrap_or(total.saturating_sub(1)).min(total.saturating_sub(1)); + if total == 0 || start > end { + return Response::builder() + .status(StatusCode::RANGE_NOT_SATISFIABLE) + .header(header::CONTENT_RANGE, format!("bytes */{total}")) + .body(Body::empty()) + .unwrap(); + } + (start, end, StatusCode::PARTIAL_CONTENT) + } + None => (0, total.saturating_sub(1), StatusCode::OK), + }; + + let length = if total == 0 { 0 } else { end - start + 1 }; + + if file.seek(std::io::SeekFrom::Start(start)).await.is_err() { + return (StatusCode::INTERNAL_SERVER_ERROR, "Failed to seek body").into_response(); + } + + let mut buf = vec![0u8; length as usize]; + if let Err(e) = file.read_exact(&mut buf).await { + return (StatusCode::INTERNAL_SERVER_ERROR, format!("Failed to read body: {e}")) + .into_response(); + } + + let mut builder = Response::builder() + .status(status) + .header(header::CONTENT_TYPE, content_type) + .header(header::ACCEPT_RANGES, "bytes") + .header(header::CONTENT_LENGTH, length); + + if status == StatusCode::PARTIAL_CONTENT { + builder = builder.header(header::CONTENT_RANGE, format!("bytes {start}-{end}/{total}")); + } + + builder.body(Body::from(buf)).unwrap() +} + +/// Parses a single `bytes=start-end` range. Multi-range requests are not +/// answered as multipart; the first range is used, which browsers accept. +fn parse_range(value: &str) -> Option<(u64, Option)> { + let spec = value.strip_prefix("bytes=")?.split(',').next()?.trim(); + let (start, end) = spec.split_once('-')?; + if start.is_empty() { + return None; + } + let start: u64 = start.parse().ok()?; + let end = if end.is_empty() { None } else { Some(end.parse().ok()?) }; + Some((start, end)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_ranges() { + assert_eq!(parse_range("bytes=0-499"), Some((0, Some(499)))); + assert_eq!(parse_range("bytes=500-"), Some((500, None))); + assert_eq!(parse_range("bytes=0-99,200-299"), Some((0, Some(99)))); + // Suffix ranges ("last 500 bytes") aren't supported; callers get the + // whole body, which is correct if wasteful. + assert_eq!(parse_range("bytes=-500"), None); + assert_eq!(parse_range("nonsense"), None); + } + + #[test] + fn token_comparison_requires_exact_match() { + assert!(constant_time_eq("abc", "abc")); + assert!(!constant_time_eq("abc", "abd")); + assert!(!constant_time_eq("abc", "abcd")); + } +} diff --git a/crates-server/yaak-server/src/main.rs b/crates-server/yaak-server/src/main.rs new file mode 100644 index 00000000..4b189b83 --- /dev/null +++ b/crates-server/yaak-server/src/main.rs @@ -0,0 +1,121 @@ +//! Yaak Bridge — the local companion that runs the real Yaak engine for a +//! browser tab. +//! +//! The tab is the Yaak UI, unchanged. Everything it cannot do in a page — +//! sending an HTTP request and seeing every response header, following +//! redirects, keeping a cookie jar, running plugins, reading a response body +//! off disk — happens in this process, over a local HTTP and WebSocket +//! connection. +//! +//! Loopback only, and every route needs the token printed at startup. + +mod events; +mod http; +mod model_writes; +mod plugin_events; +mod rpc; +mod session; +mod state; + +use clap::Parser; +use rand::Rng; +use std::net::SocketAddr; +use std::path::PathBuf; +use std::sync::Arc; + +const APP_ID: &str = "app.yaak.bridge"; + +#[derive(Parser, Debug)] +#[command(name = "yaak-bridge", about = "Run the Yaak engine for a browser tab")] +struct Args { + /// Port to listen on. Loopback only, always. + #[arg(long, default_value_t = 9444, env = "YAAK_BRIDGE_PORT")] + port: u16, + + /// Where the database, plugins and response bodies live. + #[arg(long, env = "YAAK_BRIDGE_DATA_DIR")] + data_dir: Option, + + /// Use a fixed token instead of generating one. For scripted dev loops. + #[arg(long, env = "YAAK_BRIDGE_TOKEN")] + token: Option, + + /// Where the frontend was built to. Serving it makes this the only process + /// to run; without it, point a Vite dev server at this bridge instead. + #[arg(long, env = "YAAK_BRIDGE_WEB_DIR")] + web_dir: Option, +} + +#[tokio::main] +async fn main() { + env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init(); + + let args = Args::parse(); + + let data_dir = args.data_dir.unwrap_or_else(default_data_dir); + if let Err(e) = std::fs::create_dir_all(&data_dir) { + eprintln!("Error: failed to create data dir {}: {e}", data_dir.display()); + std::process::exit(1); + } + + if let Some(web_dir) = &args.web_dir { + // Read back by the router; keeping it in the environment avoids + // threading an option through every layer for a dev-mode convenience. + unsafe { std::env::set_var("YAAK_BRIDGE_WEB_DIR", web_dir) }; + } + + let token = args.token.unwrap_or_else(generate_token); + let is_dev = cfg!(debug_assertions); + + let mut state = state::BridgeState::new(data_dir.clone(), APP_ID, token.clone(), is_dev); + state.init_plugins().await; + let state = Arc::new(state); + + let router = Arc::new(rpc::build_router()); + let app = http::build_app(state.clone(), router); + + let addr = SocketAddr::from(([127, 0, 0, 1], args.port)); + let listener = match tokio::net::TcpListener::bind(addr).await { + Ok(listener) => listener, + Err(e) => { + eprintln!("Error: failed to bind {addr}: {e}"); + std::process::exit(1); + } + }; + + let base = format!("http://127.0.0.1:{}", args.port); + println!(); + println!(" Yaak Bridge listening on {base}"); + println!(" Data dir: {}", data_dir.display()); + println!(" Plugins: {}", if state.capabilities.plugins { "running" } else { "unavailable" }); + println!(); + if std::env::var("YAAK_BRIDGE_WEB_DIR").is_ok() { + println!(" Open: {base}/?bridgeToken={token}"); + } else { + println!(" Token: {token}"); + println!(" Open your dev server with ?bridgeToken={token}"); + } + println!(); + + let shutdown_state = state.clone(); + let server = axum::serve(listener, app).with_graceful_shutdown(async move { + let _ = tokio::signal::ctrl_c().await; + log::info!("Shutting down"); + shutdown_state.shutdown().await; + }); + + if let Err(e) = server.await { + eprintln!("Error: server failed: {e}"); + std::process::exit(1); + } +} + +fn default_data_dir() -> PathBuf { + dirs::data_dir().unwrap_or_else(|| PathBuf::from(".")).join("yaak-bridge") +} + +/// A 256-bit random token, hex encoded. Per process, never written to disk. +fn generate_token() -> String { + let bytes: [u8; 32] = rand::thread_rng().r#gen(); + bytes.iter().map(|b| format!("{b:02x}")).collect() +} diff --git a/crates-server/yaak-server/src/model_writes.rs b/crates-server/yaak-server/src/model_writes.rs new file mode 100644 index 00000000..4d245d38 --- /dev/null +++ b/crates-server/yaak-server/src/model_writes.rs @@ -0,0 +1,125 @@ +//! Pushing model writes to the connected tab. +//! +//! A direct port of the desktop's two paths (see +//! crates-tauri/yaak-app-client/src/models_ext.rs), and for the same reason: +//! the in-memory channel is the fast path for writes this process made on a +//! client's behalf, while polling the `model_changes` table is what makes an +//! external writer — the CLI, a second bridge, the desktop app open on the same +//! database — show up live in the browser. Keeping both means the browser +//! behaves like the desktop rather than like a cache. + +use crate::events::EventHub; +use chrono::Utc; +use log::error; +use std::sync::mpsc::Receiver; +use std::time::Duration; +use yaak_models::query_manager::QueryManager; +use yaak_models::util::{ModelPayload, UpdateSource}; + +const MODEL_CHANGES_RETENTION_HOURS: i64 = 1; +const MODEL_CHANGES_POLL_INTERVAL_MS: u64 = 1000; +const MODEL_CHANGES_POLL_BATCH_SIZE: usize = 200; + +struct ModelChangeCursor { + created_at: String, + id: i64, +} + +impl ModelChangeCursor { + fn from_launch_time() -> Self { + Self { + created_at: Utc::now().naive_utc().format("%Y-%m-%d %H:%M:%S%.3f").to_string(), + id: 0, + } + } +} + +pub fn start(query_manager: &QueryManager, rx: Receiver, events: EventHub) { + if let Err(err) = + query_manager.connect().prune_model_changes_older_than_hours(MODEL_CHANGES_RETENTION_HOURS) + { + error!("Failed to prune model_changes rows on startup: {err:?}"); + } + + // Only stream writes that happen after this process started. + let cursor = ModelChangeCursor::from_launch_time(); + let poll_query_manager = query_manager.clone(); + let poll_events = events.clone(); + tokio::spawn(async move { + run_model_change_poller(poll_query_manager, poll_events, cursor).await; + }); + + // `init_standalone` hands back a std (blocking) receiver, so it gets a + // thread rather than a task. + std::thread::spawn(move || { + while let Ok(payload) = rx.recv() { + let mut batch: Vec = Vec::new(); + if matches!(payload.update_source, UpdateSource::Window { .. }) { + batch.push(payload); + } + // Coalesce anything already queued into the same frame. + while let Ok(next) = rx.try_recv() { + if matches!(next.update_source, UpdateSource::Window { .. }) { + batch.push(next); + } + } + if batch.is_empty() { + continue; + } + events.emit("model_writes", &batch); + } + }); +} + +async fn run_model_change_poller( + query_manager: QueryManager, + events: EventHub, + mut cursor: ModelChangeCursor, +) { + loop { + while drain_model_changes_batch(&query_manager, &events, &mut cursor) {} + tokio::time::sleep(Duration::from_millis(MODEL_CHANGES_POLL_INTERVAL_MS)).await; + } +} + +fn drain_model_changes_batch( + query_manager: &QueryManager, + events: &EventHub, + cursor: &mut ModelChangeCursor, +) -> bool { + let changes = match query_manager.connect().list_model_changes_since( + &cursor.created_at, + cursor.id, + MODEL_CHANGES_POLL_BATCH_SIZE, + ) { + Ok(changes) => changes, + Err(err) => { + error!("Failed to poll model_changes rows: {err:?}"); + return false; + } + }; + + if changes.is_empty() { + return false; + } + + let fetched_count = changes.len(); + let mut batch: Vec = Vec::with_capacity(fetched_count); + for change in changes { + cursor.created_at = change.created_at; + cursor.id = change.id; + + // Window-sourced writes already went out on the in-memory fast path. + if matches!(change.payload.update_source, UpdateSource::Window { .. }) { + continue; + } + batch.push(change.payload); + } + + // One batch per drain so bulk writes don't flood the tab. + if !batch.is_empty() { + events.emit("model_writes", &batch); + } + + fetched_count == MODEL_CHANGES_POLL_BATCH_SIZE +} diff --git a/crates-server/yaak-server/src/plugin_events.rs b/crates-server/yaak-server/src/plugin_events.rs new file mode 100644 index 00000000..6452b95e --- /dev/null +++ b/crates-server/yaak-server/src/plugin_events.rs @@ -0,0 +1,582 @@ +//! The bridge's plugin host. +//! +//! Same shape as the CLI's bridge (crates-cli/yaak-cli/src/plugin_events.rs): +//! subscribe to the plugin manager, let `handle_shared_plugin_event` answer +//! everything that is only a database question, and implement the rest here. +//! +//! Where it differs is that a UI is attached. The CLI answers a prompt from a +//! TTY and refuses when there isn't one; the bridge does what the desktop does +//! instead — pushes the event to the tab and waits for the reply keyed by the +//! event's id. Toasts, clipboard writes and external URLs go the same way, +//! because the browser is the only thing here that can show or do them. + +use crate::events::EventHub; +use crate::session::SessionStore; +use serde_json::Value; +use std::path::PathBuf; +use std::sync::Arc; +use tokio::task::JoinHandle; +use yaak::plugin_events::{ + GroupedPluginEvent, HostRequest, SharedPluginEventContext, handle_shared_plugin_event, +}; +use yaak::render::{render_grpc_request, render_http_request}; +use yaak::send::{SendHttpRequestWithPluginsParams, send_http_request_with_plugins}; +use yaak_crypto::manager::EncryptionManager; +use yaak_http::cookies::get_cookie_value_from_jar; +use yaak_http::manager::HttpConnectionManager; +use yaak_models::blob_manager::BlobManager; +use yaak_models::models::Environment; +use yaak_models::queries::any_request::AnyRequest; +use yaak_models::query_manager::QueryManager; +use yaak_models::render::make_vars_hashmap; +use yaak_models::util::UpdateSource; +use yaak_plugins::events::{ + EmptyPayload, ErrorResponse, GetCookieValueResponse, InternalEvent, InternalEventPayload, + ListCookieNamesResponse, ListOpenWorkspacesResponse, PluginContext, PromptTextResponse, + RenderGrpcRequestResponse, RenderHttpRequestResponse, SendHttpRequestResponse, + TemplateRenderResponse, WindowInfoResponse, WorkspaceInfo, +}; +use yaak_plugins::manager::PluginManager; +use yaak_plugins::plugin_handle::PluginHandle; +use yaak_plugins::template_callback::PluginTemplateCallback; +use yaak_templates::{RenderOptions, TemplateCallback, render_json_value_raw}; + +pub struct BridgePluginEventBridge { + rx_id: String, + task: JoinHandle<()>, +} + +struct BridgeHostContext { + query_manager: QueryManager, + blob_manager: BlobManager, + plugin_manager: Arc, + encryption_manager: Arc, + connection_manager: Arc, + response_dir: PathBuf, + events: EventHub, + session: SessionStore, +} + +impl BridgePluginEventBridge { + #[allow(clippy::too_many_arguments)] + pub async fn start( + plugin_manager: Arc, + query_manager: QueryManager, + blob_manager: BlobManager, + encryption_manager: Arc, + connection_manager: Arc, + data_dir: PathBuf, + events: EventHub, + session: SessionStore, + ) -> Self { + let (rx_id, mut rx) = plugin_manager.subscribe("bridge").await; + let rx_id_for_task = rx_id.clone(); + let pm = plugin_manager.clone(); + let host_context = Arc::new(BridgeHostContext { + query_manager, + blob_manager, + plugin_manager, + encryption_manager, + connection_manager, + response_dir: data_dir.join("responses"), + events, + session, + }); + + let task = tokio::spawn(async move { + while let Some(event) = rx.recv().await { + // Events with reply IDs are replies to app-originated requests. + if event.reply_id.is_some() { + continue; + } + + let Some(plugin_handle) = pm.get_plugin_by_ref_id(&event.plugin_ref_id).await + else { + log::warn!( + "Ignoring plugin event with unknown plugin ref '{}'", + event.plugin_ref_id + ); + continue; + }; + + let pm = pm.clone(); + let host_context = host_context.clone(); + + // Avoid deadlocks for nested plugin-host requests (for example, template functions + // that trigger additional host requests during render) by handling each event in + // its own task. + tokio::spawn(async move { + let plugin_name = plugin_handle.info().name; + let Some(reply_payload) = build_plugin_reply( + host_context.as_ref(), + &event, + &plugin_name, + &plugin_handle, + ) + .await + else { + return; + }; + + if let Err(err) = pm.reply(&event, &reply_payload).await { + log::warn!("Failed replying to plugin event: {err}"); + } + }); + } + + pm.unsubscribe(&rx_id_for_task).await; + }); + + Self { rx_id, task } + } + + pub async fn shutdown(self, plugin_manager: &PluginManager) { + plugin_manager.unsubscribe(&self.rx_id).await; + self.task.abort(); + let _ = self.task.await; + } +} + +async fn build_plugin_reply( + host_context: &BridgeHostContext, + event: &InternalEvent, + plugin_name: &str, + plugin_handle: &PluginHandle, +) -> Option { + let session = host_context.session.get(); + let shared_workspace_id = + event.context.workspace_id.clone().or_else(|| session.workspace_id()); + + match handle_shared_plugin_event( + &host_context.query_manager, + &event.payload, + SharedPluginEventContext { + plugin_name, + workspace_id: shared_workspace_id.as_deref(), + }, + ) { + GroupedPluginEvent::Handled(payload) => payload, + GroupedPluginEvent::ToHandle(host_request) => match host_request { + HostRequest::ErrorResponse(resp) => { + log::warn!("[plugin:{plugin_name}] error: {}", resp.error); + None + } + HostRequest::ReloadResponse(_) => None, + + // The tab owns everything the user can see or the OS can do. These + // are fire-and-forget: the plugin gets its acknowledgement as soon + // as the frame is queued, matching the desktop, which also does not + // wait for the webview to paint. + HostRequest::ShowToast(req) => { + host_context.events.emit("show_toast", &req); + Some(InternalEventPayload::ShowToastResponse(EmptyPayload {})) + } + HostRequest::CopyText(req) => { + host_context.events.emit("bridge_copy_text", &req); + Some(InternalEventPayload::CopyTextResponse(EmptyPayload {})) + } + HostRequest::OpenExternalUrl(req) => { + host_context.events.emit("bridge_open_url", &req); + Some(InternalEventPayload::OpenExternalUrlResponse(EmptyPayload {})) + } + + // Prompts are questions, so they round-trip: the tab renders the + // dialog and emits the answer back under the event's own id. + HostRequest::PromptText(_) => { + let reply = call_frontend(host_context, event).await; + Some(reply.unwrap_or(InternalEventPayload::PromptTextResponse( + PromptTextResponse { value: None }, + ))) + } + + // A form streams: the tab sends a response per interaction and the + // plugin re-renders, until one comes back marked done. + HostRequest::PromptForm(_) => { + host_context.events.emit("plugin_event", event); + if event.reply_id.is_none() { + spawn_form_reply_pump(host_context, event, plugin_handle); + } + None + } + + HostRequest::ListOpenWorkspaces(_) => { + let workspaces = match host_context.query_manager.connect().list_workspaces() { + Ok(workspaces) => workspaces + .into_iter() + .map(|w| WorkspaceInfo { + id: w.id.clone(), + name: w.name, + label: session.label.clone(), + }) + .collect(), + Err(err) => { + return Some(InternalEventPayload::ErrorResponse(ErrorResponse { + error: format!("Failed to list workspaces in bridge: {err}"), + })); + } + }; + Some(InternalEventPayload::ListOpenWorkspacesResponse(ListOpenWorkspacesResponse { + workspaces, + })) + } + + HostRequest::SendHttpRequest(req) => { + let mut http_request = req.http_request.clone(); + if http_request.workspace_id.is_empty() { + let Some(workspace_id) = shared_workspace_id.clone() else { + return Some(InternalEventPayload::ErrorResponse(ErrorResponse { + error: "workspace_id is required to send HTTP requests in bridge" + .to_string(), + })); + }; + http_request.workspace_id = workspace_id; + } + + let cookie_jar_id = match session.cookie_jar_id() { + Some(id) => Some(id), + None => match host_context + .query_manager + .connect() + .list_cookie_jars(http_request.workspace_id.as_str()) + { + Ok(jars) => { + jars.into_iter().min_by_key(|jar| jar.created_at).map(|jar| jar.id) + } + Err(err) => { + return Some(InternalEventPayload::ErrorResponse(ErrorResponse { + error: format!("Failed to list cookie jars in bridge: {err}"), + })); + } + }, + }; + + let plugin_context = PluginContext { + workspace_id: Some(http_request.workspace_id.clone()), + ..event.context.clone() + }; + + match send_http_request_with_plugins(SendHttpRequestWithPluginsParams { + query_manager: &host_context.query_manager, + blob_manager: &host_context.blob_manager, + request: http_request, + environment_id: session.environment_id().as_deref(), + update_source: UpdateSource::Plugin, + cookie_jar_id, + response_dir: &host_context.response_dir, + emit_events_to: None, + emit_response_body_chunks_to: None, + existing_response: None, + plugin_manager: host_context.plugin_manager.clone(), + encryption_manager: host_context.encryption_manager.clone(), + plugin_context: &plugin_context, + cancelled_rx: None, + connection_manager: &host_context.connection_manager, + }) + .await + { + Ok(result) => Some(InternalEventPayload::SendHttpRequestResponse( + SendHttpRequestResponse { http_response: result.response }, + )), + Err(err) => Some(InternalEventPayload::ErrorResponse(ErrorResponse { + error: format!("Failed to send HTTP request in bridge: {err}"), + })), + } + } + + HostRequest::RenderHttpRequest(req) => { + let mut http_request = req.http_request.clone(); + if http_request.workspace_id.is_empty() { + let Some(workspace_id) = shared_workspace_id.clone() else { + return Some(InternalEventPayload::ErrorResponse(ErrorResponse { + error: "workspace_id is required to render HTTP requests in bridge" + .to_string(), + })); + }; + http_request.workspace_id = workspace_id; + } + + let plugin_context = PluginContext { + workspace_id: Some(http_request.workspace_id.clone()), + ..event.context.clone() + }; + + let environment_chain = match host_context.query_manager.connect().resolve_environments( + &http_request.workspace_id, + http_request.folder_id.as_deref(), + session.environment_id().as_deref(), + ) { + Ok(chain) => chain, + Err(err) => { + return Some(InternalEventPayload::ErrorResponse(ErrorResponse { + error: format!("Failed to resolve environments in bridge: {err}"), + })); + } + }; + + let template_callback = PluginTemplateCallback::new( + host_context.plugin_manager.clone(), + host_context.encryption_manager.clone(), + &plugin_context, + req.purpose.clone(), + ); + + match render_http_request( + &http_request, + environment_chain, + &template_callback, + &RenderOptions::throw(), + ) + .await + { + Ok(http_request) => Some(InternalEventPayload::RenderHttpRequestResponse( + RenderHttpRequestResponse { http_request }, + )), + Err(err) => Some(InternalEventPayload::ErrorResponse(ErrorResponse { + error: format!("Failed to render HTTP request in bridge: {err}"), + })), + } + } + + HostRequest::RenderGrpcRequest(req) => { + let mut grpc_request = req.grpc_request.clone(); + if grpc_request.workspace_id.is_empty() { + let Some(workspace_id) = shared_workspace_id.clone() else { + return Some(InternalEventPayload::ErrorResponse(ErrorResponse { + error: "workspace_id is required to render gRPC requests in bridge" + .to_string(), + })); + }; + grpc_request.workspace_id = workspace_id; + } + + let plugin_context = PluginContext { + workspace_id: Some(grpc_request.workspace_id.clone()), + ..event.context.clone() + }; + + let environment_chain = match host_context.query_manager.connect().resolve_environments( + &grpc_request.workspace_id, + grpc_request.folder_id.as_deref(), + session.environment_id().as_deref(), + ) { + Ok(chain) => chain, + Err(err) => { + return Some(InternalEventPayload::ErrorResponse(ErrorResponse { + error: format!("Failed to resolve environments in bridge: {err}"), + })); + } + }; + + let template_callback = PluginTemplateCallback::new( + host_context.plugin_manager.clone(), + host_context.encryption_manager.clone(), + &plugin_context, + req.purpose.clone(), + ); + + match render_grpc_request( + &grpc_request, + environment_chain, + &template_callback, + &RenderOptions::throw(), + ) + .await + { + Ok(grpc_request) => Some(InternalEventPayload::RenderGrpcRequestResponse( + RenderGrpcRequestResponse { grpc_request }, + )), + Err(err) => Some(InternalEventPayload::ErrorResponse(ErrorResponse { + error: format!("Failed to render gRPC request in bridge: {err}"), + })), + } + } + + HostRequest::TemplateRender(req) => { + let Some(workspace_id) = shared_workspace_id.clone() else { + return Some(InternalEventPayload::ErrorResponse(ErrorResponse { + error: "workspace_id is required to render templates in bridge".to_string(), + })); + }; + + let plugin_context = + PluginContext { workspace_id: Some(workspace_id.clone()), ..event.context.clone() }; + + let folder_id = session.request_id().and_then(|rid| { + match host_context.query_manager.connect().get_any_request(&rid) { + Ok(AnyRequest::HttpRequest(r)) => r.folder_id, + Ok(AnyRequest::GrpcRequest(r)) => r.folder_id, + Ok(AnyRequest::WebsocketRequest(r)) => r.folder_id, + Err(_) => None, + } + }); + + let environment_chain = match host_context.query_manager.connect().resolve_environments( + &workspace_id, + folder_id.as_deref(), + session.environment_id().as_deref(), + ) { + Ok(chain) => chain, + Err(err) => { + return Some(InternalEventPayload::ErrorResponse(ErrorResponse { + error: format!("Failed to resolve environments in bridge: {err}"), + })); + } + }; + + let template_callback = PluginTemplateCallback::new( + host_context.plugin_manager.clone(), + host_context.encryption_manager.clone(), + &plugin_context, + req.purpose.clone(), + ); + + match render_json_value( + req.data.clone(), + environment_chain, + &template_callback, + &RenderOptions::throw(), + ) + .await + { + Ok(data) => { + Some(InternalEventPayload::TemplateRenderResponse(TemplateRenderResponse { + data, + })) + } + Err(err) => Some(InternalEventPayload::ErrorResponse(ErrorResponse { + error: format!("Failed to render template data in bridge: {err}"), + })), + } + } + + HostRequest::ListCookieNames(_) => { + let Some(cookie_jar_id) = session.cookie_jar_id() else { + return Some(InternalEventPayload::ListCookieNamesResponse( + ListCookieNamesResponse { names: Vec::new() }, + )); + }; + match host_context.query_manager.connect().get_cookie_jar(&cookie_jar_id) { + Ok(jar) => Some(InternalEventPayload::ListCookieNamesResponse( + ListCookieNamesResponse { + names: jar.cookies.into_iter().map(|c| c.name).collect(), + }, + )), + Err(err) => Some(InternalEventPayload::ErrorResponse(ErrorResponse { + error: format!("Failed to load cookie jar in bridge: {err}"), + })), + } + } + + HostRequest::GetCookieValue(req) => { + let Some(cookie_jar_id) = session.cookie_jar_id() else { + return Some(InternalEventPayload::GetCookieValueResponse( + GetCookieValueResponse { value: None }, + )); + }; + match host_context.query_manager.connect().get_cookie_jar(&cookie_jar_id) { + Ok(jar) => { + let value = + get_cookie_value_from_jar(jar.cookies, &req.name, req.domain.as_deref()); + Some(InternalEventPayload::GetCookieValueResponse(GetCookieValueResponse { + value, + })) + } + Err(err) => Some(InternalEventPayload::ErrorResponse(ErrorResponse { + error: format!("Failed to load cookie jar in bridge: {err}"), + })), + } + } + + HostRequest::WindowInfo(req) => { + Some(InternalEventPayload::WindowInfoResponse(WindowInfoResponse { + label: req.label.clone(), + request_id: session.request_id(), + workspace_id: shared_workspace_id.clone(), + environment_id: session.environment_id(), + })) + } + + // A tab is one window. Opening and closing them needs the + // multiWindow capability the bridge reports false. + HostRequest::OpenWindow(_) => Some(unsupported("open_window_request")), + HostRequest::CloseWindow(_) => Some(unsupported("close_window_request")), + HostRequest::OtherRequest(payload) => Some(unsupported(&payload.type_name())), + }, + } +} + +fn unsupported(type_name: &str) -> InternalEventPayload { + InternalEventPayload::ErrorResponse(ErrorResponse { + error: format!("Unsupported plugin request in bridge: {type_name}"), + }) +} + +/// Ask the tab and wait for its answer, keyed by the event's id — the same +/// contract as the desktop's `call_frontend`. +async fn call_frontend( + host_context: &BridgeHostContext, + event: &InternalEvent, +) -> Option { + // Subscribe before emitting: the tab can answer faster than this task is + // rescheduled, and a reply that arrives before the listener exists is lost. + let mut replies = host_context.events.subscribe_inbound(event.id.clone()); + host_context.events.emit("plugin_event", event); + + let value = replies.recv().await?; + match serde_json::from_value::(value) { + Ok(reply) => Some(reply.payload), + Err(e) => { + log::warn!("Failed to parse plugin reply from browser: {e}"); + None + } + } +} + +/// Forward every form response the tab sends back to the plugin, until one is +/// marked done. +fn spawn_form_reply_pump( + host_context: &BridgeHostContext, + event: &InternalEvent, + plugin_handle: &PluginHandle, +) { + let mut replies = host_context.events.subscribe_inbound(event.id.clone()); + let plugin_handle = plugin_handle.clone(); + let plugin_context = event.context.clone(); + + tokio::spawn(async move { + while let Some(value) = replies.recv().await { + let Ok(resp) = serde_json::from_value::(value) else { + log::warn!("Failed to parse form response from browser"); + continue; + }; + + let is_done = matches!( + &resp.payload, + InternalEventPayload::PromptFormResponse(r) if r.done.unwrap_or(false) + ); + + let event_to_send = plugin_handle.build_event_to_send( + &plugin_context, + &resp.payload, + Some(resp.reply_id.unwrap_or_default()), + ); + if let Err(e) = plugin_handle.send(&event_to_send).await { + log::warn!("Failed to forward form response to plugin: {e:?}"); + } + + if is_done { + break; + } + } + }); +} + +async fn render_json_value( + value: Value, + environment_chain: Vec, + cb: &T, + opt: &RenderOptions, +) -> yaak_templates::error::Result { + let vars = &make_vars_hashmap(environment_chain); + render_json_value_raw(value, vars, cb, opt).await +} diff --git a/crates-server/yaak-server/src/rpc/commands.rs b/crates-server/yaak-server/src/rpc/commands.rs new file mode 100644 index 00000000..f0e60a21 --- /dev/null +++ b/crates-server/yaak-server/src/rpc/commands.rs @@ -0,0 +1,1152 @@ +//! The implemented commands. +//! +//! Request payloads mirror the desktop's structs in +//! crates-tauri/yaak-app-client/src/rpc_ext.rs field for field, because the +//! frontend is unchanged and sends the same JSON. They are redeclared rather +//! than shared: those live in a Tauri crate this one must not depend on, and +//! they are plain data. The command *bodies* are what matter, and they call the +//! same engine functions the desktop calls. + +use super::{BridgeCtx, UNSUPPORTED_COMMANDS, unsupported_command}; +use mime_guess::{Mime, mime}; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::path::Path; +use std::str::FromStr; +use yaak::import::{ImportDataParams, import_data as import_data_shared}; +use yaak::models_ops::{delete_model, duplicate_model, upsert_model}; +use yaak::send::{SendHttpRequestWithPluginsParams, send_http_request_with_plugins}; +use yaak_core::WorkspaceContext; +use yaak_models::models::{ + AnyModel, Environment, GraphQlIntrospection, GrpcEvent, HttpRequest, HttpRequestHeader, + HttpResponse, HttpResponseEvent, HttpResponseState, Settings, WebsocketEvent, WorkspaceMeta, +}; +use yaak_models::render::make_vars_hashmap; +use yaak_models::queries::workspaces::default_headers; +use yaak_models::util::BatchUpsertResult; +use yaak_plugins::events::{ + CallFolderActionRequest, CallHttpRequestActionRequest, CallWorkspaceActionRequest, + FilterResponse, GetFolderActionsResponse, GetHttpAuthenticationConfigResponse, + GetHttpAuthenticationSummaryResponse, GetHttpRequestActionsResponse, + GetTemplateFunctionConfigResponse, GetTemplateFunctionSummaryResponse, GetThemesResponse, + GetWorkspaceActionsResponse, JsonPrimitive, RenderPurpose, +}; +use yaak_plugins::native_template_functions::{ + decrypt_secure_template_function, encrypt_secure_template_function, +}; +use yaak_plugins::plugin_meta::PluginMetadata; +use yaak_rpc::{RpcError, RpcRouter, rpc_handler_async}; +use yaak_sse::sse::ServerSentEvent; +use yaak_templates::format_json::format_json; +use yaak_templates::{ + RenderErrorBehavior, RenderOptions, TemplateCallback, Tokens, parse_and_render, + render_json_value_raw, +}; + +type Result = std::result::Result; + +/// Any engine error becomes an RPC error with its message, matching how the +/// desktop's `rpc` command flattens its error enum before it crosses the wire. +fn err(e: impl std::fmt::Display) -> RpcError { + RpcError { message: e.to_string() } +} + +/// Run database work that opens a transaction off the async runtime. +/// +/// A `rusqlite` transaction borrows a connection that is neither `Send` nor +/// `Sync`, so a future holding one cannot be spawned. Moving it to a blocking +/// thread satisfies that and is the right shape anyway — these are synchronous +/// disk writes that can cascade. +async fn blocking(f: F) -> Result +where + F: FnOnce() -> std::result::Result + Send + 'static, + T: Send + 'static, +{ + match tokio::task::spawn_blocking(f).await { + Ok(result) => result.map_err(err), + Err(e) => Err(RpcError { message: format!("Database task failed: {e}") }), + } +} + +macro_rules! rpc_commands { + ( $( $name:ident ),* $(,)? ) => { + pub fn build_router() -> RpcRouter { + let mut router = RpcRouter::new(); + $( router.register(stringify!($name), rpc_handler_async!($name)); )* + for cmd in UNSUPPORTED_COMMANDS { + router.register( + cmd, + Box::new(move |_ctx, _payload| { + let cmd = *cmd; + Box::pin(async move { Err(unsupported_command(cmd)) }) + }), + ); + } + router + } + }; +} + +// -- App metadata -- + +#[derive(Debug, Deserialize)] +pub struct EmptyReq {} + +/// Deliberately not the desktop's `AppMetaData`: that type lives in a Tauri +/// crate and half its fields are Tauri paths. The serialized shape is the same, +/// which is what the frontend reads. +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct BridgeMetaData { + is_dev: bool, + version: String, + cli_version: Option, + name: String, + app_data_dir: String, + app_log_dir: String, + vendored_plugin_dir: String, + default_project_dir: String, + feature_updater: bool, + feature_license: bool, +} + +async fn cmd_metadata(ctx: BridgeCtx, _req: EmptyReq) -> Result { + let data_dir = ctx.state.data_dir().to_string_lossy().to_string(); + Ok(BridgeMetaData { + is_dev: ctx.state.is_dev, + version: env!("CARGO_PKG_VERSION").to_string(), + cli_version: None, + name: "Yaak Bridge".to_string(), + app_data_dir: data_dir.clone(), + app_log_dir: data_dir.clone(), + vendored_plugin_dir: ctx + .state + .data_dir() + .join("vendored-plugins") + .to_string_lossy() + .to_string(), + default_project_dir: dirs::home_dir() + .map(|d| d.join("YaakProjects")) + .unwrap_or_default() + .to_string_lossy() + .to_string(), + feature_updater: false, + feature_license: false, + }) +} + +// -- Models -- + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ModelsUpsertReq { + pub model: AnyModel, +} + +async fn models_upsert(ctx: BridgeCtx, req: ModelsUpsertReq) -> Result { + let db = ctx.state.db(); + upsert_model(&db, ctx.state.blob_manager(), req.model, &ctx.update_source()).map_err(err) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ModelsDeleteReq { + pub model: AnyModel, +} + +/// Deletes run on a blocking thread, as they do on the desktop: a transaction +/// holds a raw sqlite connection, which is neither `Send` nor cheap to hold — +/// dropping a workspace with thousands of requests would otherwise stall the +/// runtime and every other request with it. +async fn models_delete(ctx: BridgeCtx, req: ModelsDeleteReq) -> Result { + let source = ctx.update_source(); + blocking(move || { + ctx.state + .query_manager() + .with_tx(|tx| delete_model(tx, ctx.state.blob_manager(), req.model, &source)) + }) + .await +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ModelsDuplicateReq { + pub model_type: String, + pub model_id: String, +} + +async fn models_duplicate(ctx: BridgeCtx, req: ModelsDuplicateReq) -> Result { + let source = ctx.update_source(); + blocking(move || { + ctx.state + .query_manager() + .with_tx(|tx| duplicate_model(tx, &req.model_type, &req.model_id, &source)) + }) + .await +} + +async fn models_get_settings(ctx: BridgeCtx, _req: EmptyReq) -> Result { + Ok(ctx.state.db().get_settings()) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ModelsWorkspaceModelsReq { + pub workspace_id: Option, +} + +/// Everything the frontend's model store needs for a workspace, as one JSON +/// string. +/// +/// The desktop escapes non-ASCII into `\uXXXX` before handing this to the +/// webview; that is a workaround for Tauri's IPC and would only corrupt a +/// perfectly good UTF-8 HTTP response body, so the bridge returns the string as +/// serialized. The frontend `JSON.parse`s either form identically. +async fn models_workspace_models(ctx: BridgeCtx, req: ModelsWorkspaceModelsReq) -> Result { + let mut l: Vec = Vec::new(); + + { + let db = ctx.state.db(); + l.push(db.get_settings().into()); + l.append(&mut db.list_workspaces().map_err(err)?.into_iter().map(Into::into).collect()); + l.append(&mut db.list_key_values().map_err(err)?.into_iter().map(Into::into).collect()); + } + + let plugins = ctx.state.db().list_plugins().map_err(err)?; + if let Some(plugin_manager) = ctx.state.plugin_manager() { + let plugins = plugin_manager.resolve_plugins_for_runtime_from_db(plugins).await; + l.append(&mut plugins.into_iter().map(Into::into).collect()); + } else { + l.append(&mut plugins.into_iter().map(Into::into).collect()); + } + + if let Some(wid) = req.workspace_id.as_deref() { + let db = ctx.state.db(); + l.append(&mut db.list_cookie_jars(wid).map_err(err)?.into_iter().map(Into::into).collect()); + l.append( + &mut db + .list_environments_ensure_base(wid) + .map_err(err)? + .into_iter() + .map(Into::into) + .collect(), + ); + l.append(&mut db.list_folders(wid).map_err(err)?.into_iter().map(Into::into).collect()); + l.append( + &mut db.list_grpc_connections(wid).map_err(err)?.into_iter().map(Into::into).collect(), + ); + l.append( + &mut db.list_grpc_requests(wid).map_err(err)?.into_iter().map(Into::into).collect(), + ); + l.append( + &mut db.list_http_requests(wid).map_err(err)?.into_iter().map(Into::into).collect(), + ); + l.append( + &mut db + .list_http_responses(wid, None) + .map_err(err)? + .into_iter() + .map(Into::into) + .collect(), + ); + l.append( + &mut db + .list_websocket_connections(wid) + .map_err(err)? + .into_iter() + .map(Into::into) + .collect(), + ); + l.append( + &mut db.list_websocket_requests(wid).map_err(err)?.into_iter().map(Into::into).collect(), + ); + l.append( + &mut db.list_workspace_metas(wid).map_err(err)?.into_iter().map(Into::into).collect(), + ); + } + + serde_json::to_string(&l).map_err(err) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ModelsWebsocketEventsReq { + pub connection_id: String, +} + +async fn models_websocket_events( + ctx: BridgeCtx, + req: ModelsWebsocketEventsReq, +) -> Result> { + ctx.state.db().list_websocket_events(&req.connection_id).map_err(err) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ModelsGrpcEventsReq { + pub connection_id: String, +} + +async fn models_grpc_events(ctx: BridgeCtx, req: ModelsGrpcEventsReq) -> Result> { + ctx.state.db().list_grpc_events(&req.connection_id).map_err(err) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ModelsGetGraphqlIntrospectionReq { + pub request_id: String, +} + +async fn models_get_graphql_introspection( + ctx: BridgeCtx, + req: ModelsGetGraphqlIntrospectionReq, +) -> Result> { + Ok(ctx.state.db().get_graphql_introspection(&req.request_id)) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ModelsUpsertGraphqlIntrospectionReq { + pub request_id: String, + pub workspace_id: String, + pub content: Option, +} + +async fn models_upsert_graphql_introspection( + ctx: BridgeCtx, + req: ModelsUpsertGraphqlIntrospectionReq, +) -> Result { + ctx.state + .db() + .upsert_graphql_introspection( + &req.workspace_id, + &req.request_id, + req.content, + &ctx.update_source(), + ) + .map_err(err) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdGetWorkspaceMetaReq { + pub workspace_id: String, +} + +async fn cmd_get_workspace_meta( + ctx: BridgeCtx, + req: CmdGetWorkspaceMetaReq, +) -> Result { + let db = ctx.state.db(); + let workspace = db.get_workspace(&req.workspace_id).map_err(err)?; + db.get_or_create_workspace_meta(&workspace.id).map_err(err) +} + +// -- Sending -- + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdSendHttpRequestReq { + pub environment_id: Option, + pub cookie_jar_id: Option, + pub request_id: String, +} + +/// Send a saved request. +/// +/// Same sequence as the desktop (crates-tauri/.../lib.rs `cmd_send_http_request`): +/// create the response row first so the UI has something to show, wire up +/// cancellation, then hand off to the engine. Nothing is streamed back to the +/// tab directly — every state change is a database write, and the model-writes +/// push carries it, which is exactly how the desktop does it too. +async fn cmd_send_http_request(ctx: BridgeCtx, req: CmdSendHttpRequestReq) -> Result { + let request = ctx.state.db().get_http_request(&req.request_id).map_err(err)?; + let source = ctx.update_source(); + + let response = ctx + .state + .db() + .upsert_http_response( + &HttpResponse { + request_id: request.id.clone(), + workspace_id: request.workspace_id.clone(), + ..Default::default() + }, + &source, + ctx.state.blob_manager(), + ) + .map_err(err)?; + + let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(false); + let mut cancels = + ctx.state.events.subscribe_inbound(format!("cancel_http_response_{}", response.id)); + tokio::spawn(async move { + if cancels.recv().await.is_some() { + let _ = cancel_tx.send(true); + } + }); + + let result = send_persisted(&ctx, request, response.clone(), &req, cancel_rx).await; + + match result { + Ok(response) => Ok(response), + Err(e) => { + // Mirror the desktop: a failure is a closed response carrying the + // error, not a rejected command, so the UI shows it in place. + let existing = ctx.state.db().get_http_response(&response.id).map_err(err)?; + ctx.state + .db() + .upsert_http_response( + &HttpResponse { + state: HttpResponseState::Closed, + error: Some(e.message), + ..existing + }, + &source, + ctx.state.blob_manager(), + ) + .map_err(err) + } + } +} + +async fn send_persisted( + ctx: &BridgeCtx, + request: HttpRequest, + response: HttpResponse, + req: &CmdSendHttpRequestReq, + cancel_rx: tokio::sync::watch::Receiver, +) -> Result { + let plugin_manager = ctx.plugins()?; + let response_dir = ctx.state.response_dir(); + + let result = send_http_request_with_plugins(SendHttpRequestWithPluginsParams { + query_manager: ctx.state.query_manager(), + blob_manager: ctx.state.blob_manager(), + request, + environment_id: req.environment_id.as_deref(), + update_source: ctx.update_source(), + cookie_jar_id: req.cookie_jar_id.clone(), + response_dir: &response_dir, + emit_events_to: None, + emit_response_body_chunks_to: None, + existing_response: Some(response), + plugin_manager, + encryption_manager: ctx.state.encryption_manager.clone(), + plugin_context: &ctx.plugin_context(), + cancelled_rx: Some(cancel_rx), + connection_manager: ctx.state.connection_manager(), + }) + .await + .map_err(err)?; + + Ok(result.response) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdSendEphemeralRequestReq { + pub request: HttpRequest, + pub environment_id: Option, + pub cookie_jar_id: Option, +} + +/// Send without saving. An empty request id keeps the engine from persisting +/// anything, so the body comes back in memory instead of on disk. +async fn cmd_send_ephemeral_request( + ctx: BridgeCtx, + req: CmdSendEphemeralRequestReq, +) -> Result { + let mut request = req.request; + request.id = String::new(); + let plugin_manager = ctx.plugins()?; + let response_dir = ctx.state.response_dir(); + + let result = send_http_request_with_plugins(SendHttpRequestWithPluginsParams { + query_manager: ctx.state.query_manager(), + blob_manager: ctx.state.blob_manager(), + request, + environment_id: req.environment_id.as_deref(), + update_source: ctx.update_source(), + cookie_jar_id: req.cookie_jar_id, + response_dir: &response_dir, + emit_events_to: None, + emit_response_body_chunks_to: None, + existing_response: Some(HttpResponse::default()), + plugin_manager, + encryption_manager: ctx.state.encryption_manager.clone(), + plugin_context: &ctx.plugin_context(), + cancelled_rx: None, + connection_manager: ctx.state.connection_manager(), + }) + .await + .map_err(err)?; + + Ok(result.response) +} + +// -- Reading responses -- + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdHttpResponseBodyReq { + pub response: HttpResponse, + pub filter: Option, +} + +async fn cmd_http_response_body( + ctx: BridgeCtx, + req: CmdHttpResponseBodyReq, +) -> Result { + let Some(body_path) = req.response.body_path else { + return Ok(FilterResponse { content: String::new(), error: None }); + }; + + let content_type = req + .response + .headers + .iter() + .find_map(|h| { + if h.name.eq_ignore_ascii_case("content-type") { Some(h.value.as_str()) } else { None } + }) + .unwrap_or_default(); + + let body = read_response_body(&body_path, content_type) + .await + .ok_or_else(|| RpcError { message: "Failed to find response body".to_string() })?; + + match req.filter.as_deref() { + Some(filter) if !filter.is_empty() => ctx + .plugins()? + .filter_data(&ctx.plugin_context(), filter, &body, content_type) + .await + .map_err(err), + _ => Ok(FilterResponse { content: body, error: None }), + } +} + +/// Decode a response body from disk using the charset its Content-Type +/// declares. Ported from crates-tauri/yaak-app-client/src/encoding.rs. +async fn read_response_body(body_path: impl AsRef, content_type: &str) -> Option { + let body = tokio::fs::read(body_path).await.ok()?; + let body_charset = parse_charset(content_type).unwrap_or_else(|| "utf-8".to_string()); + if let Some(decoder) = charset::Charset::for_label(body_charset.as_bytes()) { + let (cow, _real_encoding, _exist_replace) = decoder.decode(&body); + return Some(cow.into_owned()); + } + Some(String::from_utf8_lossy(&body).to_string()) +} + +fn parse_charset(content_type: &str) -> Option { + let mime: Mime = Mime::from_str(content_type).ok()?; + mime.get_param(mime::CHARSET).map(|v| v.to_string()) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdHttpRequestBodyReq { + pub response_id: String, +} + +async fn cmd_http_request_body( + ctx: BridgeCtx, + req: CmdHttpRequestBodyReq, +) -> Result>> { + let body_id = format!("{}.request", req.response_id); + let chunks = ctx.state.blob_manager().connect().get_chunks(&body_id).map_err(err)?; + if chunks.is_empty() { + return Ok(None); + } + Ok(Some(chunks.into_iter().flat_map(|c| c.data).collect())) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdGetHttpResponseEventsReq { + pub response_id: String, +} + +async fn cmd_get_http_response_events( + ctx: BridgeCtx, + req: CmdGetHttpResponseEventsReq, +) -> Result> { + ctx.state.db().list_http_response_events(&req.response_id).map_err(err) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdGetSseEventsReq { + pub file_path: String, +} + +async fn cmd_get_sse_events( + _ctx: BridgeCtx, + req: CmdGetSseEventsReq, +) -> Result> { + use eventsource_client::{EventParser, SSE}; + + let body = std::fs::read(&req.file_path).map_err(err)?; + let mut event_parser = EventParser::new(); + event_parser.process_bytes(body).map_err(err)?; + + let mut events = Vec::new(); + while let Some(e) = event_parser.get_event() { + if let SSE::Event(e) = e { + events.push(ServerSentEvent { + event_type: e.event_type, + data: e.data, + id: e.id, + retry: e.retry, + }); + } + } + Ok(events) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdDeleteAllHttpResponsesReq { + pub request_id: String, +} + +async fn cmd_delete_all_http_responses( + ctx: BridgeCtx, + req: CmdDeleteAllHttpResponsesReq, +) -> Result<()> { + ctx.state + .db() + .delete_all_http_responses_for_request(&req.request_id, &ctx.update_source()) + .map_err(err)?; + Ok(()) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdDeleteSendHistoryReq { + pub workspace_id: String, +} + +async fn cmd_delete_send_history(ctx: BridgeCtx, req: CmdDeleteSendHistoryReq) -> Result<()> { + let source = ctx.update_source(); + blocking(move || { + let blobs = ctx.state.blob_manager(); + let db = ctx.state.db(); + for r in db.list_http_responses(&req.workspace_id, None)? { + db.delete_http_response(&r, &source, blobs)?; + } + Ok(()) + }) + .await +} + +// -- Formatting and templates -- + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdFormatJsonReq { + pub text: String, +} + +async fn cmd_format_json(_ctx: BridgeCtx, req: CmdFormatJsonReq) -> Result { + Ok(format_json(&req.text, " ")) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdFormatGraphqlReq { + pub text: String, +} + +async fn cmd_format_graphql(_ctx: BridgeCtx, req: CmdFormatGraphqlReq) -> Result { + match pretty_graphql::format_text(&req.text, &Default::default()) { + Ok(formatted) => Ok(formatted), + Err(_) => Ok(req.text), + } +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdRenderTemplateReq { + pub template: String, + pub workspace_id: String, + pub environment_id: Option, + pub purpose: Option, + pub ignore_error: Option, +} + +async fn cmd_render_template(ctx: BridgeCtx, req: CmdRenderTemplateReq) -> Result { + let environment_chain = ctx + .state + .db() + .resolve_environments(&req.workspace_id, None, req.environment_id.as_deref()) + .map_err(err)?; + + let callback = yaak_plugins::template_callback::PluginTemplateCallback::new( + ctx.plugins()?, + ctx.state.encryption_manager.clone(), + &ctx.plugin_context(), + req.purpose.unwrap_or(RenderPurpose::Preview), + ); + + let options = RenderOptions { + error_behavior: match req.ignore_error { + Some(true) => RenderErrorBehavior::ReturnEmpty, + _ => RenderErrorBehavior::Throw, + }, + }; + let vars = make_vars_hashmap(environment_chain); + parse_and_render(&req.template, &vars, &callback, &options).await.map_err(err) +} + +async fn render_json_value( + value: serde_json::Value, + environment_chain: Vec, + cb: &T, + opt: &RenderOptions, +) -> yaak_templates::error::Result { + let vars = &make_vars_hashmap(environment_chain); + render_json_value_raw(value, vars, cb, opt).await +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdTemplateTokensToStringReq { + pub tokens: Tokens, +} + +async fn cmd_template_tokens_to_string( + _ctx: BridgeCtx, + req: CmdTemplateTokensToStringReq, +) -> Result { + Ok(req.tokens.to_string()) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdDecryptTemplateReq { + pub template: String, +} + +async fn cmd_decrypt_template(ctx: BridgeCtx, req: CmdDecryptTemplateReq) -> Result { + decrypt_secure_template_function( + &ctx.state.encryption_manager, + &ctx.plugin_context(), + &req.template, + ) + .map_err(err) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdSecureTemplateReq { + pub template: String, +} + +async fn cmd_secure_template(ctx: BridgeCtx, req: CmdSecureTemplateReq) -> Result { + encrypt_secure_template_function( + ctx.plugins()?, + ctx.state.encryption_manager.clone(), + &ctx.plugin_context(), + &req.template, + ) + .map_err(err) +} + +async fn cmd_default_headers(_ctx: BridgeCtx, _req: EmptyReq) -> Result> { + Ok(default_headers()) +} + +// -- Plugins -- + +async fn cmd_get_themes(ctx: BridgeCtx, _req: EmptyReq) -> Result> { + // Themes are optional: the TypeScript package ships defaults, and an empty + // list still renders. Don't fail boot when the runtime is down. + let Ok(plugins) = ctx.plugins() else { + return Ok(Vec::new()); + }; + plugins.get_themes(&ctx.plugin_context()).await.map_err(err) +} + +async fn cmd_plugin_init_errors(ctx: BridgeCtx, _req: EmptyReq) -> Result> { + let Ok(plugins) = ctx.plugins() else { + return Ok(Vec::new()); + }; + Ok(plugins.take_init_errors().await) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdPluginInfoReq { + pub id: String, +} + +async fn cmd_plugin_info(ctx: BridgeCtx, req: CmdPluginInfoReq) -> Result { + let plugin = ctx.state.db().get_plugin(&req.id).map_err(err)?; + let plugins = ctx.plugins()?; + let handle = plugins + .get_plugin_by_dir(&plugin.directory) + .await + .ok_or_else(|| RpcError { message: format!("Plugin not found: {}", req.id) })?; + Ok(handle.info()) +} + +async fn cmd_template_function_summaries( + ctx: BridgeCtx, + _req: EmptyReq, +) -> Result> { + ctx.plugins()?.get_template_function_summaries(&ctx.plugin_context()).await.map_err(err) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdTemplateFunctionConfigReq { + pub function_name: String, + pub values: HashMap, + pub model: AnyModel, + /// Sent by the frontend, unused here — same as the desktop, which takes it + /// as `_environment_id`. Template function values are not pre-rendered the + /// way auth values are. + #[allow(dead_code)] + pub environment_id: Option, +} + +async fn cmd_template_function_config( + ctx: BridgeCtx, + req: CmdTemplateFunctionConfigReq, +) -> Result { + ctx.plugins()? + .get_template_function_config( + &ctx.plugin_context(), + &req.function_name, + req.values, + req.model.id(), + ) + .await + .map_err(err) +} + +async fn cmd_get_http_authentication_summaries( + ctx: BridgeCtx, + _req: EmptyReq, +) -> Result> { + let results = + ctx.plugins()?.get_http_authentication_summaries(&ctx.plugin_context()).await.map_err(err)?; + Ok(results.into_iter().map(|(_, a)| a).collect()) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdGetHttpAuthenticationConfigReq { + pub auth_name: String, + pub values: HashMap, + pub model: AnyModel, + pub environment_id: Option, +} + +async fn cmd_get_http_authentication_config( + ctx: BridgeCtx, + req: CmdGetHttpAuthenticationConfigReq, +) -> Result { + let rendered_values = + render_auth_values(&ctx, &req.model, req.environment_id.as_deref(), &req.values).await?; + ctx.plugins()? + .get_http_authentication_config( + &ctx.plugin_context(), + &req.auth_name, + rendered_values, + req.model.id(), + ) + .await + .map_err(err) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdCallHttpAuthenticationActionReq { + pub auth_name: String, + pub action_index: i32, + pub values: HashMap, + pub model: AnyModel, + pub environment_id: Option, +} + +async fn cmd_call_http_authentication_action( + ctx: BridgeCtx, + req: CmdCallHttpAuthenticationActionReq, +) -> Result<()> { + let rendered_values = + render_auth_values(&ctx, &req.model, req.environment_id.as_deref(), &req.values).await?; + ctx.plugins()? + .call_http_authentication_action( + &ctx.plugin_context(), + &req.auth_name, + req.action_index, + rendered_values, + req.model.id(), + ) + .await + .map_err(err) +} + +/// Auth config values are templates, so they are rendered against the model's +/// environment chain before the plugin sees them. +async fn render_auth_values( + ctx: &BridgeCtx, + model: &AnyModel, + environment_id: Option<&str>, + values: &HashMap, +) -> Result> { + let (workspace_id, folder_id) = match model { + AnyModel::HttpRequest(r) => (r.workspace_id.clone(), r.folder_id.clone()), + AnyModel::GrpcRequest(r) => (r.workspace_id.clone(), r.folder_id.clone()), + AnyModel::WebsocketRequest(r) => (r.workspace_id.clone(), r.folder_id.clone()), + AnyModel::Folder(f) => (f.workspace_id.clone(), f.folder_id.clone()), + AnyModel::Workspace(w) => (w.id.clone(), None), + _ => { + return Err(RpcError { + message: "Unsupported model type for authentication config".to_string(), + }); + } + }; + + let environment_chain = ctx + .state + .db() + .resolve_environments(&workspace_id, folder_id.as_deref(), environment_id) + .map_err(err)?; + + let callback = yaak_plugins::template_callback::PluginTemplateCallback::new( + ctx.plugins()?, + ctx.state.encryption_manager.clone(), + &ctx.plugin_context(), + RenderPurpose::Preview, + ); + + let values_json = serde_json::to_value(values).map_err(err)?; + let rendered_json = + render_json_value(values_json, environment_chain, &callback, &RenderOptions::return_empty()) + .await + .map_err(err)?; + serde_json::from_value(rendered_json).map_err(err) +} + +// -- Plugin actions -- + +async fn cmd_http_request_actions( + ctx: BridgeCtx, + _req: EmptyReq, +) -> Result> { + ctx.plugins()?.get_http_request_actions(&ctx.plugin_context()).await.map_err(err) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdCallHttpRequestActionReq { + pub req: CallHttpRequestActionRequest, +} + +async fn cmd_call_http_request_action( + ctx: BridgeCtx, + req: CmdCallHttpRequestActionReq, +) -> Result<()> { + use yaak_plugins::events::CallHttpRequestActionArgs; + + // Resolve inherited auth and headers before handing the request to the + // plugin, so an action sees what a send would see. Scoped so the database + // connection is released before the plugin call awaits. + let http_request = { + let db = ctx.state.db(); + let mut http_request = req.req.args.http_request.clone(); + let (authentication_type, authentication, _) = + db.resolve_auth_for_http_request(&http_request).map_err(err)?; + http_request.authentication_type = authentication_type; + http_request.authentication = authentication; + http_request.headers = db.resolve_headers_for_http_request(&http_request).map_err(err)?; + http_request + }; + + ctx.plugins()? + .call_http_request_action( + &ctx.plugin_context(), + CallHttpRequestActionRequest { + args: CallHttpRequestActionArgs { http_request }, + ..req.req + }, + ) + .await + .map_err(err) +} + +async fn cmd_workspace_actions( + ctx: BridgeCtx, + _req: EmptyReq, +) -> Result> { + ctx.plugins()?.get_workspace_actions(&ctx.plugin_context()).await.map_err(err) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdCallWorkspaceActionReq { + pub req: CallWorkspaceActionRequest, +} + +async fn cmd_call_workspace_action(ctx: BridgeCtx, req: CmdCallWorkspaceActionReq) -> Result<()> { + use yaak_plugins::events::CallWorkspaceActionArgs; + + let workspace = ctx.state.db().get_workspace(&req.req.args.workspace.id).map_err(err)?; + ctx.plugins()? + .call_workspace_action( + &ctx.plugin_context(), + CallWorkspaceActionRequest { args: CallWorkspaceActionArgs { workspace }, ..req.req }, + ) + .await + .map_err(err) +} + +async fn cmd_folder_actions(ctx: BridgeCtx, _req: EmptyReq) -> Result> { + ctx.plugins()?.get_folder_actions(&ctx.plugin_context()).await.map_err(err) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdCallFolderActionReq { + pub req: CallFolderActionRequest, +} + +async fn cmd_call_folder_action(ctx: BridgeCtx, req: CmdCallFolderActionReq) -> Result<()> { + use yaak_plugins::events::CallFolderActionArgs; + + let folder = ctx.state.db().get_folder(&req.req.args.folder.id).map_err(err)?; + ctx.plugins()? + .call_folder_action( + &ctx.plugin_context(), + CallFolderActionRequest { args: CallFolderActionArgs { folder }, ..req.req }, + ) + .await + .map_err(err) +} + +// -- Import -- + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdCurlToRequestReq { + pub command: String, + pub workspace_id: String, +} + +async fn cmd_curl_to_request(ctx: BridgeCtx, req: CmdCurlToRequestReq) -> Result { + let import_result = + ctx.plugins()?.import_data(&ctx.plugin_context(), &req.command).await.map_err(err)?; + + let r = import_result + .resources + .http_requests + .first() + .ok_or_else(|| RpcError { message: "No curl command found".to_string() })?; + + let mut request = r.clone(); + request.workspace_id = req.workspace_id; + request.id = String::new(); + Ok(request) +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CmdImportDataReq { + pub file_path: String, +} + +/// Import from a path on the *bridge's* machine. +/// +/// The desktop gets this path from a native file dialog. A tab has no way to +/// produce one, so in practice this only works for a path typed by hand — which +/// is why `localFiles` is reported false. Kept registered because the command +/// itself works, and a future upload route can reuse it. +async fn cmd_import_data(ctx: BridgeCtx, req: CmdImportDataReq) -> Result { + let contents = std::fs::read_to_string(&req.file_path).map_err(|e| RpcError { + message: format!("Unable to read import file {}: {e}", req.file_path), + })?; + let plugins = ctx.plugins()?; + + import_data_shared(ImportDataParams { + query_manager: ctx.state.query_manager(), + plugin_manager: &plugins, + plugin_context: &ctx.plugin_context(), + workspace_context: WorkspaceContext { + workspace_id: ctx.session.workspace_id(), + environment_id: ctx.session.environment_id(), + cookie_jar_id: ctx.session.cookie_jar_id(), + request_id: None, + }, + contents: &contents, + }) + .await + .map_err(err) +} + +rpc_commands! { + cmd_metadata, + cmd_default_headers, + cmd_get_themes, + cmd_plugin_init_errors, + cmd_plugin_info, + + models_upsert, + models_delete, + models_duplicate, + models_get_settings, + models_workspace_models, + models_websocket_events, + models_grpc_events, + models_get_graphql_introspection, + models_upsert_graphql_introspection, + cmd_get_workspace_meta, + + cmd_send_http_request, + cmd_send_ephemeral_request, + + cmd_http_response_body, + cmd_http_request_body, + cmd_get_http_response_events, + cmd_get_sse_events, + cmd_delete_all_http_responses, + cmd_delete_send_history, + + cmd_format_json, + cmd_format_graphql, + cmd_render_template, + cmd_template_tokens_to_string, + cmd_decrypt_template, + cmd_secure_template, + + cmd_template_function_summaries, + cmd_template_function_config, + cmd_get_http_authentication_summaries, + cmd_get_http_authentication_config, + cmd_call_http_authentication_action, + + cmd_http_request_actions, + cmd_call_http_request_action, + cmd_workspace_actions, + cmd_call_workspace_action, + cmd_folder_actions, + cmd_call_folder_action, + + cmd_curl_to_request, + cmd_import_data, +} + +/// Command names this host implements, for the capability report. +pub fn implemented_commands(router: &RpcRouter) -> Vec { + let unsupported: std::collections::HashSet<&str> = + UNSUPPORTED_COMMANDS.iter().copied().collect(); + let mut names: Vec = router + .commands() + .into_iter() + .filter(|c| !unsupported.contains(c)) + .map(|c| c.to_string()) + .collect(); + names.sort(); + names +} diff --git a/crates-server/yaak-server/src/rpc/mod.rs b/crates-server/yaak-server/src/rpc/mod.rs new file mode 100644 index 00000000..92c8f0f2 --- /dev/null +++ b/crates-server/yaak-server/src/rpc/mod.rs @@ -0,0 +1,142 @@ +//! The bridge's RPC surface. +//! +//! Same envelope and same command names as the desktop, dispatched through the +//! same `RpcRouter`. Only the adapters differ: the desktop's take a Tauri +//! window and read the workspace off its URL, while these take a `BridgeCtx` +//! carrying the connected tab's reported URL. The bodies underneath call the +//! same engine functions in `yaak`, `yaak-models` and `yaak-plugins`. +//! +//! This is a subset — enough to boot, edit, send and inspect. Anything not +//! registered here still gets a well-formed answer: `unsupported_command` +//! turns it into an RPC error naming the command and this host, so the frontend +//! surfaces "not supported by the Yaak Bridge" instead of a bare failure. + +mod commands; + +pub use commands::implemented_commands; + +use crate::session::SessionContext; +use crate::state::BridgeState; +use std::sync::Arc; +use yaak_plugins::events::PluginContext; +use yaak_rpc::{RpcError, RpcRouter}; + +/// Per-call context. The tab's identity and location, plus the engine. +/// +/// Mirrors the desktop's `ClientCtx { window }`: the window there answers both +/// "who is calling" and "what are they looking at", and those are exactly the +/// two things a bridge call needs that the payload doesn't carry. +#[derive(Clone)] +pub struct BridgeCtx { + pub state: Arc, + pub session: SessionContext, +} + +impl BridgeCtx { + pub fn plugin_context(&self) -> PluginContext { + PluginContext::new(Some(self.session.label.clone()), self.session.workspace_id()) + } + + pub fn update_source(&self) -> yaak_models::util::UpdateSource { + yaak_models::util::UpdateSource::from_window_label(&self.session.label) + } + + /// The plugin runtime, or an error naming the reason it isn't there. + pub fn plugins(&self) -> Result, RpcError> { + self.state.plugin_manager().ok_or_else(|| RpcError { + message: "The plugin runtime failed to start, so this command is unavailable" + .to_string(), + }) + } +} + +pub fn build_router() -> RpcRouter { + commands::build_router() +} + +/// Every command the desktop has that the bridge does not implement. +/// +/// Registered explicitly rather than left to fall through to "unknown command", +/// so the message says *why* — the frontend can tell a host that will never +/// support git from one that is simply out of date. +pub const UNSUPPORTED_COMMANDS: &[&str] = &[ + // Multi-window. A tab is one window; Settings opens through this on the + // desktop and is therefore unreachable in the browser today. + "cmd_new_child_window", + "cmd_new_main_window", + // gRPC and WebSocket sending. + "cmd_grpc_reflect", + "cmd_grpc_go", + "cmd_grpc_request_actions", + "cmd_call_grpc_request_action", + "cmd_delete_all_grpc_connections", + "cmd_ws_connect", + "cmd_ws_send", + "cmd_ws_close", + "cmd_ws_delete_connections", + "cmd_websocket_request_actions", + "cmd_call_websocket_request_action", + // Git-backed workspaces. + "cmd_git_checkout", + "cmd_git_branch", + "cmd_git_delete_branch", + "cmd_git_delete_remote_branch", + "cmd_git_merge_branch", + "cmd_git_rename_branch", + "cmd_git_status", + "cmd_git_branch_info", + "cmd_git_worktree_status", + "cmd_git_log", + "cmd_git_log_for_file", + "cmd_git_file_diff_for_commit", + "cmd_git_initialize", + "cmd_git_clone", + "cmd_git_commit", + "cmd_git_fetch_all", + "cmd_git_push", + "cmd_git_pull", + "cmd_git_pull_force_reset", + "cmd_git_pull_merge", + "cmd_git_add", + "cmd_git_unstage", + "cmd_git_reset_changes", + "cmd_git_restore_files", + "cmd_git_restore_file_from_commit", + "cmd_git_add_credential", + "cmd_git_remotes", + "cmd_git_add_remote", + "cmd_git_rm_remote", + "cmd_git_watch_worktree_status", + // Filesystem sync. + "cmd_sync_calculate", + "cmd_sync_calculate_fs", + "cmd_sync_apply", + "cmd_sync_watch", + // Workspace encryption. + "cmd_enable_encryption", + "cmd_disable_encryption", + "cmd_reveal_workspace_key", + "cmd_set_workspace_key", + // Things that need a local filesystem the tab can point at. + "cmd_export_data", + "cmd_save_response", + "cmd_save_base64_to_binary", + "cmd_plugins_install_from_directory", + // Desktop application management. + "cmd_restart", + "cmd_check_for_updates", + "cmd_dismiss_notification", + "cmd_send_feedback", + "cmd_plugins_search", + "cmd_plugins_install", + "cmd_plugins_uninstall", + "cmd_plugins_updates", + "cmd_plugins_update_all", + "cmd_reload_plugins", +]; + +pub fn unsupported_command(cmd: &str) -> RpcError { + RpcError { + message: format!("`{cmd}` is not supported on this host (Yaak Bridge)"), + } +} diff --git a/crates-server/yaak-server/src/session.rs b/crates-server/yaak-server/src/session.rs new file mode 100644 index 00000000..390bfa5d --- /dev/null +++ b/crates-server/yaak-server/src/session.rs @@ -0,0 +1,140 @@ +//! What the connected tab is currently looking at. +//! +//! The desktop reads workspace, environment, cookie jar and request straight off +//! the window's URL (crates-tauri/yaak-tauri-utils/src/window.rs). A browser tab +//! runs the same router and so has the same URL, but the server cannot see it — +//! so the tab reports it, on connect and whenever it changes, and the same +//! parsing happens here. +//! +//! One session for the whole process: this slice serves a single tab. A second +//! tab overwrites the first's context rather than getting its own. + +use std::sync::{Arc, RwLock}; + +#[derive(Debug, Clone, Default)] +pub struct SessionContext { + /// Identifies the tab, and lands in `UpdateSource::Window { label }` so + /// model-write echo suppression works exactly as it does on the desktop. + pub label: String, + pub url: String, +} + +impl SessionContext { + pub fn workspace_id(&self) -> Option { + let rest = self.url.split("/workspaces/").nth(1)?; + let id: String = + rest.chars().take_while(|c| c.is_alphanumeric() || *c == '_').collect(); + if id.is_empty() { None } else { Some(id) } + } + + pub fn request_id(&self) -> Option { + let rest = self.url.split("/requests/").nth(1)?; + let id: String = + rest.chars().take_while(|c| c.is_alphanumeric() || *c == '_').collect(); + if id.is_empty() { None } else { Some(id) } + } + + pub fn environment_id(&self) -> Option { + self.query_param("environment_id") + } + + pub fn cookie_jar_id(&self) -> Option { + self.query_param("cookie_jar_id") + } + + fn query_param(&self, key: &str) -> Option { + let query = self.url.split('?').nth(1)?; + let value = query.split('&').find_map(|pair| { + let (k, v) = pair.split_once('=')?; + if k != key { + return None; + } + Some(percent_decode(v)) + })?; + + // The router writes `environment_id=null` when nothing is selected. + // Neither of these is an id, and treating them as one sends a lookup + // for a model that cannot exist. + if value.is_empty() || value == "null" || value == "undefined" { + return None; + } + Some(value) + } +} + +fn percent_decode(input: &str) -> String { + let bytes = input.replace('+', " ").into_bytes(); + let mut out: Vec = Vec::with_capacity(bytes.len()); + let mut i = 0; + while i < bytes.len() { + if bytes[i] == b'%' && i + 2 < bytes.len() { + let hex = std::str::from_utf8(&bytes[i + 1..i + 3]).ok(); + if let Some(byte) = hex.and_then(|h| u8::from_str_radix(h, 16).ok()) { + out.push(byte); + i += 3; + continue; + } + } + out.push(bytes[i]); + i += 1; + } + String::from_utf8_lossy(&out).to_string() +} + +#[derive(Clone, Default)] +pub struct SessionStore { + inner: Arc>, +} + +impl SessionStore { + pub fn get(&self) -> SessionContext { + match self.inner.read() { + Ok(guard) => guard.clone(), + Err(poisoned) => poisoned.into_inner().clone(), + } + } + + pub fn set(&self, context: SessionContext) { + let mut guard = match self.inner.write() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + }; + *guard = context; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ctx(url: &str) -> SessionContext { + SessionContext { label: "tab".into(), url: url.into() } + } + + #[test] + fn parses_ids_from_a_router_url() { + let c = ctx( + "http://localhost:1472/workspaces/wk_abc123/requests/rq_def456?environment_id=ev_1&cookie_jar_id=cj_2", + ); + assert_eq!(c.workspace_id().as_deref(), Some("wk_abc123")); + assert_eq!(c.request_id().as_deref(), Some("rq_def456")); + assert_eq!(c.environment_id().as_deref(), Some("ev_1")); + assert_eq!(c.cookie_jar_id().as_deref(), Some("cj_2")); + } + + #[test] + fn placeholder_query_values_are_not_ids() { + let c = ctx("http://localhost:1472/workspaces/wk_a?environment_id=null&cookie_jar_id="); + assert_eq!(c.environment_id(), None); + assert_eq!(c.cookie_jar_id(), None); + } + + #[test] + fn missing_parts_are_none() { + let c = ctx("http://localhost:1472/"); + assert_eq!(c.workspace_id(), None); + assert_eq!(c.request_id(), None); + assert_eq!(c.environment_id(), None); + assert_eq!(c.cookie_jar_id(), None); + } +} diff --git a/crates-server/yaak-server/src/state.rs b/crates-server/yaak-server/src/state.rs new file mode 100644 index 00000000..c216ae48 --- /dev/null +++ b/crates-server/yaak-server/src/state.rs @@ -0,0 +1,244 @@ +//! The bridge's engine handles, shared by every route. +//! +//! Structurally this is `CliContext` (crates-cli/yaak-cli/src/context.rs) with +//! an event hub bolted on: the same `init_standalone` database, the same +//! `PluginManager` over the same Node sidecar. What differs is that a browser +//! tab is attached, so writes have to be pushed out as they happen instead of +//! the process exiting when a command finishes. + +use crate::events::EventHub; +use crate::plugin_events::BridgePluginEventBridge; +use crate::session::SessionStore; +use include_dir::{Dir, include_dir}; +use serde::Serialize; +use std::fs; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use tokio::sync::Mutex; +use yaak_crypto::manager::EncryptionManager; +use yaak_http::manager::HttpConnectionManager; +use yaak_models::blob_manager::BlobManager; +use yaak_models::client_db::ClientDb; +use yaak_models::query_manager::QueryManager; +use yaak_plugins::events::PluginContext; +use yaak_plugins::manager::PluginManager; + +const EMBEDDED_PLUGIN_RUNTIME: &str = include_str!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/../../crates-tauri/yaak-app-client/vendored/plugin-runtime/index.cjs" +)); +static EMBEDDED_VENDORED_PLUGINS: Dir<'_> = + include_dir!("$CARGO_MANIFEST_DIR/../../crates-tauri/yaak-app-client/vendored/plugins"); + +/// What this host can do, mirroring `PlatformCapabilities` in +/// packages/platform/src/types.ts. +/// +/// Reported to the browser rather than hardcoded there, because the honest +/// answer depends on how the bridge was built — these become cargo features as +/// the surface grows, and the tab should not have to guess. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct BridgeCapabilities { + pub grpc: bool, + pub websocket: bool, + pub git: bool, + pub sync: bool, + pub tls_options: bool, + pub cookie_jar: bool, + pub local_files: bool, + pub timeline: bool, + pub multi_window: bool, + pub plugins: bool, + pub encryption: bool, + pub updater: bool, + pub clipboard_read: bool, + pub system_fonts: bool, + pub license: bool, +} + +impl BridgeCapabilities { + /// The first slice: real HTTP sending with full fidelity, real plugins, a + /// real cookie jar and timeline. Everything the bridge has no route for is + /// reported false so the UI hides it rather than calling and failing. + fn for_this_build(plugins: bool) -> Self { + Self { + grpc: false, + websocket: false, + git: false, + sync: false, + // The engine does the TLS, so client certs and custom CAs are real. + tls_options: true, + cookie_jar: true, + // The bridge has a filesystem but the tab has no way to pick a path + // on it: there is no dialog implementation on this host. + local_files: false, + timeline: true, + multi_window: false, + plugins, + encryption: false, + updater: false, + clipboard_read: false, + system_fonts: false, + license: false, + } + } +} + +pub struct BridgeState { + data_dir: PathBuf, + query_manager: QueryManager, + blob_manager: BlobManager, + pub encryption_manager: Arc, + connection_manager: Arc, + plugin_manager: Option>, + plugin_event_bridge: Mutex>, + pub events: EventHub, + pub session: SessionStore, + pub capabilities: BridgeCapabilities, + /// Dev-grade shared secret, minted per process. The seam where OTP pairing + /// and per-session keys will go; deliberately not persisted. + pub token: String, + pub is_dev: bool, +} + +impl BridgeState { + pub fn new(data_dir: PathBuf, app_id: &str, token: String, is_dev: bool) -> Self { + let db_path = data_dir.join("db.sqlite"); + let blob_path = data_dir.join("blobs.sqlite"); + let (query_manager, blob_manager, rx) = + match yaak_models::init_standalone(&db_path, &blob_path) { + Ok(v) => v, + Err(err) => { + eprintln!("Error: Failed to initialize database: {err}"); + std::process::exit(1); + } + }; + let encryption_manager = Arc::new(EncryptionManager::new(query_manager.clone(), app_id)); + let events = EventHub::new(); + + // A Settings row has to exist before the frontend's first render — the + // singular model atom throws without one. `get_settings` upserts a + // default when it finds nothing, so touching it here is enough. + let _ = query_manager.connect().get_settings(); + + crate::model_writes::start(&query_manager, rx, events.clone()); + + Self { + data_dir, + query_manager, + blob_manager, + encryption_manager, + connection_manager: Arc::new(HttpConnectionManager::new()), + plugin_manager: None, + plugin_event_bridge: Mutex::new(None), + events, + session: SessionStore::default(), + capabilities: BridgeCapabilities::for_this_build(false), + token, + is_dev, + } + } + + /// Start the Node plugin runtime and the host-request bridge. Mirrors + /// `CliContext::init_plugins`; a failure here is survivable, but sending + /// loses auth and template functions, so the capability flips off. + pub async fn init_plugins(&mut self) { + let vendored_plugin_dir = self.data_dir.join("vendored-plugins"); + let installed_plugin_dir = self.data_dir.join("installed-plugins"); + let node_bin_path = PathBuf::from("node"); + + prepare_embedded_vendored_plugins(&vendored_plugin_dir) + .expect("Failed to prepare bundled plugins"); + + let plugin_runtime_main = + std::env::var("YAAK_PLUGIN_RUNTIME").map(PathBuf::from).unwrap_or_else(|_| { + prepare_embedded_plugin_runtime(&self.data_dir) + .expect("Failed to prepare embedded plugin runtime") + }); + + match PluginManager::new( + vendored_plugin_dir, + installed_plugin_dir, + node_bin_path, + plugin_runtime_main, + &self.query_manager, + &PluginContext::new_empty(), + false, + ) + .await + { + Ok(plugin_manager) => { + let plugin_manager = Arc::new(plugin_manager); + let plugin_event_bridge = BridgePluginEventBridge::start( + plugin_manager.clone(), + self.query_manager.clone(), + self.blob_manager.clone(), + self.encryption_manager.clone(), + self.connection_manager.clone(), + self.data_dir.clone(), + self.events.clone(), + self.session.clone(), + ) + .await; + self.plugin_manager = Some(plugin_manager); + *self.plugin_event_bridge.lock().await = Some(plugin_event_bridge); + self.capabilities.plugins = true; + } + Err(err) => { + log::warn!("Failed to initialize plugins: {err}"); + self.capabilities.plugins = false; + } + } + } + + pub fn data_dir(&self) -> &Path { + &self.data_dir + } + + pub fn response_dir(&self) -> PathBuf { + self.data_dir.join("responses") + } + + pub fn db(&self) -> ClientDb<'_> { + self.query_manager.connect() + } + + pub fn query_manager(&self) -> &QueryManager { + &self.query_manager + } + + pub fn blob_manager(&self) -> &BlobManager { + &self.blob_manager + } + + pub fn connection_manager(&self) -> &HttpConnectionManager { + &self.connection_manager + } + + pub fn plugin_manager(&self) -> Option> { + self.plugin_manager.clone() + } + + pub async fn shutdown(&self) { + if let Some(plugin_manager) = &self.plugin_manager { + if let Some(plugin_event_bridge) = self.plugin_event_bridge.lock().await.take() { + plugin_event_bridge.shutdown(plugin_manager).await; + } + plugin_manager.terminate().await; + } + } +} + +fn prepare_embedded_plugin_runtime(data_dir: &Path) -> std::io::Result { + let runtime_dir = data_dir.join("vendored").join("plugin-runtime"); + fs::create_dir_all(&runtime_dir)?; + let runtime_main = runtime_dir.join("index.cjs"); + fs::write(&runtime_main, EMBEDDED_PLUGIN_RUNTIME)?; + Ok(runtime_main) +} + +fn prepare_embedded_vendored_plugins(vendored_plugin_dir: &Path) -> std::io::Result<()> { + fs::create_dir_all(vendored_plugin_dir)?; + EMBEDDED_VENDORED_PLUGINS.extract(vendored_plugin_dir)?; + Ok(()) +} diff --git a/packages/platform/src/bridge/connection.ts b/packages/platform/src/bridge/connection.ts new file mode 100644 index 00000000..7db06e46 --- /dev/null +++ b/packages/platform/src/bridge/connection.ts @@ -0,0 +1,211 @@ +import type { Unsubscribe } from "../types"; + +/** + * The wire to the Yaak Bridge: one `POST /rpc` per command, one WebSocket for + * events in both directions. + * + * The hard requirement this file exists to satisfy: the connection is opened + * asynchronously, but the host that uses it must be constructible + * *synchronously*. Boot-time modules call commands while the module graph is + * still evaluating (`lib/appInfo.ts` top-level-awaits one), so there is no + * later moment to install a host, and a registry that waited for a socket would + * deadlock. So every call made before the connection opens is queued here and + * flushed when it does. The app's own top-level await then doubles as the + * connection gate: nothing renders until the first command has answered, which + * means it has answered over a live connection. + */ + +interface EventFrame { + event: string; + payload: unknown; +} + +export interface BridgeInfo { + name: string; + version: string; + capabilities: Record; + commands: string[]; +} + +/** How long to wait before retrying a dropped connection, and the ceiling. */ +const RECONNECT_BASE_MS = 250; +const RECONNECT_MAX_MS = 5000; + +export class BridgeConnection { + readonly baseUrl: string; + readonly label: string; + /** + * Null when the user hasn't supplied one yet. The connection then never + * opens, so every call queues forever — which is exactly what the connect + * screen wants, and means "waiting for a token" and "waiting for the socket" + * are the same code path rather than two. + */ + private readonly token: string | null; + + private socket: WebSocket | null = null; + private connected = false; + private reconnectDelay = RECONNECT_BASE_MS; + + /** Frames the page tried to send before the socket opened. */ + private outboundQueue: EventFrame[] = []; + /** Resolvers for anything awaiting the first successful connection. */ + private readyWaiters: Array<() => void> = []; + private listeners = new Map void>>(); + + info: BridgeInfo | null = null; + + constructor(baseUrl: string, token: string | null, label: string) { + this.baseUrl = baseUrl.replace(/\/$/, ""); + this.token = token; + this.label = label; + if (token != null) this.openSocket(); + } + + get hasToken(): boolean { + return this.token != null; + } + + /** Resolves once the events socket is open. */ + ready(): Promise { + if (this.connected) return Promise.resolve(); + return new Promise((resolve) => this.readyWaiters.push(resolve)); + } + + /** A URL on the bridge with the token attached, for the browser to fetch directly. */ + url(path: string): string { + const url = new URL(this.baseUrl + path); + url.searchParams.set("token", this.token ?? ""); + return url.toString(); + } + + async fetch(path: string, init?: RequestInit): Promise { + const headers = new Headers(init?.headers); + headers.set("Authorization", `Bearer ${this.token ?? ""}`); + return fetch(this.baseUrl + path, { ...init, headers }); + } + + async loadInfo(): Promise { + const res = await this.fetch("/bridge/info"); + if (!res.ok) { + throw new Error(`Bridge rejected the connection (${res.status}). Is the token correct?`); + } + this.info = (await res.json()) as BridgeInfo; + return this.info; + } + + /** + * Send a command and await its result. + * + * Waits for the connection first, so a command issued during module + * evaluation queues instead of failing. Errors are carried inside the + * envelope and rethrown here, so callers see the backend's own message — + * matching what Tauri's `invoke` does with a rejected command. + */ + async rpc(cmd: string, payload: Record = {}): Promise { + await this.ready(); + + const res = await this.fetch("/rpc", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ id: crypto.randomUUID(), cmd, payload }), + }); + + if (!res.ok) { + throw new Error(`Bridge request failed (${res.status})`); + } + + const body = (await res.json()) as + | { type: "Success"; id: string; payload: T } + | { type: "Error"; id: string; error: string }; + + if (body.type === "Error") { + throw new Error(body.error); + } + return body.payload; + } + + listen(event: string, callback: (payload: unknown) => void): Unsubscribe { + let handlers = this.listeners.get(event); + if (handlers == null) { + handlers = new Set(); + this.listeners.set(event, handlers); + } + handlers.add(callback); + + // Synchronous, because callers unsubscribe from React cleanups. + return () => { + const current = this.listeners.get(event); + if (current == null) return; + current.delete(callback); + if (current.size === 0) this.listeners.delete(event); + }; + } + + emit(event: string, payload: unknown): void { + const frame: EventFrame = { event, payload }; + if (this.socket != null && this.socket.readyState === WebSocket.OPEN) { + this.socket.send(JSON.stringify(frame)); + } else { + this.outboundQueue.push(frame); + } + } + + /** Tell the bridge who and where we are — what a desktop window's URL says. */ + attach(): void { + this.emit("bridge_attach", { label: this.label, url: window.location.href }); + } + + private openSocket(): void { + const wsUrl = new URL(this.baseUrl.replace(/^http/, "ws") + "/events"); + wsUrl.searchParams.set("token", this.token ?? ""); + + const socket = new WebSocket(wsUrl.toString()); + this.socket = socket; + + socket.onopen = () => { + this.connected = true; + this.reconnectDelay = RECONNECT_BASE_MS; + this.attach(); + + for (const frame of this.outboundQueue.splice(0)) { + socket.send(JSON.stringify(frame)); + } + for (const resolve of this.readyWaiters.splice(0)) { + resolve(); + } + }; + + socket.onmessage = (message) => { + let frame: EventFrame; + try { + frame = JSON.parse(String(message.data)) as EventFrame; + } catch { + console.warn("Bridge sent a malformed event frame"); + return; + } + // Deliver the payload directly, not wrapped in Tauri's `{ payload }`. + for (const handler of this.listeners.get(frame.event) ?? []) { + try { + handler(frame.payload); + } catch (err) { + console.error("Bridge event handler threw", frame.event, err); + } + } + }; + + socket.onclose = () => { + this.connected = false; + this.socket = null; + // The server closes the socket when a tab falls too far behind to be + // consistent, so a reconnect has to re-read the workspace rather than + // resume. `bridge_reconnected` is what tells the app to do that. + window.setTimeout(() => this.openSocket(), this.reconnectDelay); + this.reconnectDelay = Math.min(this.reconnectDelay * 2, RECONNECT_MAX_MS); + }; + + socket.onerror = () => { + // `onclose` always follows, and it owns the retry. + socket.close(); + }; + } +} diff --git a/packages/platform/src/bridge/index.ts b/packages/platform/src/bridge/index.ts new file mode 100644 index 00000000..70f49be2 --- /dev/null +++ b/packages/platform/src/bridge/index.ts @@ -0,0 +1,290 @@ +import type { + DragDropEvent, + OsType, + Platform, + PlatformCapabilities, + PlatformWindow, + RpcPayload, + RpcStreamHandle, + Unsubscribe, +} from "../types"; +import { BridgeConnection } from "./connection"; + +/** + * The browser host: the Yaak UI in a tab, with the real engine running in the + * Yaak Bridge next to it. + * + * Everything the desktop gets from Tauri comes over one HTTP connection + * instead. The parts a page genuinely cannot do — a native file dialog, a + * second window, reading the clipboard unprompted — are not faked. They report + * false through `capabilities` and throw if called anyway, so a missing feature + * surfaces as a disabled control rather than a silent no-op. + */ + +/** + * Until the bridge answers, assume nothing works. + * + * These are replaced wholesale by the server's own report as soon as + * `/bridge/info` returns, which happens before the app's first render — the + * boot sequence top-level-awaits a command, and that command cannot resolve + * before the connection is up. Starting pessimistic means that if that ordering + * ever changes, the UI hides a feature it should have shown instead of offering + * one that will fail. + */ +const NO_CAPABILITIES: PlatformCapabilities = { + grpc: false, + websocket: false, + git: false, + sync: false, + tlsOptions: false, + cookieJar: false, + localFiles: false, + timeline: false, + multiWindow: false, + plugins: false, + encryption: false, + updater: false, + clipboardRead: false, + systemFonts: false, + license: false, +}; + +function unsupported(what: string): Error { + return new Error(`${what} is not supported in the browser`); +} + +/** Match `@tauri-apps/plugin-os` spellings so layout code needs no new branch. */ +function detectOsType(): OsType { + const platform = navigator.userAgent; + if (/Mac|iPhone|iPad|iPod/.test(platform)) return "macos"; + if (/Win/.test(platform)) return "windows"; + if (/Android/.test(platform)) return "android"; + return "linux"; +} + +/** + * A response body path is an opaque handle the backend minted, and the bridge + * writes them as `/responses/`. Taking the last segment + * turns it back into the id the `/responses/{id}/body` route wants, which keeps + * the server from ever being asked for a path chosen by the page. + */ +function responseIdFromBodyPath(path: string): string { + const segments = path.split(/[/\\]/); + return segments[segments.length - 1] ?? path; +} + +/** + * Answer the Tauri host-plugin commands, which ride outside the RPC envelope. + * + * `set_title` has a real browser equivalent. `set_theme` paints the native + * window frame behind the webview, which a tab has no equivalent of and does + * not need. Everything else is a desktop-only feature; rejecting is correct, + * and the callers already gate on the matching capability. + */ +async function handleHostPluginCommand(cmd: string, payload?: RpcPayload): Promise { + switch (cmd) { + case "plugin:yaak-mac-window|set_title": { + const title = payload?.title; + document.title = typeof title === "string" ? title : "Yaak"; + return undefined as T; + } + case "plugin:yaak-mac-window|set_theme": + return undefined as T; + default: + throw unsupported(`\`${cmd}\``); + } +} + +function createWindow(connection: BridgeConnection): PlatformWindow { + const noop = async () => {}; + + return { + label: connection.label, + + // A tab manages its own frame. These exist because the interface names + // them; the UI only reaches for them behind `multiWindow`. + show: noop, + close: noop, + minimize: noop, + maximize: noop, + unmaximize: noop, + isMaximized: async () => false, + isFullscreen: async () => document.fullscreenElement != null, + setZoom: noop, + + // Null means "no opinion, let CSS decide". The desktop returns a real value + // because applying a theme forces the window appearance and poisons the + // media query; nothing does that here, so `prefers-color-scheme` is the + // honest answer and the theme package already falls back to it. + theme: async () => null, + + onThemeChanged(callback) { + const media = window.matchMedia("(prefers-color-scheme: dark)"); + const listener = () => callback(media.matches ? "dark" : "light"); + media.addEventListener("change", listener); + return () => media.removeEventListener("change", listener); + }, + + onFocusChanged(callback) { + const onFocus = () => callback(true); + const onBlur = () => callback(false); + window.addEventListener("focus", onFocus); + window.addEventListener("blur", onBlur); + return () => { + window.removeEventListener("focus", onFocus); + window.removeEventListener("blur", onBlur); + }; + }, + + // Native drag-and-drop reports OS paths, which a page never sees. The DOM's + // own drag events are a different thing and the components that need them + // use them directly. + onDragDrop(_callback: (event: DragDropEvent) => void): Unsubscribe { + return () => {}; + }, + }; +} + +/** + * Keep the bridge told where the tab is. + * + * The desktop reads the workspace, environment, cookie jar and request straight + * off the window's URL whenever a plugin asks. The bridge can't, so the tab + * pushes it on every navigation. The router uses the History API, which fires + * no event of its own on push, hence the wrapping. + */ +function trackNavigation(connection: BridgeConnection): void { + const report = () => connection.attach(); + + for (const method of ["pushState", "replaceState"] as const) { + const original = history[method]; + history[method] = function (this: History, ...args: Parameters) { + const result = original.apply(this, args); + report(); + return result; + }; + } + + window.addEventListener("popstate", report); + window.addEventListener("hashchange", report); +} + +export function createBridgePlatform(baseUrl: string, token: string | null): Platform { + const label = `tab_${crypto.randomUUID().slice(0, 8)}`; + const connection = new BridgeConnection(baseUrl, token, label); + + // Mutated in place once the bridge reports, because `platform.capabilities` + // hands out this object and callers hold the reference. + const capabilities: PlatformCapabilities = { ...NO_CAPABILITIES }; + + if (connection.hasToken) { + void connection + .loadInfo() + .then((info) => Object.assign(capabilities, info.capabilities)) + .catch((err) => console.error("Failed to read bridge capabilities", err)); + } + + trackNavigation(connection); + + // Two host requests the plugin runtime makes that only a page can carry out. + connection.listen("bridge_copy_text", (payload) => { + const text = (payload as { text?: string } | null)?.text; + if (typeof text === "string") void navigator.clipboard.writeText(text); + }); + connection.listen("bridge_open_url", (payload) => { + const url = (payload as { url?: string } | null)?.url; + if (typeof url === "string") window.open(url, "_blank", "noopener,noreferrer"); + }); + + const platformWindow = createWindow(connection); + + return { + capabilities, + window: platformWindow, + + clipboard: { + writeText: (text) => navigator.clipboard.writeText(text), + // Reading needs a permission prompt the moment the page paints, which is + // a bad ask for an app people paste bearer tokens into. `clipboardRead` + // is false and the one caller is gated on it. + readText: async () => { + throw unsupported("Reading the clipboard"); + }, + clear: async () => { + throw unsupported("Clearing the clipboard"); + }, + }, + + dialog: { + open: (async () => null) as Platform["dialog"]["open"], + save: async () => null, + }, + + files: { + async readFile(path) { + const res = await connection.fetch(`/responses/${responseIdFromBodyPath(path)}/body`); + if (!res.ok) { + throw new Error(`Failed to read response body (${res.status})`); + } + return new Uint8Array(await res.arrayBuffer()); + }, + + readDir: async () => { + throw unsupported("Browsing the filesystem"); + }, + + // The ``/`