diff --git a/apps/yaak-client/components/Sidebar.tsx b/apps/yaak-client/components/Sidebar.tsx index 9e897418..196e116e 100644 --- a/apps/yaak-client/components/Sidebar.tsx +++ b/apps/yaak-client/components/Sidebar.tsx @@ -1,5 +1,6 @@ import type { Extension } from "@codemirror/state"; import { Compartment } from "@codemirror/state"; +import { getCurrentWebviewWindow } from "@tauri-apps/api/webviewWindow"; import { debounce } from "@yaakapp-internal/lib"; import { gitMutations } from "@yaakapp-internal/git"; import type { GitStatus } from "@yaakapp-internal/git"; @@ -131,11 +132,17 @@ function Sidebar({ className }: { className?: string }) { if (!didFocus) filterRef.current?.focus(); }, []); - // Focus any new sidebar models when created - useListenToTauriEvent("model_write", ({ payload }) => { - if (!isSidebarLeafModel(payload.model)) return; - if (!(payload.change.type === "upsert" && payload.change.created)) return; - treeRef.current?.selectItem(payload.model.id, true); + // Focus new sidebar models created by the user in this window. Writes from other + // sources (import, sync, CLI) can carry thousands of models and shouldn't move + // the selection. + useListenToTauriEvent("model_writes", ({ payload: payloads }) => { + for (const payload of payloads) { + if (payload.updateSource.type !== "window") continue; + if (payload.updateSource.label !== getCurrentWebviewWindow().label) continue; + if (!isSidebarLeafModel(payload.model)) continue; + if (!(payload.change.type === "upsert" && payload.change.created)) continue; + treeRef.current?.selectItem(payload.model.id, true); + } }); useEffect(() => { diff --git a/apps/yaak-client/components/WorkspaceActionsDropdown.tsx b/apps/yaak-client/components/WorkspaceActionsDropdown.tsx index a7a234c7..16ff8550 100644 --- a/apps/yaak-client/components/WorkspaceActionsDropdown.tsx +++ b/apps/yaak-client/components/WorkspaceActionsDropdown.tsx @@ -16,6 +16,7 @@ import { useCreateWorkspace } from "../hooks/useCreateWorkspace"; import { useDeleteSendHistory } from "../hooks/useDeleteSendHistory"; import { useWorkspaceActions } from "../hooks/useWorkspaceActions"; import { showDialog } from "../lib/dialog"; +import { importData } from "../lib/importData"; import { jotaiStore } from "../lib/jotai"; import { revealInFinderText } from "../lib/reveal"; import { CloneGitRepositoryDialog } from "./CloneGitRepositoryDialog"; @@ -90,6 +91,11 @@ export const WorkspaceActionsDropdown = memo(function WorkspaceActionsDropdown({ leftSlot: , onSelect: openCloneGitRepositoryDialog, }, + { + label: "Import Data", + leftSlot: , + onSelect: () => importData.mutate(), + }, ], }, ]; diff --git a/apps/yaak-client/font-size.ts b/apps/yaak-client/font-size.ts index 4055f35c..6a5dbf2f 100644 --- a/apps/yaak-client/font-size.ts +++ b/apps/yaak-client/font-size.ts @@ -8,10 +8,12 @@ function setFontSizeOnDocument(fontSize: number) { document.documentElement.style.fontSize = `${fontSize}px`; } -listen("model_write", async (event) => { - if (event.payload.change.type !== "upsert") return; - if (event.payload.model.model !== "settings") return; - setFontSizeOnDocument(event.payload.model.interfaceFontSize); +listen("model_writes", async (event) => { + for (const payload of event.payload) { + if (payload.change.type !== "upsert") continue; + if (payload.model.model !== "settings") continue; + setFontSizeOnDocument(payload.model.interfaceFontSize); + } }).catch(console.error); fireAndForget(getSettings().then((settings) => setFontSizeOnDocument(settings.interfaceFontSize))); diff --git a/apps/yaak-client/font.ts b/apps/yaak-client/font.ts index dbdde529..96913d0a 100644 --- a/apps/yaak-client/font.ts +++ b/apps/yaak-client/font.ts @@ -12,10 +12,12 @@ function setFonts(settings: Settings) { ); } -listen("model_write", async (event) => { - if (event.payload.change.type !== "upsert") return; - if (event.payload.model.model !== "settings") return; - setFonts(event.payload.model); +listen("model_writes", async (event) => { + for (const payload of event.payload) { + if (payload.change.type !== "upsert") continue; + if (payload.model.model !== "settings") continue; + setFonts(payload.model); + } }).catch(console.error); fireAndForget(getSettings().then((settings) => setFonts(settings))); diff --git a/apps/yaak-client/hooks/useRequestUpdateKey.ts b/apps/yaak-client/hooks/useRequestUpdateKey.ts index 9546f8ff..dbd48609 100644 --- a/apps/yaak-client/hooks/useRequestUpdateKey.ts +++ b/apps/yaak-client/hooks/useRequestUpdateKey.ts @@ -7,24 +7,33 @@ import { jotaiStore } from "../lib/jotai"; const requestUpdateKeyAtom = atom>({}); getCurrentWebviewWindow() - .listen("model_write", ({ payload }) => { - if (payload.change.type !== "upsert") return; + .listen("model_writes", ({ payload: payloads }) => { + const changedIds: string[] = []; + for (const payload of payloads) { + if (payload.change.type !== "upsert") continue; - if ( - (payload.model.model === "http_request" || - payload.model.model === "grpc_request" || - payload.model.model === "websocket_request") && - ((payload.updateSource.type === "window" && - payload.updateSource.label !== getCurrentWebviewWindow().label) || - payload.updateSource.type !== "window") - ) { - wasUpdatedExternally(payload.model.id); + if ( + (payload.model.model === "http_request" || + payload.model.model === "grpc_request" || + payload.model.model === "websocket_request") && + ((payload.updateSource.type === "window" && + payload.updateSource.label !== getCurrentWebviewWindow().label) || + payload.updateSource.type !== "window") + ) { + changedIds.push(payload.model.id); + } } + if (changedIds.length > 0) wasUpdatedExternally(changedIds); }) .catch(console.error); -export function wasUpdatedExternally(changedRequestId: string) { - jotaiStore.set(requestUpdateKeyAtom, (m) => ({ ...m, [changedRequestId]: generateId() })); +export function wasUpdatedExternally(changedRequestIds: string | string[]) { + const ids = Array.isArray(changedRequestIds) ? changedRequestIds : [changedRequestIds]; + jotaiStore.set(requestUpdateKeyAtom, (m) => { + const next = { ...m }; + for (const id of ids) next[id] = generateId(); + return next; + }); } export function useRequestUpdateKey(requestId: string | null) { diff --git a/apps/yaak-client/init/sync.ts b/apps/yaak-client/init/sync.ts index 61412711..f81b1220 100644 --- a/apps/yaak-client/init/sync.ts +++ b/apps/yaak-client/init/sync.ts @@ -33,8 +33,8 @@ const syncAfterModelWrite = eagerDebounceAsync(sync, 1000); * simply add long-lived subscribers for the lifetime of the app. */ function initModelListeners() { - listenToTauriEvent("model_write", (p) => { - if (isModelRelevant(p.payload.model)) syncAfterModelWrite(); + listenToTauriEvent("model_writes", (p) => { + if (p.payload.some((payload) => isModelRelevant(payload.model))) syncAfterModelWrite(); }); } diff --git a/apps/yaak-client/theme.ts b/apps/yaak-client/theme.ts index 3d01a3d4..858b1879 100644 --- a/apps/yaak-client/theme.ts +++ b/apps/yaak-client/theme.ts @@ -46,11 +46,13 @@ async function configureThemeAndShow() { } // Listen for settings changes, the re-compute theme -listen("model_write", async (event) => { - if (event.payload.change.type !== "upsert") return; - - const model = event.payload.model.model; - if (model !== "settings" && model !== "plugin") return; +listen("model_writes", async (event) => { + const relevant = event.payload.some( + (p) => + p.change.type === "upsert" && + (p.model.model === "settings" || p.model.model === "plugin"), + ); + if (!relevant) return; await configureThemeAndShow(); }).catch(console.error); diff --git a/crates-cli/yaak-cli/src/commands/workspace.rs b/crates-cli/yaak-cli/src/commands/workspace.rs index 38bfcc01..9a05219b 100644 --- a/crates-cli/yaak-cli/src/commands/workspace.rs +++ b/crates-cli/yaak-cli/src/commands/workspace.rs @@ -131,7 +131,7 @@ fn delete(ctx: &CliContext, workspace_id: &str, yes: bool) -> CommandResult { let deleted = ctx .db() - .delete_workspace_by_id(workspace_id, &UpdateSource::Sync) + .delete_workspace_by_id(workspace_id, &UpdateSource::Sync, ctx.blob_manager()) .map_err(|e| format!("Failed to delete workspace: {e}"))?; println!("Deleted workspace: {}", deleted.id); Ok(()) diff --git a/crates-tauri/yaak-app-client/src/models_ext.rs b/crates-tauri/yaak-app-client/src/models_ext.rs index e210cc97..219d4573 100644 --- a/crates-tauri/yaak-app-client/src/models_ext.rs +++ b/crates-tauri/yaak-app-client/src/models_ext.rs @@ -14,7 +14,7 @@ use yaak_models::client_db::ClientDb; use yaak_models::error::Result; use yaak_models::models::{AnyModel, GraphQlIntrospection, GrpcEvent, Settings, WebsocketEvent}; use yaak_models::query_manager::QueryManager; -use yaak_models::util::UpdateSource; +use yaak_models::util::{ModelPayload, UpdateSource}; use yaak_plugins::manager::PluginManager; const MODEL_CHANGES_RETENTION_HOURS: i64 = 1; @@ -57,6 +57,7 @@ fn drain_model_changes_batch( } let fetched_count = changes.len(); + let mut batch: Vec = Vec::with_capacity(fetched_count); for change in changes { cursor.created_at = change.created_at; cursor.id = change.id; @@ -66,8 +67,14 @@ fn drain_model_changes_batch( if matches!(change.payload.update_source, UpdateSource::Window { .. }) { continue; } - if let Err(err) = app_handle.emit("model_write", change.payload) { - error!("Failed to emit model_write event: {err:?}"); + batch.push(change.payload); + } + + // Emit as a single batch so bulk writes (imports, sync, CLI) don't flood the + // frontend with per-model events. + if !batch.is_empty() { + if let Err(err) = app_handle.emit("model_writes", batch) { + error!("Failed to emit model_writes event: {err:?}"); } } @@ -162,33 +169,39 @@ pub(crate) fn models_upsert( Ok(id) } +// Async so cascading deletes (e.g. a workspace with thousands of requests) run on a +// blocking thread instead of stalling the main thread and all other IPC. #[tauri::command] -pub(crate) fn models_delete( +pub(crate) async fn models_delete( window: WebviewWindow, model: AnyModel, ) -> Result { use yaak_models::error::Error::GenericError; - let blobs = window.blob_manager(); - // 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)?.id, - a => return Err(GenericError(format!("Cannot delete AnyModel {a:?})"))), - }; - Ok(id) + tauri::async_runtime::spawn_blocking(move || { + let blobs = window.blob_manager(); + // 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) + }) }) + .await + .map_err(|e| GenericError(format!("Delete task failed: {e}")))? } #[tauri::command] @@ -364,6 +377,20 @@ pub fn init() -> TauriPlugin { let poll_query_manager = query_manager.clone(); + // GC response bodies orphaned by cascade deletes, which historically + // didn't clean the blob DB or responses directory + let gc_query_manager = query_manager.clone(); + let gc_blob_manager = blob_manager.clone(); + let gc_responses_dir = app_path.join("responses"); + tauri::async_runtime::spawn_blocking(move || { + let db = gc_query_manager.connect(); + match db.delete_orphaned_response_bodies(&gc_blob_manager, &gc_responses_dir) { + Ok(0) => {} + Ok(n) => log::info!("Deleted {n} orphaned response bodies"), + Err(e) => error!("Failed to delete orphaned response bodies: {e:?}"), + } + }); + app_handle.manage(query_manager); app_handle.manage(blob_manager); @@ -378,12 +405,22 @@ pub fn init() -> TauriPlugin { // current sync-model UX snappy, while DB polling handles external writers (CLI). let app_handle_local = app_handle.clone(); tauri::async_runtime::spawn(async move { - for payload in rx { - if !matches!(payload.update_source, UpdateSource::Window { .. }) { + while let Ok(payload) = rx.recv() { + let mut batch: Vec = Vec::new(); + if matches!(payload.update_source, UpdateSource::Window { .. }) { + batch.push(payload); + } + // Coalesce any writes already queued into the same emit + while let Ok(next) = rx.try_recv() { + if matches!(next.update_source, UpdateSource::Window { .. }) { + batch.push(next); + } + } + if batch.is_empty() { continue; } - if let Err(err) = app_handle_local.emit("model_write", payload) { - error!("Failed to emit local model_write event: {err:?}"); + if let Err(err) = app_handle_local.emit("model_writes", batch) { + error!("Failed to emit local model_writes event: {err:?}"); } } }); diff --git a/crates-tauri/yaak-app-client/src/plugins_ext.rs b/crates-tauri/yaak-app-client/src/plugins_ext.rs index 3fcc8fe2..fb00da96 100644 --- a/crates-tauri/yaak-app-client/src/plugins_ext.rs +++ b/crates-tauri/yaak-app-client/src/plugins_ext.rs @@ -285,6 +285,19 @@ pub fn init() -> TauriPlugin { #[cfg(not(target_os = "windows"))] let node_bin_name = "yaaknode"; + // In dev, spawn yaaknode from the source vendored dir, not the copy under + // target/. tauri-build rewrites the target copy in place when the source + // changes (e.g. a Node version bump between branches), and on macOS an + // in-place write permanently taints the inode's cached code signature — + // every later spawn of it dies with SIGKILL (Code Signature Invalid). + // vendor-node.cjs always recreates the source file on a fresh inode. + #[cfg(debug_assertions)] + let node_bin_path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("vendored") + .join("node") + .join(node_bin_name); + + #[cfg(not(debug_assertions))] let node_bin_path = app_handle .path() .resolve(format!("vendored/node/{}", node_bin_name), BaseDirectory::Resource) diff --git a/crates-tauri/yaak-app-client/src/sync_ext.rs b/crates-tauri/yaak-app-client/src/sync_ext.rs index c5f54b5f..f38446f0 100644 --- a/crates-tauri/yaak-app-client/src/sync_ext.rs +++ b/crates-tauri/yaak-app-client/src/sync_ext.rs @@ -3,7 +3,7 @@ //! This module provides the Tauri commands for sync functionality. use crate::error::Result; -use crate::models_ext::QueryManagerExt; +use crate::models_ext::{BlobManagerExt, QueryManagerExt}; use chrono::Utc; use log::warn; use serde::{Deserialize, Serialize}; @@ -55,7 +55,8 @@ pub(crate) async fn cmd_sync_apply( workspace_id: &str, ) -> Result<()> { let db = app_handle.db(); - let sync_state_ops = apply_sync_ops(&db, workspace_id, sync_dir, sync_ops)?; + let blobs = app_handle.blob_manager(); + let sync_state_ops = apply_sync_ops(&db, &blobs, workspace_id, sync_dir, sync_ops)?; apply_sync_state_ops(&db, workspace_id, sync_dir, sync_state_ops)?; Ok(()) } diff --git a/crates/common/yaak-database/src/db_context.rs b/crates/common/yaak-database/src/db_context.rs index 8303e89d..eabafef4 100644 --- a/crates/common/yaak-database/src/db_context.rs +++ b/crates/common/yaak-database/src/db_context.rs @@ -160,6 +160,24 @@ impl<'a> DbContext<'a> { Ok((m, created)) } + /// Bulk-delete all rows matching a column value with a single statement. + /// Returns the number of rows deleted. + pub fn delete_many( + &self, + col: impl IntoColumnRef, + value: impl Into, + ) -> Result + where + M: UpsertModelInfo, + { + let (sql, params) = Query::delete() + .from_table(M::table_name()) + .cond_where(Expr::col(col).eq(value)) + .build_rusqlite(SqliteQueryBuilder); + let count = self.conn.execute(sql.as_str(), &*params.as_params())?; + Ok(count) + } + /// Delete a model by its ID. Returns the number of rows deleted. pub fn delete(&self, m: &M) -> Result where diff --git a/crates/common/yaak-database/src/update_source.rs b/crates/common/yaak-database/src/update_source.rs index 18afea4f..c378d5d3 100644 --- a/crates/common/yaak-database/src/update_source.rs +++ b/crates/common/yaak-database/src/update_source.rs @@ -21,5 +21,9 @@ impl UpdateSource { #[serde(rename_all = "snake_case", tag = "type")] pub enum ModelChangeEvent { Upsert { created: bool }, + /// A delete for a workspace implies deletion of every model in that + /// workspace — children are bulk-deleted without their own change rows or + /// events, and consumers must prune the subtree themselves (the frontend + /// model store does this centrally). Delete, } diff --git a/crates/yaak-models/guest-js/store.ts b/crates/yaak-models/guest-js/store.ts index 404dd85b..d913064d 100644 --- a/crates/yaak-models/guest-js/store.ts +++ b/crates/yaak-models/guest-js/store.ts @@ -17,28 +17,91 @@ export function initModelStore(store: JotaiStore) { window.addEventListener("beforeunload", flushAllPendingPatches); getCurrentWebviewWindow() - .listen("model_write", ({ payload }) => { - if (shouldIgnoreModel(payload)) return; - + .listen("model_writes", ({ payload: payloads }) => { mustStore().set(modelStoreDataAtom, (prev: ModelStoreData) => { - if (payload.change.type === "upsert") { - return { - ...prev, - [payload.model.model]: { - ...prev[payload.model.model], - [payload.model.id]: payload.model, - }, - }; - } else { - const modelData = { ...prev[payload.model.model] }; - delete modelData[payload.model.id]; - return { ...prev, [payload.model.model]: modelData }; + // Apply the entire batch in one update, cloning each touched bucket only + // once. Bulk writes (imports, sync, CLI) can carry hundreds of models. + const next = { ...prev }; + const clonedBuckets = new Set(); + let changed = false; + + for (const payload of payloads) { + if (shouldIgnoreModel(payload)) continue; + if (isUnsafeObjectKey(payload.model.model)) continue; + if (isUnsafeObjectKey(payload.model.id)) continue; + + if (payload.change.type === "upsert") { + const modelType = payload.model.model; + if (!clonedBuckets.has(modelType)) { + next[modelType] = { ...next[modelType] } as never; + clonedBuckets.add(modelType); + } + (next[modelType] as Record)[payload.model.id] = payload.model; + changed = true; + } else { + changed = applyModelDelete(next, clonedBuckets, payload.model) || changed; + } } + + return changed ? next : prev; }); }) .catch(console.error); } +/** + * Model buckets are plain objects keyed by model id, so ids that collide with + * Object.prototype members ("__proto__" etc.) could pollute the prototype. + * Real model ids are backend-generated and never look like this. + */ +function isUnsafeObjectKey(key: string): boolean { + return key === "__proto__" || key === "constructor" || key === "prototype"; +} + +function deleteFromBucket( + next: ModelStoreData, + clonedBuckets: Set, + modelType: AnyModel["model"], + id: string, +): boolean { + if (isUnsafeObjectKey(modelType)) return false; + if (isUnsafeObjectKey(id)) return false; + if (!Object.prototype.hasOwnProperty.call(next[modelType], id)) return false; + if (!clonedBuckets.has(modelType)) { + next[modelType] = { ...next[modelType] } as never; + clonedBuckets.add(modelType); + } + delete (next[modelType] as Record)[id]; + return true; +} + +/** + * Apply a model delete to store data, mutating `next` in place (buckets are + * cloned once, tracked via `clonedBuckets`). A workspace delete implies its + * entire subtree: the backend bulk-deletes children and records/emits only the + * workspace event (see ModelChangeEvent), so prune them here. + */ +function applyModelDelete( + next: ModelStoreData, + clonedBuckets: Set, + model: AnyModel, +): boolean { + let changed = deleteFromBucket(next, clonedBuckets, model.model, model.id); + + if (model.model === "workspace") { + for (const modelType of Object.keys(next) as AnyModel["model"][]) { + const bucket = next[modelType] as Record; + for (const [id, m] of Object.entries(bucket)) { + if ("workspaceId" in m && m.workspaceId === model.id) { + changed = deleteFromBucket(next, clonedBuckets, modelType, id) || changed; + } + } + } + } + + return changed; +} + function mustStore(): JotaiStore { if (_store == null) { throw new Error("Model store was not initialized"); @@ -243,6 +306,15 @@ export async function deleteModel("models_delete", { model })); + + // Apply the delete locally right away so callers can rely on the store once the + // promise resolves. The backend echo arrives async, so anything that reads the + // store immediately after awaiting (e.g. redirecting away from a deleted + // workspace) would otherwise race it. The echo re-applying later is a no-op. + mustStore().set(modelStoreDataAtom, (prev: ModelStoreData) => { + const next = { ...prev }; + return applyModelDelete(next, new Set(), model as AnyModel) ? next : prev; + }); } export async function duplicateModel< diff --git a/crates/yaak-models/src/blob_manager.rs b/crates/yaak-models/src/blob_manager.rs index 31d4b155..871cbf75 100644 --- a/crates/yaak-models/src/blob_manager.rs +++ b/crates/yaak-models/src/blob_manager.rs @@ -79,6 +79,15 @@ impl BlobContext { Ok(chunks) } + /// List all distinct body IDs in the blob database. + pub fn list_body_ids(&self) -> Result> { + let mut stmt = self.conn.prepare("SELECT DISTINCT body_id FROM body_chunks")?; + let ids = stmt + .query_map([], |row| row.get(0))? + .collect::, _>>()?; + Ok(ids) + } + /// Delete all chunks for a body. pub fn delete_chunks(&self, body_id: &str) -> Result<()> { self.conn.execute("DELETE FROM body_chunks WHERE body_id = ?1", params![body_id])?; diff --git a/crates/yaak-models/src/client_db.rs b/crates/yaak-models/src/client_db.rs index 470a99a8..e4023e42 100644 --- a/crates/yaak-models/src/client_db.rs +++ b/crates/yaak-models/src/client_db.rs @@ -65,6 +65,21 @@ impl<'a> ClientDb<'a> { Ok(self.ctx.find_many(col, value, limit)?) } + /// Bulk-delete all rows matching a column value WITHOUT recording model + /// changes or emitting events. Only use for cascades whose deletion is + /// implied by a recorded parent delete (e.g. workspace children — see + /// [`ModelChangeEvent::Delete`]). + pub(crate) fn delete_many_untracked( + &self, + col: impl IntoColumnRef, + value: impl Into, + ) -> Result + where + M: UpsertModelInfo, + { + Ok(self.ctx.delete_many::(col, value)?) + } + // --- Write operations (with event recording) --- pub(crate) fn upsert(&self, model: &M, source: &UpdateSource) -> Result diff --git a/crates/yaak-models/src/queries/http_responses.rs b/crates/yaak-models/src/queries/http_responses.rs index 3530929f..a1b2010c 100644 --- a/crates/yaak-models/src/queries/http_responses.rs +++ b/crates/yaak-models/src/queries/http_responses.rs @@ -44,6 +44,56 @@ impl<'a> ClientDb<'a> { Ok(count) } + /// Delete response body data (blob chunks and body files) whose owning HTTP + /// response row no longer exists. Cascaded deletes (request, folder, + /// workspace) historically never cleaned the blob DB or the responses + /// directory, so orphans accumulate; this runs in the background at startup. + /// + /// Safe against in-flight sends: the response row is created before its + /// body file or chunks are written. + /// + /// Returns the number of orphaned bodies deleted. + pub fn delete_orphaned_response_bodies( + &self, + blobs: &BlobManager, + responses_dir: &std::path::Path, + ) -> Result { + let mut deleted = 0; + + // Blob chunks are keyed "{response_id}.request" + let blob_ctx = blobs.connect(); + for body_id in blob_ctx.list_body_ids()? { + let response_id = body_id.split('.').next().unwrap_or_default(); + if self.find_optional::(HttpResponseIden::Id, response_id).is_some() { + continue; + } + blob_ctx.delete_chunks(&body_id)?; + deleted += 1; + } + + // Body files are stored as {responses_dir}/{response_id} + if let Ok(entries) = fs::read_dir(responses_dir) { + for entry in entries.flatten() { + let path = entry.path(); + if !path.is_file() { + continue; + } + let Some(response_id) = path.file_name().and_then(|n| n.to_str()) else { + continue; + }; + if self.find_optional::(HttpResponseIden::Id, response_id).is_some() + { + continue; + } + if fs::remove_file(&path).is_ok() { + deleted += 1; + } + } + } + + Ok(deleted) + } + /// Returns the number of responses deleted. pub fn delete_all_http_responses_for_workspace( &self, @@ -118,3 +168,68 @@ impl<'a> ClientDb<'a> { if response.id.is_empty() { Ok(response.clone()) } else { self.upsert(response, source) } } } + +#[cfg(test)] +mod tests { + use crate::blob_manager::BodyChunk; + use crate::init_in_memory; + use crate::models::{HttpRequest, HttpResponse, Workspace}; + use crate::util::UpdateSource; + + #[test] + fn deletes_orphaned_response_bodies() { + let (query_manager, blob_manager, _rx) = init_in_memory().expect("Failed to init DB"); + let db = query_manager.connect(); + + let source = &UpdateSource::Background; + let workspace = db + .upsert_workspace(&Workspace { name: "GC Test".to_string(), ..Default::default() }, source) + .expect("Failed to upsert workspace"); + let request = db + .upsert_http_request( + &HttpRequest { workspace_id: workspace.id.clone(), ..Default::default() }, + source, + ) + .expect("Failed to upsert request"); + + let live = db + .upsert_http_response( + &HttpResponse { + request_id: request.id.clone(), + workspace_id: workspace.id.clone(), + ..Default::default() + }, + source, + &blob_manager, + ) + .expect("Failed to upsert response"); + + let live_body_id = format!("{}.request", live.id); + { + // Scope the connection: the in-memory pool only has one, and the GC + // needs to take it + let blob_ctx = blob_manager.connect(); + blob_ctx.insert_chunk(&BodyChunk::new(&live_body_id, 0, b"live".to_vec())).unwrap(); + blob_ctx.insert_chunk(&BodyChunk::new("rs_gone.request", 0, b"dead".to_vec())).unwrap(); + } + + let dir = std::env::temp_dir().join(format!("yaak-blob-gc-test-{}", live.id)); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::write(dir.join(&live.id), b"live").unwrap(); + std::fs::write(dir.join("rs_gone"), b"dead").unwrap(); + + let deleted = db + .delete_orphaned_response_bodies(&blob_manager, &dir) + .expect("Failed to GC response bodies"); + assert_eq!(deleted, 2); + + // Live data survives, orphans are gone + let blob_ctx = blob_manager.connect(); + assert!(blob_ctx.body_exists(&live_body_id).unwrap()); + assert!(!blob_ctx.body_exists("rs_gone.request").unwrap()); + assert!(dir.join(&live.id).exists()); + assert!(!dir.join("rs_gone").exists()); + + std::fs::remove_dir_all(&dir).ok(); + } +} diff --git a/crates/yaak-models/src/queries/model_changes.rs b/crates/yaak-models/src/queries/model_changes.rs index 386511ea..987118ad 100644 --- a/crates/yaak-models/src/queries/model_changes.rs +++ b/crates/yaak-models/src/queries/model_changes.rs @@ -103,7 +103,7 @@ mod tests { #[test] fn records_model_changes_for_upsert_and_delete() { - let (query_manager, _blob_manager, _rx) = init_in_memory().expect("Failed to init DB"); + let (query_manager, blob_manager, _rx) = init_in_memory().expect("Failed to init DB"); let db = query_manager.connect(); let workspace = db @@ -128,7 +128,7 @@ mod tests { )); assert!(matches!(created_changes[0].payload.update_source, UpdateSource::Sync)); - db.delete_workspace_by_id(&workspace.id, &UpdateSource::Sync) + db.delete_workspace_by_id(&workspace.id, &UpdateSource::Sync, &blob_manager) .expect("Failed to delete workspace"); let all_changes = db.list_model_changes_after(0, 10).expect("Failed to list changes"); @@ -178,7 +178,7 @@ mod tests { #[test] fn list_model_changes_since_uses_timestamp_with_id_tiebreaker() { - let (query_manager, _blob_manager, _rx) = init_in_memory().expect("Failed to init DB"); + let (query_manager, blob_manager, _rx) = init_in_memory().expect("Failed to init DB"); let db = query_manager.connect(); let workspace = db @@ -192,7 +192,7 @@ mod tests { &UpdateSource::Sync, ) .expect("Failed to upsert workspace"); - db.delete_workspace_by_id(&workspace.id, &UpdateSource::Sync) + db.delete_workspace_by_id(&workspace.id, &UpdateSource::Sync, &blob_manager) .expect("Failed to delete workspace"); let all = db.list_model_changes_after(0, 10).expect("Failed to list changes"); diff --git a/crates/yaak-models/src/queries/workspaces.rs b/crates/yaak-models/src/queries/workspaces.rs index c77d7e38..b7874c35 100644 --- a/crates/yaak-models/src/queries/workspaces.rs +++ b/crates/yaak-models/src/queries/workspaces.rs @@ -1,10 +1,17 @@ +use crate::blob_manager::BlobManager; use crate::client_db::ClientDb; use crate::error::Result; use crate::models::{ - AnyModel, EnvironmentIden, FolderIden, GrpcRequestIden, HttpRequestHeader, HttpRequestIden, - ResolvedHttpRequestSettings, ResolvedSetting, WebsocketRequestIden, Workspace, WorkspaceIden, + AnyModel, CookieJar, CookieJarIden, Environment, EnvironmentIden, Folder, FolderIden, + GraphQlIntrospection, GraphQlIntrospectionIden, GrpcConnection, GrpcConnectionIden, GrpcEvent, + GrpcEventIden, GrpcRequest, GrpcRequestIden, HttpRequest, HttpRequestHeader, HttpRequestIden, + HttpResponse, HttpResponseEvent, HttpResponseEventIden, HttpResponseIden, + ResolvedHttpRequestSettings, ResolvedSetting, SyncState, SyncStateIden, WebsocketConnection, + WebsocketConnectionIden, WebsocketEvent, WebsocketEventIden, WebsocketRequest, + WebsocketRequestIden, Workspace, WorkspaceIden, WorkspaceMeta, WorkspaceMetaIden, }; use crate::util::UpdateSource; +use log::warn; use serde_json::Value; use std::collections::BTreeMap; @@ -32,37 +39,99 @@ impl<'a> ClientDb<'a> { Ok(workspaces) } + /// Delete a workspace and everything in it. + /// + /// Children are bulk-deleted with one statement per table and are NOT + /// individually recorded in model_changes or emitted as events — the single + /// workspace delete event implies the subtree (see [`ModelChangeEvent::Delete`]). + /// This keeps huge workspaces (thousands of requests) fast and avoids + /// flooding event consumers. pub fn delete_workspace( &self, workspace: &Workspace, source: &UpdateSource, + blobs: &BlobManager, ) -> Result { - for m in self.find_many(HttpRequestIden::WorkspaceId, &workspace.id, None)? { - self.delete_http_request(&m, source)?; + let wid = workspace.id.as_str(); + + // Collect response cleanup targets before their rows disappear. The actual + // cleanup runs at the end: response bodies live on disk and in the blob DB, + // which don't participate in this transaction, so removing them must wait + // until every statement that could fail (and roll back the rows) is done. + let responses = self.find_many::(HttpResponseIden::WorkspaceId, wid, None)?; + + // Sync and the CLI call this on a plain connection where each statement + // would otherwise commit on its own, leaving a partially-deleted workspace + // if one fails. A savepoint makes the cascade atomic there, and nests + // harmlessly inside the interactive path's transaction. + let conn = self.conn().resolve(); + conn.execute_batch("SAVEPOINT delete_workspace")?; + + let result: Result = (|| { + self.delete_many_untracked::( + HttpResponseEventIden::WorkspaceId, + wid, + )?; + self.delete_many_untracked::(HttpResponseIden::WorkspaceId, wid)?; + self.delete_many_untracked::(HttpRequestIden::WorkspaceId, wid)?; + self.delete_many_untracked::(GrpcEventIden::WorkspaceId, wid)?; + self.delete_many_untracked::(GrpcConnectionIden::WorkspaceId, wid)?; + self.delete_many_untracked::(GrpcRequestIden::WorkspaceId, wid)?; + self.delete_many_untracked::(WebsocketEventIden::WorkspaceId, wid)?; + self.delete_many_untracked::( + WebsocketConnectionIden::WorkspaceId, + wid, + )?; + self.delete_many_untracked::(WebsocketRequestIden::WorkspaceId, wid)?; + self.delete_many_untracked::( + GraphQlIntrospectionIden::WorkspaceId, + wid, + )?; + self.delete_many_untracked::(FolderIden::WorkspaceId, wid)?; + self.delete_many_untracked::(EnvironmentIden::WorkspaceId, wid)?; + self.delete_many_untracked::(CookieJarIden::WorkspaceId, wid)?; + self.delete_many_untracked::(SyncStateIden::WorkspaceId, wid)?; + self.delete_many_untracked::(WorkspaceMetaIden::WorkspaceId, wid)?; + self.delete(workspace, source) + })(); + + let deleted = match result { + Ok(deleted) => { + conn.execute_batch("RELEASE delete_workspace")?; + deleted + } + Err(e) => { + let _ = conn + .execute_batch("ROLLBACK TO delete_workspace; RELEASE delete_workspace"); + return Err(e); + } + }; + + // Best-effort cleanup of response bodies (disk files and blob chunks). + // Failures only orphan unreferenced data, and are logged. + let blob_ctx = blobs.connect(); + for m in responses { + if let Some(p) = m.body_path { + if let Err(e) = std::fs::remove_file(&p) { + warn!("Failed to delete response body file {p:?}: {e}"); + } + } + if let Err(e) = blob_ctx.delete_chunks_like(&format!("{}.%", m.id)) { + warn!("Failed to delete blobs for response {}: {e}", m.id); + } } - for m in self.find_many(GrpcRequestIden::WorkspaceId, &workspace.id, None)? { - self.delete_grpc_request(&m, source)?; - } - - for m in self.find_many(WebsocketRequestIden::FolderId, &workspace.id, None)? { - self.delete_websocket_request(&m, source)?; - } - - for m in self.find_many(FolderIden::WorkspaceId, &workspace.id, None)? { - self.delete_folder(&m, source)?; - } - - for m in self.find_many(EnvironmentIden::WorkspaceId, &workspace.id, None)? { - self.delete_environment(&m, source)?; - } - - self.delete(workspace, source) + Ok(deleted) } - pub fn delete_workspace_by_id(&self, id: &str, source: &UpdateSource) -> Result { + pub fn delete_workspace_by_id( + &self, + id: &str, + source: &UpdateSource, + blobs: &BlobManager, + ) -> Result { let workspace = self.get_workspace(id)?; - self.delete_workspace(&workspace, source) + self.delete_workspace(&workspace, source, blobs) } pub fn upsert_workspace(&self, w: &Workspace, source: &UpdateSource) -> Result { diff --git a/crates/yaak-sync/src/sync.rs b/crates/yaak-sync/src/sync.rs index b2021cb3..c775d6e1 100644 --- a/crates/yaak-sync/src/sync.rs +++ b/crates/yaak-sync/src/sync.rs @@ -1,4 +1,5 @@ use crate::error::Result; +use yaak_models::blob_manager::BlobManager; use crate::models::SyncModel; use chrono::Utc; use log::{info, warn}; @@ -339,6 +340,7 @@ fn workspace_models(db: &ClientDb, version: &str, workspace_id: &str) -> Result< /// Returns a list of SyncStateOps that should be applied afterward. pub fn apply_sync_ops( db: &ClientDb, + blobs: &BlobManager, workspace_id: &str, sync_dir: &Path, sync_ops: Vec, @@ -435,7 +437,7 @@ pub fn apply_sync_ops( } } SyncOp::DbDelete { model, state } => { - delete_model(db, &model)?; + delete_model(db, blobs, &model)?; SyncStateOp::Delete { state: state.to_owned() } } SyncOp::IgnorePrivate { .. } => SyncStateOp::NoOp, @@ -547,10 +549,10 @@ fn derive_model_filename(m: &SyncModel) -> PathBuf { Path::new(&rel).to_path_buf() } -fn delete_model(db: &ClientDb, model: &SyncModel) -> Result<()> { +fn delete_model(db: &ClientDb, blobs: &BlobManager, model: &SyncModel) -> Result<()> { match model { SyncModel::Workspace(m) => { - db.delete_workspace(&m, &UpdateSource::Sync)?; + db.delete_workspace(&m, &UpdateSource::Sync, blobs)?; } SyncModel::Environment(m) => { db.delete_environment(&m, &UpdateSource::Sync)?;