Compare commits

..
47 changed files with 4897 additions and 1062 deletions
Generated
+127 -4
View File
@@ -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",
@@ -11035,6 +11106,7 @@ dependencies = [
"yaak-models",
"yaak-plugins",
"yaak-rpc",
"yaak-rpc-schema",
"yaak-sse",
"yaak-sync",
"yaak-system-appearance",
@@ -11317,7 +11389,7 @@ dependencies = [
"sha2",
"thiserror 2.0.17",
"tokio",
"tokio-tungstenite",
"tokio-tungstenite 0.26.2",
"ts-rs",
"yaak-common",
"yaak-crypto",
@@ -11372,6 +11444,57 @@ dependencies = [
"ts-rs",
]
[[package]]
name = "yaak-rpc-schema"
version = "0.0.0"
dependencies = [
"serde",
"ts-rs",
"yaak-git",
"yaak-grpc",
"yaak-models",
"yaak-plugins",
"yaak-sse",
"yaak-sync",
"yaak-templates",
"yaak-ws",
]
[[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-rpc-schema",
"yaak-sse",
"yaak-templates",
]
[[package]]
name = "yaak-sse"
version = "0.1.0"
@@ -11472,7 +11595,7 @@ dependencies = [
"serde_json",
"thiserror 2.0.17",
"tokio",
"tokio-tungstenite",
"tokio-tungstenite 0.26.2",
"url",
"yaak-http",
"yaak-models",
+4
View File
@@ -5,6 +5,7 @@ members = [
# Common/foundation crates
"crates/common/yaak-database",
"crates/common/yaak-rpc",
"crates/common/yaak-rpc-schema",
# Shared crates (no Tauri dependency)
"crates/yaak-core",
"crates/yaak-common",
@@ -25,6 +26,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",
@@ -63,6 +66,7 @@ ts-rs = "11.1.0"
# Internal crates - common/foundation
yaak-database = { path = "crates/common/yaak-database" }
yaak-rpc = { path = "crates/common/yaak-rpc" }
yaak-rpc-schema = { path = "crates/common/yaak-rpc-schema" }
# Internal crates - shared
yaak-core = { path = "crates/yaak-core" }
+1 -1
View File
@@ -1,6 +1,6 @@
import type { RpcPayload } from "@yaakapp-internal/platform";
import { platform } from "@yaakapp-internal/platform";
import type { RpcSchema } from "@yaakapp-internal/tauri-client";
import type { RpcSchema } from "@yaakapp-internal/rpc-schema";
/**
* Every backend command the app can call: the generated wire schema, one field
+1 -1
View File
@@ -1,5 +1,5 @@
import type { HttpRequest } from "@yaakapp-internal/models";
import type { EphemeralHttpResponse } from "@yaakapp-internal/tauri-client";
import type { EphemeralHttpResponse } from "@yaakapp-internal/rpc-schema";
import { getActiveCookieJar } from "../hooks/useActiveCookieJar";
import { rpc } from "./rpc";
+48
View File
@@ -0,0 +1,48 @@
[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-rpc-schema = { workspace = true }
yaak-sse = { workspace = true }
yaak-templates = { workspace = true }
+94
View File
@@ -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=<token>`. 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
`<img src>`-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.
+117
View File
@@ -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<EventFrame>,
/// Listeners waiting on a named event from the client, keyed by event name.
inbound: Arc<Mutex<HashMap<String, Vec<mpsc::UnboundedSender<serde_json::Value>>>>>,
}
/// 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<serde_json::Value>,
inbound: Arc<Mutex<HashMap<String, Vec<mpsc::UnboundedSender<serde_json::Value>>>>>,
}
impl InboundSubscription {
pub async fn recv(&mut self) -> Option<serde_json::Value> {
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<T: Serialize>(&self, event: impl Into<String>, 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<EventFrame> {
self.outbound.subscribe()
}
/// Listen for a named event sent *by* the client.
pub fn subscribe_inbound(&self, event: impl Into<String>) -> 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);
}
}
}
+354
View File
@@ -0,0 +1,354 @@
//! 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<BridgeState>,
pub router: Arc<RpcRouter<BridgeCtx>>,
}
pub fn build_app(state: Arc<BridgeState>, router: Arc<RpcRouter<BridgeCtx>>) -> 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<String>,
}
/// 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 `<img src>`
/// 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<Arc<BridgeState>>,
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::<TokenQuery>(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<String>,
}
async fn bridge_info(State(app): State<AppState>) -> Json<BridgeInfo> {
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<AppState>,
Json(req): Json<RpcRequest>,
) -> Json<RpcResponse> {
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<AppState>, 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::<EventFrame>(&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::<AttachPayload>(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<String>,
}
/// 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<AppState>,
Path(id): Path<String>,
Query(_q): Query<BodyQuery>,
headers: HeaderMap,
) -> Response {
let location = match app.state.locate_response_body(&id) {
Ok(location) => location,
Err(_) => return (StatusCode::NOT_FOUND, "No such response").into_response(),
};
let Some(body_path) = location.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 = if location.content_type.is_empty() {
"application/octet-stream".to_string()
} else {
location.content_type
};
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<u64>)> {
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"));
}
}
+121
View File
@@ -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<PathBuf>,
/// Use a fixed token instead of generating one. For scripted dev loops.
#[arg(long, env = "YAAK_BRIDGE_TOKEN")]
token: Option<String>,
/// 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<PathBuf>,
}
#[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()
}
@@ -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<ModelPayload>, 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<ModelPayload> = 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<ModelPayload> = 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
}
@@ -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<PluginManager>,
encryption_manager: Arc<EncryptionManager>,
connection_manager: Arc<HttpConnectionManager>,
response_dir: PathBuf,
events: EventHub,
session: SessionStore,
}
impl BridgePluginEventBridge {
#[allow(clippy::too_many_arguments)]
pub async fn start(
plugin_manager: Arc<PluginManager>,
query_manager: QueryManager,
blob_manager: BlobManager,
encryption_manager: Arc<EncryptionManager>,
connection_manager: Arc<HttpConnectionManager>,
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<InternalEventPayload> {
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<InternalEventPayload> {
// 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::<InternalEvent>(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::<InternalEvent>(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<T: TemplateCallback>(
value: Value,
environment_chain: Vec<Environment>,
cb: &T,
opt: &RenderOptions,
) -> yaak_templates::error::Result<Value> {
let vars = &make_vars_hashmap(environment_chain);
render_json_value_raw(value, vars, cb, opt).await
}
@@ -0,0 +1,985 @@
//! 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_command};
use mime_guess::{Mime, mime};
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::{ResponseBody, 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_rpc_schema::*;
use yaak_sse::sse::ServerSentEvent;
use yaak_templates::format_json::format_json;
use yaak_templates::{
RenderErrorBehavior, RenderOptions, TemplateCallback, parse_and_render,
render_json_value_raw,
};
type Result<T> = std::result::Result<T, RpcError>;
/// 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<T, F>(f: F) -> Result<T>
where
F: FnOnce() -> std::result::Result<T, yaak_models::error::Error> + 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}") }),
}
}
// -- App metadata --
async fn cmd_metadata(ctx: BridgeCtx, _req: CmdMetadataReq) -> Result<AppMetaData> {
let data_dir = ctx.state.data_dir().to_string_lossy().to_string();
Ok(AppMetaData {
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 --
async fn models_upsert(ctx: BridgeCtx, req: ModelsUpsertReq) -> Result<String> {
let db = ctx.state.db();
upsert_model(&db, ctx.state.blob_manager(), req.model, &ctx.update_source()).map_err(err)
}
/// 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<String> {
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
}
async fn models_duplicate(ctx: BridgeCtx, req: ModelsDuplicateReq) -> Result<String> {
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: ModelsGetSettingsReq) -> Result<Settings> {
Ok(ctx.state.db().get_settings())
}
/// 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<String> {
let mut l: Vec<AnyModel> = 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)
}
async fn models_websocket_events(
ctx: BridgeCtx,
req: ModelsWebsocketEventsReq,
) -> Result<Vec<WebsocketEvent>> {
ctx.state.db().list_websocket_events(&req.connection_id).map_err(err)
}
async fn models_grpc_events(ctx: BridgeCtx, req: ModelsGrpcEventsReq) -> Result<Vec<GrpcEvent>> {
ctx.state.db().list_grpc_events(&req.connection_id).map_err(err)
}
async fn models_get_graphql_introspection(
ctx: BridgeCtx,
req: ModelsGetGraphqlIntrospectionReq,
) -> Result<Option<GraphQlIntrospection>> {
Ok(ctx.state.db().get_graphql_introspection(&req.request_id))
}
async fn models_upsert_graphql_introspection(
ctx: BridgeCtx,
req: ModelsUpsertGraphqlIntrospectionReq,
) -> Result<GraphQlIntrospection> {
ctx.state
.db()
.upsert_graphql_introspection(
&req.workspace_id,
&req.request_id,
req.content,
&ctx.update_source(),
)
.map_err(err)
}
async fn cmd_get_workspace_meta(
ctx: BridgeCtx,
req: CmdGetWorkspaceMetaReq,
) -> Result<WorkspaceMeta> {
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 --
/// 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<HttpResponse> {
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<bool>,
) -> Result<HttpResponse> {
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)
}
/// Send without saving. An empty request id keeps the engine from persisting
/// anything, so the body comes back in memory and rides along with the
/// response — there is no row to look up later and no file to serve.
async fn cmd_send_ephemeral_request(
ctx: BridgeCtx,
req: CmdSendEphemeralRequestReq,
) -> Result<EphemeralHttpResponse> {
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)?;
// Blanking the request id above is what makes this send unsaved, so the
// engine always hands the body back. Failing loudly beats returning an
// empty body that reads as "the server sent nothing".
let ResponseBody::Returned(body) = result.response_body else {
return Err(RpcError { message: "Unsaved response did not return a body".to_string() });
};
Ok(EphemeralHttpResponse { response: result.response, body })
}
// -- Reading responses --
/// The frontend hands back an id and never a path, so the only bodies reachable
/// here are ones the engine wrote and the database still knows about.
async fn cmd_http_response_body(
ctx: BridgeCtx,
req: CmdHttpResponseBodyReq,
) -> Result<FilterResponse> {
let location = ctx.state.locate_response_body(&req.response_id).map_err(err)?;
let Some(body_path) = location.path else {
return Ok(FilterResponse { content: String::new(), error: None });
};
let content_type = location.content_type.as_str();
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 }),
}
}
/// The desktop host uses this to open the file itself. A tab cannot open a
/// path, so the bridge's browser host never calls it — it fetches
/// `/responses/:id/body` instead — but the command answers honestly for any
/// client that does, with the path on the bridge's machine.
async fn cmd_http_response_body_path(
ctx: BridgeCtx,
req: CmdHttpResponseBodyPathReq,
) -> Result<Option<String>> {
let location = ctx.state.locate_response_body(&req.response_id).map_err(err)?;
Ok(location.path.map(|p| p.to_string_lossy().to_string()))
}
/// 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<Path>, content_type: &str) -> Option<String> {
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<String> {
let mime: Mime = Mime::from_str(content_type).ok()?;
mime.get_param(mime::CHARSET).map(|v| v.to_string())
}
async fn cmd_http_request_body(
ctx: BridgeCtx,
req: CmdHttpRequestBodyReq,
) -> Result<Option<Vec<u8>>> {
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()))
}
async fn cmd_get_http_response_events(
ctx: BridgeCtx,
req: CmdGetHttpResponseEventsReq,
) -> Result<Vec<HttpResponseEvent>> {
ctx.state.db().list_http_response_events(&req.response_id).map_err(err)
}
async fn cmd_get_sse_events(
ctx: BridgeCtx,
req: CmdGetSseEventsReq,
) -> Result<Vec<ServerSentEvent>> {
use eventsource_client::{EventParser, SSE};
let Some(body_path) = ctx.state.locate_response_body(&req.response_id).map_err(err)?.path
else {
return Ok(Vec::new());
};
let body = std::fs::read(&body_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)
}
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(())
}
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 --
async fn cmd_format_json(_ctx: BridgeCtx, req: CmdFormatJsonReq) -> Result<String> {
Ok(format_json(&req.text, " "))
}
async fn cmd_format_graphql(_ctx: BridgeCtx, req: CmdFormatGraphqlReq) -> Result<String> {
match pretty_graphql::format_text(&req.text, &Default::default()) {
Ok(formatted) => Ok(formatted),
Err(_) => Ok(req.text),
}
}
async fn cmd_render_template(ctx: BridgeCtx, req: CmdRenderTemplateReq) -> Result<String> {
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<T: TemplateCallback>(
value: serde_json::Value,
environment_chain: Vec<Environment>,
cb: &T,
opt: &RenderOptions,
) -> yaak_templates::error::Result<serde_json::Value> {
let vars = &make_vars_hashmap(environment_chain);
render_json_value_raw(value, vars, cb, opt).await
}
async fn cmd_template_tokens_to_string(
_ctx: BridgeCtx,
req: CmdTemplateTokensToStringReq,
) -> Result<String> {
Ok(req.tokens.to_string())
}
async fn cmd_decrypt_template(ctx: BridgeCtx, req: CmdDecryptTemplateReq) -> Result<String> {
decrypt_secure_template_function(
&ctx.state.encryption_manager,
&ctx.plugin_context(),
&req.template,
)
.map_err(err)
}
async fn cmd_secure_template(ctx: BridgeCtx, req: CmdSecureTemplateReq) -> Result<String> {
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: CmdDefaultHeadersReq) -> Result<Vec<HttpRequestHeader>> {
Ok(default_headers())
}
// -- Plugins --
async fn cmd_get_themes(ctx: BridgeCtx, _req: CmdGetThemesReq) -> Result<Vec<GetThemesResponse>> {
// 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: CmdPluginInitErrorsReq) -> Result<Vec<(String, String)>> {
let Ok(plugins) = ctx.plugins() else {
return Ok(Vec::new());
};
Ok(plugins.take_init_errors().await)
}
async fn cmd_plugin_info(ctx: BridgeCtx, req: CmdPluginInfoReq) -> Result<PluginMetadata> {
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: CmdTemplateFunctionSummariesReq,
) -> Result<Vec<GetTemplateFunctionSummaryResponse>> {
ctx.plugins()?.get_template_function_summaries(&ctx.plugin_context()).await.map_err(err)
}
async fn cmd_template_function_config(
ctx: BridgeCtx,
req: CmdTemplateFunctionConfigReq,
) -> Result<GetTemplateFunctionConfigResponse> {
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: CmdGetHttpAuthenticationSummariesReq,
) -> Result<Vec<GetHttpAuthenticationSummaryResponse>> {
let results =
ctx.plugins()?.get_http_authentication_summaries(&ctx.plugin_context()).await.map_err(err)?;
Ok(results.into_iter().map(|(_, a)| a).collect())
}
async fn cmd_get_http_authentication_config(
ctx: BridgeCtx,
req: CmdGetHttpAuthenticationConfigReq,
) -> Result<GetHttpAuthenticationConfigResponse> {
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)
}
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<String, JsonPrimitive>,
) -> Result<HashMap<String, JsonPrimitive>> {
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: CmdHttpRequestActionsReq,
) -> Result<Vec<GetHttpRequestActionsResponse>> {
ctx.plugins()?.get_http_request_actions(&ctx.plugin_context()).await.map_err(err)
}
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: CmdWorkspaceActionsReq,
) -> Result<Vec<GetWorkspaceActionsResponse>> {
ctx.plugins()?.get_workspace_actions(&ctx.plugin_context()).await.map_err(err)
}
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: CmdFolderActionsReq) -> Result<Vec<GetFolderActionsResponse>> {
ctx.plugins()?.get_folder_actions(&ctx.plugin_context()).await.map_err(err)
}
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 --
async fn cmd_curl_to_request(ctx: BridgeCtx, req: CmdCurlToRequestReq) -> Result<HttpRequest> {
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)
}
/// 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<BatchUpsertResult> {
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)
}
// -- Not on this host --
/// Commands the bridge does not implement. Each still gets an adapter, so the
/// schema stays fully covered and the frontend receives a structured error
/// naming the command and this host rather than a bare "unknown command".
///
/// One list, two uses: `unsupported_commands!` emits both the adapters and the
/// `UNSUPPORTED_COMMANDS` array `implemented_commands` subtracts.
macro_rules! unsupported_commands {
( $( $name:ident ( $req:ty ) ),* $(,)? ) => {
// The stub never produces a value, so it doesn't need to name the
// response type — which keeps git, gRPC and WebSocket crates out of a
// binary that will never call them. `Never` serializes fine.
$( async fn $name(_ctx: BridgeCtx, _req: $req) -> Result<Never> {
Err(unsupported_command(stringify!($name)))
} )*
pub const UNSUPPORTED_COMMANDS: &[&str] = &[ $( stringify!($name), )* ];
};
}
/// A value that cannot exist. The unsupported adapters return `Result<Never>`
/// and always take the `Err` branch, so `rpc_handler_async!` has something
/// serializable to name without a real response type ever being constructed.
#[derive(serde::Serialize)]
enum Never {}
unsupported_commands! {
// 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(CmdNewChildWindowReq),
cmd_new_main_window(CmdNewMainWindowReq),
// gRPC and WebSocket sending.
cmd_grpc_reflect(CmdGrpcReflectReq),
cmd_grpc_go(CmdGrpcGoReq),
cmd_grpc_request_actions(CmdGrpcRequestActionsReq),
cmd_call_grpc_request_action(CmdCallGrpcRequestActionReq),
cmd_delete_all_grpc_connections(CmdDeleteAllGrpcConnectionsReq),
cmd_ws_connect(CmdWsConnectReq),
cmd_ws_send(CmdWsSendReq),
cmd_ws_close(CmdWsCloseReq),
cmd_ws_delete_connections(CmdWsDeleteConnectionsReq),
cmd_websocket_request_actions(CmdWebsocketRequestActionsReq),
cmd_call_websocket_request_action(CmdCallWebsocketRequestActionReq),
// Git-backed workspaces.
cmd_git_checkout(CmdGitCheckoutReq),
cmd_git_branch(CmdGitBranchReq),
cmd_git_delete_branch(CmdGitDeleteBranchReq),
cmd_git_delete_remote_branch(CmdGitDeleteRemoteBranchReq),
cmd_git_merge_branch(CmdGitMergeBranchReq),
cmd_git_rename_branch(CmdGitRenameBranchReq),
cmd_git_status(CmdGitStatusReq),
cmd_git_branch_info(CmdGitBranchInfoReq),
cmd_git_worktree_status(CmdGitWorktreeStatusReq),
cmd_git_log(CmdGitLogReq),
cmd_git_log_for_file(CmdGitLogForFileReq),
cmd_git_file_diff_for_commit(CmdGitFileDiffForCommitReq),
cmd_git_initialize(CmdGitInitializeReq),
cmd_git_clone(CmdGitCloneReq),
cmd_git_commit(CmdGitCommitReq),
cmd_git_fetch_all(CmdGitFetchAllReq),
cmd_git_push(CmdGitPushReq),
cmd_git_pull(CmdGitPullReq),
cmd_git_pull_force_reset(CmdGitPullForceResetReq),
cmd_git_pull_merge(CmdGitPullMergeReq),
cmd_git_add(CmdGitAddReq),
cmd_git_unstage(CmdGitUnstageReq),
cmd_git_reset_changes(CmdGitResetChangesReq),
cmd_git_restore_files(CmdGitRestoreFilesReq),
cmd_git_restore_file_from_commit(CmdGitRestoreFileFromCommitReq),
cmd_git_add_credential(CmdGitAddCredentialReq),
cmd_git_remotes(CmdGitRemotesReq),
cmd_git_add_remote(CmdGitAddRemoteReq),
cmd_git_rm_remote(CmdGitRmRemoteReq),
cmd_git_watch_worktree_status(CmdGitWatchWorktreeStatusReq),
// Filesystem sync.
cmd_sync_calculate(CmdSyncCalculateReq),
cmd_sync_calculate_fs(CmdSyncCalculateFsReq),
cmd_sync_apply(CmdSyncApplyReq),
cmd_sync_watch(CmdSyncWatchReq),
// Workspace encryption.
cmd_enable_encryption(CmdEnableEncryptionReq),
cmd_disable_encryption(CmdDisableEncryptionReq),
cmd_reveal_workspace_key(CmdRevealWorkspaceKeyReq),
cmd_set_workspace_key(CmdSetWorkspaceKeyReq),
// Things that need a local filesystem the tab can point at.
cmd_export_data(CmdExportDataReq),
cmd_save_response(CmdSaveResponseReq),
cmd_save_base64_to_binary(CmdSaveBase64ToBinaryReq),
cmd_plugins_install_from_directory(CmdPluginsInstallFromDirectoryReq),
cmd_import_url(CmdImportUrlReq),
// Desktop application management.
cmd_restart(CmdRestartReq),
cmd_check_for_updates(CmdCheckForUpdatesReq),
cmd_dismiss_notification(CmdDismissNotificationReq),
cmd_send_feedback(CmdSendFeedbackReq),
cmd_plugins_search(CmdPluginsSearchReq),
cmd_plugins_install(CmdPluginsInstallReq),
cmd_plugins_uninstall(CmdPluginsUninstallReq),
cmd_plugins_updates(CmdPluginsUpdatesReq),
cmd_plugins_update_all(CmdPluginsUpdateAllReq),
cmd_reload_plugins(CmdReloadPluginsReq),
}
// -- The router --
/// Every command in the schema, wired to an adapter here.
///
/// The list comes from `yaak_rpc_schema`, so this host cannot silently miss a
/// command the frontend knows about: a schema entry with no adapter below is a
/// compile error, not a runtime "unknown command". Commands the bridge does not
/// support still get an adapter — one that says so — which is what lets the
/// frontend tell a host that will never do git from one that is out of date.
macro_rules! register_commands {
( $( $name:ident ( $req:ty ) -> $res:ty ),* $(,)? ) => {
pub fn build_router() -> RpcRouter<BridgeCtx> {
let mut router = RpcRouter::new();
$( router.register(stringify!($name), rpc_handler_async!($name)); )*
router
}
};
}
yaak_rpc_schema::with_commands!(register_commands);
/// The names of the commands this host actually implements — everything in
/// the schema minus the ones whose adapter is `unsupported`. Reported to the
/// browser so it can fail fast with a clear message.
pub fn implemented_commands(router: &RpcRouter<BridgeCtx>) -> Vec<String> {
let unsupported: std::collections::HashSet<&str> =
UNSUPPORTED_COMMANDS.iter().copied().collect();
let mut names: Vec<String> = router
.commands()
.into_iter()
.filter(|c| !unsupported.contains(c))
.map(|c| c.to_string())
.collect();
names.sort();
names
}
+63
View File
@@ -0,0 +1,63 @@
//! The bridge's RPC surface.
//!
//! Same envelope, same command names, same request and response types as the
//! desktop — all of that comes from `yaak_rpc_schema` — 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`.
//!
//! The router is built from the schema's full command list, so every command
//! the frontend knows has an adapter here — the ones this host doesn't
//! implement return a structured error naming the command and the host, and
//! the frontend surfaces "not supported by the Yaak Bridge" instead of a bare
//! failure. Enough is implemented to boot, edit, send and inspect.
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<BridgeState>,
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<Arc<yaak_plugins::manager::PluginManager>, 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<BridgeCtx> {
commands::build_router()
}
pub fn unsupported_command(cmd: &str) -> RpcError {
RpcError {
message: format!("`{cmd}` is not supported on this host (Yaak Bridge)"),
}
}
+140
View File
@@ -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<String> {
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<String> {
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<String> {
self.query_param("environment_id")
}
pub fn cookie_jar_id(&self) -> Option<String> {
self.query_param("cookie_jar_id")
}
fn query_param(&self, key: &str) -> Option<String> {
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<u8> = 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<RwLock<SessionContext>>,
}
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);
}
}
+274
View File
@@ -0,0 +1,274 @@
//! 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,
}
}
}
/// Where a response's body is, and what it is meant to be read as.
pub struct ResponseBodyLocation {
/// None when the response has no stored body.
pub path: Option<PathBuf>,
/// The response's declared `Content-Type`, empty when it has none.
pub content_type: String,
}
pub struct BridgeState {
data_dir: PathBuf,
query_manager: QueryManager,
blob_manager: BlobManager,
pub encryption_manager: Arc<EncryptionManager>,
connection_manager: Arc<HttpConnectionManager>,
plugin_manager: Option<Arc<PluginManager>>,
plugin_event_bridge: Mutex<Option<BridgePluginEventBridge>>,
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")
}
/// Find a response's body from its id alone.
///
/// The tab hands back an id and never a path, so the only bodies reachable
/// through the bridge are ones the engine wrote and the database still
/// knows about. Every route and command that reads a body goes through
/// here for that reason.
pub fn locate_response_body(
&self,
response_id: &str,
) -> yaak_models::error::Result<ResponseBodyLocation> {
let response = self.db().get_http_response(response_id)?;
Ok(ResponseBodyLocation {
path: response.body_path.map(PathBuf::from),
content_type: response
.headers
.iter()
.find(|h| h.name.eq_ignore_ascii_case("content-type"))
.map(|h| h.value.clone())
.unwrap_or_default(),
})
}
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<Arc<PluginManager>> {
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<PathBuf> {
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(())
}
+1
View File
@@ -73,6 +73,7 @@ url = "2"
tokio-util = { version = "0.7", features = ["codec"] }
ts-rs = { workspace = true }
yaak-rpc = { workspace = true }
yaak-rpc-schema = { workspace = true }
uuid = "1.12.1"
yaak-api = { workspace = true }
yaak-common = { workspace = true }
-4
View File
@@ -1,7 +1,5 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
export type GitWatchResult = { unlistenEvent: string, };
export type PluginUpdateInfo = { name: string, currentVersion: string, latestVersion: string, };
export type PluginUpdateNotification = { updateCount: number, plugins: Array<PluginUpdateInfo>, };
@@ -12,8 +10,6 @@ export type UpdateResponse = { "type": "ack" } | { "type": "action", action: Upd
export type UpdateResponseAction = "install" | "skip";
export type WatchResult = { unlistenEvent: string, };
export type YaakNotification = { timestamp: string, timeout: number | null, id: string, title: string | null, message: string, color: string | null, action: YaakNotificationAction | null, };
export type YaakNotificationAction = { label: string, url: string, };
+3 -3
View File
@@ -1,4 +1,4 @@
// ts-rs owns bindings/index.ts and rewrites it on export, so this hand-written
// entry point is where the generated files come together.
export * from "./bindings/gen_rpc";
// ts-rs owns bindings/index.ts and rewrites it on export. What remains here
// after the RPC schema moved to @yaakapp-internal/rpc-schema is the
// desktop-only surface: updater and notification types.
export * from "./bindings/index";
@@ -2,7 +2,6 @@ use crate::error::{Error, Result};
use chrono::Utc;
use log::{debug, error, warn};
use notify::Watcher;
use serde::{Deserialize, Serialize};
use std::path::Path;
use std::sync::mpsc;
use std::time::Duration;
@@ -10,18 +9,11 @@ use tauri::{AppHandle, Listener, Runtime};
use tokio::select;
use tokio::sync::watch;
use tokio::time::sleep;
use ts_rs::TS;
use yaak_git::{GitWorktreeStatus, git_path_is_ignored, git_repository_paths, git_worktree_status};
use yaak_rpc_schema::GitWatchResult;
const GIT_STATUS_COALESCE_WINDOW: Duration = Duration::from_millis(250);
#[derive(Debug, Clone, Serialize, Deserialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export, export_to = "index.ts")]
pub(crate) struct GitWatchResult {
unlisten_event: String,
}
pub(crate) async fn watch_git_worktree_status<R, F>(
app_handle: AppHandle<R>,
dir: &Path,
+1 -17
View File
@@ -8,7 +8,6 @@ use crate::import::{import_data, import_url};
use crate::models_ext::{BlobManagerExt, QueryManagerExt};
use crate::notifications::YaakNotifier;
use crate::render::{render_grpc_request, render_json_value, render_template};
use crate::rpc_ext::EphemeralHttpResponse;
use crate::updates::{UpdateMode, UpdateTrigger, YaakUpdater};
use crate::uri_scheme::handle_deep_link;
use error::Result as YaakResult;
@@ -57,6 +56,7 @@ use yaak_plugins::events::{
use yaak_plugins::manager::PluginManager;
use yaak_plugins::plugin_meta::{PluginMetadata, get_plugin_meta};
use yaak_plugins::template_callback::PluginTemplateCallback;
use yaak_rpc_schema::{AppMetaData, EphemeralHttpResponse};
use yaak_sse::sse::ServerSentEvent;
use yaak_tauri_utils::window::WorkspaceWindowTrait;
use yaak_templates::format_json::format_json;
@@ -184,22 +184,6 @@ impl<R: Runtime> PluginContextExt<R> for WebviewWindow<R> {
}
}
#[derive(serde::Serialize, ts_rs::TS)]
#[serde(default, rename_all = "camelCase")]
#[ts(export, export_to = "gen_rpc.ts")]
pub struct AppMetaData {
is_dev: bool,
version: String,
cli_version: Option<String>,
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<R: Runtime>(app_handle: AppHandle<R>) -> YaakResult<AppMetaData> {
let app_data_dir = app_handle.path().app_data_dir()?;
let app_log_dir = app_handle.path().app_log_dir()?;
+3 -56
View File
@@ -144,28 +144,10 @@ pub(crate) fn models_upsert<R: Runtime>(
window: WebviewWindow<R>,
model: AnyModel,
) -> Result<String> {
use yaak_models::error::Error::GenericError;
let db = window.db();
let blobs = window.blob_manager();
let source = &UpdateSource::from_window_label(window.label());
let id = match model {
AnyModel::CookieJar(m) => db.upsert_cookie_jar(&m, source)?.id,
AnyModel::Environment(m) => db.upsert_environment(&m, source)?.id,
AnyModel::Folder(m) => db.upsert_folder(&m, source)?.id,
AnyModel::GrpcRequest(m) => db.upsert_grpc_request(&m, source)?.id,
AnyModel::HttpRequest(m) => db.upsert_http_request(&m, source)?.id,
AnyModel::HttpResponse(m) => db.upsert_http_response(&m, source, &blobs)?.id,
AnyModel::KeyValue(m) => db.upsert_key_value(&m, source)?.id,
AnyModel::Plugin(m) => db.upsert_plugin(&m, source)?.id,
AnyModel::Settings(m) => db.upsert_settings(&m, source)?.id,
AnyModel::WebsocketRequest(m) => db.upsert_websocket_request(&m, source)?.id,
AnyModel::Workspace(m) => db.upsert_workspace(&m, source)?.id,
AnyModel::WorkspaceMeta(m) => db.upsert_workspace_meta(&m, source)?.id,
a => return Err(GenericError(format!("Cannot upsert AnyModel {a:?})"))),
};
Ok(id)
yaak::models_ops::upsert_model(&db, &blobs, model, source)
}
// Async so cascading deletes (e.g. a workspace with thousands of requests) run on a
@@ -181,21 +163,7 @@ pub(crate) async fn models_delete<R: Runtime>(
// Use transaction for deletions because it might recurse
window.with_tx(|tx| {
let source = &UpdateSource::from_window_label(window.label());
let id = match model {
AnyModel::CookieJar(m) => tx.delete_cookie_jar(&m, source)?.id,
AnyModel::Environment(m) => tx.delete_environment(&m, source)?.id,
AnyModel::Folder(m) => tx.delete_folder(&m, source)?.id,
AnyModel::GrpcConnection(m) => tx.delete_grpc_connection(&m, source)?.id,
AnyModel::GrpcRequest(m) => tx.delete_grpc_request(&m, source)?.id,
AnyModel::HttpRequest(m) => tx.delete_http_request(&m, source)?.id,
AnyModel::HttpResponse(m) => tx.delete_http_response(&m, source, &blobs)?.id,
AnyModel::Plugin(m) => tx.delete_plugin(&m, source)?.id,
AnyModel::WebsocketConnection(m) => tx.delete_websocket_connection(&m, source)?.id,
AnyModel::WebsocketRequest(m) => tx.delete_websocket_request(&m, source)?.id,
AnyModel::Workspace(m) => tx.delete_workspace(&m, source, &blobs)?.id,
a => return Err(GenericError(format!("Cannot delete AnyModel {a:?})"))),
};
Ok(id)
yaak::models_ops::delete_model(tx, &blobs, model, source)
})
})
.await
@@ -207,31 +175,10 @@ pub(crate) fn models_duplicate<R: Runtime>(
model_type: String,
model_id: String,
) -> Result<String> {
use yaak_models::error::Error::GenericError;
// Use transaction for duplications because it might recurse
window.with_tx(|tx| {
let source = &UpdateSource::from_window_label(window.label());
// Fetch the model fresh from the DB so the duplicate doesn't come from
// a stale frontend snapshot
let id = match model_type.as_str() {
"environment" => {
tx.duplicate_environment(&tx.get_environment(&model_id)?, source)?.id
}
"folder" => tx.duplicate_folder(&tx.get_folder(&model_id)?, source)?.id,
"grpc_request" => {
tx.duplicate_grpc_request(&tx.get_grpc_request(&model_id)?, source)?.id
}
"http_request" => {
tx.duplicate_http_request(&tx.get_http_request(&model_id)?, source)?.id
}
"websocket_request" => {
tx.duplicate_websocket_request(&tx.get_websocket_request(&model_id)?, source)?.id
}
t => return Err(GenericError(format!("Cannot duplicate model type {t}"))),
};
Ok(id)
yaak::models_ops::duplicate_model(tx, &model_type, &model_id, source)
})
}
File diff suppressed because it is too large Load Diff
+1 -9
View File
@@ -6,11 +6,10 @@ use crate::error::Result;
use crate::models_ext::{BlobManagerExt, QueryManagerExt};
use chrono::Utc;
use log::warn;
use serde::{Deserialize, Serialize};
use std::path::Path;
use tauri::{AppHandle, Listener, Runtime};
use tokio::sync::watch;
use ts_rs::TS;
use yaak_rpc_schema::WatchResult;
use yaak_sync::error::Error::InvalidSyncDirectory;
use yaak_sync::sync::{
FsCandidate, SyncOp, apply_sync_ops, apply_sync_state_ops, compute_sync_ops, get_db_candidates,
@@ -57,13 +56,6 @@ pub(crate) async fn cmd_sync_apply<R: Runtime>(
Ok(())
}
#[derive(Debug, Clone, Serialize, Deserialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export, export_to = "index.ts")]
pub(crate) struct WatchResult {
unlisten_event: String,
}
pub(crate) async fn sync_watch<R, F>(
app_handle: AppHandle<R>,
sync_dir: &Path,
+18
View File
@@ -0,0 +1,18 @@
[package]
name = "yaak-rpc-schema"
version = "0.0.0"
edition = "2024"
authors = ["Gregory Schier"]
publish = false
[dependencies]
serde = { workspace = true, features = ["derive"] }
ts-rs = { workspace = true }
yaak-git = { workspace = true }
yaak-grpc = { workspace = true }
yaak-models = { workspace = true }
yaak-plugins = { workspace = true }
yaak-sse = { workspace = true }
yaak-sync = { workspace = true }
yaak-templates = { workspace = true }
yaak-ws = { workspace = true }
+44
View File
@@ -0,0 +1,44 @@
# yaak-rpc-schema
The wire schema for the app's RPC surface: every command name, its request
payload, and its response type, declared once.
Every host that serves the Yaak UI — the desktop app today, the browser bridge
and anything after it — imports these types and implements the commands against
them. That is what keeps a request's shape from drifting between hosts, and it
is why the TypeScript bindings (`bindings/gen_rpc.ts`, exposed to the frontend
as `@yaakapp-internal/rpc-schema`) are generated from one place.
Nothing here depends on Tauri or on any host. Request structs are plain data,
and so are the few response types declared here rather than in an engine crate.
Command *bodies* live with the host that runs them.
## Adding a command
1. Add its request struct and an entry in `with_commands!` in `src/lib.rs`.
2. Write the adapter in each host — the desktop's live in
`crates-tauri/yaak-app-client/src/rpc_ext.rs`. A host that does not support
the command still has to say so; a missing adapter fails to compile.
3. Regenerate the bindings: `cargo test -p yaak-rpc-schema` writes
`bindings/gen_rpc.ts`, which is committed.
## How hosts consume the list
`with_commands!` takes the name of a `macro_rules!` macro and calls it with the
full `name(Req) -> Res` list. Each host writes a small macro that receives that
list and builds its router:
```rust
macro_rules! register_commands {
( $( $name:ident ( $req:ty ) -> $res:ty ),* $(,)? ) => {
pub fn build_router() -> RpcRouter<MyCtx> {
let mut router = RpcRouter::new();
$( router.register(stringify!($name), rpc_handler_async!($name)); )*
router
}
};
}
yaak_rpc_schema::with_commands!(register_commands);
```
The schema decides *what* commands exist; the host decides *how* each one runs.
File diff suppressed because one or more lines are too long
+4
View File
@@ -0,0 +1,4 @@
// The RPC wire schema, generated by ts-rs from the Rust declarations in
// src/lib.rs. `RpcSchema` maps every command name to its (request, response)
// pair; the app's `rpc()` helper derives its command union from it.
export * from "./bindings/gen_rpc";
@@ -0,0 +1,6 @@
{
"name": "@yaakapp-internal/rpc-schema",
"version": "1.0.0",
"private": true,
"main": "index.ts"
}
File diff suppressed because it is too large Load Diff
+1 -1
View File
@@ -1,5 +1,5 @@
import { platform } from "@yaakapp-internal/platform";
import type { WatchResult } from "@yaakapp-internal/tauri-client";
import type { WatchResult } from "@yaakapp-internal/rpc-schema";
import { SyncOp } from "./bindings/gen_sync";
import { WatchEvent } from "./bindings/gen_watch";
+1
View File
@@ -1,6 +1,7 @@
pub mod error;
pub mod export;
pub mod import;
pub mod models_ops;
pub mod plugin_events;
pub mod render;
pub mod send;
+89
View File
@@ -0,0 +1,89 @@
//! Generic model writes, shared by every host.
//!
//! `upsert`, `delete` and `duplicate` take an `AnyModel` and fan out to the
//! typed query for its variant. That fan-out is long, mechanical, and has to
//! grow a new arm every time a model is added — exactly the code that should
//! not exist twice. The host supplies the database handles and the
//! `UpdateSource` identifying who is writing; nothing here knows whether the
//! caller is a desktop window or an HTTP request.
use yaak_models::blob_manager::BlobManager;
use yaak_models::client_db::ClientDb;
use yaak_models::error::Error::GenericError;
use yaak_models::error::Result;
use yaak_models::models::AnyModel;
use yaak_models::util::UpdateSource;
pub fn upsert_model(
db: &ClientDb,
blobs: &BlobManager,
model: AnyModel,
source: &UpdateSource,
) -> Result<String> {
let id = match model {
AnyModel::CookieJar(m) => db.upsert_cookie_jar(&m, source)?.id,
AnyModel::Environment(m) => db.upsert_environment(&m, source)?.id,
AnyModel::Folder(m) => db.upsert_folder(&m, source)?.id,
AnyModel::GrpcRequest(m) => db.upsert_grpc_request(&m, source)?.id,
AnyModel::HttpRequest(m) => db.upsert_http_request(&m, source)?.id,
AnyModel::HttpResponse(m) => db.upsert_http_response(&m, source, blobs)?.id,
AnyModel::KeyValue(m) => db.upsert_key_value(&m, source)?.id,
AnyModel::Plugin(m) => db.upsert_plugin(&m, source)?.id,
AnyModel::Settings(m) => db.upsert_settings(&m, source)?.id,
AnyModel::WebsocketRequest(m) => db.upsert_websocket_request(&m, source)?.id,
AnyModel::Workspace(m) => db.upsert_workspace(&m, source)?.id,
AnyModel::WorkspaceMeta(m) => db.upsert_workspace_meta(&m, source)?.id,
a => return Err(GenericError(format!("Cannot upsert AnyModel {a:?})"))),
};
Ok(id)
}
/// Deletes cascade, so callers run this inside a transaction.
pub fn delete_model(
tx: &ClientDb,
blobs: &BlobManager,
model: AnyModel,
source: &UpdateSource,
) -> Result<String> {
let id = match model {
AnyModel::CookieJar(m) => tx.delete_cookie_jar(&m, source)?.id,
AnyModel::Environment(m) => tx.delete_environment(&m, source)?.id,
AnyModel::Folder(m) => tx.delete_folder(&m, source)?.id,
AnyModel::GrpcConnection(m) => tx.delete_grpc_connection(&m, source)?.id,
AnyModel::GrpcRequest(m) => tx.delete_grpc_request(&m, source)?.id,
AnyModel::HttpRequest(m) => tx.delete_http_request(&m, source)?.id,
AnyModel::HttpResponse(m) => tx.delete_http_response(&m, source, blobs)?.id,
AnyModel::Plugin(m) => tx.delete_plugin(&m, source)?.id,
AnyModel::WebsocketConnection(m) => tx.delete_websocket_connection(&m, source)?.id,
AnyModel::WebsocketRequest(m) => tx.delete_websocket_request(&m, source)?.id,
AnyModel::Workspace(m) => tx.delete_workspace(&m, source, blobs)?.id,
a => return Err(GenericError(format!("Cannot delete AnyModel {a:?})"))),
};
Ok(id)
}
/// Duplicates recurse, so callers run this inside a transaction.
///
/// The model is re-read from the database rather than taken from the caller, so
/// a duplicate never comes from a stale frontend snapshot.
pub fn duplicate_model(
tx: &ClientDb,
model_type: &str,
model_id: &str,
source: &UpdateSource,
) -> Result<String> {
let id = match model_type {
"environment" => tx.duplicate_environment(&tx.get_environment(model_id)?, source)?.id,
"folder" => tx.duplicate_folder(&tx.get_folder(model_id)?, source)?.id,
"grpc_request" => tx.duplicate_grpc_request(&tx.get_grpc_request(model_id)?, source)?.id,
"http_request" => tx.duplicate_http_request(&tx.get_http_request(model_id)?, source)?.id,
"websocket_request" => {
tx.duplicate_websocket_request(&tx.get_websocket_request(model_id)?, source)?.id
}
t => return Err(GenericError(format!("Cannot duplicate model type {t}"))),
};
Ok(id)
}
+1
View File
@@ -58,6 +58,7 @@
"crates-tauri/yaak-fonts",
"crates-tauri/yaak-license",
"crates-tauri/yaak-mac-window",
"crates/common/yaak-rpc-schema",
"crates/yaak-crypto",
"crates/yaak-git",
"crates/yaak-models",
+211
View File
@@ -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<string, boolean>;
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<string, Set<(payload: unknown) => 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<void> {
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<Response> {
const headers = new Headers(init?.headers);
headers.set("Authorization", `Bearer ${this.token ?? ""}`);
return fetch(this.baseUrl + path, { ...init, headers });
}
async loadInfo(): Promise<BridgeInfo> {
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<T>(cmd: string, payload: Record<string, unknown> = {}): Promise<T> {
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();
};
}
}
+293
View File
@@ -0,0 +1,293 @@
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";
}
/**
* 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<T>(cmd: string, payload?: RpcPayload): Promise<T> {
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<History["pushState"]>) {
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: {
readDir: async () => {
throw unsupported("Browsing the filesystem");
},
// Only ever called with a path this host handed out, and this host has
// no dialog or drag-drop to hand one out with.
readText: async () => {
throw unsupported("Reading a local file");
},
// No filesystem here, so a path is just a string this host echoes back.
url: (path) => path,
basename: async (path) => path.split(/[/\\]/).pop() ?? path,
resolveResource: async (path) => path,
},
// Bodies live on the bridge's disk and are addressed by response id. The
// server resolves the id through its database, so a page can only ever
// reach a body the engine actually wrote.
blobs: {
async read(id) {
const res = await connection.fetch(`/responses/${encodeURIComponent(id)}/body`);
if (res.status === 404) return null;
if (!res.ok) {
throw new Error(`Failed to read response body (${res.status})`);
}
return new Uint8Array(await res.arrayBuffer());
},
// The `<img src>`/`<video src>` equivalent of Tauri's `convertFileSrc`.
// The token rides in the query because the browser makes these requests
// itself and the page cannot add a header to them.
async url(id) {
return connection.url(`/responses/${encodeURIComponent(id)}/body`);
},
},
rpc: <T,>(cmd: string, payload?: RpcPayload): Promise<T> => {
// `plugin:`-prefixed commands are Tauri host plugins, not engine
// commands, so they never reach the RpcRouter. Two of them are window
// chrome the tab can do itself; the rest belong to features this host
// reports false for, and saying so beats a confusing "unknown command".
if (cmd.startsWith("plugin:")) {
return handleHostPluginCommand<T>(cmd, payload);
}
return connection.rpc<T>(cmd, payload);
},
async rpcStream<T, M>(
cmd: string,
payload: RpcPayload,
onMessage: (message: M) => void,
): Promise<RpcStreamHandle<T>> {
// Caller-minted id, subscribed before dispatch, exactly as on the
// desktop: the command can emit its first message before it returns.
const streamId = crypto.randomUUID();
const unlisten = connection.listen(`stream_${streamId}`, (p) => onMessage(p as M));
try {
const result = await connection.rpc<T>(cmd, { ...payload, streamId });
return { result, unlisten };
} catch (err) {
unlisten();
throw err;
}
},
listen: <T,>(event: string, callback: (payload: T) => void): Unsubscribe =>
connection.listen(event, (payload) => callback(payload as T)),
emit: async (event, payload) => connection.emit(event, payload),
openUrl: async (url) => {
window.open(url, "_blank", "noopener,noreferrer");
},
revealItemInDir: async () => {
throw unsupported("Revealing a file");
},
osType: detectOsType,
appIdentifier: async () => "app.yaak.bridge",
};
}
+106
View File
@@ -0,0 +1,106 @@
/**
* Deciding which host to install, and getting a bridge token when there isn't
* one yet.
*
* Dev-grade on purpose. The token is a shared secret the bridge prints at
* startup, passed in the URL and kept for the session. OTP pairing and request
* encryption replace this whole file; the seam is that nothing outside it knows
* how the token was obtained.
*/
export interface BridgeConfig {
url: string;
token: string;
}
const TOKEN_STORAGE_KEY = "yaak.bridge.token";
const TOKEN_QUERY_PARAM = "bridgeToken";
declare global {
interface Window {
__TAURI_INTERNALS__?: unknown;
}
}
function bridgeUrl(): string {
// Set when the frontend runs on a Vite dev server and the bridge is on its
// own port. When the bridge serves the built app, they share an origin.
// Vite inlines `import.meta.env` at build time. Read it through a cast so
// this package typechecks on its own without depending on Vite's types, and
// still picks up the real declaration when the app compiles it.
const env = (import.meta as { env?: Record<string, string | undefined> }).env;
const configured = env?.VITE_YAAK_BRIDGE_URL;
return (configured ?? window.location.origin).replace(/\/$/, "");
}
/**
* The bridge token, or null if the user hasn't supplied one.
*
* A token in the URL is consumed and stashed: leaving it in the address bar
* means it lands in the router's own history entries and in anything the user
* copies out of the bar.
*/
function readToken(): string | null {
const url = new URL(window.location.href);
const fromQuery = url.searchParams.get(TOKEN_QUERY_PARAM);
if (fromQuery != null && fromQuery !== "") {
sessionStorage.setItem(TOKEN_STORAGE_KEY, fromQuery);
url.searchParams.delete(TOKEN_QUERY_PARAM);
history.replaceState(null, "", url.toString());
return fromQuery;
}
return sessionStorage.getItem(TOKEN_STORAGE_KEY);
}
/** Whether this build should talk to a bridge at all. */
export function shouldUseBridge(): boolean {
if (typeof window === "undefined") return false;
// Running inside the desktop app: Tauri always wins.
if (window.__TAURI_INTERNALS__ != null) return false;
return true;
}
export function bridgeConfig(): BridgeConfig | null {
const token = readToken();
if (token == null) return null;
return { url: bridgeUrl(), token };
}
/**
* Put the connect form on screen.
*
* Synchronous, and it does not stop anything by itself — the caller pairs it
* with a host that never connects, so the app's own boot-time await is what
* holds. Submitting reloads with the token in the query, which `readToken`
* then consumes.
*/
export function promptForToken(): void {
document.body.innerHTML = `
<div style="font-family: system-ui, sans-serif; max-width: 26rem; margin: 15vh auto; padding: 0 1.5rem; color: #d5d3e0">
<h1 style="font-size: 1.25rem; margin: 0 0 0.5rem">Connect to the Yaak Bridge</h1>
<p style="margin: 0 0 1.25rem; line-height: 1.5; color: #9a97ad">
Paste the token the bridge printed when it started.
</p>
<form id="yaak-bridge-connect" style="display: flex; gap: 0.5rem">
<input name="token" autofocus autocomplete="off" spellcheck="false" placeholder="Bridge token"
style="flex: 1; padding: 0.5rem 0.65rem; border-radius: 0.375rem; border: 1px solid #3b3950; background: #232135; color: inherit; font-family: ui-monospace, monospace" />
<button type="submit"
style="padding: 0.5rem 1rem; border-radius: 0.375rem; border: 0; background: #6d5ef0; color: white; font-weight: 500; cursor: pointer">
Connect
</button>
</form>
</div>
`;
document.documentElement.style.background = "#1b1a29";
document.getElementById("yaak-bridge-connect")?.addEventListener("submit", (e) => {
e.preventDefault();
const token = new FormData(e.target as HTMLFormElement).get("token");
if (typeof token !== "string" || token === "") return;
const url = new URL(window.location.href);
url.searchParams.set(TOKEN_QUERY_PARAM, token);
window.location.href = url.toString();
});
}
+23 -7
View File
@@ -1,14 +1,30 @@
import { createBridgePlatform } from "./bridge";
import { bridgeConfig, promptForToken, shouldUseBridge } from "./connect";
import { setPlatform } from "./registry";
import { createTauriPlatform } from "./tauri";
// Desktop is the only host today, so it is installed unconditionally and
// synchronously — several modules call commands while the module graph is still
// evaluating, so there is no later moment to do this in.
// This line is the swap point, and it has to run synchronously: several modules
// call commands while the module graph is still evaluating, so there is no
// later moment to install a host in.
//
// This line is the swap point. A browser build selects its own host here, and
// because nothing else in the app imports a host directly, that is the whole
// change.
setPlatform(createTauriPlatform());
// Both hosts are constructible without waiting for anything. The bridge host
// opens its connection in the background and queues calls made before it lands,
// which is why picking a host here does not mean blocking on one.
if (shouldUseBridge()) {
const config = bridgeConfig();
if (config == null) {
// No token yet: put the connect form on screen and install a host that
// never connects. Boot then stalls at its own top-level await rather than
// failing somewhere that doesn't explain itself — and it stalls in the one
// place already designed to wait, which keeps this file synchronous.
promptForToken();
setPlatform(createBridgePlatform(window.location.origin, null));
} else {
setPlatform(createBridgePlatform(config.url, config.token));
}
} else {
setPlatform(createTauriPlatform());
}
export * from "./capabilities";
export { platform, setPlatform } from "./registry";