Compare commits

..
Author SHA1 Message Date
Gregory Schier b2fa604022 Refresh PR #657 screenshot with operator-neutral disclosure 2026-09-14 21:26:26 -07:00
Gregory Schier 9d16dc87fe Refresh PR #657 screenshot with the external link icon 2026-09-14 21:20:19 -07:00
Gregory Schier ba6a8e39eb Refresh PR #657 screenshot with approved Yaak Web copy 2026-09-14 21:16:10 -07:00
Gregory Schier 8eb2368151 Refresh PR #657 screenshot with clearer secrets wording 2026-09-14 18:45:10 -07:00
Gregory Schier 34822442b2 Update PR #657 screenshot with storage details and tooltip 2026-09-14 18:41:04 -07:00
Gregory Schier 5149427c1b Update PR #657 screenshot with shorter notice 2026-09-14 18:37:35 -07:00
Gregory Schier bcb8b2e729 Add proxy notice screenshot for PR #657 2026-09-14 18:35:36 -07:00
Gregory Schier 7c660f4528 Correct the language table in the HTTP Snippet README (#656) 2026-09-14 13:41:33 -07:00
Gregory Schier 0337ef35c9 Describe the request template functions (#655) 2026-09-14 12:38:21 -07:00
Gregory SchierandClaude Fable 5 602bd1bb45 Move Install under the screenshot and License to the end of the README
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-09-13 11:03:59 -07:00
Gregory SchierandClaude Fable 5 57ab349901 Add install and license sections to the README
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-09-13 11:02:35 -07:00
Gregory SchierandClaude Opus 5 7a6d275fdb feat(auth-oauth2): custom parameters for authorization, token, and refresh requests (#629)
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
2026-09-13 09:15:25 -07:00
Gregory Schier 7d0926c961 fix(input): opt plain inputs out of macOS Writing Tools suggestions (#646) 2026-09-13 09:15:15 -07:00
Gregory SchierandClaude Opus 5 522b898620 feat(auth): add HTTP Digest authentication plugin (#628)
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
2026-09-13 09:15:06 -07:00
Gregory SchierandClaude Fable 5.1 cb7ed9264d feat(client): show an indicator on the Info tab when a request has a description (#645)
Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-13 09:12:42 -07:00
Gregory SchierandClaude Fable 5.1 90cb578e26 Route all database writes through one connection (#642)
Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-13 08:12:41 -07:00
Gregory SchierandClaude Fable 5.1 ff7eebf3cd fix(importer-curl): join line continuations that have trailing whitespace (#644)
Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-12 23:33:54 -07:00
Ngo Quoc Viet eff0c06cb0 fix(importer-postman): stop swapping the API key name and its value (#643) 2026-09-12 12:52:20 -07:00
dependabot[bot] c6e09a5af3 Bump vite-plus from 0.2.9 to 0.3.0 in the npm-development group (#632)
Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2026-09-11 08:17:56 -07:00
dependabot[bot] b05071ef4b Bump cpy-cli from 5.0.0 to 7.0.0 (#633)
Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2026-09-11 08:17:40 -07:00
dependabot[bot] a44f31eeb2 Bump cpx2 from 8.0.2 to 9.0.0 (#634)
Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2026-09-11 08:17:26 -07:00
dependabot[bot] 1103d05915 Bump zip from 4.0.0 to 7.0.0 (#637)
Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2026-09-11 08:17:15 -07:00
Gregory SchierandClaude Fable 5.1 2f5639a26f chore: cargo fmt
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-10 19:35:33 -07:00
142 changed files with 5954 additions and 3340 deletions

No files matched your search

Binary file not shown.

After

Width:  |  Height:  |  Size: 65 KiB

Generated
+51 -29
View File
@@ -960,15 +960,6 @@ dependencies = [
"serde",
]
[[package]]
name = "bzip2"
version = "0.5.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "49ecfb22d906f800d4fe833b6282cf4dc1c298f5057ca0b5445e5c209735ca47"
dependencies = [
"bzip2-sys",
]
[[package]]
name = "bzip2"
version = "0.6.1"
@@ -978,16 +969,6 @@ dependencies = [
"libbz2-rs-sys",
]
[[package]]
name = "bzip2-sys"
version = "0.1.13+1.0.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "225bff33b2141874fe80d71e07d6eec4f85c5c216453dd96388240f96e1acc14"
dependencies = [
"cc",
"pkg-config",
]
[[package]]
name = "cairo-rs"
version = "0.18.5"
@@ -2114,7 +2095,7 @@ version = "2.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c169adf067a787e1f1c58ed62906a557de85388bee4b54fb878b722ff606b113"
dependencies = [
"bzip2 0.6.1",
"bzip2",
"clap",
"fs-err",
"indicatif",
@@ -2914,11 +2895,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "26145e563e54f2cadc477553f1ec5ee650b00862f0a58bcd12cbdc5f0ea2d2f4"
dependencies = [
"cfg-if",
"js-sys",
"libc",
"r-efi",
"wasi 0.14.2+wasi-0.2.4",
"wasm-bindgen",
]
[[package]]
@@ -4194,6 +4173,15 @@ dependencies = [
"value-bag",
]
[[package]]
name = "lzma-rust2"
version = "0.15.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e20f57f9918e5bd7bc58c22cdd70a6afc7375d4dd9683af5f2b34bd3d2bba619"
dependencies = [
"sha2",
]
[[package]]
name = "mac"
version = "0.1.1"
@@ -6174,6 +6162,12 @@ version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391"
[[package]]
name = "ppmd-rust"
version = "1.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9e9219bcb9d7aca6b2f63c83cf100cf78bcd619ac46e6ecbd0dd90869a39345d"
[[package]]
name = "ppv-lite86"
version = "0.2.21"
@@ -8906,7 +8900,7 @@ dependencies = [
"tokio",
"url",
"windows-sys 0.60.2",
"zip",
"zip 4.6.1",
]
[[package]]
@@ -11376,7 +11370,7 @@ dependencies = [
"yaak-models",
"yaak-plugins",
"yaak-templates",
"zip",
"zip 7.0.0",
]
[[package]]
@@ -11595,7 +11589,6 @@ dependencies = [
"serde",
"serde_json",
"sha2",
"tempfile",
"thiserror 2.0.17",
"ts-rs",
"urlencoding",
@@ -12047,13 +12040,13 @@ dependencies = [
[[package]]
name = "zip"
version = "4.0.0"
version = "4.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "153a6fff49d264c4babdcfa6b4d534747f520e56e8f0f384f3b808c4b64cc1fd"
checksum = "caa8cd6af31c3b31c6631b8f483848b91589021b28fffe50adada48d4f4d2ed1"
dependencies = [
"aes",
"arbitrary",
"bzip2 0.5.2",
"bzip2",
"constant_time_eq 0.3.1",
"crc32fast",
"deflate64",
@@ -12064,6 +12057,35 @@ dependencies = [
"liblzma",
"memchr",
"pbkdf2",
"ppmd-rust",
"sha1",
"time",
"zeroize",
"zopfli",
"zstd",
]
[[package]]
name = "zip"
version = "7.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bdd8a47718a4ee5fe78e07667cd36f3de80e7c2bfe727c7074245ffc7303c037"
dependencies = [
"aes",
"arbitrary",
"bzip2",
"constant_time_eq 0.3.1",
"crc32fast",
"deflate64",
"flate2",
"generic-array",
"getrandom 0.3.3",
"hmac",
"indexmap 2.14.0",
"lzma-rust2",
"memchr",
"pbkdf2",
"ppmd-rust",
"sha1",
"time",
"zeroize",
@@ -12079,7 +12101,7 @@ checksum = "aed5f10c571472911e37d8f7601a8dfba52b4f7f73a344015291b82ab292faf6"
dependencies = [
"log 0.4.29",
"thiserror 2.0.17",
"zip",
"zip 4.6.1",
]
[[package]]
+10
View File
@@ -25,6 +25,10 @@
![Yaak API Client](https://yaak.app/static/screenshot.png)
## Install
Download Yaak for macOS, Windows, or Linux from [yaak.app/download](https://yaak.app/download). Package managers and other install methods are in the [installation docs](https://yaak.app/docs/getting-started/installation).
## Features
Yaak is an offline-first API client designed to stay out of your way while giving you everything you need when you need it.
@@ -68,3 +72,9 @@ Built with [Tauri](https://tauri.app), Rust, and React, it’s fast, lightweight
- [Yaak vs Postman](https://yaak.app/alternatives/postman)
- [Yaak vs Bruno](https://yaak.app/alternatives/bruno)
- [Yaak vs Insomnia](https://yaak.app/alternatives/insomnia)
## License
The source is [MIT licensed](LICENSE) and can be built and run for free, for both personal and commercial use.
The prebuilt binaries follow Yaak's [pricing terms](https://yaak.app/pricing): free for personal use, license required for commercial use.
@@ -133,6 +133,8 @@ export function GrpcRequestPane({
onSend({ message: activeRequest.message });
}, [activeRequest, onSend]);
const hasDescription = activeRequest.description.trim().length > 0;
const tabs: TabItem[] = useMemo(
() => [
{ value: TAB_MESSAGE, label: "Message" },
@@ -146,10 +148,10 @@ export function GrpcRequestPane({
{
value: TAB_DESCRIPTION,
label: "Info",
rightSlot: activeRequest.description && <CountBadge count={true} />,
rightSlot: hasDescription && <CountBadge count={true} />,
},
],
[activeRequest.description, authTab, metadataTab, numSettingsOverrides],
[authTab, hasDescription, metadataTab, numSettingsOverrides],
);
const handleMetadataChange = useCallback(
@@ -165,6 +165,8 @@ export function HttpRequestPane({ style, fullHeight, className, activeRequest }:
: 0;
}
const hasDescription = activeRequest.description.trim().length > 0;
const tabs = useMemo<TabItem[]>(
() => [
{
@@ -260,6 +262,7 @@ export function HttpRequestPane({ style, fullHeight, className, activeRequest }:
{
value: TAB_DESCRIPTION,
label: "Info",
rightSlot: hasDescription && <CountBadge count={true} />,
},
],
[
@@ -267,6 +270,7 @@ export function HttpRequestPane({ style, fullHeight, className, activeRequest }:
authTab,
contentType,
handleContentTypeChange,
hasDescription,
headersTab,
numParams,
numSettingsOverrides,
@@ -110,6 +110,8 @@ export function WebsocketRequestPane({ style, fullHeight, className, activeReque
[activeRequest.url, activeRequest.urlParameters, handleRenamePathPlaceholder],
);
const hasDescription = activeRequest.description.trim().length > 0;
const tabs = useMemo<TabItem[]>(() => {
return [
{
@@ -131,9 +133,10 @@ export function WebsocketRequestPane({ style, fullHeight, className, activeReque
{
value: TAB_DESCRIPTION,
label: "Info",
rightSlot: hasDescription && <CountBadge count={true} />,
},
];
}, [authTab, headersTab, numSettingsOverrides, urlParameterPairs.length]);
}, [authTab, hasDescription, headersTab, numSettingsOverrides, urlParameterPairs.length]);
const { activeResponse } = usePinnedHttpResponse(activeRequestId);
const { mutate: cancelResponse } = useCancelHttpResponse(activeResponse?.id ?? null);
@@ -205,6 +205,7 @@ export const PlainInput = forwardRef<{ focus: () => void }, PlainInputProps>(fun
autoComplete="off"
autoCapitalize="off"
autoCorrect="off"
writingsuggestions="false"
inputMode={inputMode}
onChange={(e) => handleChange(e.target.value)}
onPaste={(e) => onPaste?.(e.clipboardData.getData("Text"))}
+1 -1
View File
@@ -99,6 +99,6 @@
"vite-plugin-static-copy": "^3.3.0",
"vite-plugin-svgr": "^4.5.0",
"vite-plugin-wasm": "^3.5.0",
"vite-plus": "^0.2.9"
"vite-plus": "^0.3.0"
}
}
+8
View File
@@ -0,0 +1,8 @@
import "react";
declare module "react" {
interface HTMLAttributes<T> {
// Not in @types/react yet; opts editable elements out of macOS Writing Tools
writingsuggestions?: "true" | "false";
}
}
+1 -1
View File
@@ -32,6 +32,6 @@
"babel-plugin-react-compiler": "^1.0.0",
"typescript": "^5.8.3",
"vite": "npm:@voidzero-dev/vite-plus-core@^0.2.9",
"vite-plus": "^0.2.9"
"vite-plus": "^0.3.0"
}
}
+1 -1
View File
@@ -39,7 +39,7 @@ tokio = { workspace = true, features = [
] }
walkdir = "2"
webbrowser = "1"
zip = "4"
zip = "7"
yaak = { workspace = true }
yaak-api = { workspace = true }
yaak-core = { workspace = true }
@@ -22,7 +22,6 @@ fn list(ctx: &CliContext, workspace_id: Option<&str>) -> CommandResult {
let workspace_id = resolve_workspace_id(ctx, workspace_id, "cookie-jar list")?;
let cookie_jars = ctx
.db()
.map_err(|e| e.to_string())?
.list_cookie_jars(&workspace_id)
.map_err(|e| format!("Failed to list cookie jars: {e}"))?;
+13 -17
View File
@@ -49,9 +49,11 @@ fn schema(pretty: bool) -> CommandResult {
fn list(ctx: &CliContext, workspace_id: Option<&str>) -> CommandResult {
let workspace_id = resolve_workspace_id(ctx, workspace_id, "environment list")?;
let environments = ctx
.db()
.map_err(|e| e.to_string())?
.list_environments_ensure_base(&workspace_id)
.query_manager()
.with_tx(|tx| {
tx.ensure_base_environment(&workspace_id)?;
tx.list_environments(&workspace_id)
})
.map_err(|e| format!("Failed to list environments: {e}"))?;
if environments.is_empty() {
@@ -67,7 +69,6 @@ fn list(ctx: &CliContext, workspace_id: Option<&str>) -> CommandResult {
fn show(ctx: &CliContext, environment_id: &str) -> CommandResult {
let environment = ctx
.db()
.map_err(|e| e.to_string())?
.get_environment(environment_id)
.map_err(|e| format!("Failed to get environment: {e}"))?;
let output = serde_json::to_string_pretty(&environment)
@@ -113,9 +114,8 @@ fn create(
}
let created = ctx
.db()
.map_err(|e| e.to_string())?
.upsert_environment(&environment, &UpdateSource::Sync)
.query_manager()
.with_tx(|tx| tx.upsert_environment(&environment, &UpdateSource::Sync))
.map_err(|e| format!("Failed to create environment: {e}"))?;
println!("Created environment: {}", created.id);
@@ -136,9 +136,8 @@ fn create(
};
let created = ctx
.db()
.map_err(|e| e.to_string())?
.upsert_environment(&environment, &UpdateSource::Sync)
.query_manager()
.with_tx(|tx| tx.upsert_environment(&environment, &UpdateSource::Sync))
.map_err(|e| format!("Failed to create environment: {e}"))?;
println!("Created environment: {}", created.id);
@@ -151,15 +150,13 @@ fn update(ctx: &CliContext, json: Option<String>, json_input: Option<String>) ->
let existing = ctx
.db()
.map_err(|e| e.to_string())?
.get_environment(&id)
.map_err(|e| format!("Failed to get environment for update: {e}"))?;
let updated = apply_merge_patch(&existing, &patch, &id, "environment update")?;
let saved = ctx
.db()
.map_err(|e| e.to_string())?
.upsert_environment(&updated, &UpdateSource::Sync)
.query_manager()
.with_tx(|tx| tx.upsert_environment(&updated, &UpdateSource::Sync))
.map_err(|e| format!("Failed to update environment: {e}"))?;
println!("Updated environment: {}", saved.id);
@@ -173,9 +170,8 @@ fn delete(ctx: &CliContext, environment_id: &str, yes: bool) -> CommandResult {
}
let deleted = ctx
.db()
.map_err(|e| e.to_string())?
.delete_environment_by_id(environment_id, &UpdateSource::Sync)
.query_manager()
.with_tx(|tx| tx.delete_environment_by_id(environment_id, &UpdateSource::Sync))
.map_err(|e| format!("Failed to delete environment: {e}"))?;
println!("Deleted environment: {}", deleted.id);
+14 -27
View File
@@ -48,11 +48,8 @@ fn schema(pretty: bool) -> CommandResult {
fn list(ctx: &CliContext, workspace_id: Option<&str>) -> CommandResult {
let workspace_id = resolve_workspace_id(ctx, workspace_id, "folder list")?;
let folders = ctx
.db()
.map_err(|e| e.to_string())?
.list_folders(&workspace_id)
.map_err(|e| format!("Failed to list folders: {e}"))?;
let folders =
ctx.db().list_folders(&workspace_id).map_err(|e| format!("Failed to list folders: {e}"))?;
if folders.is_empty() {
println!("No folders found in workspace {}", workspace_id);
} else {
@@ -64,11 +61,8 @@ fn list(ctx: &CliContext, workspace_id: Option<&str>) -> CommandResult {
}
fn show(ctx: &CliContext, folder_id: &str) -> CommandResult {
let folder = ctx
.db()
.map_err(|e| e.to_string())?
.get_folder(folder_id)
.map_err(|e| format!("Failed to get folder: {e}"))?;
let folder =
ctx.db().get_folder(folder_id).map_err(|e| format!("Failed to get folder: {e}"))?;
let output = serde_json::to_string_pretty(&folder)
.map_err(|e| format!("Failed to serialize folder: {e}"))?;
println!("{output}");
@@ -108,9 +102,8 @@ fn create(
)?;
let created = ctx
.db()
.map_err(|e| e.to_string())?
.upsert_folder(&folder, &UpdateSource::Sync)
.query_manager()
.with_tx(|tx| tx.upsert_folder(&folder, &UpdateSource::Sync))
.map_err(|e| format!("Failed to create folder: {e}"))?;
println!("Created folder: {}", created.id);
@@ -125,9 +118,8 @@ fn create(
let folder = Folder { workspace_id, name, ..Default::default() };
let created = ctx
.db()
.map_err(|e| e.to_string())?
.upsert_folder(&folder, &UpdateSource::Sync)
.query_manager()
.with_tx(|tx| tx.upsert_folder(&folder, &UpdateSource::Sync))
.map_err(|e| format!("Failed to create folder: {e}"))?;
println!("Created folder: {}", created.id);
@@ -138,17 +130,13 @@ fn update(ctx: &CliContext, json: Option<String>, json_input: Option<String>) ->
let patch = parse_required_json(json, json_input, "folder update")?;
let id = require_id(&patch, "folder update")?;
let existing = ctx
.db()
.map_err(|e| e.to_string())?
.get_folder(&id)
.map_err(|e| format!("Failed to get folder for update: {e}"))?;
let existing =
ctx.db().get_folder(&id).map_err(|e| format!("Failed to get folder for update: {e}"))?;
let updated = apply_merge_patch(&existing, &patch, &id, "folder update")?;
let saved = ctx
.db()
.map_err(|e| e.to_string())?
.upsert_folder(&updated, &UpdateSource::Sync)
.query_manager()
.with_tx(|tx| tx.upsert_folder(&updated, &UpdateSource::Sync))
.map_err(|e| format!("Failed to update folder: {e}"))?;
println!("Updated folder: {}", saved.id);
@@ -162,9 +150,8 @@ fn delete(ctx: &CliContext, folder_id: &str, yes: bool) -> CommandResult {
}
let deleted = ctx
.db()
.map_err(|e| e.to_string())?
.delete_folder_by_id(folder_id, &UpdateSource::Sync)
.query_manager()
.with_tx(|tx| tx.delete_folder_by_id(folder_id, &UpdateSource::Sync))
.map_err(|e| format!("Failed to delete folder: {e}"))?;
println!("Deleted folder: {}", deleted.id);
@@ -47,7 +47,6 @@ async fn import(
) -> CommandResult<(BatchUpsertResult, Vec<ImportPlanItem>)> {
if let Some(workspace_id) = args.workspace_id.as_deref() {
ctx.db()
.map_err(|e| e.to_string())?
.get_workspace(workspace_id)
.map_err(|e| format!("Failed to get workspace '{workspace_id}': {e}"))?;
}
@@ -69,7 +68,9 @@ async fn import(
);
}
let destination = match workspace_id {
Some(workspace_id) => ImportDestination::ExistingWorkspace { workspace_id, folder_id: None },
Some(workspace_id) => {
ImportDestination::ExistingWorkspace { workspace_id, folder_id: None }
}
None => ImportDestination::NewWorkspace,
};
let plan = import::plan_import_resources(
@@ -147,11 +148,8 @@ fn resolve_export_workspace_ids(
all: bool,
) -> CommandResult<Vec<String>> {
if all {
let workspaces = ctx
.db()
.map_err(|e| e.to_string())?
.list_workspaces()
.map_err(|e| format!("Failed to list workspaces: {e}"))?;
let workspaces =
ctx.db().list_workspaces().map_err(|e| format!("Failed to list workspaces: {e}"))?;
if workspaces.is_empty() {
return Err("No workspaces found to export".to_string());
}
@@ -164,7 +162,6 @@ fn resolve_export_workspace_ids(
for workspace_id in &workspace_ids {
ctx.db()
.map_err(|e| e.to_string())?
.get_workspace(workspace_id)
.map_err(|e| format!("Failed to get workspace '{workspace_id}': {e}"))?;
}
+15 -17
View File
@@ -348,18 +348,19 @@ async fn install_from_directory(context: &CliContext, source: &str) -> CommandRe
ui::info(&format!("Installing plugin from directory {}", plugin_dir.display()));
let plugin = context
.db()
.map_err(|e| e.to_string())?
.upsert_plugin(
&Plugin {
directory: plugin_dir_str,
url: None,
enabled: true,
source: PluginSource::Filesystem,
..Default::default()
},
&UpdateSource::Background,
)
.query_manager()
.with_tx(|tx| {
tx.upsert_plugin(
&Plugin {
directory: plugin_dir_str,
url: None,
enabled: true,
source: PluginSource::Filesystem,
..Default::default()
},
&UpdateSource::Background,
)
})
.map_err(|err| format!("Failed to save plugin in database: {err}"))?;
let plugin_context = PluginContext::new(Some("cli".to_string()), None);
@@ -884,11 +885,8 @@ mod tests {
fs::create_dir_all(root.join("build")).expect("create build");
fs::create_dir_all(root.join("vendor")).expect("create vendor");
fs::write(root.join("vendor/core_bg.wasm"), "asset").expect("write asset");
fs::write(
root.join("package.json"),
r#"{"yaak":{"buildAssets":["vendor/core_bg.wasm"]}}"#,
)
.expect("write package.json");
fs::write(root.join("package.json"), r#"{"yaak":{"buildAssets":["vendor/core_bg.wasm"]}}"#)
.expect("write package.json");
copy_build_assets(root).expect("copy assets");
+12 -25
View File
@@ -71,7 +71,6 @@ fn list(ctx: &CliContext, workspace_id: Option<&str>) -> CommandResult {
let workspace_id = resolve_workspace_id(ctx, workspace_id, "request list")?;
let requests = ctx
.db()
.map_err(|e| e.to_string())?
.list_http_requests(&workspace_id)
.map_err(|e| format!("Failed to list requests: {e}"))?;
if requests.is_empty() {
@@ -425,9 +424,8 @@ fn create(
)?;
let created = ctx
.db()
.map_err(|e| e.to_string())?
.upsert_http_request(&request, &UpdateSource::Sync)
.query_manager()
.with_tx(|tx| tx.upsert_http_request(&request, &UpdateSource::Sync))
.map_err(|e| format!("Failed to create request: {e}"))?;
println!("Created request: {}", created.id);
@@ -445,9 +443,8 @@ fn create(
}
let created = ctx
.db()
.map_err(|e| e.to_string())?
.upsert_http_request(&request, &UpdateSource::Sync)
.query_manager()
.with_tx(|tx| tx.upsert_http_request(&request, &UpdateSource::Sync))
.map_err(|e| format!("Failed to create request: {e}"))?;
println!("Created request: {}", created.id);
@@ -461,15 +458,13 @@ fn update(ctx: &CliContext, json: Option<String>, json_input: Option<String>) ->
let existing = ctx
.db()
.map_err(|e| e.to_string())?
.get_http_request(&id)
.map_err(|e| format!("Failed to get request for update: {e}"))?;
let updated = apply_merge_patch(&existing, &patch, &id, "request update")?;
let saved = ctx
.db()
.map_err(|e| e.to_string())?
.upsert_http_request(&updated, &UpdateSource::Sync)
.query_manager()
.with_tx(|tx| tx.upsert_http_request(&updated, &UpdateSource::Sync))
.map_err(|e| format!("Failed to update request: {e}"))?;
println!("Updated request: {}", saved.id);
@@ -477,11 +472,8 @@ fn update(ctx: &CliContext, json: Option<String>, json_input: Option<String>) ->
}
fn show(ctx: &CliContext, request_id: &str) -> CommandResult {
let request = ctx
.db()
.map_err(|e| e.to_string())?
.get_http_request(request_id)
.map_err(|e| format!("Failed to get request: {e}"))?;
let request =
ctx.db().get_http_request(request_id).map_err(|e| format!("Failed to get request: {e}"))?;
let output = serde_json::to_string_pretty(&request)
.map_err(|e| format!("Failed to serialize request: {e}"))?;
println!("{output}");
@@ -495,9 +487,8 @@ fn delete(ctx: &CliContext, request_id: &str, yes: bool) -> CommandResult {
}
let deleted = ctx
.db()
.map_err(|e| e.to_string())?
.delete_http_request_by_id(request_id, &UpdateSource::Sync)
.query_manager()
.with_tx(|tx| tx.delete_http_request_by_id(request_id, &UpdateSource::Sync))
.map_err(|e| format!("Failed to delete request: {e}"))?;
println!("Deleted request: {}", deleted.id);
Ok(())
@@ -511,11 +502,8 @@ pub async fn send_request_by_id(
cookie_jar_id: Option<&str>,
verbose: bool,
) -> Result<(), String> {
let request = ctx
.db()
.map_err(|e| e.to_string())?
.get_any_request(request_id)
.map_err(|e| format!("Failed to get request: {e}"))?;
let request =
ctx.db().get_any_request(request_id).map_err(|e| format!("Failed to get request: {e}"))?;
match request {
AnyRequest::HttpRequest(http_request) => {
send_http_request_by_id(
@@ -623,7 +611,6 @@ pub(crate) fn resolve_cookie_jar_id(
let default_cookie_jar = ctx
.db()
.map_err(|e| e.to_string())?
.list_cookie_jars(workspace_id)
.map_err(|e| format!("Failed to list cookie jars: {e}"))?
.into_iter()
+10 -13
View File
@@ -27,14 +27,13 @@ pub fn run(ctx: &CliContext, args: ResponseArgs) -> i32 {
/// Accepts a response ID, or a request ID meaning "the latest response for that request".
fn resolve_response(ctx: &CliContext, id: &str) -> CommandResult<HttpResponse> {
if let Ok(response) = ctx.db().map_err(|e| e.to_string())?.get_http_response(id) {
if let Ok(response) = ctx.db().get_http_response(id) {
return Ok(response);
}
if ctx.db().map_err(|e| e.to_string())?.get_http_request(id).is_ok() {
if ctx.db().get_http_request(id).is_ok() {
return ctx
.db()
.map_err(|e| e.to_string())?
.list_http_responses_for_request(id, Some(1))
.map_err(|e| format!("Failed to list responses: {e}"))?
.into_iter()
@@ -48,15 +47,13 @@ fn resolve_response(ctx: &CliContext, id: &str) -> CommandResult<HttpResponse> {
fn list(ctx: &CliContext, id: Option<&str>, limit: Option<u64>) -> CommandResult {
// An explicit request ID scopes to that request; anything else lists the workspace.
let responses = match id {
Some(id) if ctx.db().map_err(|e| e.to_string())?.get_http_request(id).is_ok() => ctx
Some(id) if ctx.db().get_http_request(id).is_ok() => ctx
.db()
.map_err(|e| e.to_string())?
.list_http_responses_for_request(id, limit)
.map_err(|e| format!("Failed to list responses: {e}"))?,
other => {
let workspace_id = resolve_workspace_id(ctx, other, "response list")?;
ctx.db()
.map_err(|e| e.to_string())?
.list_http_responses(&workspace_id, limit)
.map_err(|e| format!("Failed to list responses: {e}"))?
}
@@ -115,15 +112,14 @@ fn body(ctx: &CliContext, id: &str) -> CommandResult {
fn delete(ctx: &CliContext, id: &str, yes: bool) -> CommandResult {
// Deleting is scoped to a whole request or workspace, since a single stored
// response is rarely the thing someone wants to remove.
if ctx.db().map_err(|e| e.to_string())?.get_http_request(id).is_ok() {
if ctx.db().get_http_request(id).is_ok() {
if !yes && !confirm_delete("responses for request", id) {
println!("Aborted");
return Ok(());
}
let count = ctx
.db()
.map_err(|e| e.to_string())?
.delete_all_http_responses_for_request(id, &UpdateSource::Sync)
.query_manager()
.with_tx(|tx| tx.delete_all_http_responses_for_request(id, &UpdateSource::Sync))
.map_err(|e| format!("Failed to delete responses: {e}"))?;
println!("Deleted {count} responses for request {id}");
return Ok(());
@@ -135,9 +131,10 @@ fn delete(ctx: &CliContext, id: &str, yes: bool) -> CommandResult {
return Ok(());
}
let count = ctx
.db()
.map_err(|e| e.to_string())?
.delete_all_http_responses_for_workspace(&workspace_id, &UpdateSource::Sync)
.query_manager()
.with_tx(|tx| {
tx.delete_all_http_responses_for_workspace(&workspace_id, &UpdateSource::Sync)
})
.map_err(|e| format!("Failed to delete responses: {e}"))?;
println!("Deleted {count} responses for workspace {workspace_id}");
Ok(())
+3 -9
View File
@@ -34,7 +34,7 @@ async fn send_target(
) -> Result<(), String> {
let mode = if args.parallel { ExecutionMode::Parallel } else { ExecutionMode::Sequential };
if let Ok(request) = ctx.db().map_err(|e| e.to_string())?.get_any_request(&args.id) {
if let Ok(request) = ctx.db().get_any_request(&args.id) {
let workspace_id = match &request {
AnyRequest::HttpRequest(r) => r.workspace_id.clone(),
AnyRequest::GrpcRequest(r) => r.workspace_id.clone(),
@@ -53,7 +53,7 @@ async fn send_target(
.await;
}
if let Ok(folder) = ctx.db().map_err(|e| e.to_string())?.get_folder(&args.id) {
if let Ok(folder) = ctx.db().get_folder(&args.id) {
let resolved_cookie_jar_id =
request::resolve_cookie_jar_id(ctx, &folder.workspace_id, cookie_jar_id)?;
@@ -74,7 +74,7 @@ async fn send_target(
.await;
}
if let Ok(workspace) = ctx.db().map_err(|e| e.to_string())?.get_workspace(&args.id) {
if let Ok(workspace) = ctx.db().get_workspace(&args.id) {
let resolved_cookie_jar_id =
request::resolve_cookie_jar_id(ctx, &workspace.id, cookie_jar_id)?;
@@ -103,7 +103,6 @@ fn collect_folder_request_ids(ctx: &CliContext, folder_id: &str) -> Result<Vec<S
let mut http_ids = ctx
.db()
.map_err(|e| e.to_string())?
.list_http_requests_for_folder_recursive(folder_id)
.map_err(|e| format!("Failed to list HTTP requests in folder: {e}"))?
.into_iter()
@@ -113,7 +112,6 @@ fn collect_folder_request_ids(ctx: &CliContext, folder_id: &str) -> Result<Vec<S
let mut grpc_ids = ctx
.db()
.map_err(|e| e.to_string())?
.list_grpc_requests_for_folder_recursive(folder_id)
.map_err(|e| format!("Failed to list gRPC requests in folder: {e}"))?
.into_iter()
@@ -123,7 +121,6 @@ fn collect_folder_request_ids(ctx: &CliContext, folder_id: &str) -> Result<Vec<S
let mut websocket_ids = ctx
.db()
.map_err(|e| e.to_string())?
.list_websocket_requests_for_folder_recursive(folder_id)
.map_err(|e| format!("Failed to list WebSocket requests in folder: {e}"))?
.into_iter()
@@ -142,7 +139,6 @@ fn collect_workspace_request_ids(
let mut http_ids = ctx
.db()
.map_err(|e| e.to_string())?
.list_http_requests(workspace_id)
.map_err(|e| format!("Failed to list HTTP requests in workspace: {e}"))?
.into_iter()
@@ -152,7 +148,6 @@ fn collect_workspace_request_ids(
let mut grpc_ids = ctx
.db()
.map_err(|e| e.to_string())?
.list_grpc_requests(workspace_id)
.map_err(|e| format!("Failed to list gRPC requests in workspace: {e}"))?
.into_iter()
@@ -162,7 +157,6 @@ fn collect_workspace_request_ids(
let mut websocket_ids = ctx
.db()
.map_err(|e| e.to_string())?
.list_websocket_requests(workspace_id)
.map_err(|e| format!("Failed to list WebSocket requests in workspace: {e}"))?
.into_iter()
+12 -19
View File
@@ -43,11 +43,8 @@ fn schema(pretty: bool) -> CommandResult {
}
fn list(ctx: &CliContext) -> CommandResult {
let workspaces = ctx
.db()
.map_err(|e| e.to_string())?
.list_workspaces()
.map_err(|e| format!("Failed to list workspaces: {e}"))?;
let workspaces =
ctx.db().list_workspaces().map_err(|e| format!("Failed to list workspaces: {e}"))?;
if workspaces.is_empty() {
println!("No workspaces found");
} else {
@@ -61,7 +58,6 @@ fn list(ctx: &CliContext) -> CommandResult {
fn show(ctx: &CliContext, workspace_id: &str) -> CommandResult {
let workspace = ctx
.db()
.map_err(|e| e.to_string())?
.get_workspace(workspace_id)
.map_err(|e| format!("Failed to get workspace: {e}"))?;
let output = serde_json::to_string_pretty(&workspace)
@@ -88,9 +84,8 @@ fn create(
.map_err(|e| format!("Failed to parse workspace create JSON: {e}"))?;
let created = ctx
.db()
.map_err(|e| e.to_string())?
.upsert_workspace(&workspace, &UpdateSource::Sync)
.query_manager()
.with_tx(|tx| tx.upsert_workspace(&workspace, &UpdateSource::Sync))
.map_err(|e| format!("Failed to create workspace: {e}"))?;
println!("Created workspace: {}", created.id);
return Ok(());
@@ -102,9 +97,8 @@ fn create(
let workspace = Workspace { name, ..Default::default() };
let created = ctx
.db()
.map_err(|e| e.to_string())?
.upsert_workspace(&workspace, &UpdateSource::Sync)
.query_manager()
.with_tx(|tx| tx.upsert_workspace(&workspace, &UpdateSource::Sync))
.map_err(|e| format!("Failed to create workspace: {e}"))?;
println!("Created workspace: {}", created.id);
Ok(())
@@ -116,15 +110,13 @@ fn update(ctx: &CliContext, json: Option<String>, json_input: Option<String>) ->
let existing = ctx
.db()
.map_err(|e| e.to_string())?
.get_workspace(&id)
.map_err(|e| format!("Failed to get workspace for update: {e}"))?;
let updated = apply_merge_patch(&existing, &patch, &id, "workspace update")?;
let saved = ctx
.db()
.map_err(|e| e.to_string())?
.upsert_workspace(&updated, &UpdateSource::Sync)
.query_manager()
.with_tx(|tx| tx.upsert_workspace(&updated, &UpdateSource::Sync))
.map_err(|e| format!("Failed to update workspace: {e}"))?;
println!("Updated workspace: {}", saved.id);
@@ -138,9 +130,10 @@ fn delete(ctx: &CliContext, workspace_id: &str, yes: bool) -> CommandResult {
}
let deleted = ctx
.db()
.map_err(|e| e.to_string())?
.delete_workspace_by_id(workspace_id, &UpdateSource::Sync, ctx.blob_manager())
.query_manager()
.with_tx(|tx| {
tx.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(())
+4 -4
View File
@@ -51,9 +51,9 @@ impl CliContext {
};
// Guest: the desktop may have this DB open, so only what's safe beside a live session
if let Ok(db) = query_manager.connect() {
let _ = yaak_lifecycle::on_launch(&yaak_lifecycle::Host::guest(), &db, &blob_manager);
}
let _ = query_manager.with_tx(|tx| {
yaak_lifecycle::on_launch(&yaak_lifecycle::Host::guest(), tx, &blob_manager)
});
let encryption_manager = Arc::new(EncryptionManager::new(query_manager.clone(), app_id));
@@ -118,7 +118,7 @@ impl CliContext {
&self.data_dir
}
pub fn db(&self) -> yaak_models::error::Result<ClientDb<'_>> {
pub fn db(&self) -> ClientDb<'_> {
self.query_manager.connect()
}
+3 -5
View File
@@ -203,7 +203,7 @@ fn resolve_send_execution_context(
environment: Option<&str>,
explicit_cookie_jar_id: Option<&str>,
) -> Result<CliExecutionContext, String> {
if let Ok(request) = context.db().map_err(|e| e.to_string())?.get_any_request(id) {
if let Ok(request) = context.db().get_any_request(id) {
let (request_id, workspace_id) = match request {
AnyRequest::HttpRequest(r) => (Some(r.id), r.workspace_id),
AnyRequest::GrpcRequest(r) => (Some(r.id), r.workspace_id),
@@ -218,7 +218,7 @@ fn resolve_send_execution_context(
});
}
if let Ok(folder) = context.db().map_err(|e| e.to_string())?.get_folder(id) {
if let Ok(folder) = context.db().get_folder(id) {
let cookie_jar_id =
resolve_cookie_jar_id(context, &folder.workspace_id, explicit_cookie_jar_id)?;
return Ok(CliExecutionContext {
@@ -229,7 +229,7 @@ fn resolve_send_execution_context(
});
}
if let Ok(workspace) = context.db().map_err(|e| e.to_string())?.get_workspace(id) {
if let Ok(workspace) = context.db().get_workspace(id) {
let cookie_jar_id = resolve_cookie_jar_id(context, &workspace.id, explicit_cookie_jar_id)?;
return Ok(CliExecutionContext {
request_id: None,
@@ -250,7 +250,6 @@ fn resolve_request_execution_context(
) -> Result<CliExecutionContext, String> {
let request = context
.db()
.map_err(|e| e.to_string())?
.get_any_request(request_id)
.map_err(|e| format!("Failed to get request: {e}"))?;
@@ -280,7 +279,6 @@ fn resolve_cookie_jar_id(
let default_cookie_jar = context
.db()
.map_err(|e| e.to_string())?
.list_cookie_jars(workspace_id)
.map_err(|e| format!("Failed to list cookie jars: {e}"))?
.into_iter()
+51 -68
View File
@@ -150,11 +150,7 @@ async fn build_plugin_reply(
Some(InternalEventPayload::ShowToastResponse(EmptyPayload {}))
}
HostRequest::ListOpenWorkspaces(_) => {
let workspaces = match host_context
.query_manager
.connect()
.and_then(|db| db.list_workspaces())
{
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: w.id })
@@ -193,7 +189,7 @@ async fn build_plugin_reply(
match host_context
.query_manager
.connect()
.and_then(|db| db.list_cookie_jars(http_request.workspace_id.as_str()))
.list_cookie_jars(http_request.workspace_id.as_str())
{
Ok(cookie_jars) => cookie_jars
.into_iter()
@@ -268,20 +264,19 @@ async fn build_plugin_reply(
..event.context.clone()
};
let environment_chain = match host_context.query_manager.connect().and_then(|db| {
db.resolve_environments(
let environment_chain =
match host_context.query_manager.connect().resolve_environments(
&grpc_request.workspace_id,
grpc_request.folder_id.as_deref(),
execution_context.environment_id.as_deref(),
)
}) {
Ok(chain) => chain,
Err(err) => {
return Some(InternalEventPayload::ErrorResponse(ErrorResponse {
error: format!("Failed to resolve environments in CLI: {err}"),
}));
}
};
) {
Ok(chain) => chain,
Err(err) => {
return Some(InternalEventPayload::ErrorResponse(ErrorResponse {
error: format!("Failed to resolve environments in CLI: {err}"),
}));
}
};
let template_callback = PluginTemplateCallback::new(
host_context.plugin_manager.clone(),
@@ -329,20 +324,19 @@ async fn build_plugin_reply(
..event.context.clone()
};
let environment_chain = match host_context.query_manager.connect().and_then(|db| {
db.resolve_environments(
let environment_chain =
match host_context.query_manager.connect().resolve_environments(
&http_request.workspace_id,
http_request.folder_id.as_deref(),
execution_context.environment_id.as_deref(),
)
}) {
Ok(chain) => chain,
Err(err) => {
return Some(InternalEventPayload::ErrorResponse(ErrorResponse {
error: format!("Failed to resolve environments in CLI: {err}"),
}));
}
};
) {
Ok(chain) => chain,
Err(err) => {
return Some(InternalEventPayload::ErrorResponse(ErrorResponse {
error: format!("Failed to resolve environments in CLI: {err}"),
}));
}
};
let template_callback = PluginTemplateCallback::new(
host_context.plugin_manager.clone(),
@@ -386,11 +380,7 @@ async fn build_plugin_reply(
};
let folder_id = execution_context.request_id.as_ref().and_then(|rid| {
match host_context
.query_manager
.connect()
.and_then(|db| db.get_any_request(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,
@@ -398,20 +388,19 @@ async fn build_plugin_reply(
}
});
let environment_chain = match host_context.query_manager.connect().and_then(|db| {
db.resolve_environments(
let environment_chain =
match host_context.query_manager.connect().resolve_environments(
&workspace_id,
folder_id.as_deref(),
execution_context.environment_id.as_deref(),
)
}) {
Ok(chain) => chain,
Err(err) => {
return Some(InternalEventPayload::ErrorResponse(ErrorResponse {
error: format!("Failed to resolve environments in CLI: {err}"),
}));
}
};
) {
Ok(chain) => chain,
Err(err) => {
return Some(InternalEventPayload::ErrorResponse(ErrorResponse {
error: format!("Failed to resolve environments in CLI: {err}"),
}));
}
};
let template_callback = PluginTemplateCallback::new(
host_context.plugin_manager.clone(),
@@ -487,18 +476,15 @@ async fn build_plugin_reply(
));
};
let cookie_jar = match host_context
.query_manager
.connect()
.and_then(|db| db.get_cookie_jar(cookie_jar_id))
{
Ok(cookie_jar) => cookie_jar,
Err(err) => {
return Some(InternalEventPayload::ErrorResponse(ErrorResponse {
error: format!("Failed to load cookie jar in CLI: {err}"),
}));
}
};
let cookie_jar =
match host_context.query_manager.connect().get_cookie_jar(cookie_jar_id) {
Ok(cookie_jar) => cookie_jar,
Err(err) => {
return Some(InternalEventPayload::ErrorResponse(ErrorResponse {
error: format!("Failed to load cookie jar in CLI: {err}"),
}));
}
};
let names = cookie_jar.cookies.into_iter().map(|c| c.name).collect();
@@ -513,18 +499,15 @@ async fn build_plugin_reply(
));
};
let cookie_jar = match host_context
.query_manager
.connect()
.and_then(|db| db.get_cookie_jar(cookie_jar_id))
{
Ok(cookie_jar) => cookie_jar,
Err(err) => {
return Some(InternalEventPayload::ErrorResponse(ErrorResponse {
error: format!("Failed to load cookie jar in CLI: {err}"),
}));
}
};
let cookie_jar =
match host_context.query_manager.connect().get_cookie_jar(cookie_jar_id) {
Ok(cookie_jar) => cookie_jar,
Err(err) => {
return Some(InternalEventPayload::ErrorResponse(ErrorResponse {
error: format!("Failed to load cookie jar in CLI: {err}"),
}));
}
};
let value =
get_cookie_value_from_jar(cookie_jar.cookies, &req.name, req.domain.as_deref());
+2 -5
View File
@@ -9,11 +9,8 @@ pub fn resolve_workspace_id(
return Ok(workspace_id.to_string());
}
let workspaces = ctx
.db()
.map_err(|e| e.to_string())?
.list_workspaces()
.map_err(|e| format!("Failed to list workspaces: {e}"))?;
let workspaces =
ctx.db().list_workspaces().map_err(|e| format!("Failed to list workspaces: {e}"))?;
match workspaces.as_slice() {
[] => Err(format!("No workspaces found. {command_name} requires a workspace ID.")),
[workspace] => Ok(workspace.id.clone()),
+5 -15
View File
@@ -40,9 +40,7 @@ pub fn seed_workspace(data_dir: &Path, workspace_id: &str) {
};
query_manager(data_dir)
.connect()
.unwrap()
.upsert_workspace(&workspace, &UpdateSource::Sync)
.with_tx(|tx| tx.upsert_workspace(&workspace, &UpdateSource::Sync))
.expect("Failed to seed workspace");
}
@@ -57,9 +55,7 @@ pub fn seed_request(data_dir: &Path, workspace_id: &str, request_id: &str) {
};
query_manager(data_dir)
.connect()
.unwrap()
.upsert_http_request(&request, &UpdateSource::Sync)
.with_tx(|tx| tx.upsert_http_request(&request, &UpdateSource::Sync))
.expect("Failed to seed request");
}
@@ -72,9 +68,7 @@ pub fn seed_folder(data_dir: &Path, workspace_id: &str, folder_id: &str) {
};
query_manager(data_dir)
.connect()
.unwrap()
.upsert_folder(&folder, &UpdateSource::Sync)
.with_tx(|tx| tx.upsert_folder(&folder, &UpdateSource::Sync))
.expect("Failed to seed folder");
}
@@ -88,9 +82,7 @@ pub fn seed_grpc_request(data_dir: &Path, workspace_id: &str, request_id: &str)
};
query_manager(data_dir)
.connect()
.unwrap()
.upsert_grpc_request(&request, &UpdateSource::Sync)
.with_tx(|tx| tx.upsert_grpc_request(&request, &UpdateSource::Sync))
.expect("Failed to seed gRPC request");
}
@@ -104,8 +96,6 @@ pub fn seed_websocket_request(data_dir: &Path, workspace_id: &str, request_id: &
};
query_manager(data_dir)
.connect()
.unwrap()
.upsert_websocket_request(&request, &UpdateSource::Sync)
.with_tx(|tx| tx.upsert_websocket_request(&request, &UpdateSource::Sync))
.expect("Failed to seed WebSocket request");
}
@@ -42,7 +42,7 @@ fn create_list_show_delete_round_trip() {
.success()
.stdout(contains(format!("Deleted environment: {environment_id}")));
assert!(query_manager(data_dir).connect().unwrap().get_environment(&environment_id).is_err());
assert!(query_manager(data_dir).connect().get_environment(&environment_id).is_err());
}
#[test]
+1 -1
View File
@@ -36,7 +36,7 @@ fn create_list_show_delete_round_trip() {
.success()
.stdout(contains(format!("Deleted folder: {folder_id}")));
assert!(query_manager(data_dir).connect().unwrap().get_folder(&folder_id).is_err());
assert!(query_manager(data_dir).connect().get_folder(&folder_id).is_err());
}
#[test]
@@ -81,7 +81,7 @@ fn import_reads_yaak_workspace_file() {
.stdout(contains("Imported 1 workspace, 1 HTTP request"));
let query_manager = query_manager(data_dir);
let db = query_manager.connect().unwrap();
let db = query_manager.connect();
let workspaces = db.list_workspaces().expect("list imported workspaces");
let workspace = workspaces
.iter()
@@ -160,9 +160,8 @@ fn import_postman_environment_uses_workspace_id() {
.stdout(contains("Imported 1 environment"));
let query_manager = query_manager(data_dir);
let db = query_manager.connect().unwrap();
let environments =
db.list_environments_ensure_base(&workspace_id).expect("list imported environments");
let db = query_manager.connect();
let environments = db.list_environments(&workspace_id).expect("list imported environments");
let imported_environment =
environments.iter().find(|e| e.name == "Local").expect("postman environment imported");
@@ -210,14 +209,17 @@ fn re_import_merges_into_linked_workspace() {
],
);
cli_cmd(data_dir)
.args(["import", import_path.to_str().expect("import path is utf-8")])
.args([
"import",
import_path.to_str().expect("import path is utf-8"),
])
.assert()
.success()
.stdout(contains("Imported 1 workspace, 2 HTTP requests"));
let workspace_id = {
let query_manager = query_manager(data_dir);
let db = query_manager.connect().unwrap();
let db = query_manager.connect();
db.list_workspaces()
.expect("list workspaces")
.into_iter()
@@ -248,7 +250,7 @@ fn re_import_merges_into_linked_workspace() {
.stdout(contains("Skipped 1 removed from source"));
let query_manager = query_manager(data_dir);
let db = query_manager.connect().unwrap();
let db = query_manager.connect();
let requests = db.list_http_requests(&workspace_id).expect("list requests");
assert_eq!(requests.len(), 3, "merge must not duplicate: {requests:?}");
assert_eq!(
@@ -273,13 +275,16 @@ fn re_import_leaves_deleted_resources_alone() {
],
);
cli_cmd(data_dir)
.args(["import", import_path.to_str().expect("import path is utf-8")])
.args([
"import",
import_path.to_str().expect("import path is utf-8"),
])
.assert()
.success();
let workspace_id = {
let query_manager = query_manager(data_dir);
let db = query_manager.connect().unwrap();
let db = query_manager.connect();
let workspace_id = db
.list_workspaces()
.expect("list workspaces")
@@ -293,7 +298,9 @@ fn re_import_leaves_deleted_resources_alone() {
.into_iter()
.find(|r| r.name == "Request B")
.expect("request B imported");
db.delete_http_request_by_id(&request_b.id, &UpdateSource::Sync)
drop(db);
query_manager
.with_tx(|tx| tx.delete_http_request_by_id(&request_b.id, &UpdateSource::Sync))
.expect("delete request B");
workspace_id
};
@@ -311,7 +318,7 @@ fn re_import_leaves_deleted_resources_alone() {
let query_manager = query_manager(data_dir);
let requests =
query_manager.connect().unwrap().list_http_requests(&workspace_id).expect("list requests");
query_manager.connect().list_http_requests(&workspace_id).expect("list requests");
assert_eq!(requests.len(), 1, "a deleted request must not come back: {requests:?}");
assert_eq!(requests[0].name, "Request A");
}
@@ -43,7 +43,7 @@ fn show_and_delete_yes_round_trip() {
.success()
.stdout(contains(format!("Deleted request: {request_id}")));
assert!(query_manager(data_dir).connect().unwrap().get_http_request(&request_id).is_err());
assert!(query_manager(data_dir).connect().get_http_request(&request_id).is_err());
}
#[test]
@@ -61,11 +61,7 @@ fn delete_without_yes_fails_in_non_interactive_mode() {
.stderr(contains("Refusing to delete in non-interactive mode without --yes"));
assert!(
query_manager(data_dir)
.connect()
.unwrap()
.get_http_request("rq_seed_delete_noninteractive")
.is_ok()
query_manager(data_dir).connect().get_http_request("rq_seed_delete_noninteractive").is_ok()
);
}
@@ -126,7 +122,6 @@ fn create_allows_workspace_only_with_empty_defaults() {
let request = query_manager(data_dir)
.connect()
.unwrap()
.get_http_request(&request_id)
.expect("Failed to load created request");
assert_eq!(request.workspace_id, "wk_test");
@@ -212,7 +207,7 @@ fn request_send_persists_response_body_and_events() {
.stdout(contains("hello from integration test"));
let qm = query_manager(data_dir);
let db = qm.connect().unwrap();
let db = qm.connect();
let responses =
db.list_http_responses_for_request(&request_id, None).expect("Failed to load responses");
assert_eq!(responses.len(), 1, "expected exactly one persisted response");
+1 -3
View File
@@ -25,9 +25,7 @@ fn top_level_send_folder_sends_http_requests_and_prints_summary() {
..Default::default()
};
query_manager(data_dir)
.connect()
.unwrap()
.upsert_http_request(&request, &UpdateSource::Sync)
.with_tx(|tx| tx.upsert_http_request(&request, &UpdateSource::Sync))
.expect("Failed to seed folder request");
cli_cmd(data_dir)
@@ -26,7 +26,7 @@ fn create_show_delete_round_trip() {
.success()
.stdout(contains(format!("Deleted workspace: {workspace_id}")));
assert!(query_manager(data_dir).connect().unwrap().get_workspace(&workspace_id).is_err());
assert!(query_manager(data_dir).connect().get_workspace(&workspace_id).is_err());
}
#[test]
+2 -5
View File
@@ -145,11 +145,8 @@ async fn cache_control(req: Request, next: Next) -> Response {
.get(header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
.is_some_and(|v| v.starts_with("text/html"));
let value = if hashed_name && !is_html {
"public, max-age=31536000, immutable"
} else {
"no-cache"
};
let value =
if hashed_name && !is_html { "public, max-age=31536000, immutable" } else { "no-cache" };
res.headers_mut().insert(header::CACHE_CONTROL, HeaderValue::from_static(value));
res
}
-1
View File
@@ -19,7 +19,6 @@ pub(crate) fn metadata_to_map(metadata: MetadataMap) -> BTreeMap<String, String>
entries
}
pub(crate) async fn build_metadata<R: Runtime>(
window: &WebviewWindow<R>,
request: &GrpcRequest,
+4 -5
View File
@@ -27,9 +27,12 @@ pub fn get_or_upsert_launch_info<R: Runtime>(app_handle: &AppHandle<R>) -> &Laun
LAUNCH_INFO.get_or_init(|| {
let now = Utc::now().naive_utc();
let mut info = LaunchEventInfo {
version_since: app_handle.db().get_key_value_dte(NAMESPACE, VERSION_SINCE_KEY, now),
current_version: app_handle.package_info().version.to_string(),
user_since: app_handle.db().get_settings().created_at,
num_launches: app_handle.db().get_key_value_int(NAMESPACE, NUM_LAUNCHES_KEY, 0) + 1,
// The rest is read and written inside the transaction below
// The rest will be set below
..Default::default()
};
@@ -39,10 +42,6 @@ pub fn get_or_upsert_launch_info<R: Runtime>(app_handle: &AppHandle<R>) -> &Laun
app_handle
.with_tx(|tx| {
info.version_since = tx.get_key_value_dte(NAMESPACE, VERSION_SINCE_KEY, now);
info.user_since = tx.get_settings().created_at;
info.num_launches = tx.get_key_value_int(NAMESPACE, NUM_LAUNCHES_KEY, 0) + 1;
// Load the previously tracked version
let curr_db = tx.get_key_value_str(NAMESPACE, LAST_VERSION_KEY, "");
let prev_db = tx.get_key_value_str(NAMESPACE, PREV_VERSION_KEY, "");
@@ -177,4 +177,3 @@ async fn send_http_request_inner<R: Runtime>(
Ok(SentHttpRequest { response: result.response, body: result.response_body })
}
+139 -159
View File
@@ -29,16 +29,16 @@ use tokio::sync::Mutex;
use tokio::task::block_in_place;
use tokio::time;
use yaak::send::ResponseBody;
use yaak_commands::responses::locate_response_body;
use yaak_commands::resolve::resolve_grpc_request;
use yaak_commands::responses::locate_response_body;
use yaak_common::command::new_checked_command;
use yaak_crypto::manager::EncryptionManager;
use yaak_grpc::manager::{GrpcConfig, GrpcHandle};
use yaak_grpc::{Code, ServiceDefinition};
use yaak_mac_window::AppHandleMacWindowExt;
use yaak_models::models::{
CookieJar, Environment, GrpcConnection, GrpcConnectionState, GrpcEvent,
GrpcEventType, HttpRequest, HttpResponse, HttpResponseState, Workspace,
CookieJar, Environment, GrpcConnection, GrpcConnectionState, GrpcEvent, GrpcEventType,
HttpRequest, HttpResponse, HttpResponseState, Workspace,
};
use yaak_models::util::{BatchUpsertResult, ImportDestination, ImportPlan, UpdateSource};
use yaak_plugins::events::{
@@ -237,17 +237,17 @@ async fn cmd_grpc_reflect<R: Runtime>(
app_handle: AppHandle<R>,
grpc_handle: State<'_, Mutex<GrpcHandle>>,
) -> YaakResult<Vec<ServiceDefinition>> {
let unrendered_request = app_handle.db()?.get_grpc_request(request_id)?;
let unrendered_request = app_handle.db().get_grpc_request(request_id)?;
let (resolved_request, auth_context_id) =
resolve_grpc_request(&window.db()?, &unrendered_request)?;
resolve_grpc_request(&window.db(), &unrendered_request)?;
let environment_chain = app_handle.db()?.resolve_environments(
let environment_chain = app_handle.db().resolve_environments(
&unrendered_request.workspace_id,
unrendered_request.folder_id.as_deref(),
environment_id,
)?;
let resolved_settings =
app_handle.db()?.resolve_settings_for_grpc_request(&unrendered_request)?;
app_handle.db().resolve_settings_for_grpc_request(&unrendered_request)?;
let plugin_manager = Arc::new(crate::plugins_ext::plugin_manager(&app_handle).await?);
let encryption_manager = Arc::new((*app_handle.state::<EncryptionManager>()).clone());
@@ -266,7 +266,7 @@ async fn cmd_grpc_reflect<R: Runtime>(
let uri = safe_uri(&req.url);
let metadata = build_metadata(&window, &req, &auth_context_id).await?;
let settings = window.db()?.get_settings();
let settings = window.db().get_settings();
let client_certificate =
find_client_certificate(req.url.as_str(), &settings.client_certificates);
let proto_files: Vec<PathBuf> =
@@ -298,16 +298,16 @@ async fn cmd_grpc_go<R: Runtime>(
window: WebviewWindow<R>,
grpc_handle: State<'_, Mutex<GrpcHandle>>,
) -> YaakResult<String> {
let unrendered_request = app_handle.db()?.get_grpc_request(request_id)?;
let unrendered_request = app_handle.db().get_grpc_request(request_id)?;
let (resolved_request, auth_context_id) =
resolve_grpc_request(&window.db()?, &unrendered_request)?;
let environment_chain = app_handle.db()?.resolve_environments(
resolve_grpc_request(&window.db(), &unrendered_request)?;
let environment_chain = app_handle.db().resolve_environments(
&unrendered_request.workspace_id,
unrendered_request.folder_id.as_deref(),
environment_id,
)?;
let resolved_settings =
app_handle.db()?.resolve_settings_for_grpc_request(&unrendered_request)?;
app_handle.db().resolve_settings_for_grpc_request(&unrendered_request)?;
let plugin_manager = Arc::new(crate::plugins_ext::plugin_manager(&app_handle).await?);
let encryption_manager = Arc::new((*app_handle.state::<EncryptionManager>()).clone());
@@ -327,21 +327,23 @@ async fn cmd_grpc_go<R: Runtime>(
let metadata = build_metadata(&window, &request, &auth_context_id).await?;
// Find matching client certificate for this URL
let settings = app_handle.db()?.get_settings();
let settings = app_handle.db().get_settings();
let client_cert = find_client_certificate(&request.url, &settings.client_certificates);
let conn = app_handle.db()?.upsert_grpc_connection(
&GrpcConnection {
workspace_id: request.workspace_id.clone(),
request_id: request.id.clone(),
status: -1,
elapsed: 0,
state: GrpcConnectionState::Initialized,
url: request.url.clone(),
..Default::default()
},
&UpdateSource::from_window_label(window.label()),
)?;
let conn = app_handle.with_tx(|tx| {
tx.upsert_grpc_connection(
&GrpcConnection {
workspace_id: request.workspace_id.clone(),
request_id: request.id.clone(),
status: -1,
elapsed: 0,
state: GrpcConnectionState::Initialized,
url: request.url.clone(),
..Default::default()
},
&UpdateSource::from_window_label(window.label()),
)
})?;
let conn_id = conn.id.clone();
@@ -386,15 +388,17 @@ async fn cmd_grpc_go<R: Runtime>(
let connection = match connection {
Ok(c) => c,
Err(err) => {
app_handle.db()?.upsert_grpc_connection(
&GrpcConnection {
elapsed: start.elapsed().as_millis() as i32,
error: Some(err.to_string()),
state: GrpcConnectionState::Closed,
..conn.clone()
},
&UpdateSource::from_window_label(window.label()),
)?;
app_handle.with_tx(|tx| {
tx.upsert_grpc_connection(
&GrpcConnection {
elapsed: start.elapsed().as_millis() as i32,
error: Some(err.to_string()),
state: GrpcConnectionState::Closed,
..conn.clone()
},
&UpdateSource::from_window_label(window.label()),
)
})?;
return Ok(conn_id);
}
};
@@ -495,15 +499,17 @@ async fn cmd_grpc_go<R: Runtime>(
.await?;
let msg = strip_json_comments(&msg);
app_handle.db()?.upsert_grpc_event(
&GrpcEvent {
content: format!("Connecting to {}", req.url),
event_type: GrpcEventType::ConnectionStart,
metadata: metadata.clone(),
..base_event.clone()
},
&UpdateSource::from_window_label(window.label()),
)?;
app_handle.with_tx(|tx| {
tx.upsert_grpc_event(
&GrpcEvent {
content: format!("Connecting to {}", req.url),
event_type: GrpcEventType::ConnectionStart,
metadata: metadata.clone(),
..base_event.clone()
},
&UpdateSource::from_window_label(window.label()),
)
})?;
async move {
// Create callback for streaming methods that handles both success and error
@@ -513,8 +519,8 @@ async fn cmd_grpc_go<R: Runtime>(
let window_label = window.label().to_string();
move |result: std::result::Result<String, String>| match result {
Ok(msg) => {
let _ = app_handle.db().and_then(|db| {
db.upsert_grpc_event(
let _ = app_handle.with_tx(|tx| {
tx.upsert_grpc_event(
&GrpcEvent {
content: msg,
event_type: GrpcEventType::ClientMessage,
@@ -525,8 +531,8 @@ async fn cmd_grpc_go<R: Runtime>(
});
}
Err(error) => {
let _ = app_handle.db().and_then(|db| {
db.upsert_grpc_event(
let _ = app_handle.with_tx(|tx| {
tx.upsert_grpc_event(
&GrpcEvent {
content: format!("Failed to send message: {}", error),
event_type: GrpcEventType::Error,
@@ -587,9 +593,8 @@ async fn cmd_grpc_go<R: Runtime>(
if !method_desc.is_client_streaming() {
app_handle
.db()
.and_then(|db| {
db.upsert_grpc_event(
.with_tx(|tx| {
tx.upsert_grpc_event(
&GrpcEvent {
event_type: GrpcEventType::ClientMessage,
content: msg,
@@ -604,9 +609,8 @@ async fn cmd_grpc_go<R: Runtime>(
match maybe_msg {
Some(Ok(msg)) => {
app_handle
.db()
.and_then(|db| {
db.upsert_grpc_event(
.with_tx(|tx| {
tx.upsert_grpc_event(
&GrpcEvent {
metadata: metadata_to_map(msg.metadata().clone()),
content: if msg.metadata().len() == 0 {
@@ -630,9 +634,8 @@ async fn cmd_grpc_go<R: Runtime>(
Ok(content) => content,
Err(err) => {
app_handle
.db()
.and_then(|db| {
db.upsert_grpc_event(
.with_tx(|tx| {
tx.upsert_grpc_event(
&GrpcEvent {
content: "Failed to read response".to_string(),
error: Some(err.to_string()),
@@ -648,9 +651,8 @@ async fn cmd_grpc_go<R: Runtime>(
}
};
app_handle
.db()
.and_then(|db| {
db.upsert_grpc_event(
.with_tx(|tx| {
tx.upsert_grpc_event(
&GrpcEvent {
content,
event_type: GrpcEventType::ServerMessage,
@@ -661,9 +663,8 @@ async fn cmd_grpc_go<R: Runtime>(
})
.unwrap();
app_handle
.db()
.and_then(|db| {
db.upsert_grpc_event(
.with_tx(|tx| {
tx.upsert_grpc_event(
&GrpcEvent {
content: "Connection complete".to_string(),
event_type: GrpcEventType::ConnectionEnd,
@@ -677,9 +678,8 @@ async fn cmd_grpc_go<R: Runtime>(
}
Some(Err(yaak_grpc::error::Error::GrpcStreamError(e))) => {
app_handle
.db()
.and_then(|db| {
db.upsert_grpc_event(
.with_tx(|tx| {
tx.upsert_grpc_event(
&(match e.status {
Some(s) => GrpcEvent {
error: Some(s.message().to_string()),
@@ -704,9 +704,8 @@ async fn cmd_grpc_go<R: Runtime>(
}
Some(Err(e)) => {
app_handle
.db()
.and_then(|db| {
db.upsert_grpc_event(
.with_tx(|tx| {
tx.upsert_grpc_event(
&GrpcEvent {
error: Some(e.to_string()),
status: Some(Code::Unknown as i32),
@@ -727,9 +726,8 @@ async fn cmd_grpc_go<R: Runtime>(
let mut stream = match maybe_stream {
Some(Ok(stream)) => {
app_handle
.db()
.and_then(|db| {
db.upsert_grpc_event(
.with_tx(|tx| {
tx.upsert_grpc_event(
&GrpcEvent {
metadata: metadata_to_map(stream.metadata().clone()),
content: if stream.metadata().len() == 0 {
@@ -750,9 +748,8 @@ async fn cmd_grpc_go<R: Runtime>(
Some(Err(yaak_grpc::error::Error::GrpcStreamError(e))) => {
warn!("GRPC stream error {e:?}");
app_handle
.db()
.and_then(|db| {
db.upsert_grpc_event(
.with_tx(|tx| {
tx.upsert_grpc_event(
&(match e.status {
Some(s) => GrpcEvent {
error: Some(s.message().to_string()),
@@ -778,9 +775,8 @@ async fn cmd_grpc_go<R: Runtime>(
}
Some(Err(e)) => {
app_handle
.db()
.and_then(|db| {
db.upsert_grpc_event(
.with_tx(|tx| {
tx.upsert_grpc_event(
&GrpcEvent {
error: Some(e.to_string()),
status: Some(Code::Unknown as i32),
@@ -807,9 +803,8 @@ async fn cmd_grpc_go<R: Runtime>(
Ok(message) => message,
Err(err) => {
app_handle
.db()
.and_then(|db| {
db.upsert_grpc_event(
.with_tx(|tx| {
tx.upsert_grpc_event(
&GrpcEvent {
content: "Failed to read response".to_string(),
error: Some(err.to_string()),
@@ -825,9 +820,8 @@ async fn cmd_grpc_go<R: Runtime>(
}
};
app_handle
.db()
.and_then(|db| {
db.upsert_grpc_event(
.with_tx(|tx| {
tx.upsert_grpc_event(
&GrpcEvent {
content: message,
event_type: GrpcEventType::ServerMessage,
@@ -842,9 +836,8 @@ async fn cmd_grpc_go<R: Runtime>(
let trailers =
stream.trailers().await.unwrap_or_default().unwrap_or_default();
app_handle
.db()
.and_then(|db| {
db.upsert_grpc_event(
.with_tx(|tx| {
tx.upsert_grpc_event(
&GrpcEvent {
content: "Connection complete".to_string(),
status: Some(Code::Ok as i32),
@@ -860,9 +853,8 @@ async fn cmd_grpc_go<R: Runtime>(
}
Err(status) => {
app_handle
.db()
.and_then(|db| {
db.upsert_grpc_event(
.with_tx(|tx| {
tx.upsert_grpc_event(
&GrpcEvent {
content: "Stream failed".to_string(),
error: Some(status.message().to_string()),
@@ -888,7 +880,7 @@ async fn cmd_grpc_go<R: Runtime>(
let w = app_handle.clone();
tokio::select! {
_ = grpc_listen => {
let events = w.db().and_then(|db| db.list_grpc_events(&conn_id)).unwrap();
let events = w.db().list_grpc_events(&conn_id).unwrap();
let closed_event = events
.iter()
.find(|e| GrpcEventType::ConnectionEnd == e.event_type);
@@ -906,21 +898,24 @@ async fn cmd_grpc_go<R: Runtime>(
}).unwrap();
},
_ = cancelled_rx.changed() => {
w.db().and_then(|db| db.upsert_grpc_event(
&GrpcEvent {
content: "Cancelled".to_string(),
event_type: GrpcEventType::ConnectionEnd,
status: Some(Code::Cancelled as i32),
..base_msg.clone()
},
&UpdateSource::from_window_label(window.label()),
)).unwrap();
w.with_tx(|tx| {
tx.upsert_grpc_event(
&GrpcEvent {
content: "Cancelled".to_string(),
event_type: GrpcEventType::ConnectionEnd,
status: Some(Code::Cancelled as i32),
..base_msg.clone()
},
&UpdateSource::from_window_label(window.label()),
)
})
.unwrap();
w.with_tx(|c| {
c.upsert_grpc_connection(
&GrpcConnection{
elapsed: start.elapsed().as_millis() as i32,
status: Code::Cancelled as i32,
state: GrpcConnectionState::Closed,
elapsed: start.elapsed().as_millis() as i32,
status: Code::Cancelled as i32,
state: GrpcConnectionState::Closed,
..c.get_grpc_connection( &conn_id).unwrap().clone()
},
&UpdateSource::from_window_label(window.label()),
@@ -955,11 +950,11 @@ async fn cmd_send_ephemeral_request<R: Runtime>(
let response = HttpResponse::default();
request.id = "".to_string();
let environment = match environment_id {
Some(id) => Some(app_handle.db()?.get_environment(id)?),
Some(id) => Some(app_handle.db().get_environment(id)?),
None => None,
};
let cookie_jar = match cookie_jar_id {
Some(id) => Some(app_handle.db()?.get_cookie_jar(id)?),
Some(id) => Some(app_handle.db().get_cookie_jar(id)?),
None => None,
};
@@ -996,7 +991,7 @@ async fn cmd_http_response_body<R: Runtime>(
response_id: &str,
filter: Option<&str>,
) -> YaakResult<FilterResponse> {
let location = locate_response_body(&window.db()?, response_id)?;
let location = locate_response_body(&window.db(), response_id)?;
let Some(body_path) = location.path else {
return Ok(FilterResponse { content: String::new(), error: None });
};
@@ -1019,7 +1014,7 @@ async fn cmd_get_sse_events<R: Runtime>(
app_handle: AppHandle<R>,
response_id: &str,
) -> YaakResult<Vec<ServerSentEvent>> {
let Some(body_path) = locate_response_body(&app_handle.db()?, response_id)?.path else {
let Some(body_path) = locate_response_body(&app_handle.db(), response_id)?.path else {
return Ok(Vec::new());
};
@@ -1065,20 +1060,6 @@ async fn cmd_commit_import<R: Runtime>(
commit_import(&window, plan)
}
/// Decodes base64 and writes the bytes to a file the user picked.
///
/// The webview can't do this itself: its `fs` permissions are read-only and scoped to the app
@@ -1109,18 +1090,20 @@ async fn cmd_send_http_request<R: Runtime>(
cookie_jar_id: Option<&str>,
request_id: String,
) -> YaakResult<HttpResponse> {
let request = app_handle.db()?.get_http_request(&request_id)?;
let request = app_handle.db().get_http_request(&request_id)?;
let blobs = app_handle.blob_manager();
let response = app_handle.db()?.upsert_http_response(
&HttpResponse {
request_id: request.id.clone(),
workspace_id: request.workspace_id.clone(),
..Default::default()
},
&UpdateSource::from_window_label(window.label()),
&blobs,
)?;
let response = app_handle.with_tx(|tx| {
tx.upsert_http_response(
&HttpResponse {
request_id: request.id.clone(),
workspace_id: request.workspace_id.clone(),
..Default::default()
},
&UpdateSource::from_window_label(window.label()),
&blobs,
)
})?;
let (cancel_tx, mut cancel_rx) = tokio::sync::watch::channel(false);
app_handle.listen_any(format!("cancel_http_response_{}", response.id), move |_event| {
@@ -1130,7 +1113,7 @@ async fn cmd_send_http_request<R: Runtime>(
});
let environment = match environment_id {
Some(id) => match app_handle.db()?.get_environment(id) {
Some(id) => match app_handle.db().get_environment(id) {
Ok(env) => Some(env),
Err(e) => {
warn!("Failed to find environment by id {id} {}", e);
@@ -1141,7 +1124,7 @@ async fn cmd_send_http_request<R: Runtime>(
};
let cookie_jar = match cookie_jar_id {
Some(id) => Some(app_handle.db()?.get_cookie_jar(id)?),
Some(id) => Some(app_handle.db().get_cookie_jar(id)?),
None => None,
};
@@ -1157,23 +1140,24 @@ async fn cmd_send_http_request<R: Runtime>(
{
Ok(sent) => sent.response,
Err(e) => {
let resp = app_handle.db()?.get_http_response(&response.id)?;
app_handle.db()?.upsert_http_response(
&HttpResponse {
state: HttpResponseState::Closed,
error: Some(e.to_string()),
..resp
},
&UpdateSource::from_window_label(window.label()),
&blobs,
)?
let resp = app_handle.db().get_http_response(&response.id)?;
app_handle.with_tx(|tx| {
tx.upsert_http_response(
&HttpResponse {
state: HttpResponseState::Closed,
error: Some(e.to_string()),
..resp
},
&UpdateSource::from_window_label(window.label()),
&blobs,
)
})?
}
};
Ok(r)
}
async fn cmd_new_child_window<R: Runtime>(
parent_window: WebviewWindow<R>,
url: &str,
@@ -1181,7 +1165,7 @@ async fn cmd_new_child_window<R: Runtime>(
title: &str,
inner_size: (f64, f64),
) -> YaakResult<()> {
let use_native_titlebar = parent_window.app_handle().db()?.get_settings().use_native_titlebar;
let use_native_titlebar = parent_window.app_handle().db().get_settings().use_native_titlebar;
let initialization_script = initial_appearance_script(&parent_window.app_handle());
let win = yaak_window::window::create_child_window(
&parent_window,
@@ -1197,7 +1181,7 @@ async fn cmd_new_child_window<R: Runtime>(
}
async fn cmd_new_main_window<R: Runtime>(app_handle: AppHandle<R>, url: &str) -> YaakResult<()> {
let use_native_titlebar = app_handle.db()?.get_settings().use_native_titlebar;
let use_native_titlebar = app_handle.db().get_settings().use_native_titlebar;
let initialization_script = initial_appearance_script(&app_handle);
let win = yaak_window::window::create_main_window(
&app_handle,
@@ -1214,7 +1198,7 @@ async fn cmd_check_for_updates<R: Runtime>(
yaak_updater: State<'_, Mutex<YaakUpdater>>,
) -> YaakResult<bool> {
let update_mode = get_update_mode(&window).await?;
let settings = window.db()?.get_settings();
let settings = window.db().get_settings();
Ok(yaak_updater
.lock()
.await
@@ -1301,8 +1285,8 @@ pub fn run() {
.setup(|app| {
let lifecycle_host = yaak_lifecycle::Host::owner()
.with_responses_dir(app.path().app_data_dir()?.join("responses"));
if let Err(e) =
yaak_lifecycle::on_launch(&lifecycle_host, &app.db()?, &app.blob_manager())
if let Err(e) = app
.with_tx(|tx| yaak_lifecycle::on_launch(&lifecycle_host, tx, &app.blob_manager()))
{
error!("on_launch hook failed: {e:?}");
}
@@ -1378,7 +1362,7 @@ pub fn run() {
app.manage(Mutex::new(ws_manager));
// Specific settings
let settings = app.db()?.get_settings();
let settings = app.db().get_settings();
app.app_handle().set_native_titlebar(settings.use_native_titlebar);
monitor_plugin_events(&app.app_handle().clone());
@@ -1391,8 +1375,7 @@ pub fn run() {
.run(|app_handle, event| {
match event {
RunEvent::Ready => {
let settings = app_handle.db().map(|db| db.get_settings()).unwrap_or_default();
let use_native_titlebar = settings.use_native_titlebar;
let use_native_titlebar = app_handle.db().get_settings().use_native_titlebar;
let initialization_script = initial_appearance_script(app_handle);
if let Ok(win) = yaak_window::window::create_main_window(
app_handle,
@@ -1431,10 +1414,7 @@ pub fn run() {
let w = app_handle.get_webview_window(&label).unwrap();
let h = app_handle.clone();
tauri::async_runtime::spawn(async move {
let Ok(settings) = w.db().map(|db| db.get_settings()) else {
warn!("Failed to read settings for update check");
return;
};
let settings = w.db().get_settings();
if settings.autoupdate {
time::sleep(Duration::from_secs(3)).await; // Wait a bit so it's not so jarring
let val: State<'_, Mutex<YaakUpdater>> = h.state();
@@ -1470,7 +1450,7 @@ pub fn run() {
}
async fn get_update_mode<R: Runtime>(window: &WebviewWindow<R>) -> YaakResult<UpdateMode> {
let settings = window.db()?.get_settings();
let settings = window.db().get_settings();
Ok(UpdateMode::new(settings.update_channel.as_str()))
}
@@ -1597,13 +1577,13 @@ fn get_window_from_plugin_context<R: Runtime>(
}
fn workspace_from_window<R: Runtime>(window: &WebviewWindow<R>) -> Option<Workspace> {
window.workspace_id().and_then(|id| window.db().and_then(|db| db.get_workspace(&id)).ok())
window.workspace_id().and_then(|id| window.db().get_workspace(&id).ok())
}
fn environment_from_window<R: Runtime>(window: &WebviewWindow<R>) -> Option<Environment> {
window.environment_id().and_then(|id| window.db().and_then(|db| db.get_environment(&id)).ok())
window.environment_id().and_then(|id| window.db().get_environment(&id).ok())
}
fn cookie_jar_from_window<R: Runtime>(window: &WebviewWindow<R>) -> Option<CookieJar> {
window.cookie_jar_id().and_then(|id| window.db().and_then(|db| db.get_cookie_jar(&id)).ok())
window.cookie_jar_id().and_then(|id| window.db().get_cookie_jar(&id).ok())
}
+10 -8
View File
@@ -10,7 +10,7 @@ use tauri::plugin::TauriPlugin;
use tauri::{Emitter, Manager, Runtime, State};
use tauri_plugin_dialog::{DialogExt, MessageDialogKind};
use yaak_models::blob_manager::BlobManager;
use yaak_models::client_db::ClientDb;
use yaak_models::client_db::{ClientDb, WriteDb};
use yaak_models::error::Result;
use yaak_models::query_manager::QueryManager;
use yaak_models::util::{ModelPayload, UpdateSource};
@@ -37,9 +37,11 @@ fn drain_model_changes_batch<R: Runtime>(
app_handle: &tauri::AppHandle<R>,
cursor: &mut ModelChangeCursor,
) -> bool {
let changes = match query_manager.connect().and_then(|db| {
db.list_model_changes_since(&cursor.created_at, cursor.id, MODEL_CHANGES_POLL_BATCH_SIZE)
}) {
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:?}");
@@ -90,10 +92,10 @@ async fn run_model_change_poller<R: Runtime>(
/// Extension trait for accessing the QueryManager from Tauri Manager types.
pub trait QueryManagerExt<'a, R> {
fn db_manager(&'a self) -> State<'a, QueryManager>;
fn db(&'a self) -> Result<ClientDb<'a>>;
fn db(&'a self) -> ClientDb<'a>;
fn with_tx<F, T>(&'a self, func: F) -> Result<T>
where
F: FnOnce(&ClientDb) -> Result<T>;
F: FnOnce(&WriteDb) -> Result<T>;
}
impl<'a, R: Runtime, M: Manager<R>> QueryManagerExt<'a, R> for M {
@@ -101,14 +103,14 @@ impl<'a, R: Runtime, M: Manager<R>> QueryManagerExt<'a, R> for M {
self.state::<QueryManager>()
}
fn db(&'a self) -> Result<ClientDb<'a>> {
fn db(&'a self) -> ClientDb<'a> {
let qm = self.state::<QueryManager>();
qm.inner().connect()
}
fn with_tx<F, T>(&'a self, func: F) -> Result<T>
where
F: FnOnce(&ClientDb) -> Result<T>,
F: FnOnce(&WriteDb) -> Result<T>,
{
let qm = self.state::<QueryManager>();
qm.inner().with_tx(func)
@@ -55,12 +55,15 @@ impl YaakNotifier {
seen.push(id.to_string());
debug!("Marked notification as seen {}", id);
let seen_json = serde_json::to_string(&seen)?;
window.db()?.set_key_value_raw(
KV_NAMESPACE,
KV_KEY,
seen_json.as_str(),
&UpdateSource::from_window_label(window.label()),
);
window.with_tx(|tx| {
tx.set_key_value_raw(
KV_NAMESPACE,
KV_KEY,
seen_json.as_str(),
&UpdateSource::from_window_label(window.label()),
);
Ok(())
})?;
Ok(())
}
@@ -74,7 +77,7 @@ impl YaakNotifier {
self.last_check = Some(Instant::now());
if !app_handle.db()?.get_settings().check_notifications {
if !app_handle.db().get_settings().check_notifications {
info!("Notifications are disabled. Skipping check.");
return Ok(());
}
@@ -139,7 +142,7 @@ impl YaakNotifier {
}
async fn get_kv<R: Runtime>(app_handle: &AppHandle<R>) -> Result<Vec<String>> {
match app_handle.db()?.get_key_value_raw("notifications", "seen") {
match app_handle.db().get_key_value_raw("notifications", "seen") {
None => Ok(Vec::new()),
Some(v) => Ok(serde_json::from_str(&v.value)?),
}
@@ -155,7 +158,7 @@ fn get_updater_status<R: Runtime>(app_handle: &AppHandle<R>) -> &'static str {
#[cfg(all(feature = "updater", target_os = "linux"))]
{
let settings = app_handle.db()?.get_settings();
let settings = app_handle.db().get_settings();
if !settings.autoupdate {
// Updates are explicitly disabled
"disabled"
@@ -170,7 +173,7 @@ fn get_updater_status<R: Runtime>(app_handle: &AppHandle<R>) -> &'static str {
#[cfg(all(feature = "updater", not(target_os = "linux")))]
{
let settings = app_handle.db()?.get_settings();
let settings = app_handle.db().get_settings();
if settings.autoupdate { "enabled" } else { "disabled" }
}
}
@@ -104,14 +104,14 @@ async fn handle_host_plugin_request<R: Runtime>(
Box::pin(handle_plugin_event(app_handle, &toast_event, plugin_handle)).await
}
HostRequest::ReloadResponse(req) => {
let plugins = app_handle.db()?.list_plugins()?;
let plugins = app_handle.db().list_plugins()?;
for plugin in plugins {
if plugin.directory != plugin_handle.dir {
continue;
}
let new_plugin = Plugin { updated_at: Utc::now().naive_utc(), ..plugin };
app_handle.db()?.upsert_plugin(&new_plugin, &UpdateSource::Plugin)?;
app_handle.with_tx(|tx| tx.upsert_plugin(&new_plugin, &UpdateSource::Plugin))?;
}
if !req.silent {
@@ -199,7 +199,7 @@ async fn handle_host_plugin_request<R: Runtime>(
let workspace =
workspace_from_window(&window).expect("Failed to get workspace_id from window URL");
let environment_id = environment_from_window(&window).map(|e| e.id);
let environment_chain = window.db()?.resolve_environments(
let environment_chain = window.db().resolve_environments(
&workspace.id,
req.grpc_request.folder_id.as_deref(),
environment_id.as_deref(),
@@ -225,7 +225,7 @@ async fn handle_host_plugin_request<R: Runtime>(
let workspace =
workspace_from_window(&window).expect("Failed to get workspace_id from window URL");
let environment_id = environment_from_window(&window).map(|e| e.id);
let environment_chain = window.db()?.resolve_environments(
let environment_chain = window.db().resolve_environments(
&workspace.id,
req.http_request.folder_id.as_deref(),
environment_id.as_deref(),
@@ -252,7 +252,7 @@ async fn handle_host_plugin_request<R: Runtime>(
workspace_from_window(&window).expect("Failed to get workspace_id from window URL");
let environment_id = environment_from_window(&window).map(|e| e.id);
let folder_id = if let Some(id) = window.request_id() {
match window.db()?.get_any_request(&id) {
match window.db().get_any_request(&id) {
Ok(AnyRequest::HttpRequest(r)) => r.folder_id,
Ok(AnyRequest::GrpcRequest(r)) => r.folder_id,
Ok(AnyRequest::WebsocketRequest(r)) => r.folder_id,
@@ -261,7 +261,7 @@ async fn handle_host_plugin_request<R: Runtime>(
} else {
None
};
let environment_chain = window.db()?.resolve_environments(
let environment_chain = window.db().resolve_environments(
&workspace.id,
folder_id.as_deref(),
environment_id.as_deref(),
@@ -294,15 +294,17 @@ async fn handle_host_plugin_request<R: Runtime>(
HttpResponse::default()
} else {
let blobs = window.blob_manager();
window.db()?.upsert_http_response(
&HttpResponse {
request_id: http_request.id.clone(),
workspace_id: http_request.workspace_id.clone(),
..Default::default()
},
&UpdateSource::from_window_label(window.label()),
&blobs,
)?
window.with_tx(|tx| {
tx.upsert_http_response(
&HttpResponse {
request_id: http_request.id.clone(),
workspace_id: http_request.workspace_id.clone(),
..Default::default()
},
&UpdateSource::from_window_label(window.label()),
&blobs,
)
})?
};
let http_response = send_http_request_with_context(
@@ -328,7 +330,7 @@ async fn handle_host_plugin_request<R: Runtime>(
HostRequest::OpenWindow(req) => {
let (navigation_tx, mut navigation_rx) = tokio::sync::mpsc::channel(128);
let (close_tx, mut close_rx) = tokio::sync::mpsc::channel(128);
let use_native_titlebar = app_handle.db()?.get_settings().use_native_titlebar;
let use_native_titlebar = app_handle.db().get_settings().use_native_titlebar;
let win_config = CreateWindowConfig {
url: &req.url,
label: &req.label,
@@ -438,7 +440,7 @@ async fn handle_host_plugin_request<R: Runtime>(
let environment_id = environment_from_window(&w).map(|m| m.id);
let workspace_id = workspace_from_window(&w).map(|m| m.id);
let request_id =
match app_handle.db()?.get_any_request(&w.request_id().unwrap_or_default()) {
match app_handle.db().get_any_request(&w.request_id().unwrap_or_default()) {
Ok(AnyRequest::HttpRequest(r)) => Some(r.id),
Ok(AnyRequest::WebsocketRequest(r)) => Some(r.id),
Ok(AnyRequest::GrpcRequest(r)) => Some(r.id),
+22 -19
View File
@@ -16,8 +16,7 @@ use std::time::{Duration, Instant};
use tauri::path::BaseDirectory;
use tauri::plugin::{Builder, TauriPlugin};
use tauri::{
AppHandle, Emitter, Manager, RunEvent, Runtime, State, WebviewWindow, WindowEvent,
is_dev,
AppHandle, Emitter, Manager, RunEvent, Runtime, State, WebviewWindow, WindowEvent, is_dev,
};
use tokio::sync::Mutex;
use ts_rs::TS;
@@ -28,10 +27,10 @@ use yaak_plugins::api::{
PluginNameVersion, PluginSearchResponse, PluginUpdatesResponse, check_plugin_updates,
search_plugins,
};
use yaak_plugins::error::Error::PluginErr;
use yaak_plugins::events::{Color, PluginContext, ShowToastRequest};
use yaak_plugins::install::{delete_and_uninstall, download_and_install};
use yaak_plugins::manager::PluginManager;
use yaak_plugins::error::Error::PluginErr;
use yaak_plugins::plugin_meta::get_plugin_meta;
static EXITING: AtomicBool = AtomicBool::new(false);
@@ -107,7 +106,7 @@ impl PluginUpdater {
let app_version = window.app_handle().package_info().version.to_string();
let http_client = yaak_api_client(ApiClientKind::App, &app_version)?;
let plugins = window.app_handle().db()?.list_plugins()?;
let plugins = window.app_handle().db().list_plugins()?;
let updates = check_plugin_updates(&http_client, plugins.clone()).await?;
if updates.plugins.is_empty() {
@@ -203,16 +202,18 @@ pub async fn cmd_plugins_install_from_directory<R: Runtime>(
// Resolve the manager before writing the row so startup's plugin snapshot
// can't include it and boot it a second time
let plugin_manager = Arc::new(plugin_manager(&window).await?);
let plugin = window.db()?.upsert_plugin(
&Plugin {
directory: directory.into(),
url: None,
enabled: true,
source: PluginSource::Filesystem,
..Default::default()
},
&UpdateSource::from_window_label(window.label()),
)?;
let plugin = window.with_tx(|tx| {
tx.upsert_plugin(
&Plugin {
directory: directory.into(),
url: None,
enabled: true,
source: PluginSource::Filesystem,
..Default::default()
},
&UpdateSource::from_window_label(window.label()),
)
})?;
plugin_manager.add_plugin(&window.plugin_context(), &plugin).await?;
@@ -234,7 +235,7 @@ pub async fn cmd_plugins_updates<R: Runtime>(
) -> Result<PluginUpdatesResponse> {
let app_version = app_handle.package_info().version.to_string();
let http_client = yaak_api_client(ApiClientKind::App, &app_version)?;
let plugins = app_handle.db()?.list_plugins()?;
let plugins = app_handle.db().list_plugins()?;
Ok(check_plugin_updates(&http_client, plugins).await?)
}
@@ -243,7 +244,7 @@ pub async fn cmd_plugins_update_all<R: Runtime>(
) -> Result<Vec<PluginNameVersion>> {
let app_version = window.app_handle().package_info().version.to_string();
let http_client = yaak_api_client(ApiClientKind::App, &app_version)?;
let plugins = window.db()?.list_plugins()?;
let plugins = window.db().list_plugins()?;
// Get list of available updates (already filtered to only registry plugins)
let updates = check_plugin_updates(&http_client, plugins).await?;
@@ -355,9 +356,11 @@ pub fn init<R: Runtime>() -> TauriPlugin<R> {
),
)
.await
.unwrap_or_else(|_| Err(yaak_plugins::error::Error::PluginErr(
"Timed out starting the plugin runtime".to_string(),
)));
.unwrap_or_else(|_| {
Err(yaak_plugins::error::Error::PluginErr(
"Timed out starting the plugin runtime".to_string(),
))
});
let manager = match result {
Ok(manager) => manager,
+1 -1
View File
@@ -4,5 +4,5 @@
//! `yaak-commands` when the template commands did. Callers in this crate do not
//! need to track which is which.
pub use yaak_models::render::{render_grpc_request, render_http_request};
pub use yaak_commands::render::{render_json_value, render_template};
pub use yaak_models::render::{render_grpc_request, render_http_request};
+465 -116
View File
@@ -21,9 +21,9 @@ use crate::notifications::YaakNotifier;
use crate::updates::YaakUpdater;
use log::warn;
use serde::Serialize;
use tauri::{Manager, Runtime, State, WebviewWindow};
use std::collections::HashMap;
use std::sync::Arc;
use tauri::{Manager, Runtime, State, WebviewWindow};
use tokio::sync::Mutex;
use yaak_commands::{Host, PluginHost};
use yaak_core::WorkspaceContext;
@@ -32,8 +32,8 @@ use yaak_git::{
BranchDeleteResult, CloneResult, GitBranchInfo, GitCommit, GitFileDiff, GitRemote,
GitStatusSummary, GitWorktreeStatus, PullResult, PushResult,
};
use yaak_grpc::manager::GrpcHandle;
use yaak_grpc::ServiceDefinition;
use yaak_grpc::manager::GrpcHandle;
use yaak_models::blob_manager::BlobManager;
use yaak_models::models::{
GraphQlIntrospection, GrpcEvent, HttpRequest, HttpRequestHeader, HttpResponse,
@@ -231,17 +231,15 @@ impl<R: Runtime> PluginHost for ClientCtx<R> {
Ok(self.pm().await?.call_workspace_action(&self.plugin_context(), req).await?)
}
async fn call_folder_action(
&self,
req: CallFolderActionRequest,
) -> yaak_commands::Result<()> {
async fn call_folder_action(&self, req: CallFolderActionRequest) -> yaak_commands::Result<()> {
Ok(self.pm().await?.call_folder_action(&self.plugin_context(), req).await?)
}
async fn http_authentication_summaries(
&self,
) -> yaak_commands::Result<Vec<GetHttpAuthenticationSummaryResponse>> {
let results = self.pm().await?.get_http_authentication_summaries(&self.plugin_context()).await?;
let results =
self.pm().await?.get_http_authentication_summaries(&self.plugin_context()).await?;
Ok(results.into_iter().map(|(_, a)| a).collect())
}
@@ -393,11 +391,17 @@ async fn cmd_metadata<R: Runtime>(ctx: ClientCtx<R>, _req: CmdMetadataReq) -> Re
Ok(crate::cmd_metadata(ctx.window.app_handle().clone()).await?)
}
async fn cmd_template_tokens_to_string<R: Runtime>(ctx: ClientCtx<R>, req: CmdTemplateTokensToStringReq) -> Result<String> {
async fn cmd_template_tokens_to_string<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdTemplateTokensToStringReq,
) -> Result<String> {
Ok(yaak_commands::templates::cmd_template_tokens_to_string(ctx, req).await?)
}
async fn cmd_render_template<R: Runtime>(ctx: ClientCtx<R>, req: CmdRenderTemplateReq) -> Result<String> {
async fn cmd_render_template<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdRenderTemplateReq,
) -> Result<String> {
Ok(yaak_commands::templates::cmd_render_template(ctx, req).await?)
}
@@ -405,55 +409,114 @@ async fn cmd_send_feedback<R: Runtime>(ctx: ClientCtx<R>, req: CmdSendFeedbackRe
Ok(crate::cmd_send_feedback(ctx.window.app_handle().clone(), req.feature, req.text).await?)
}
async fn cmd_dismiss_notification<R: Runtime>(ctx: ClientCtx<R>, req: CmdDismissNotificationReq) -> Result<()> {
Ok(crate::cmd_dismiss_notification(ctx.window.clone(), &req.notification_id, ctx.window.app_handle().state::<Mutex<YaakNotifier>>()).await?)
async fn cmd_dismiss_notification<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdDismissNotificationReq,
) -> Result<()> {
Ok(crate::cmd_dismiss_notification(
ctx.window.clone(),
&req.notification_id,
ctx.window.app_handle().state::<Mutex<YaakNotifier>>(),
)
.await?)
}
async fn cmd_grpc_reflect<R: Runtime>(ctx: ClientCtx<R>, req: CmdGrpcReflectReq) -> Result<Vec<ServiceDefinition>> {
Ok(crate::cmd_grpc_reflect(&req.request_id, req.environment_id.as_deref(), req.proto_files, ctx.window.clone(), ctx.window.app_handle().clone(), ctx.window.app_handle().state::<Mutex<GrpcHandle>>()).await?)
async fn cmd_grpc_reflect<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdGrpcReflectReq,
) -> Result<Vec<ServiceDefinition>> {
Ok(crate::cmd_grpc_reflect(
&req.request_id,
req.environment_id.as_deref(),
req.proto_files,
ctx.window.clone(),
ctx.window.app_handle().clone(),
ctx.window.app_handle().state::<Mutex<GrpcHandle>>(),
)
.await?)
}
async fn cmd_grpc_go<R: Runtime>(ctx: ClientCtx<R>, req: CmdGrpcGoReq) -> Result<String> {
Ok(crate::cmd_grpc_go(&req.request_id, req.environment_id.as_deref(), req.proto_files, ctx.window.app_handle().clone(), ctx.window.clone(), ctx.window.app_handle().state::<Mutex<GrpcHandle>>()).await?)
Ok(crate::cmd_grpc_go(
&req.request_id,
req.environment_id.as_deref(),
req.proto_files,
ctx.window.app_handle().clone(),
ctx.window.clone(),
ctx.window.app_handle().state::<Mutex<GrpcHandle>>(),
)
.await?)
}
async fn cmd_restart<R: Runtime>(ctx: ClientCtx<R>, _req: CmdRestartReq) -> Result<()> {
Ok(crate::cmd_restart(ctx.window.app_handle().clone()).await?)
}
async fn cmd_send_ephemeral_request<R: Runtime>(ctx: ClientCtx<R>, req: CmdSendEphemeralRequestReq) -> Result<EphemeralHttpResponse> {
Ok(crate::cmd_send_ephemeral_request(req.request, req.environment_id.as_deref(), req.cookie_jar_id.as_deref(), ctx.window.clone(), ctx.window.app_handle().clone()).await?)
async fn cmd_send_ephemeral_request<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdSendEphemeralRequestReq,
) -> Result<EphemeralHttpResponse> {
Ok(crate::cmd_send_ephemeral_request(
req.request,
req.environment_id.as_deref(),
req.cookie_jar_id.as_deref(),
ctx.window.clone(),
ctx.window.app_handle().clone(),
)
.await?)
}
async fn cmd_format_json<R: Runtime>(ctx: ClientCtx<R>, req: CmdFormatJsonReq) -> Result<String> {
Ok(yaak_commands::data::cmd_format_json(ctx, req).await?)
}
async fn cmd_format_graphql<R: Runtime>(_ctx: ClientCtx<R>, req: CmdFormatGraphqlReq) -> Result<String> {
async fn cmd_format_graphql<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdFormatGraphqlReq,
) -> Result<String> {
Ok(crate::cmd_format_graphql(&req.text).await?)
}
async fn cmd_http_response_body<R: Runtime>(ctx: ClientCtx<R>, req: CmdHttpResponseBodyReq) -> Result<FilterResponse> {
Ok(crate::cmd_http_response_body(ctx.window.clone(), &req.response_id, req.filter.as_deref()).await?)
async fn cmd_http_response_body<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdHttpResponseBodyReq,
) -> Result<FilterResponse> {
Ok(crate::cmd_http_response_body(ctx.window.clone(), &req.response_id, req.filter.as_deref())
.await?)
}
async fn cmd_http_response_body_path<R: Runtime>(ctx: ClientCtx<R>, req: CmdHttpResponseBodyPathReq) -> Result<Option<String>> {
async fn cmd_http_response_body_path<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdHttpResponseBodyPathReq,
) -> Result<Option<String>> {
Ok(yaak_commands::responses::cmd_http_response_body_path(ctx, req).await?)
}
async fn cmd_http_request_body<R: Runtime>(ctx: ClientCtx<R>, req: CmdHttpRequestBodyReq) -> Result<Option<Vec<u8>>> {
async fn cmd_http_request_body<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdHttpRequestBodyReq,
) -> Result<Option<Vec<u8>>> {
Ok(yaak_commands::responses::cmd_http_request_body(ctx, req).await?)
}
async fn cmd_get_sse_events<R: Runtime>(ctx: ClientCtx<R>, req: CmdGetSseEventsReq) -> Result<Vec<ServerSentEvent>> {
async fn cmd_get_sse_events<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdGetSseEventsReq,
) -> Result<Vec<ServerSentEvent>> {
Ok(crate::cmd_get_sse_events(ctx.window.app_handle().clone(), &req.response_id).await?)
}
async fn cmd_get_http_response_events<R: Runtime>(ctx: ClientCtx<R>, req: CmdGetHttpResponseEventsReq) -> Result<Vec<HttpResponseEvent>> {
async fn cmd_get_http_response_events<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdGetHttpResponseEventsReq,
) -> Result<Vec<HttpResponseEvent>> {
Ok(yaak_commands::responses::cmd_get_http_response_events(ctx, req).await?)
}
async fn cmd_import_data<R: Runtime>(ctx: ClientCtx<R>, req: CmdImportDataReq) -> Result<ImportPlan> {
async fn cmd_import_data<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdImportDataReq,
) -> Result<ImportPlan> {
Ok(crate::cmd_import_data(ctx.window.clone(), &req.file_path, req.destination).await?)
}
@@ -461,16 +524,25 @@ async fn cmd_import_url<R: Runtime>(ctx: ClientCtx<R>, req: CmdImportUrlReq) ->
Ok(crate::cmd_import_url(ctx.window.clone(), &req.url, req.destination).await?)
}
async fn cmd_commit_import<R: Runtime>(ctx: ClientCtx<R>, req: CmdCommitImportReq) -> Result<BatchUpsertResult> {
async fn cmd_commit_import<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdCommitImportReq,
) -> Result<BatchUpsertResult> {
Ok(crate::cmd_commit_import(ctx.window.clone(), req.plan).await?)
}
async fn cmd_list_import_sources<R: Runtime>(ctx: ClientCtx<R>, req: CmdListImportSourcesReq) -> Result<Vec<ImportSource>> {
async fn cmd_list_import_sources<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdListImportSourcesReq,
) -> Result<Vec<ImportSource>> {
use crate::models_ext::QueryManagerExt;
Ok(ctx.window.db()?.list_import_sources(&req.workspace_id)?)
Ok(ctx.window.db().list_import_sources(&req.workspace_id)?)
}
async fn cmd_import_sources_for_origin<R: Runtime>(ctx: ClientCtx<R>, req: CmdImportSourcesForOriginReq) -> Result<Vec<ImportSource>> {
async fn cmd_import_sources_for_origin<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdImportSourcesForOriginReq,
) -> Result<Vec<ImportSource>> {
use crate::models_ext::QueryManagerExt;
let origin = match (req.file_path, req.url) {
(Some(file_path), _) => crate::import::file_origin(&file_path).origin,
@@ -480,70 +552,118 @@ async fn cmd_import_sources_for_origin<R: Runtime>(ctx: ClientCtx<R>, req: CmdIm
},
(None, None) => return Ok(Vec::new()),
};
Ok(ctx.window.db()?.list_import_sources_by_origin(&origin)?)
Ok(ctx.window.db().list_import_sources_by_origin(&origin)?)
}
async fn cmd_http_request_actions<R: Runtime>(ctx: ClientCtx<R>, req: CmdHttpRequestActionsReq) -> Result<Vec<GetHttpRequestActionsResponse>> {
async fn cmd_http_request_actions<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdHttpRequestActionsReq,
) -> Result<Vec<GetHttpRequestActionsResponse>> {
Ok(yaak_commands::actions::cmd_http_request_actions(ctx, req).await?)
}
async fn cmd_websocket_request_actions<R: Runtime>(ctx: ClientCtx<R>, req: CmdWebsocketRequestActionsReq) -> Result<Vec<GetWebsocketRequestActionsResponse>> {
async fn cmd_websocket_request_actions<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdWebsocketRequestActionsReq,
) -> Result<Vec<GetWebsocketRequestActionsResponse>> {
Ok(yaak_commands::actions::cmd_websocket_request_actions(ctx, req).await?)
}
async fn cmd_call_websocket_request_action<R: Runtime>(ctx: ClientCtx<R>, req: CmdCallWebsocketRequestActionReq) -> Result<()> {
async fn cmd_call_websocket_request_action<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdCallWebsocketRequestActionReq,
) -> Result<()> {
Ok(yaak_commands::actions::cmd_call_websocket_request_action(ctx, req).await?)
}
async fn cmd_workspace_actions<R: Runtime>(ctx: ClientCtx<R>, req: CmdWorkspaceActionsReq) -> Result<Vec<GetWorkspaceActionsResponse>> {
async fn cmd_workspace_actions<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdWorkspaceActionsReq,
) -> Result<Vec<GetWorkspaceActionsResponse>> {
Ok(yaak_commands::actions::cmd_workspace_actions(ctx, req).await?)
}
async fn cmd_call_workspace_action<R: Runtime>(ctx: ClientCtx<R>, req: CmdCallWorkspaceActionReq) -> Result<()> {
async fn cmd_call_workspace_action<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdCallWorkspaceActionReq,
) -> Result<()> {
Ok(yaak_commands::actions::cmd_call_workspace_action(ctx, req).await?)
}
async fn cmd_folder_actions<R: Runtime>(ctx: ClientCtx<R>, req: CmdFolderActionsReq) -> Result<Vec<GetFolderActionsResponse>> {
async fn cmd_folder_actions<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdFolderActionsReq,
) -> Result<Vec<GetFolderActionsResponse>> {
Ok(yaak_commands::actions::cmd_folder_actions(ctx, req).await?)
}
async fn cmd_call_folder_action<R: Runtime>(ctx: ClientCtx<R>, req: CmdCallFolderActionReq) -> Result<()> {
async fn cmd_call_folder_action<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdCallFolderActionReq,
) -> Result<()> {
Ok(yaak_commands::actions::cmd_call_folder_action(ctx, req).await?)
}
async fn cmd_grpc_request_actions<R: Runtime>(ctx: ClientCtx<R>, req: CmdGrpcRequestActionsReq) -> Result<Vec<GetGrpcRequestActionsResponse>> {
async fn cmd_grpc_request_actions<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdGrpcRequestActionsReq,
) -> Result<Vec<GetGrpcRequestActionsResponse>> {
Ok(yaak_commands::actions::cmd_grpc_request_actions(ctx, req).await?)
}
async fn cmd_template_function_summaries<R: Runtime>(ctx: ClientCtx<R>, req: CmdTemplateFunctionSummariesReq) -> Result<Vec<GetTemplateFunctionSummaryResponse>> {
async fn cmd_template_function_summaries<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdTemplateFunctionSummariesReq,
) -> Result<Vec<GetTemplateFunctionSummaryResponse>> {
Ok(yaak_commands::templates::cmd_template_function_summaries(ctx, req).await?)
}
async fn cmd_template_function_config<R: Runtime>(ctx: ClientCtx<R>, req: CmdTemplateFunctionConfigReq) -> Result<GetTemplateFunctionConfigResponse> {
async fn cmd_template_function_config<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdTemplateFunctionConfigReq,
) -> Result<GetTemplateFunctionConfigResponse> {
Ok(yaak_commands::templates::cmd_template_function_config(ctx, req).await?)
}
async fn cmd_get_http_authentication_summaries<R: Runtime>(ctx: ClientCtx<R>, req: CmdGetHttpAuthenticationSummariesReq) -> Result<Vec<GetHttpAuthenticationSummaryResponse>> {
async fn cmd_get_http_authentication_summaries<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdGetHttpAuthenticationSummariesReq,
) -> Result<Vec<GetHttpAuthenticationSummaryResponse>> {
Ok(yaak_commands::auth::cmd_get_http_authentication_summaries(ctx, req).await?)
}
async fn cmd_get_http_authentication_config<R: Runtime>(ctx: ClientCtx<R>, req: CmdGetHttpAuthenticationConfigReq) -> Result<GetHttpAuthenticationConfigResponse> {
async fn cmd_get_http_authentication_config<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdGetHttpAuthenticationConfigReq,
) -> Result<GetHttpAuthenticationConfigResponse> {
Ok(yaak_commands::auth::cmd_get_http_authentication_config(ctx, req).await?)
}
async fn cmd_call_http_request_action<R: Runtime>(ctx: ClientCtx<R>, req: CmdCallHttpRequestActionReq) -> Result<()> {
async fn cmd_call_http_request_action<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdCallHttpRequestActionReq,
) -> Result<()> {
Ok(yaak_commands::actions::cmd_call_http_request_action(ctx, req).await?)
}
async fn cmd_call_grpc_request_action<R: Runtime>(ctx: ClientCtx<R>, req: CmdCallGrpcRequestActionReq) -> Result<()> {
async fn cmd_call_grpc_request_action<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdCallGrpcRequestActionReq,
) -> Result<()> {
Ok(yaak_commands::actions::cmd_call_grpc_request_action(ctx, req).await?)
}
async fn cmd_call_http_authentication_action<R: Runtime>(ctx: ClientCtx<R>, req: CmdCallHttpAuthenticationActionReq) -> Result<()> {
async fn cmd_call_http_authentication_action<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdCallHttpAuthenticationActionReq,
) -> Result<()> {
Ok(yaak_commands::auth::cmd_call_http_authentication_action(ctx, req).await?)
}
async fn cmd_curl_to_request<R: Runtime>(ctx: ClientCtx<R>, req: CmdCurlToRequestReq) -> Result<HttpRequest> {
async fn cmd_curl_to_request<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdCurlToRequestReq,
) -> Result<HttpRequest> {
Ok(yaak_commands::actions::cmd_curl_to_request(ctx, req).await?)
}
@@ -551,83 +671,159 @@ async fn cmd_export_data<R: Runtime>(ctx: ClientCtx<R>, req: CmdExportDataReq) -
Ok(yaak_commands::data::cmd_export_data(ctx, req).await?)
}
async fn cmd_save_base64_to_binary<R: Runtime>(ctx: ClientCtx<R>, req: CmdSaveBase64ToBinaryReq) -> Result<()> {
Ok(crate::cmd_save_base64_to_binary(ctx.window.app_handle().clone(), &req.filepath, &req.data).await?)
async fn cmd_save_base64_to_binary<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdSaveBase64ToBinaryReq,
) -> Result<()> {
Ok(crate::cmd_save_base64_to_binary(ctx.window.app_handle().clone(), &req.filepath, &req.data)
.await?)
}
async fn cmd_save_response<R: Runtime>(ctx: ClientCtx<R>, req: CmdSaveResponseReq) -> Result<()> {
Ok(yaak_commands::responses::cmd_save_response(ctx, req).await?)
}
async fn cmd_send_http_request<R: Runtime>(ctx: ClientCtx<R>, req: CmdSendHttpRequestReq) -> Result<HttpResponse> {
Ok(crate::cmd_send_http_request(ctx.window.app_handle().clone(), ctx.window.clone(), req.environment_id.as_deref(), req.cookie_jar_id.as_deref(), req.request_id).await?)
async fn cmd_send_http_request<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdSendHttpRequestReq,
) -> Result<HttpResponse> {
Ok(crate::cmd_send_http_request(
ctx.window.app_handle().clone(),
ctx.window.clone(),
req.environment_id.as_deref(),
req.cookie_jar_id.as_deref(),
req.request_id,
)
.await?)
}
async fn cmd_reload_plugins<R: Runtime>(ctx: ClientCtx<R>, req: CmdReloadPluginsReq) -> Result<Vec<(String, String)>> {
async fn cmd_reload_plugins<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdReloadPluginsReq,
) -> Result<Vec<(String, String)>> {
Ok(yaak_commands::actions::cmd_reload_plugins(ctx, req).await?)
}
async fn cmd_plugin_info<R: Runtime>(ctx: ClientCtx<R>, req: CmdPluginInfoReq) -> Result<PluginMetadata> {
async fn cmd_plugin_info<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdPluginInfoReq,
) -> Result<PluginMetadata> {
Ok(yaak_commands::plugins::cmd_plugin_info(ctx, req).await?)
}
async fn cmd_delete_all_grpc_connections<R: Runtime>(ctx: ClientCtx<R>, req: CmdDeleteAllGrpcConnectionsReq) -> Result<()> {
async fn cmd_delete_all_grpc_connections<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdDeleteAllGrpcConnectionsReq,
) -> Result<()> {
Ok(yaak_commands::models::cmd_delete_all_grpc_connections(ctx, req).await?)
}
async fn cmd_delete_send_history<R: Runtime>(ctx: ClientCtx<R>, req: CmdDeleteSendHistoryReq) -> Result<()> {
async fn cmd_delete_send_history<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdDeleteSendHistoryReq,
) -> Result<()> {
Ok(yaak_commands::models::cmd_delete_send_history(ctx, req).await?)
}
async fn cmd_delete_all_http_responses<R: Runtime>(ctx: ClientCtx<R>, req: CmdDeleteAllHttpResponsesReq) -> Result<()> {
async fn cmd_delete_all_http_responses<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdDeleteAllHttpResponsesReq,
) -> Result<()> {
Ok(yaak_commands::models::cmd_delete_all_http_responses(ctx, req).await?)
}
async fn cmd_get_workspace_meta<R: Runtime>(ctx: ClientCtx<R>, req: CmdGetWorkspaceMetaReq) -> Result<WorkspaceMeta> {
async fn cmd_get_workspace_meta<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdGetWorkspaceMetaReq,
) -> Result<WorkspaceMeta> {
Ok(yaak_commands::models::cmd_get_workspace_meta(ctx, req).await?)
}
async fn cmd_new_child_window<R: Runtime>(ctx: ClientCtx<R>, req: CmdNewChildWindowReq) -> Result<()> {
Ok(crate::cmd_new_child_window(ctx.window.clone(), &req.url, &req.label, &req.title, req.inner_size).await?)
async fn cmd_new_child_window<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdNewChildWindowReq,
) -> Result<()> {
Ok(crate::cmd_new_child_window(
ctx.window.clone(),
&req.url,
&req.label,
&req.title,
req.inner_size,
)
.await?)
}
async fn cmd_new_main_window<R: Runtime>(ctx: ClientCtx<R>, req: CmdNewMainWindowReq) -> Result<()> {
async fn cmd_new_main_window<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdNewMainWindowReq,
) -> Result<()> {
Ok(crate::cmd_new_main_window(ctx.window.app_handle().clone(), &req.url).await?)
}
async fn cmd_check_for_updates<R: Runtime>(ctx: ClientCtx<R>, _req: CmdCheckForUpdatesReq) -> Result<bool> {
Ok(crate::cmd_check_for_updates(ctx.window.clone(), ctx.window.app_handle().state::<Mutex<YaakUpdater>>()).await?)
async fn cmd_check_for_updates<R: Runtime>(
ctx: ClientCtx<R>,
_req: CmdCheckForUpdatesReq,
) -> Result<bool> {
Ok(crate::cmd_check_for_updates(
ctx.window.clone(),
ctx.window.app_handle().state::<Mutex<YaakUpdater>>(),
)
.await?)
}
async fn cmd_decrypt_template<R: Runtime>(ctx: ClientCtx<R>, req: CmdDecryptTemplateReq) -> Result<String> {
async fn cmd_decrypt_template<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdDecryptTemplateReq,
) -> Result<String> {
Ok(yaak_commands::encryption::cmd_decrypt_template(ctx, req).await?)
}
async fn cmd_secure_template<R: Runtime>(ctx: ClientCtx<R>, req: CmdSecureTemplateReq) -> Result<String> {
async fn cmd_secure_template<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdSecureTemplateReq,
) -> Result<String> {
Ok(yaak_commands::encryption::cmd_secure_template(ctx, req).await?)
}
async fn cmd_get_themes<R: Runtime>(ctx: ClientCtx<R>, req: CmdGetThemesReq) -> Result<Vec<GetThemesResponse>> {
async fn cmd_get_themes<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdGetThemesReq,
) -> Result<Vec<GetThemesResponse>> {
Ok(yaak_commands::templates::cmd_get_themes(ctx, req).await?)
}
async fn cmd_enable_encryption<R: Runtime>(ctx: ClientCtx<R>, req: CmdEnableEncryptionReq) -> Result<()> {
async fn cmd_enable_encryption<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdEnableEncryptionReq,
) -> Result<()> {
Ok(yaak_commands::encryption::cmd_enable_encryption(ctx, req).await?)
}
async fn cmd_reveal_workspace_key<R: Runtime>(ctx: ClientCtx<R>, req: CmdRevealWorkspaceKeyReq) -> Result<String> {
async fn cmd_reveal_workspace_key<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdRevealWorkspaceKeyReq,
) -> Result<String> {
Ok(yaak_commands::encryption::cmd_reveal_workspace_key(ctx, req).await?)
}
async fn cmd_set_workspace_key<R: Runtime>(ctx: ClientCtx<R>, req: CmdSetWorkspaceKeyReq) -> Result<()> {
async fn cmd_set_workspace_key<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdSetWorkspaceKeyReq,
) -> Result<()> {
Ok(yaak_commands::encryption::cmd_set_workspace_key(ctx, req).await?)
}
async fn cmd_disable_encryption<R: Runtime>(ctx: ClientCtx<R>, req: CmdDisableEncryptionReq) -> Result<()> {
async fn cmd_disable_encryption<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdDisableEncryptionReq,
) -> Result<()> {
Ok(yaak_commands::encryption::cmd_disable_encryption(ctx, req).await?)
}
async fn cmd_default_headers<R: Runtime>(ctx: ClientCtx<R>, req: CmdDefaultHeadersReq) -> Result<Vec<HttpRequestHeader>> {
async fn cmd_default_headers<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdDefaultHeadersReq,
) -> Result<Vec<HttpRequestHeader>> {
Ok(yaak_commands::models::cmd_default_headers(ctx, req).await?)
}
@@ -649,27 +845,45 @@ async fn models_delete<R: Runtime>(ctx: ClientCtx<R>, req: ModelsDeleteReq) -> R
Ok(deleted?)
}
async fn models_duplicate<R: Runtime>(ctx: ClientCtx<R>, req: ModelsDuplicateReq) -> Result<String> {
async fn models_duplicate<R: Runtime>(
ctx: ClientCtx<R>,
req: ModelsDuplicateReq,
) -> Result<String> {
Ok(yaak_commands::models::models_duplicate(ctx, req).await?)
}
async fn models_websocket_events<R: Runtime>(ctx: ClientCtx<R>, req: ModelsWebsocketEventsReq) -> Result<Vec<WebsocketEvent>> {
async fn models_websocket_events<R: Runtime>(
ctx: ClientCtx<R>,
req: ModelsWebsocketEventsReq,
) -> Result<Vec<WebsocketEvent>> {
Ok(yaak_commands::models::models_websocket_events(ctx, req).await?)
}
async fn models_grpc_events<R: Runtime>(ctx: ClientCtx<R>, req: ModelsGrpcEventsReq) -> Result<Vec<GrpcEvent>> {
async fn models_grpc_events<R: Runtime>(
ctx: ClientCtx<R>,
req: ModelsGrpcEventsReq,
) -> Result<Vec<GrpcEvent>> {
Ok(yaak_commands::models::models_grpc_events(ctx, req).await?)
}
async fn models_get_settings<R: Runtime>(ctx: ClientCtx<R>, req: ModelsGetSettingsReq) -> Result<Settings> {
async fn models_get_settings<R: Runtime>(
ctx: ClientCtx<R>,
req: ModelsGetSettingsReq,
) -> Result<Settings> {
Ok(yaak_commands::models::models_get_settings(ctx, req).await?)
}
async fn models_get_graphql_introspection<R: Runtime>(ctx: ClientCtx<R>, req: ModelsGetGraphqlIntrospectionReq) -> Result<Option<GraphQlIntrospection>> {
async fn models_get_graphql_introspection<R: Runtime>(
ctx: ClientCtx<R>,
req: ModelsGetGraphqlIntrospectionReq,
) -> Result<Option<GraphQlIntrospection>> {
Ok(yaak_commands::models::models_get_graphql_introspection(ctx, req).await?)
}
async fn models_upsert_graphql_introspection<R: Runtime>(ctx: ClientCtx<R>, req: ModelsUpsertGraphqlIntrospectionReq) -> Result<GraphQlIntrospection> {
async fn models_upsert_graphql_introspection<R: Runtime>(
ctx: ClientCtx<R>,
req: ModelsUpsertGraphqlIntrospectionReq,
) -> Result<GraphQlIntrospection> {
Ok(yaak_commands::models::models_upsert_graphql_introspection(ctx, req).await?)
}
@@ -680,7 +894,10 @@ async fn models_upsert_graphql_introspection<R: Runtime>(ctx: ClientCtx<R>, req:
/// freezes the app"). Escape sequences sidestep it. This is a quirk of the
/// webview transport, not of the data, so it lives in the adapter rather than
/// the shared handler.
async fn models_workspace_models<R: Runtime>(ctx: ClientCtx<R>, req: ModelsWorkspaceModelsReq) -> Result<String> {
async fn models_workspace_models<R: Runtime>(
ctx: ClientCtx<R>,
req: ModelsWorkspaceModelsReq,
) -> Result<String> {
let json = yaak_commands::models::models_workspace_models(ctx, req).await?;
Ok(escape_str_for_webview(&json))
}
@@ -706,7 +923,10 @@ fn escape_str_for_webview(input: &str) -> String {
.collect()
}
async fn cmd_git_checkout<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitCheckoutReq) -> Result<String> {
async fn cmd_git_checkout<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitCheckoutReq,
) -> Result<String> {
Ok(crate::git_ext::cmd_git_checkout(&req.dir, &req.branch, req.force).await?)
}
@@ -714,31 +934,52 @@ async fn cmd_git_branch<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitBranchReq) ->
Ok(crate::git_ext::cmd_git_branch(&req.dir, &req.branch, req.base.as_deref()).await?)
}
async fn cmd_git_delete_branch<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitDeleteBranchReq) -> Result<BranchDeleteResult> {
async fn cmd_git_delete_branch<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitDeleteBranchReq,
) -> Result<BranchDeleteResult> {
Ok(crate::git_ext::cmd_git_delete_branch(&req.dir, &req.branch, req.force).await?)
}
async fn cmd_git_delete_remote_branch<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitDeleteRemoteBranchReq) -> Result<()> {
async fn cmd_git_delete_remote_branch<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitDeleteRemoteBranchReq,
) -> Result<()> {
Ok(crate::git_ext::cmd_git_delete_remote_branch(&req.dir, &req.branch).await?)
}
async fn cmd_git_merge_branch<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitMergeBranchReq) -> Result<()> {
async fn cmd_git_merge_branch<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitMergeBranchReq,
) -> Result<()> {
Ok(crate::git_ext::cmd_git_merge_branch(&req.dir, &req.branch).await?)
}
async fn cmd_git_rename_branch<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitRenameBranchReq) -> Result<()> {
async fn cmd_git_rename_branch<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitRenameBranchReq,
) -> Result<()> {
Ok(crate::git_ext::cmd_git_rename_branch(&req.dir, &req.old_name, &req.new_name).await?)
}
async fn cmd_git_status<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitStatusReq) -> Result<GitStatusSummary> {
async fn cmd_git_status<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitStatusReq,
) -> Result<GitStatusSummary> {
Ok(crate::git_ext::cmd_git_status(&req.dir).await?)
}
async fn cmd_git_branch_info<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitBranchInfoReq) -> Result<GitBranchInfo> {
async fn cmd_git_branch_info<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitBranchInfoReq,
) -> Result<GitBranchInfo> {
Ok(crate::git_ext::cmd_git_branch_info(&req.dir).await?)
}
async fn cmd_git_worktree_status<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitWorktreeStatusReq) -> Result<GitWorktreeStatus> {
async fn cmd_git_worktree_status<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitWorktreeStatusReq,
) -> Result<GitWorktreeStatus> {
Ok(crate::git_ext::cmd_git_worktree_status(&req.dir).await?)
}
@@ -746,15 +987,25 @@ async fn cmd_git_log<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitLogReq) -> Resul
Ok(crate::git_ext::cmd_git_log(&req.dir).await?)
}
async fn cmd_git_log_for_file<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitLogForFileReq) -> Result<Vec<GitCommit>> {
async fn cmd_git_log_for_file<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitLogForFileReq,
) -> Result<Vec<GitCommit>> {
Ok(crate::git_ext::cmd_git_log_for_file(&req.dir, req.rela_path).await?)
}
async fn cmd_git_file_diff_for_commit<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitFileDiffForCommitReq) -> Result<GitFileDiff> {
Ok(crate::git_ext::cmd_git_file_diff_for_commit(&req.dir, &req.commit_oid, req.rela_path).await?)
async fn cmd_git_file_diff_for_commit<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitFileDiffForCommitReq,
) -> Result<GitFileDiff> {
Ok(crate::git_ext::cmd_git_file_diff_for_commit(&req.dir, &req.commit_oid, req.rela_path)
.await?)
}
async fn cmd_git_initialize<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitInitializeReq) -> Result<()> {
async fn cmd_git_initialize<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitInitializeReq,
) -> Result<()> {
Ok(crate::git_ext::cmd_git_initialize(&req.dir).await?)
}
@@ -778,11 +1029,17 @@ async fn cmd_git_pull<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitPullReq) -> Res
Ok(crate::git_ext::cmd_git_pull(&req.dir).await?)
}
async fn cmd_git_pull_force_reset<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitPullForceResetReq) -> Result<PullResult> {
async fn cmd_git_pull_force_reset<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitPullForceResetReq,
) -> Result<PullResult> {
Ok(crate::git_ext::cmd_git_pull_force_reset(&req.dir, &req.remote, &req.branch).await?)
}
async fn cmd_git_pull_merge<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitPullMergeReq) -> Result<PullResult> {
async fn cmd_git_pull_merge<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitPullMergeReq,
) -> Result<PullResult> {
Ok(crate::git_ext::cmd_git_pull_merge(&req.dir, &req.remote, &req.branch).await?)
}
@@ -794,27 +1051,47 @@ async fn cmd_git_unstage<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitUnstageReq)
Ok(crate::git_ext::cmd_git_unstage(&req.dir, req.rela_paths).await?)
}
async fn cmd_git_reset_changes<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitResetChangesReq) -> Result<()> {
async fn cmd_git_reset_changes<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitResetChangesReq,
) -> Result<()> {
Ok(crate::git_ext::cmd_git_reset_changes(&req.dir).await?)
}
async fn cmd_git_restore_files<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitRestoreFilesReq) -> Result<()> {
async fn cmd_git_restore_files<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitRestoreFilesReq,
) -> Result<()> {
Ok(crate::git_ext::cmd_git_restore_files(&req.dir, req.rela_paths).await?)
}
async fn cmd_git_restore_file_from_commit<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitRestoreFileFromCommitReq) -> Result<()> {
Ok(crate::git_ext::cmd_git_restore_file_from_commit(&req.dir, &req.commit_oid, req.rela_path).await?)
async fn cmd_git_restore_file_from_commit<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitRestoreFileFromCommitReq,
) -> Result<()> {
Ok(crate::git_ext::cmd_git_restore_file_from_commit(&req.dir, &req.commit_oid, req.rela_path)
.await?)
}
async fn cmd_git_add_credential<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitAddCredentialReq) -> Result<()> {
Ok(crate::git_ext::cmd_git_add_credential(&req.remote_url, &req.username, &req.password).await?)
async fn cmd_git_add_credential<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitAddCredentialReq,
) -> Result<()> {
Ok(crate::git_ext::cmd_git_add_credential(&req.remote_url, &req.username, &req.password)
.await?)
}
async fn cmd_git_remotes<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitRemotesReq) -> Result<Vec<GitRemote>> {
async fn cmd_git_remotes<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitRemotesReq,
) -> Result<Vec<GitRemote>> {
Ok(crate::git_ext::cmd_git_remotes(&req.dir).await?)
}
async fn cmd_git_add_remote<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitAddRemoteReq) -> Result<GitRemote> {
async fn cmd_git_add_remote<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdGitAddRemoteReq,
) -> Result<GitRemote> {
Ok(crate::git_ext::cmd_git_add_remote(&req.dir, &req.name, &req.url).await?)
}
@@ -822,58 +1099,130 @@ async fn cmd_git_rm_remote<R: Runtime>(_ctx: ClientCtx<R>, req: CmdGitRmRemoteRe
Ok(crate::git_ext::cmd_git_rm_remote(&req.dir, &req.name).await?)
}
async fn cmd_sync_calculate<R: Runtime>(ctx: ClientCtx<R>, req: CmdSyncCalculateReq) -> Result<Vec<SyncOp>> {
Ok(crate::sync_ext::cmd_sync_calculate(ctx.window.app_handle().clone(), &req.workspace_id, &req.sync_dir).await?)
async fn cmd_sync_calculate<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdSyncCalculateReq,
) -> Result<Vec<SyncOp>> {
Ok(crate::sync_ext::cmd_sync_calculate(
ctx.window.app_handle().clone(),
&req.workspace_id,
&req.sync_dir,
)
.await?)
}
async fn cmd_sync_calculate_fs<R: Runtime>(_ctx: ClientCtx<R>, req: CmdSyncCalculateFsReq) -> Result<Vec<SyncOp>> {
async fn cmd_sync_calculate_fs<R: Runtime>(
_ctx: ClientCtx<R>,
req: CmdSyncCalculateFsReq,
) -> Result<Vec<SyncOp>> {
Ok(crate::sync_ext::cmd_sync_calculate_fs(&req.dir).await?)
}
async fn cmd_sync_apply<R: Runtime>(ctx: ClientCtx<R>, req: CmdSyncApplyReq) -> Result<()> {
Ok(crate::sync_ext::cmd_sync_apply(ctx.window.app_handle().clone(), req.sync_ops, &req.sync_dir, &req.workspace_id).await?)
Ok(crate::sync_ext::cmd_sync_apply(
ctx.window.app_handle().clone(),
req.sync_ops,
&req.sync_dir,
&req.workspace_id,
)
.await?)
}
async fn cmd_ws_delete_connections<R: Runtime>(ctx: ClientCtx<R>, req: CmdWsDeleteConnectionsReq) -> Result<()> {
async fn cmd_ws_delete_connections<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdWsDeleteConnectionsReq,
) -> Result<()> {
Ok(yaak_commands::models::cmd_ws_delete_connections(ctx, req).await?)
}
async fn cmd_ws_send<R: Runtime>(ctx: ClientCtx<R>, req: CmdWsSendReq) -> Result<WebsocketConnection> {
Ok(crate::ws_ext::cmd_ws_send(&req.connection_id, req.environment_id.as_deref(), ctx.window.app_handle().clone(), ctx.window.clone(), ctx.window.app_handle().state::<Mutex<WebsocketManager>>()).await?)
async fn cmd_ws_send<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdWsSendReq,
) -> Result<WebsocketConnection> {
Ok(crate::ws_ext::cmd_ws_send(
&req.connection_id,
req.environment_id.as_deref(),
ctx.window.app_handle().clone(),
ctx.window.clone(),
ctx.window.app_handle().state::<Mutex<WebsocketManager>>(),
)
.await?)
}
async fn cmd_ws_close<R: Runtime>(ctx: ClientCtx<R>, req: CmdWsCloseReq) -> Result<WebsocketConnection> {
Ok(crate::ws_ext::cmd_ws_close(&req.connection_id, ctx.window.app_handle().clone(), ctx.window.clone(), ctx.window.app_handle().state::<Mutex<WebsocketManager>>()).await?)
async fn cmd_ws_close<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdWsCloseReq,
) -> Result<WebsocketConnection> {
Ok(crate::ws_ext::cmd_ws_close(
&req.connection_id,
ctx.window.app_handle().clone(),
ctx.window.clone(),
ctx.window.app_handle().state::<Mutex<WebsocketManager>>(),
)
.await?)
}
async fn cmd_ws_connect<R: Runtime>(ctx: ClientCtx<R>, req: CmdWsConnectReq) -> Result<WebsocketConnection> {
Ok(crate::ws_ext::cmd_ws_connect(&req.request_id, req.environment_id.as_deref(), req.cookie_jar_id.as_deref(), ctx.window.app_handle().clone(), ctx.window.clone(), ctx.window.app_handle().state::<Mutex<WebsocketManager>>()).await?)
async fn cmd_ws_connect<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdWsConnectReq,
) -> Result<WebsocketConnection> {
Ok(crate::ws_ext::cmd_ws_connect(
&req.request_id,
req.environment_id.as_deref(),
req.cookie_jar_id.as_deref(),
ctx.window.app_handle().clone(),
ctx.window.clone(),
ctx.window.app_handle().state::<Mutex<WebsocketManager>>(),
)
.await?)
}
async fn cmd_plugins_search<R: Runtime>(ctx: ClientCtx<R>, req: CmdPluginsSearchReq) -> Result<PluginSearchResponse> {
async fn cmd_plugins_search<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdPluginsSearchReq,
) -> Result<PluginSearchResponse> {
Ok(crate::plugins_ext::cmd_plugins_search(ctx.window.app_handle().clone(), &req.query).await?)
}
async fn cmd_plugins_install<R: Runtime>(ctx: ClientCtx<R>, req: CmdPluginsInstallReq) -> Result<()> {
async fn cmd_plugins_install<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdPluginsInstallReq,
) -> Result<()> {
Ok(crate::plugins_ext::cmd_plugins_install(ctx.window.clone(), &req.name, req.version).await?)
}
async fn cmd_plugins_install_from_directory<R: Runtime>(ctx: ClientCtx<R>, req: CmdPluginsInstallFromDirectoryReq) -> Result<Plugin> {
Ok(crate::plugins_ext::cmd_plugins_install_from_directory(ctx.window.clone(), &req.directory).await?)
async fn cmd_plugins_install_from_directory<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdPluginsInstallFromDirectoryReq,
) -> Result<Plugin> {
Ok(crate::plugins_ext::cmd_plugins_install_from_directory(ctx.window.clone(), &req.directory)
.await?)
}
async fn cmd_plugins_uninstall<R: Runtime>(ctx: ClientCtx<R>, req: CmdPluginsUninstallReq) -> Result<Plugin> {
async fn cmd_plugins_uninstall<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdPluginsUninstallReq,
) -> Result<Plugin> {
Ok(crate::plugins_ext::cmd_plugins_uninstall(&req.plugin_id, ctx.window.clone()).await?)
}
async fn cmd_plugin_init_errors<R: Runtime>(ctx: ClientCtx<R>, req: CmdPluginInitErrorsReq) -> Result<Vec<(String, String)>> {
async fn cmd_plugin_init_errors<R: Runtime>(
ctx: ClientCtx<R>,
req: CmdPluginInitErrorsReq,
) -> Result<Vec<(String, String)>> {
Ok(yaak_commands::plugins::cmd_plugin_init_errors(ctx, req).await?)
}
async fn cmd_plugins_updates<R: Runtime>(ctx: ClientCtx<R>, _req: CmdPluginsUpdatesReq) -> Result<PluginUpdatesResponse> {
async fn cmd_plugins_updates<R: Runtime>(
ctx: ClientCtx<R>,
_req: CmdPluginsUpdatesReq,
) -> Result<PluginUpdatesResponse> {
Ok(crate::plugins_ext::cmd_plugins_updates(ctx.window.app_handle().clone()).await?)
}
async fn cmd_plugins_update_all<R: Runtime>(ctx: ClientCtx<R>, _req: CmdPluginsUpdateAllReq) -> Result<Vec<PluginNameVersion>> {
async fn cmd_plugins_update_all<R: Runtime>(
ctx: ClientCtx<R>,
_req: CmdPluginsUpdateAllReq,
) -> Result<Vec<PluginNameVersion>> {
Ok(crate::plugins_ext::cmd_plugins_update_all(ctx.window.clone()).await?)
}
+11 -9
View File
@@ -12,8 +12,8 @@ use tokio::sync::watch;
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,
get_fs_candidates,
FsCandidate, SyncOp, apply_db_sync_ops, apply_fs_sync_ops, apply_sync_state_ops,
compute_sync_ops, get_db_candidates, get_fs_candidates,
};
use yaak_sync::watch::{WatchEvent, watch_directory};
@@ -26,10 +26,9 @@ pub(crate) async fn cmd_sync_calculate<R: Runtime>(
return Err(InvalidSyncDirectory(sync_dir.to_string_lossy().to_string()).into());
}
let db = app_handle.db();
let version = app_handle.package_info().version.to_string();
// Scoped so the pooled connection is back before the directory walk below, which is
// as slow as the sync directory is large.
let db_candidates = get_db_candidates(&app_handle.db()?, &version, workspace_id, sync_dir)?;
let db_candidates = get_db_candidates(&db, &version, workspace_id, sync_dir)?;
let fs_candidates = get_fs_candidates(sync_dir)?
.into_iter()
// Only keep items in the same workspace
@@ -50,11 +49,14 @@ pub(crate) async fn cmd_sync_apply<R: Runtime>(
sync_dir: &Path,
workspace_id: &str,
) -> Result<()> {
let db = app_handle.db()?;
// Files first, so the write transaction never waits on the filesystem
let pending = apply_fs_sync_ops(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(())
app_handle.db_manager().with_tx(|tx| {
let sync_state_ops = apply_db_sync_ops(tx, &blobs, workspace_id, sync_dir, pending)?;
apply_sync_state_ops(tx, workspace_id, sync_dir, sync_state_ops)?;
Ok(())
})
}
pub(crate) async fn sync_watch<R, F>(
+1 -1
View File
@@ -75,7 +75,7 @@ impl YaakUpdater {
auto_download: bool,
update_trigger: UpdateTrigger,
) -> Result<bool> {
let settings = window.db()?.get_settings();
let settings = window.db().get_settings();
let update_key = format!("{:x}", md5::compute(settings.id));
self.last_check = Some(Instant::now());
+148 -139
View File
@@ -36,25 +36,27 @@ pub async fn cmd_ws_send<R: Runtime>(
window: WebviewWindow<R>,
ws_manager: State<'_, Mutex<WebsocketManager>>,
) -> Result<WebsocketConnection> {
let connection = app_handle.db()?.get_websocket_connection(connection_id)?;
let connection = app_handle.db().get_websocket_connection(connection_id)?;
match send_websocket_message(&connection, environment_id, &app_handle, &window, &ws_manager)
.await
{
Ok(connection) => Ok(connection),
Err(e) => {
app_handle.db()?.upsert_websocket_event(
&WebsocketEvent {
connection_id: connection.id.clone(),
request_id: connection.request_id.clone(),
workspace_id: connection.workspace_id.clone(),
is_server: false,
message_type: WebsocketEventType::Error,
message: e.to_string().into(),
..Default::default()
},
&UpdateSource::from_window_label(window.label()),
)?;
app_handle.with_tx(|tx| {
tx.upsert_websocket_event(
&WebsocketEvent {
connection_id: connection.id.clone(),
request_id: connection.request_id.clone(),
workspace_id: connection.workspace_id.clone(),
is_server: false,
message_type: WebsocketEventType::Error,
message: e.to_string().into(),
..Default::default()
},
&UpdateSource::from_window_label(window.label()),
)
})?;
Ok(connection)
}
@@ -68,14 +70,14 @@ async fn send_websocket_message<R: Runtime>(
window: &WebviewWindow<R>,
ws_manager: &Mutex<WebsocketManager>,
) -> Result<WebsocketConnection> {
let unrendered_request = app_handle.db()?.get_websocket_request(&connection.request_id)?;
let environment_chain = app_handle.db()?.resolve_environments(
let unrendered_request = app_handle.db().get_websocket_request(&connection.request_id)?;
let environment_chain = app_handle.db().resolve_environments(
&unrendered_request.workspace_id,
unrendered_request.folder_id.as_deref(),
environment_id,
)?;
let (resolved_request, _auth_context_id) =
resolve_websocket_request(&window.db()?, &unrendered_request)?;
resolve_websocket_request(&window.db(), &unrendered_request)?;
let plugin_manager = Arc::new(crate::plugins_ext::plugin_manager(app_handle).await?);
let encryption_manager = Arc::new((*app_handle.state::<EncryptionManager>()).clone());
let request = render_websocket_request(
@@ -96,18 +98,20 @@ async fn send_websocket_message<R: Runtime>(
let mut ws_manager = ws_manager.lock().await;
ws_manager.send(&connection.id, Message::Text(message.clone().into())).await?;
app_handle.db()?.upsert_websocket_event(
&WebsocketEvent {
connection_id: connection.id.clone(),
request_id: request.id.clone(),
workspace_id: connection.workspace_id.clone(),
is_server: false,
message_type: WebsocketEventType::Text,
message: message.into(),
..Default::default()
},
&UpdateSource::from_window_label(window.label()),
)?;
app_handle.with_tx(|tx| {
tx.upsert_websocket_event(
&WebsocketEvent {
connection_id: connection.id.clone(),
request_id: request.id.clone(),
workspace_id: connection.workspace_id.clone(),
is_server: false,
message_type: WebsocketEventType::Text,
message: message.into(),
..Default::default()
},
&UpdateSource::from_window_label(window.label()),
)
})?;
Ok(connection.clone())
}
@@ -118,14 +122,13 @@ pub async fn cmd_ws_close<R: Runtime>(
window: WebviewWindow<R>,
ws_manager: State<'_, Mutex<WebsocketManager>>,
) -> Result<WebsocketConnection> {
let connection = {
let db = app_handle.db()?;
let connection = db.get_websocket_connection(connection_id)?;
db.upsert_websocket_connection(
let connection = app_handle.with_tx(|tx| {
let connection = tx.get_websocket_connection(connection_id)?;
tx.upsert_websocket_connection(
&WebsocketConnection { state: WebsocketConnectionState::Closing, ..connection },
&UpdateSource::from_window_label(window.label()),
)?
};
)
})?;
let mut ws_manager = ws_manager.lock().await;
if let Err(e) = ws_manager.close(&connection.id).await {
@@ -143,17 +146,17 @@ pub async fn cmd_ws_connect<R: Runtime>(
window: WebviewWindow<R>,
ws_manager: State<'_, Mutex<WebsocketManager>>,
) -> Result<WebsocketConnection> {
let unrendered_request = app_handle.db()?.get_websocket_request(request_id)?;
let environment_chain = app_handle.db()?.resolve_environments(
let unrendered_request = app_handle.db().get_websocket_request(request_id)?;
let environment_chain = app_handle.db().resolve_environments(
&unrendered_request.workspace_id,
unrendered_request.folder_id.as_deref(),
environment_id,
)?;
let resolved_settings =
app_handle.db()?.resolve_settings_for_websocket_request(&unrendered_request)?;
let settings = app_handle.db()?.get_settings();
app_handle.db().resolve_settings_for_websocket_request(&unrendered_request)?;
let settings = app_handle.db().get_settings();
let (resolved_request, auth_context_id) =
resolve_websocket_request(&window.db()?, &unrendered_request)?;
resolve_websocket_request(&window.db(), &unrendered_request)?;
let plugin_manager = Arc::new(crate::plugins_ext::plugin_manager(&app_handle).await?);
let encryption_manager = Arc::new((*app_handle.state::<EncryptionManager>()).clone());
let request = render_websocket_request(
@@ -169,14 +172,16 @@ pub async fn cmd_ws_connect<R: Runtime>(
)
.await?;
let connection = app_handle.db()?.upsert_websocket_connection(
&WebsocketConnection {
workspace_id: request.workspace_id.clone(),
request_id: request_id.to_string(),
..Default::default()
},
&UpdateSource::from_window_label(window.label()),
)?;
let connection = app_handle.with_tx(|tx| {
tx.upsert_websocket_connection(
&WebsocketConnection {
workspace_id: request.workspace_id.clone(),
request_id: request_id.to_string(),
..Default::default()
},
&UpdateSource::from_window_label(window.label()),
)
})?;
let (mut url, url_parameters) = apply_path_placeholders(&request.url, &request.url_parameters);
if !url.starts_with("ws://") && !url.starts_with("wss://") {
@@ -187,14 +192,16 @@ pub async fn cmd_ws_connect<R: Runtime>(
let mut url = match Url::parse(&url) {
Ok(url) => url,
Err(e) => {
return Ok(app_handle.db()?.upsert_websocket_connection(
&WebsocketConnection {
error: Some(format!("Failed to parse URL {}", e.to_string())),
state: WebsocketConnectionState::Closed,
..connection
},
&UpdateSource::from_window_label(window.label()),
)?);
return Ok(app_handle.with_tx(|tx| {
tx.upsert_websocket_connection(
&WebsocketConnection {
error: Some(format!("Failed to parse URL {}", e.to_string())),
state: WebsocketConnectionState::Closed,
..connection
},
&UpdateSource::from_window_label(window.label()),
)
})?);
}
};
@@ -268,7 +275,7 @@ pub async fn cmd_ws_connect<R: Runtime>(
resolved_settings.send_cookies.value || resolved_settings.store_cookies.value,
cookie_jar_id,
) {
(true, Some(id)) => Some(app_handle.db()?.get_cookie_jar(id)?),
(true, Some(id)) => Some(app_handle.db().get_cookie_jar(id)?),
_ => None,
};
let cookie_store =
@@ -321,28 +328,32 @@ pub async fn cmd_ws_connect<R: Runtime>(
{
Ok(r) => r,
Err(e) => {
return Ok(app_handle.db()?.upsert_websocket_connection(
&WebsocketConnection {
error: Some(e.to_string()),
state: WebsocketConnectionState::Closed,
..connection
},
&UpdateSource::from_window_label(window.label()),
)?);
return Ok(app_handle.with_tx(|tx| {
tx.upsert_websocket_connection(
&WebsocketConnection {
error: Some(e.to_string()),
state: WebsocketConnectionState::Closed,
..connection
},
&UpdateSource::from_window_label(window.label()),
)
})?);
}
};
app_handle.db()?.upsert_websocket_event(
&WebsocketEvent {
connection_id: connection.id.clone(),
request_id: request.id.clone(),
workspace_id: connection.workspace_id.clone(),
is_server: false,
message_type: WebsocketEventType::Open,
..Default::default()
},
&UpdateSource::from_window_label(window.label()),
)?;
app_handle.with_tx(|tx| {
tx.upsert_websocket_event(
&WebsocketEvent {
connection_id: connection.id.clone(),
request_id: request.id.clone(),
workspace_id: connection.workspace_id.clone(),
is_server: false,
message_type: WebsocketEventType::Open,
..Default::default()
},
&UpdateSource::from_window_label(window.label()),
)
})?;
let response_headers = response
.headers()
@@ -366,20 +377,22 @@ pub async fn cmd_ws_connect<R: Runtime>(
if !set_cookie_headers.is_empty() {
store.store_cookies_from_response(&convert_ws_url_to_http(&url), &set_cookie_headers);
cookie_jar.cookies = store.get_all_cookies();
app_handle.db()?.upsert_cookie_jar(cookie_jar, &UpdateSource::Background)?;
app_handle.with_tx(|tx| tx.upsert_cookie_jar(cookie_jar, &UpdateSource::Background))?;
}
}
let connection = app_handle.db()?.upsert_websocket_connection(
&WebsocketConnection {
state: WebsocketConnectionState::Connected,
headers: response_headers,
status: response.status().as_u16() as i32,
url: request.url.clone(),
..connection
},
&UpdateSource::from_window_label(window.label()),
)?;
let connection = app_handle.with_tx(|tx| {
tx.upsert_websocket_connection(
&WebsocketConnection {
state: WebsocketConnectionState::Connected,
headers: response_headers,
status: response.status().as_u16() as i32,
url: request.url.clone(),
..connection
},
&UpdateSource::from_window_label(window.label()),
)
})?;
{
let connection_id = connection.id.clone();
@@ -394,72 +407,68 @@ pub async fn cmd_ws_connect<R: Runtime>(
has_written_close = true;
}
let event = || {
app_handle.db()?.upsert_websocket_event(
&WebsocketEvent {
connection_id: connection_id.clone(),
request_id: request_id.clone(),
workspace_id: workspace_id.clone(),
is_server: true,
message_type: match message {
Message::Text(_) => WebsocketEventType::Text,
Message::Binary(_) => WebsocketEventType::Binary,
Message::Ping(_) => WebsocketEventType::Ping,
Message::Pong(_) => WebsocketEventType::Pong,
Message::Close(_) => WebsocketEventType::Close,
// Raw frame will never happen during a read
Message::Frame(_) => WebsocketEventType::Frame,
app_handle
.with_tx(|tx| {
tx.upsert_websocket_event(
&WebsocketEvent {
connection_id: connection_id.clone(),
request_id: request_id.clone(),
workspace_id: workspace_id.clone(),
is_server: true,
message_type: match message {
Message::Text(_) => WebsocketEventType::Text,
Message::Binary(_) => WebsocketEventType::Binary,
Message::Ping(_) => WebsocketEventType::Ping,
Message::Pong(_) => WebsocketEventType::Pong,
Message::Close(_) => WebsocketEventType::Close,
// Raw frame will never happen during a read
Message::Frame(_) => WebsocketEventType::Frame,
},
message: message.into_data().into(),
..Default::default()
},
message: message.into_data().into(),
..Default::default()
},
&UpdateSource::from_window_label(&window_label),
)
};
if let Err(e) = event() {
warn!("Failed to store WebSocket event: {e}");
}
&UpdateSource::from_window_label(&window_label),
)
})
.unwrap();
}
info!("Websocket connection closed");
if !has_written_close {
let close_event = app_handle.db().and_then(|db| {
db.upsert_websocket_event(
&WebsocketEvent {
connection_id: connection_id.clone(),
request_id: request_id.clone(),
workspace_id: workspace_id.clone(),
is_server: true,
message_type: WebsocketEventType::Close,
..Default::default()
app_handle
.with_tx(|tx| {
tx.upsert_websocket_event(
&WebsocketEvent {
connection_id: connection_id.clone(),
request_id: request_id.clone(),
workspace_id: workspace_id.clone(),
is_server: true,
message_type: WebsocketEventType::Close,
..Default::default()
},
&UpdateSource::from_window_label(&window_label),
)
})
.unwrap();
}
app_handle
.with_tx(|tx| {
tx.upsert_websocket_connection(
&WebsocketConnection {
workspace_id: request.workspace_id.clone(),
request_id: request_id.to_string(),
state: WebsocketConnectionState::Closed,
..connection
},
&UpdateSource::from_window_label(&window_label),
)
});
if let Err(e) = close_event {
warn!("Failed to store WebSocket close event: {e}");
}
}
let closed = app_handle.db().and_then(|db| {
db.upsert_websocket_connection(
&WebsocketConnection {
workspace_id: request.workspace_id.clone(),
request_id: request_id.to_string(),
state: WebsocketConnectionState::Closed,
..connection
},
&UpdateSource::from_window_label(&window_label),
)
});
if let Err(e) = closed {
warn!("Failed to mark WebSocket connection closed: {e}");
}
})
.unwrap();
});
}
Ok(connection)
}
/// Convert WS URL to HTTP URL for cookie filtering
/// WebSocket upgrade requests are HTTP requests initially, so HttpOnly cookies should apply
fn convert_ws_url_to_http(ws_url: &Url) -> Url {
+35 -19
View File
@@ -9,21 +9,33 @@ use tauri::{AppHandle, Emitter, Manager, Runtime, WebviewWindow, is_dev};
use ts_rs::TS;
use yaak_api::{ApiClientKind, yaak_api_client};
use yaak_common::platform::get_os_str;
use yaak_models::client_db::ClientDb;
use yaak_models::client_db::{ClientDb, WriteDb};
use yaak_models::query_manager::QueryManager;
use yaak_models::util::UpdateSource;
/// Extension trait for accessing the QueryManager from Tauri Manager types.
/// This is needed temporarily until all crates are refactored to not use Tauri.
trait QueryManagerExt<'a, R> {
fn db(&'a self) -> yaak_models::error::Result<ClientDb<'a>>;
fn db(&'a self) -> ClientDb<'a>;
fn with_tx<T>(
&'a self,
func: impl FnOnce(&WriteDb) -> yaak_models::error::Result<T>,
) -> yaak_models::error::Result<T>;
}
impl<'a, R: Runtime, M: Manager<R>> QueryManagerExt<'a, R> for M {
fn db(&'a self) -> yaak_models::error::Result<ClientDb<'a>> {
fn db(&'a self) -> ClientDb<'a> {
let qm = self.state::<QueryManager>();
qm.inner().connect()
}
fn with_tx<T>(
&'a self,
func: impl FnOnce(&WriteDb) -> yaak_models::error::Result<T>,
) -> yaak_models::error::Result<T> {
let qm = self.state::<QueryManager>();
qm.inner().with_tx(func)
}
}
const KV_NAMESPACE: &str = "license";
@@ -137,12 +149,17 @@ pub async fn activate_license<R: Runtime>(
}
let body: ActivateLicenseResponsePayload = response.json().await?;
window.app_handle().db()?.set_key_value_str(
KV_ACTIVATION_ID_KEY,
KV_NAMESPACE,
body.activation_id.as_str(),
&UpdateSource::from_window_label(window.label()),
);
if let Err(e) = window.app_handle().with_tx(|tx| {
tx.set_key_value_str(
KV_ACTIVATION_ID_KEY,
KV_NAMESPACE,
body.activation_id.as_str(),
&UpdateSource::from_window_label(window.label()),
);
Ok(())
}) {
warn!("Failed to store license activation: {e}");
}
if let Err(e) = window.emit("license-activated", true) {
warn!("Failed to emit check-license event: {}", e);
@@ -172,11 +189,13 @@ pub async fn deactivate_license<R: Runtime>(window: &WebviewWindow<R>) -> Result
return Err(ServerError);
}
app_handle.db()?.delete_key_value(
KV_ACTIVATION_ID_KEY,
KV_NAMESPACE,
&UpdateSource::from_window_label(window.label()),
)?;
app_handle.with_tx(|tx| {
tx.delete_key_value(
KV_ACTIVATION_ID_KEY,
KV_NAMESPACE,
&UpdateSource::from_window_label(window.label()),
)
})?;
if let Err(e) = app_handle.emit("license-deactivated", true) {
warn!("Failed to emit deactivate-license event: {}", e);
@@ -191,7 +210,7 @@ pub async fn check_license<R: Runtime>(window: &WebviewWindow<R>) -> Result<Lice
CheckActivationRequestPayload { app_platform: get_os_str().to_string(), app_version };
let activation_id = get_activation_id(window.app_handle()).await;
let settings = window.db()?.get_settings();
let settings = window.db().get_settings();
let trial_end = settings.created_at.add(Duration::from_secs(TRIAL_SECONDS)).and_utc();
let has_activation_id = !activation_id.is_empty();
@@ -238,8 +257,5 @@ fn build_url(path: &str) -> String {
}
pub async fn get_activation_id<R: Runtime>(app_handle: &AppHandle<R>) -> String {
app_handle
.db()
.map(|db| db.get_key_value_str(KV_ACTIVATION_ID_KEY, KV_NAMESPACE, ""))
.unwrap_or_default()
app_handle.db().get_key_value_str(KV_ACTIVATION_ID_KEY, KV_NAMESPACE, "")
}
+1 -1
View File
@@ -12,7 +12,7 @@ pub use connection_or_tx::ConnectionOrTx;
pub use db_context::DbContext;
pub use error::{Error, Result};
pub use migrate::run_migrations;
pub use pool::{PoolError, PoolStatus, SqliteConn, SqlitePool, status as pool_status};
pub use pool::{PoolError, SqliteConn, SqlitePool};
pub use traits::{UpsertModelInfo, upsert_date};
pub use update_source::{ModelChangeEvent, UpdateSource};
pub use util::{generate_id, generate_id_of_length, generate_prefixed_id};
+2 -28
View File
@@ -21,41 +21,20 @@
//! the second ask. Sharing the handle instead makes nested *reads* work the way
//! they do on the desktop; nested *write transactions* fail on both, only
//! differently (here SQLite refuses the inner `BEGIN`; natively the inner
//! connection blocks on `busy_timeout` and then fails).
/// What a pool looks like from outside at one moment. Enough to tell "everything is
/// checked out" apart from "nothing could be opened", which are unrelated problems that
/// both surface as an acquire timeout.
#[derive(Debug, Clone, Copy)]
pub struct PoolStatus {
pub connections: u32,
pub idle: u32,
}
impl std::fmt::Display for PoolStatus {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{} open, {} idle", self.connections, self.idle)
}
}
//! call waits for the one writer connection, which the outer call holds, and
//! times out).
#[cfg(not(target_arch = "wasm32"))]
mod imp {
use super::PoolStatus;
use r2d2_sqlite::SqliteConnectionManager;
pub type SqlitePool = r2d2::Pool<SqliteConnectionManager>;
pub type SqliteConn = r2d2::PooledConnection<SqliteConnectionManager>;
pub type PoolError = r2d2::Error;
pub fn status(pool: &SqlitePool) -> PoolStatus {
let state = pool.state();
PoolStatus { connections: state.connections, idle: state.idle_connections }
}
}
#[cfg(target_arch = "wasm32")]
mod imp {
use super::PoolStatus;
use rusqlite::Connection;
use std::ops::Deref;
use std::rc::Rc;
@@ -88,11 +67,6 @@ mod imp {
#[derive(Debug, thiserror::Error)]
pub enum PoolError {}
/// The one connection, which is never checked out and never unavailable.
pub fn status(_pool: &SqlitePool) -> PoolStatus {
PoolStatus { connections: 1, idle: 1 }
}
#[derive(Debug)]
pub struct SqliteConn(Rc<Connection>);
@@ -20,7 +20,9 @@ impl UpdateSource {
#[derive(Debug, Clone, Serialize, Deserialize, TS)]
#[serde(rename_all = "snake_case", tag = "type")]
pub enum ModelChangeEvent {
Upsert { created: bool },
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
+4 -5
View File
@@ -29,11 +29,10 @@ use yaak_plugins::api::{PluginNameVersion, PluginSearchResponse, PluginUpdatesRe
use yaak_plugins::events::{
CallFolderActionRequest, CallGrpcRequestActionRequest, CallHttpRequestActionRequest,
CallWebsocketRequestActionRequest, CallWorkspaceActionRequest, FilterResponse,
GetFolderActionsResponse, GetGrpcRequestActionsResponse,
GetHttpAuthenticationConfigResponse, GetHttpAuthenticationSummaryResponse,
GetHttpRequestActionsResponse, GetTemplateFunctionConfigResponse,
GetTemplateFunctionSummaryResponse, GetThemesResponse, GetWebsocketRequestActionsResponse,
GetWorkspaceActionsResponse, JsonPrimitive, RenderPurpose,
GetFolderActionsResponse, GetGrpcRequestActionsResponse, GetHttpAuthenticationConfigResponse,
GetHttpAuthenticationSummaryResponse, GetHttpRequestActionsResponse,
GetTemplateFunctionConfigResponse, GetTemplateFunctionSummaryResponse, GetThemesResponse,
GetWebsocketRequestActionsResponse, GetWorkspaceActionsResponse, JsonPrimitive, RenderPurpose,
};
use yaak_plugins::plugin_meta::PluginMetadata;
use yaak_sse::sse::ServerSentEvent;
+5 -5
View File
@@ -13,8 +13,9 @@ pub type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send + 'static>>;
/// `dispatch` call frame, so it cannot borrow, and contexts are cheap clones
/// (handles and `Arc`s). Synchronous handlers wrap into this via `rpc_handler!`
/// with no visible change.
type HandlerFn<Ctx> =
Box<dyn Fn(Ctx, serde_json::Value) -> BoxFuture<Result<serde_json::Value, RpcError>> + Send + Sync>;
type HandlerFn<Ctx> = Box<
dyn Fn(Ctx, serde_json::Value) -> BoxFuture<Result<serde_json::Value, RpcError>> + Send + Sync,
>;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RpcError {
@@ -249,9 +250,8 @@ macro_rules! rpc_handler_async {
Box::new(|ctx, payload| {
Box::pin(async move {
let req = serde_json::from_value(payload).map_err($crate::RpcError::from)?;
let res = $f(ctx, req)
.await
.map_err(|e| $crate::RpcError { message: e.to_string() })?;
let res =
$f(ctx, req).await.map_err(|e| $crate::RpcError { message: e.to_string() })?;
serde_json::to_value(res).map_err($crate::RpcError::from)
})
})
+6 -6
View File
@@ -63,7 +63,7 @@ pub async fn cmd_call_http_request_action<H: PluginHost>(
req: CmdCallHttpRequestActionReq,
) -> Result<()> {
let inner = req.req;
let http_request = resolve_http_request(&host.db()?, &inner.args.http_request)?.0;
let http_request = resolve_http_request(&host.db(), &inner.args.http_request)?.0;
host.call_http_request_action(CallHttpRequestActionRequest {
args: CallHttpRequestActionArgs { http_request },
..inner
@@ -76,7 +76,7 @@ pub async fn cmd_call_grpc_request_action<H: PluginHost>(
req: CmdCallGrpcRequestActionReq,
) -> Result<()> {
let inner = req.req;
let grpc_request = resolve_grpc_request(&host.db()?, &inner.args.grpc_request)?.0;
let grpc_request = resolve_grpc_request(&host.db(), &inner.args.grpc_request)?.0;
host.call_grpc_request_action(CallGrpcRequestActionRequest {
args: CallGrpcRequestActionArgs { grpc_request, ..inner.args },
..inner
@@ -89,7 +89,7 @@ pub async fn cmd_call_websocket_request_action<H: PluginHost>(
req: CmdCallWebsocketRequestActionReq,
) -> Result<()> {
let inner = req.req;
let websocket_request = host.db()?.get_websocket_request(&inner.args.websocket_request.id)?;
let websocket_request = host.db().get_websocket_request(&inner.args.websocket_request.id)?;
host.call_websocket_request_action(CallWebsocketRequestActionRequest {
args: CallWebsocketRequestActionArgs { websocket_request },
..inner
@@ -102,7 +102,7 @@ pub async fn cmd_call_workspace_action<H: PluginHost>(
req: CmdCallWorkspaceActionReq,
) -> Result<()> {
let inner = req.req;
let workspace = host.db()?.get_workspace(&inner.args.workspace.id)?;
let workspace = host.db().get_workspace(&inner.args.workspace.id)?;
host.call_workspace_action(CallWorkspaceActionRequest {
args: CallWorkspaceActionArgs { workspace },
..inner
@@ -115,7 +115,7 @@ pub async fn cmd_call_folder_action<H: PluginHost>(
req: CmdCallFolderActionReq,
) -> Result<()> {
let inner = req.req;
let folder = host.db()?.get_folder(&inner.args.folder.id)?;
let folder = host.db().get_folder(&inner.args.folder.id)?;
host.call_folder_action(CallFolderActionRequest {
args: CallFolderActionArgs { folder },
..inner
@@ -152,6 +152,6 @@ pub async fn cmd_reload_plugins<H: PluginHost>(
host: H,
_req: CmdReloadPluginsReq,
) -> Result<Vec<(String, String)>> {
let plugins = host.db()?.list_plugins()?;
let plugins = host.db().list_plugins()?;
Ok(host.reload_plugins(plugins).await)
}
+2 -2
View File
@@ -68,11 +68,11 @@ pub trait Host: Clone {
PluginContext::new(Some(self.client_id().to_string()), self.session().workspace_id)
}
fn db(&self) -> yaak_models::error::Result<ClientDb<'_>> {
fn db(&self) -> ClientDb<'_> {
self.query_manager().connect()
}
fn blobs(&self) -> yaak_models::error::Result<BlobContext> {
fn blobs(&self) -> BlobContext<'_> {
self.blob_manager().connect()
}
}
+32 -26
View File
@@ -11,10 +11,11 @@ use yaak_models::queries::workspaces::default_headers;
use yaak_rpc_schema::*;
pub async fn models_upsert<H: Host>(host: H, req: ModelsUpsertReq) -> Result<String> {
let db = host.db()?;
let blobs = host.blob_manager();
let source = host.update_source();
Ok(yaak_models::models_ops::upsert_model(&db, blobs, req.model, &source)?)
Ok(host
.query_manager()
.with_tx(|tx| yaak_models::models_ops::upsert_model(tx, blobs, req.model, &source))?)
}
/// Deletes cascade — a workspace can hold thousands of requests — and run in a
@@ -49,25 +50,25 @@ pub async fn models_websocket_events<H: Host>(
host: H,
req: ModelsWebsocketEventsReq,
) -> Result<Vec<WebsocketEvent>> {
Ok(host.db()?.list_websocket_events(&req.connection_id)?)
Ok(host.db().list_websocket_events(&req.connection_id)?)
}
pub async fn models_grpc_events<H: Host>(
host: H,
req: ModelsGrpcEventsReq,
) -> Result<Vec<GrpcEvent>> {
Ok(host.db()?.list_grpc_events(&req.connection_id)?)
Ok(host.db().list_grpc_events(&req.connection_id)?)
}
pub async fn models_get_settings<H: Host>(host: H, _req: ModelsGetSettingsReq) -> Result<Settings> {
Ok(host.db()?.get_settings())
Ok(host.db().get_settings())
}
pub async fn models_get_graphql_introspection<H: Host>(
host: H,
req: ModelsGetGraphqlIntrospectionReq,
) -> Result<Option<GraphQlIntrospection>> {
Ok(host.db()?.get_graphql_introspection(&req.request_id))
Ok(host.db().get_graphql_introspection(&req.request_id))
}
pub async fn models_upsert_graphql_introspection<H: Host>(
@@ -75,12 +76,9 @@ pub async fn models_upsert_graphql_introspection<H: Host>(
req: ModelsUpsertGraphqlIntrospectionReq,
) -> Result<GraphQlIntrospection> {
let source = host.update_source();
Ok(host.db()?.upsert_graphql_introspection(
&req.workspace_id,
&req.request_id,
req.content,
&source,
)?)
Ok(host.query_manager().with_tx(|tx| {
tx.upsert_graphql_introspection(&req.workspace_id, &req.request_id, req.content, &source)
})?)
}
/// Everything the frontend's model store needs to boot, as one JSON string.
@@ -96,14 +94,14 @@ pub async fn models_workspace_models<H: PluginHost>(
// Add the global models
{
let db = host.db()?;
let db = host.db();
l.push(db.get_settings().into());
l.append(&mut db.list_workspaces()?.into_iter().map(Into::into).collect());
l.append(&mut db.list_key_values()?.into_iter().map(Into::into).collect());
}
let plugins = {
let db = host.db()?;
let db = host.db();
db.list_plugins()?
};
@@ -112,9 +110,16 @@ pub async fn models_workspace_models<H: PluginHost>(
// Add the workspace children
if let Some(wid) = req.workspace_id.as_deref() {
let db = host.db()?;
// Opening a workspace is where the rows it is assumed to have get created
host.query_manager().with_tx(|tx| {
tx.ensure_base_environment(wid)?;
tx.ensure_default_cookie_jar(wid)?;
tx.ensure_workspace_meta(wid)?;
Ok::<(), yaak_models::error::Error>(())
})?;
let db = host.db();
l.append(&mut db.list_cookie_jars(wid)?.into_iter().map(Into::into).collect());
l.append(&mut db.list_environments_ensure_base(wid)?.into_iter().map(Into::into).collect());
l.append(&mut db.list_environments(wid)?.into_iter().map(Into::into).collect());
l.append(&mut db.list_folders(wid)?.into_iter().map(Into::into).collect());
l.append(&mut db.list_grpc_connections(wid)?.into_iter().map(Into::into).collect());
l.append(&mut db.list_grpc_requests(wid)?.into_iter().map(Into::into).collect());
@@ -132,25 +137,26 @@ pub async fn cmd_get_workspace_meta<H: Host>(
host: H,
req: CmdGetWorkspaceMetaReq,
) -> Result<WorkspaceMeta> {
let db = host.db()?;
let workspace = db.get_workspace(&req.workspace_id)?;
Ok(db.get_or_create_workspace_meta(&workspace.id)?)
let workspace = host.db().get_workspace(&req.workspace_id)?;
Ok(host.query_manager().with_tx(|tx| tx.ensure_workspace_meta(&workspace.id))?)
}
pub async fn cmd_delete_all_grpc_connections<H: Host>(
host: H,
req: CmdDeleteAllGrpcConnectionsReq,
) -> Result<()> {
Ok(host
.db()?
.delete_all_grpc_connections_for_request(&req.request_id, &host.update_source())?)
Ok(host.query_manager().with_tx(|tx| {
tx.delete_all_grpc_connections_for_request(&req.request_id, &host.update_source())
})?)
}
pub async fn cmd_delete_all_http_responses<H: Host>(
host: H,
req: CmdDeleteAllHttpResponsesReq,
) -> Result<()> {
host.db()?.delete_all_http_responses_for_request(&req.request_id, &host.update_source())?;
host.query_manager().with_tx(|tx| {
tx.delete_all_http_responses_for_request(&req.request_id, &host.update_source())
})?;
Ok(())
}
@@ -158,9 +164,9 @@ pub async fn cmd_ws_delete_connections<H: Host>(
host: H,
req: CmdWsDeleteConnectionsReq,
) -> Result<()> {
Ok(host
.db()?
.delete_all_websocket_connections_for_request(&req.request_id, &host.update_source())?)
Ok(host.query_manager().with_tx(|tx| {
tx.delete_all_websocket_connections_for_request(&req.request_id, &host.update_source())
})?)
}
pub async fn cmd_delete_send_history<H: Host>(host: H, req: CmdDeleteSendHistoryReq) -> Result<()> {
+1 -1
View File
@@ -10,7 +10,7 @@ pub async fn cmd_plugin_info<H: PluginHost>(
host: H,
req: CmdPluginInfoReq,
) -> Result<PluginMetadata> {
let plugin = host.db()?.get_plugin(&req.id)?;
let plugin = host.db().get_plugin(&req.id)?;
if let Some(metadata) = host.loaded_plugin_metadata(&plugin.directory).await {
return Ok(metadata);
}
+1 -1
View File
@@ -64,7 +64,7 @@ pub(crate) async fn render_form_values<H: PluginHost>(
};
let environment_chain =
host.db()?.resolve_environments(&workspace_id, folder_id.as_deref(), environment_id)?;
host.db().resolve_environments(&workspace_id, folder_id.as_deref(), environment_id)?;
let cb = host.template_callback(purpose).await?;
let rendered =
+4 -4
View File
@@ -41,7 +41,7 @@ pub async fn cmd_get_http_response_events<H: Host>(
host: H,
req: CmdGetHttpResponseEventsReq,
) -> Result<Vec<HttpResponseEvent>> {
let events: Vec<HttpResponseEvent> = host.db()?.list_http_response_events(&req.response_id)?;
let events: Vec<HttpResponseEvent> = host.db().list_http_response_events(&req.response_id)?;
Ok(events)
}
@@ -55,7 +55,7 @@ pub async fn cmd_http_response_body_path<H: Host>(
host: H,
req: CmdHttpResponseBodyPathReq,
) -> Result<Option<String>> {
let location = locate_response_body(&host.db()?, &req.response_id)?;
let location = locate_response_body(&host.db(), &req.response_id)?;
Ok(location.path.map(|p| p.to_string_lossy().to_string()))
}
@@ -64,7 +64,7 @@ pub async fn cmd_http_request_body<H: Host>(
req: CmdHttpRequestBodyReq,
) -> Result<Option<Vec<u8>>> {
let body_id = format!("{}.request", req.response_id);
let chunks = host.blobs()?.get_chunks(&body_id)?;
let chunks = host.blobs().get_chunks(&body_id)?;
if chunks.is_empty() {
return Ok(None);
@@ -76,7 +76,7 @@ pub async fn cmd_http_request_body<H: Host>(
}
pub async fn cmd_save_response<H: Host>(host: H, req: CmdSaveResponseReq) -> Result<()> {
let response = host.db()?.get_http_response(&req.response_id)?;
let response = host.db().get_http_response(&req.response_id)?;
let body_path =
response.body_path.ok_or(Error::Generic("Response does not have a body".to_string()))?;
+1 -1
View File
@@ -20,7 +20,7 @@ pub async fn cmd_render_template<H: PluginHost>(
req: CmdRenderTemplateReq,
) -> Result<String> {
let environment_chain =
host.db()?.resolve_environments(&req.workspace_id, None, req.environment_id.as_deref())?;
host.db().resolve_environments(&req.workspace_id, None, req.environment_id.as_deref())?;
let cb = host.template_callback(req.purpose.unwrap_or(RenderPurpose::Preview)).await?;
let options = RenderOptions {
// A preview that throws would show the user an error where they expect
+71 -67
View File
@@ -118,6 +118,7 @@ impl Host for TestHost {
#[tokio::test(flavor = "multi_thread")]
async fn writes_carry_the_client_id() {
let host = TestHost::new();
host.drain_writes(); // the rows startup creates
let workspace = Workspace { name: "From a test".to_string(), ..Default::default() };
let id = models_upsert(host.clone(), ModelsUpsertReq { model: AnyModel::Workspace(workspace) })
@@ -141,13 +142,13 @@ async fn writes_carry_the_client_id() {
// Deletes cascade inside a transaction; make sure that path works with no
// host doing anything special around it.
let workspace = host.db().unwrap().get_workspace(&id).expect("get workspace");
let workspace = host.db().get_workspace(&id).expect("get workspace");
let deleted =
models_delete(host.clone(), ModelsDeleteReq { model: AnyModel::Workspace(workspace) })
.await
.expect("delete");
assert_eq!(deleted, id);
assert!(host.db().unwrap().get_workspace(&id).is_err(), "workspace should be gone");
assert!(host.db().get_workspace(&id).is_err(), "workspace should be gone");
}
#[tokio::test]
@@ -406,22 +407,23 @@ async fn a_single_threaded_host_can_implement_the_trait() {
// shared code; only the callback came from the host. Rendering a real
// variable is what proves the chain was resolved rather than skipped.
let environment = host
.db()
.unwrap()
.upsert_environment(
&Environment {
workspace_id: id.clone(),
name: "Test env".to_string(),
variables: vec![EnvironmentVariable {
enabled: true,
name: "greeting".to_string(),
value: "hello".to_string(),
id: None,
}],
..Default::default()
},
&host.update_source(),
)
.query_manager()
.with_tx(|tx| {
tx.upsert_environment(
&Environment {
workspace_id: id.clone(),
name: "Test env".to_string(),
variables: vec![EnvironmentVariable {
enabled: true,
name: "greeting".to_string(),
value: "hello".to_string(),
id: None,
}],
..Default::default()
},
&host.update_source(),
)
})
.expect("seed environment");
let rendered = cmd_render_template(
@@ -440,7 +442,7 @@ async fn a_single_threaded_host_can_implement_the_trait() {
// The delete path too, since it is the one that used to reach for a
// blocking thread this host does not have.
let workspace = host.db().unwrap().get_workspace(&id).expect("get workspace");
let workspace = host.db().get_workspace(&id).expect("get workspace");
let deleted = models_delete(host, ModelsDeleteReq { model: AnyModel::Workspace(workspace) })
.await
.expect("delete");
@@ -460,32 +462,33 @@ async fn auth_values_are_rendered_before_the_host_sees_them() {
};
let workspace = host
.db()
.unwrap()
.upsert_workspace(
&Workspace { name: "Auth".to_string(), ..Default::default() },
&host.update_source(),
)
.query_manager()
.with_tx(|tx| {
tx.upsert_workspace(
&Workspace { name: "Auth".to_string(), ..Default::default() },
&host.update_source(),
)
})
.expect("workspace");
host.db()
.unwrap()
.upsert_environment(
&Environment {
workspace_id: workspace.id.clone(),
name: "Env".to_string(),
variables: vec![EnvironmentVariable {
enabled: true,
name: "token".to_string(),
value: "s3cret".to_string(),
id: None,
}],
..Default::default()
},
&host.update_source(),
)
host.query_manager()
.with_tx(|tx| {
tx.upsert_environment(
&Environment {
workspace_id: workspace.id.clone(),
name: "Env".to_string(),
variables: vec![EnvironmentVariable {
enabled: true,
name: "token".to_string(),
value: "s3cret".to_string(),
id: None,
}],
..Default::default()
},
&host.update_source(),
)
})
.expect("environment");
let environment =
host.db().unwrap().list_environments_ensure_base(&workspace.id).expect("list").remove(0);
let environment = host.db().list_environments(&workspace.id).expect("list").remove(0);
let mut values = HashMap::new();
values.insert("password".to_string(), JsonPrimitive::String("${[ token ]}".to_string()));
@@ -524,32 +527,33 @@ async fn template_function_values_are_rendered_before_the_host_sees_them() {
};
let workspace = host
.db()
.unwrap()
.upsert_workspace(
&Workspace { name: "Functions".to_string(), ..Default::default() },
&host.update_source(),
)
.query_manager()
.with_tx(|tx| {
tx.upsert_workspace(
&Workspace { name: "Functions".to_string(), ..Default::default() },
&host.update_source(),
)
})
.expect("workspace");
host.db()
.unwrap()
.upsert_environment(
&Environment {
workspace_id: workspace.id.clone(),
name: "Env".to_string(),
variables: vec![EnvironmentVariable {
enabled: true,
name: "1PASSWORD_TOKEN".to_string(),
value: "ops_abc123".to_string(),
id: None,
}],
..Default::default()
},
&host.update_source(),
)
host.query_manager()
.with_tx(|tx| {
tx.upsert_environment(
&Environment {
workspace_id: workspace.id.clone(),
name: "Env".to_string(),
variables: vec![EnvironmentVariable {
enabled: true,
name: "1PASSWORD_TOKEN".to_string(),
value: "ops_abc123".to_string(),
id: None,
}],
..Default::default()
},
&host.update_source(),
)
})
.expect("environment");
let environment =
host.db().unwrap().list_environments_ensure_base(&workspace.id).expect("list").remove(0);
let environment = host.db().list_environments(&workspace.id).expect("list").remove(0);
let mut values = HashMap::new();
values.insert("token".to_string(), JsonPrimitive::String("${[1PASSWORD_TOKEN]}".to_string()));
+10 -10
View File
@@ -51,7 +51,7 @@ impl EncryptionManager {
pub fn set_human_key(&self, workspace_id: &str, human_key: &str) -> Result<WorkspaceMeta> {
let wkey = WorkspaceKey::from_human(human_key)?;
let workspace = self.query_manager.connect()?.get_workspace(workspace_id)?;
let workspace = self.query_manager.connect().get_workspace(workspace_id)?;
let encryption_key_challenge = match workspace.encryption_key_challenge {
None => return self.set_workspace_key(workspace_id, &wkey),
Some(c) => c,
@@ -83,7 +83,7 @@ impl EncryptionManager {
let workspace_meta = self.query_manager.with_tx::<WorkspaceMeta, Error>(|tx| {
let workspace = tx.get_workspace(workspace_id)?;
let workspace_meta = tx.get_or_create_workspace_meta(workspace_id)?;
let workspace_meta = tx.ensure_workspace_meta(workspace_id)?;
tx.upsert_workspace(
&Workspace { encryption_key_challenge, ..workspace },
&UpdateSource::Background,
@@ -103,7 +103,7 @@ impl EncryptionManager {
pub fn ensure_workspace_key(&self, workspace_id: &str) -> Result<WorkspaceMeta> {
let workspace_meta =
self.query_manager.connect()?.get_or_create_workspace_meta(workspace_id)?;
self.query_manager.with_tx(|tx| tx.ensure_workspace_meta(workspace_id))?;
// Already exists
if let Some(_) = workspace_meta.encryption_key {
@@ -120,7 +120,7 @@ impl EncryptionManager {
self.query_manager.with_tx::<(), Error>(|tx| {
let workspace = tx.get_workspace(workspace_id)?;
let workspace_meta = tx.get_or_create_workspace_meta(workspace_id)?;
let workspace_meta = tx.ensure_workspace_meta(workspace_id)?;
// Clear encryption challenge on workspace
tx.upsert_workspace(
@@ -152,12 +152,12 @@ impl EncryptionManager {
}
};
// Scoped so the pooled connection is back before `get_master_key` below, which
// can block on the OS keyring for as long as it likes.
let workspace_meta =
self.query_manager.connect()?.get_or_create_workspace_meta(workspace_id)?;
let key = match workspace_meta.encryption_key {
let key = match self
.query_manager
.connect()
.get_workspace_meta(workspace_id)
.and_then(|m| m.encryption_key)
{
None => return Err(MissingWorkspaceKey),
Some(k) => k,
};
+33 -37
View File
@@ -7,7 +7,7 @@
use log::info;
use std::path::PathBuf;
use yaak_models::blob_manager::BlobManager;
use yaak_models::client_db::ClientDb;
use yaak_models::client_db::WriteDb;
use yaak_models::error::Result;
const MODEL_CHANGES_RETENTION_HOURS: i64 = 1;
@@ -44,7 +44,8 @@ impl Host {
}
/// Run once after the database is open, before the host answers anything.
pub fn on_launch(host: &Host, db: &ClientDb, blobs: &BlobManager) -> Result<()> {
/// Takes the write handle: a launch closes what the last session left open.
pub fn on_launch(host: &Host, db: &WriteDb, blobs: &BlobManager) -> Result<()> {
db.prune_model_changes_older_than_hours(MODEL_CHANGES_RETENTION_HOURS)?;
if host.role == Role::Owner {
@@ -77,54 +78,49 @@ mod tests {
#[test]
fn only_the_owner_closes_what_the_last_session_left_open() {
let (query_manager, blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
let db = query_manager.connect().unwrap();
let source = &UpdateSource::Background;
let workspace = db
.upsert_workspace(
&Workspace { name: "Hooks".to_string(), ..Default::default() },
source,
)
.unwrap();
let request = db
.upsert_http_request(
&HttpRequest { workspace_id: workspace.id.clone(), ..Default::default() },
source,
)
.unwrap();
let pending = db
.upsert_http_response(
&HttpResponse {
request_id: request.id.clone(),
workspace_id: workspace.id.clone(),
state: HttpResponseState::Connected,
..Default::default()
},
source,
&blob_manager,
)
let pending = query_manager
.with_tx(|db| {
let workspace = db.upsert_workspace(
&Workspace { name: "Hooks".to_string(), ..Default::default() },
source,
)?;
let request = db.upsert_http_request(
&HttpRequest { workspace_id: workspace.id.clone(), ..Default::default() },
source,
)?;
db.upsert_http_response(
&HttpResponse {
request_id: request.id.clone(),
workspace_id: workspace.id.clone(),
state: HttpResponseState::Connected,
..Default::default()
},
source,
&blob_manager,
)
})
.unwrap();
on_launch(&Host::guest(), &db, &blob_manager).unwrap();
let response = db.get_http_response(&pending.id).unwrap();
query_manager.with_tx(|db| on_launch(&Host::guest(), db, &blob_manager)).unwrap();
let response = query_manager.connect().get_http_response(&pending.id).unwrap();
assert!(matches!(response.state, HttpResponseState::Connected));
on_launch(&Host::owner(), &db, &blob_manager).unwrap();
let response = db.get_http_response(&pending.id).unwrap();
query_manager.with_tx(|db| on_launch(&Host::owner(), db, &blob_manager)).unwrap();
let response = query_manager.connect().get_http_response(&pending.id).unwrap();
assert!(matches!(response.state, HttpResponseState::Closed));
}
#[test]
fn owner_without_a_filesystem_still_sweeps_blobs() {
let (query_manager, blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
let db = query_manager.connect().unwrap();
{
let blob_ctx = blob_manager.connect().unwrap();
blob_ctx.insert_chunk(&BodyChunk::new("rs_gone", 0, b"dead".to_vec())).unwrap();
}
blob_manager
.with_tx(|b| b.insert_chunk(&BodyChunk::new("rs_gone", 0, b"dead".to_vec())))
.unwrap();
on_launch(&Host::owner(), &db, &blob_manager).unwrap();
query_manager.with_tx(|db| on_launch(&Host::owner(), db, &blob_manager)).unwrap();
assert!(!blob_manager.connect().unwrap().body_exists("rs_gone").unwrap());
assert!(!blob_manager.connect().body_exists("rs_gone").unwrap());
}
}
-3
View File
@@ -27,6 +27,3 @@ yaak-templates = { path = "../yaak-templates", default-features = false }
[target.'cfg(not(target_arch = "wasm32"))'.dependencies]
r2d2 = "0.8.10"
r2d2_sqlite = { version = "0.32" }
[dev-dependencies]
tempfile = "3"
+165 -110
View File
@@ -1,9 +1,11 @@
use crate::error::Error::GenericError;
use crate::error::Result;
use crate::util::generate_prefixed_id;
use include_dir::{Dir, include_dir};
use log::{debug, info};
use rusqlite::{OptionalExtension, params};
use yaak_database::{SqliteConn, SqlitePool};
use rusqlite::{OptionalExtension, Transaction, TransactionBehavior, params};
use std::ops::Deref;
use yaak_database::{ConnectionOrTx, SqlitePool};
static BLOB_MIGRATIONS_DIR: Dir = include_dir!("$CARGO_MANIFEST_DIR/blob_migrations");
@@ -22,41 +24,64 @@ impl BodyChunk {
}
}
/// Manages the blob database connection pool.
// Pool is internally synchronized — don't wrap it in a Mutex. A Mutex held across the
/// Manages the blob database: a reader pool and a single writer, for the
/// same reason as [`crate::query_manager::QueryManager`].
// Pools are internally synchronized — don't wrap them in a Mutex. A Mutex held across the
// blocking `get()` serializes every blob access behind the slowest waiter, freezing the
// whole app whenever the pool is exhausted.
#[derive(Debug, Clone)]
pub struct BlobManager {
pool: SqlitePool,
readers: SqlitePool,
writer: SqlitePool,
}
impl BlobManager {
pub fn new(pool: SqlitePool) -> Self {
Self { pool }
/// `writer` must be a pool with a single connection.
pub fn new(readers: SqlitePool, writer: SqlitePool) -> Self {
Self { readers, writer }
}
pub fn connect(&self) -> Result<BlobContext> {
let conn = crate::pool::acquire(&self.pool, "blob")?;
Ok(BlobContext { conn })
/// A read handle from the reader pool.
pub fn connect(&self) -> BlobContext<'_> {
let conn = self.readers.get().expect("Failed to get blob DB connection from pool");
BlobContext { conn: ConnectionOrTx::Connection(conn) }
}
/// Run `func` in a transaction on the writer connection.
pub fn with_tx<T, E>(
&self,
func: impl FnOnce(&BlobWriter) -> std::result::Result<T, E>,
) -> std::result::Result<T, E>
where
E: From<crate::error::Error>,
{
let conn = self.writer.get().map_err(crate::error::Error::SqlPoolError)?;
let tx = Transaction::new_unchecked(&conn, TransactionBehavior::Immediate)
.map_err(crate::error::Error::SqlError)?;
let writer = BlobWriter { ctx: BlobContext { conn: ConnectionOrTx::Transaction(&tx) } };
match func(&writer) {
Ok(val) => {
tx.commit().map_err(|e| {
GenericError(format!("Failed to commit blob transaction {e:?}"))
})?;
Ok(val)
}
Err(e) => {
tx.rollback().map_err(|e| {
GenericError(format!("Failed to rollback blob transaction {e:?}"))
})?;
Err(e)
}
}
}
}
/// Context for blob database operations.
pub struct BlobContext {
conn: SqliteConn,
/// Read handle for the blob database.
pub struct BlobContext<'a> {
conn: ConnectionOrTx<'a>,
}
impl BlobContext {
/// Insert a single chunk.
pub fn insert_chunk(&self, chunk: &BodyChunk) -> Result<()> {
self.conn.execute(
"INSERT INTO body_chunks (id, body_id, chunk_index, data) VALUES (?1, ?2, ?3, ?4)",
params![chunk.id, chunk.body_id, chunk.chunk_index, chunk.data],
)?;
Ok(())
}
impl<'a> BlobContext<'a> {
/// Get all chunks for a body, ordered by chunk_index.
pub fn get_chunks(&self, body_id: &str) -> Result<Vec<BodyChunk>> {
let mut stmt = self.conn.prepare(
@@ -87,25 +112,11 @@ impl BlobContext {
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])?;
Ok(())
}
/// Delete all chunks matching a body_id prefix (e.g., "rs_abc123.%" to delete all bodies for a response).
pub fn delete_chunks_like(&self, body_id_prefix: &str) -> Result<()> {
self.conn
.execute("DELETE FROM body_chunks WHERE body_id LIKE ?1", params![body_id_prefix])?;
Ok(())
}
}
/// Get total size of a body without loading data.
impl BlobContext {
/// Get total size of a body without loading data.
pub fn get_body_size(&self, body_id: &str) -> Result<usize> {
let size: i64 = self
.conn
.resolve()
.query_row(
"SELECT COALESCE(SUM(LENGTH(data)), 0) FROM body_chunks WHERE body_id = ?1",
params![body_id],
@@ -119,6 +130,7 @@ impl BlobContext {
pub fn body_exists(&self, body_id: &str) -> Result<bool> {
let count: i64 = self
.conn
.resolve()
.query_row(
"SELECT COUNT(*) FROM body_chunks WHERE body_id = ?1",
params![body_id],
@@ -129,6 +141,44 @@ impl BlobContext {
}
}
/// Write handle for the blob database. Derefs to [`BlobContext`] for reads.
pub struct BlobWriter<'a> {
ctx: BlobContext<'a>,
}
impl<'a> Deref for BlobWriter<'a> {
type Target = BlobContext<'a>;
fn deref(&self) -> &BlobContext<'a> {
&self.ctx
}
}
impl<'a> BlobWriter<'a> {
/// Insert a single chunk.
pub fn insert_chunk(&self, chunk: &BodyChunk) -> Result<()> {
self.conn.execute(
"INSERT INTO body_chunks (id, body_id, chunk_index, data) VALUES (?1, ?2, ?3, ?4)",
params![chunk.id, chunk.body_id, chunk.chunk_index, chunk.data],
)?;
Ok(())
}
/// 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])?;
Ok(())
}
/// Delete all chunks matching a body_id prefix (e.g., "rs_abc123.%" to delete all
/// bodies for a response).
pub fn delete_chunks_like(&self, body_id_prefix: &str) -> Result<()> {
self.conn
.execute("DELETE FROM body_chunks WHERE body_id LIKE ?1", params![body_id_prefix])?;
Ok(())
}
}
/// Run migrations for the blob database.
pub fn migrate_blob_db(pool: &SqlitePool) -> Result<()> {
info!("Running blob database migrations");
@@ -196,28 +246,39 @@ pub fn migrate_blob_db(pool: &SqlitePool) -> Result<()> {
#[cfg(test)]
mod tests {
use super::*;
use crate::error::Error;
fn create_test_pool() -> SqlitePool {
fn create_test_manager() -> BlobManager {
let manager = r2d2_sqlite::SqliteConnectionManager::memory();
let pool = r2d2::Pool::builder().max_size(1).build(manager).unwrap();
migrate_blob_db(&pool).unwrap();
pool
BlobManager::new(pool.clone(), pool)
}
fn insert(manager: &BlobManager, chunks: &[BodyChunk]) {
manager
.with_tx(|b| {
for c in chunks {
b.insert_chunk(c)?;
}
Ok::<_, Error>(())
})
.unwrap();
}
#[test]
fn test_insert_and_get_chunks() {
let pool = create_test_pool();
let manager = BlobManager::new(pool);
let ctx = manager.connect().unwrap();
let manager = create_test_manager();
let body_id = "rs_test123.request";
let chunk1 = BodyChunk::new(body_id, 0, b"Hello, ".to_vec());
let chunk2 = BodyChunk::new(body_id, 1, b"World!".to_vec());
insert(
&manager,
&[
BodyChunk::new(body_id, 0, b"Hello, ".to_vec()),
BodyChunk::new(body_id, 1, b"World!".to_vec()),
],
);
ctx.insert_chunk(&chunk1).unwrap();
ctx.insert_chunk(&chunk2).unwrap();
let chunks = ctx.get_chunks(body_id).unwrap();
let chunks = manager.connect().get_chunks(body_id).unwrap();
assert_eq!(chunks.len(), 2);
assert_eq!(chunks[0].chunk_index, 0);
assert_eq!(chunks[0].data, b"Hello, ");
@@ -227,18 +288,19 @@ mod tests {
#[test]
fn test_get_chunks_ordered_by_index() {
let pool = create_test_pool();
let manager = BlobManager::new(pool);
let ctx = manager.connect().unwrap();
let manager = create_test_manager();
let body_id = "rs_test123.request";
// Insert out of order
ctx.insert_chunk(&BodyChunk::new(body_id, 2, b"C".to_vec())).unwrap();
ctx.insert_chunk(&BodyChunk::new(body_id, 0, b"A".to_vec())).unwrap();
ctx.insert_chunk(&BodyChunk::new(body_id, 1, b"B".to_vec())).unwrap();
insert(
&manager,
&[
BodyChunk::new(body_id, 2, b"C".to_vec()),
BodyChunk::new(body_id, 0, b"A".to_vec()),
BodyChunk::new(body_id, 1, b"B".to_vec()),
],
);
let chunks = ctx.get_chunks(body_id).unwrap();
let chunks = manager.connect().get_chunks(body_id).unwrap();
assert_eq!(chunks.len(), 3);
assert_eq!(chunks[0].data, b"A");
assert_eq!(chunks[1].data, b"B");
@@ -247,89 +309,85 @@ mod tests {
#[test]
fn test_delete_chunks() {
let pool = create_test_pool();
let manager = BlobManager::new(pool);
let ctx = manager.connect().unwrap();
let manager = create_test_manager();
let body_id = "rs_test123.request";
ctx.insert_chunk(&BodyChunk::new(body_id, 0, b"data".to_vec())).unwrap();
insert(&manager, &[BodyChunk::new(body_id, 0, b"data".to_vec())]);
assert!(manager.connect().body_exists(body_id).unwrap());
assert!(ctx.body_exists(body_id).unwrap());
ctx.delete_chunks(body_id).unwrap();
manager.with_tx(|b| b.delete_chunks(body_id)).unwrap();
let ctx = manager.connect();
assert!(!ctx.body_exists(body_id).unwrap());
assert_eq!(ctx.get_chunks(body_id).unwrap().len(), 0);
}
#[test]
fn test_delete_chunks_like() {
let pool = create_test_pool();
let manager = BlobManager::new(pool);
let ctx = manager.connect().unwrap();
let manager = create_test_manager();
// Insert chunks for same response but different body types
ctx.insert_chunk(&BodyChunk::new("rs_abc.request", 0, b"req".to_vec())).unwrap();
ctx.insert_chunk(&BodyChunk::new("rs_abc.response", 0, b"resp".to_vec())).unwrap();
ctx.insert_chunk(&BodyChunk::new("rs_other.request", 0, b"other".to_vec())).unwrap();
insert(
&manager,
&[
BodyChunk::new("rs_abc.request", 0, b"req".to_vec()),
BodyChunk::new("rs_abc.response", 0, b"resp".to_vec()),
BodyChunk::new("rs_other.request", 0, b"other".to_vec()),
],
);
// Delete all bodies for rs_abc
ctx.delete_chunks_like("rs_abc.%").unwrap();
manager.with_tx(|b| b.delete_chunks_like("rs_abc.%")).unwrap();
// rs_abc bodies should be gone
let ctx = manager.connect();
assert!(!ctx.body_exists("rs_abc.request").unwrap());
assert!(!ctx.body_exists("rs_abc.response").unwrap());
// rs_other should still exist
assert!(ctx.body_exists("rs_other.request").unwrap());
}
#[test]
fn test_get_body_size() {
let pool = create_test_pool();
let manager = BlobManager::new(pool);
let ctx = manager.connect().unwrap();
let manager = create_test_manager();
let body_id = "rs_test123.request";
ctx.insert_chunk(&BodyChunk::new(body_id, 0, b"Hello".to_vec())).unwrap();
ctx.insert_chunk(&BodyChunk::new(body_id, 1, b"World".to_vec())).unwrap();
insert(
&manager,
&[
BodyChunk::new(body_id, 0, b"Hello".to_vec()),
BodyChunk::new(body_id, 1, b"World".to_vec()),
],
);
let size = ctx.get_body_size(body_id).unwrap();
let size = manager.connect().get_body_size(body_id).unwrap();
assert_eq!(size, 10); // "Hello" + "World" = 10 bytes
}
#[test]
fn test_get_body_size_empty() {
let pool = create_test_pool();
let manager = BlobManager::new(pool);
let ctx = manager.connect().unwrap();
let size = ctx.get_body_size("nonexistent").unwrap();
let manager = create_test_manager();
let size = manager.connect().get_body_size("nonexistent").unwrap();
assert_eq!(size, 0);
}
#[test]
fn test_body_exists() {
let pool = create_test_pool();
let manager = BlobManager::new(pool);
let ctx = manager.connect().unwrap();
let manager = create_test_manager();
assert!(!manager.connect().body_exists("rs_test.request").unwrap());
assert!(!ctx.body_exists("rs_test.request").unwrap());
insert(&manager, &[BodyChunk::new("rs_test.request", 0, b"data".to_vec())]);
ctx.insert_chunk(&BodyChunk::new("rs_test.request", 0, b"data".to_vec())).unwrap();
assert!(ctx.body_exists("rs_test.request").unwrap());
assert!(manager.connect().body_exists("rs_test.request").unwrap());
}
#[test]
fn test_multiple_bodies_isolated() {
let pool = create_test_pool();
let manager = BlobManager::new(pool);
let ctx = manager.connect().unwrap();
ctx.insert_chunk(&BodyChunk::new("body1", 0, b"data1".to_vec())).unwrap();
ctx.insert_chunk(&BodyChunk::new("body2", 0, b"data2".to_vec())).unwrap();
let manager = create_test_manager();
insert(
&manager,
&[
BodyChunk::new("body1", 0, b"data1".to_vec()),
BodyChunk::new("body2", 0, b"data2".to_vec()),
],
);
let ctx = manager.connect();
let chunks1 = ctx.get_chunks("body1").unwrap();
let chunks2 = ctx.get_chunks("body2").unwrap();
@@ -341,16 +399,13 @@ mod tests {
#[test]
fn test_large_chunk() {
let pool = create_test_pool();
let manager = BlobManager::new(pool);
let ctx = manager.connect().unwrap();
let manager = create_test_manager();
// 1MB chunk
let large_data: Vec<u8> = (0..1024 * 1024).map(|i| (i % 256) as u8).collect();
let body_id = "rs_large.request";
insert(&manager, &[BodyChunk::new(body_id, 0, large_data.clone())]);
ctx.insert_chunk(&BodyChunk::new(body_id, 0, large_data.clone())).unwrap();
let ctx = manager.connect();
let chunks = ctx.get_chunks(body_id).unwrap();
assert_eq!(chunks.len(), 1);
assert_eq!(chunks[0].data, large_data);
+45 -9
View File
@@ -3,18 +3,25 @@ use crate::models::{AnyModel, UpsertModelInfo};
use crate::util::{ModelChangeEvent, ModelPayload, UpdateSource};
use rusqlite::params;
use sea_query::{IntoColumnRef, IntoIden, SimpleExpr};
use std::cell::RefCell;
use std::fmt::Debug;
use std::ops::Deref;
use std::sync::mpsc;
use yaak_database::DbContext;
/// A read handle. Comes from the reader pool and can only query.
///
/// Anything that changes a row lives on [`WriteDb`], which is only ever handed
/// out inside a transaction on the single writer connection. That split is
/// what keeps the pool from filling with writers waiting on each other: there
/// is one writer, so there is never a second one to wait for.
pub struct ClientDb<'a> {
pub(crate) ctx: DbContext<'a>,
pub(crate) events_tx: mpsc::Sender<ModelPayload>,
}
impl<'a> ClientDb<'a> {
pub fn new(ctx: DbContext<'a>, events_tx: mpsc::Sender<ModelPayload>) -> Self {
Self { ctx, events_tx }
pub fn new(ctx: DbContext<'a>) -> Self {
Self { ctx }
}
/// Access the underlying connection for custom queries.
@@ -22,8 +29,6 @@ impl<'a> ClientDb<'a> {
self.ctx.conn()
}
// --- Read delegates (thin wrappers over DbContext) ---
pub(crate) fn find_one<M>(
&self,
col: impl IntoColumnRef + IntoIden + Clone,
@@ -64,6 +69,39 @@ impl<'a> ClientDb<'a> {
{
Ok(self.ctx.find_many(col, value, limit)?)
}
}
/// A write handle: a [`ClientDb`] on the writer connection, inside a
/// transaction, that can also change rows. Derefs to [`ClientDb`] so every
/// query is available while writing, and reads inside the transaction see
/// its own uncommitted writes.
///
/// Model events are held back until the transaction commits; a rollback
/// discards them along with the rows.
pub struct WriteDb<'a> {
db: ClientDb<'a>,
events_tx: mpsc::Sender<ModelPayload>,
pending_events: RefCell<Vec<ModelPayload>>,
}
impl<'a> Deref for WriteDb<'a> {
type Target = ClientDb<'a>;
fn deref(&self) -> &ClientDb<'a> {
&self.db
}
}
impl<'a> WriteDb<'a> {
pub fn new(ctx: DbContext<'a>, events_tx: mpsc::Sender<ModelPayload>) -> Self {
Self { db: ClientDb::new(ctx), events_tx, pending_events: RefCell::new(Vec::new()) }
}
/// The events for everything written so far, to send once the
/// transaction has committed.
pub(crate) fn into_events(self) -> Vec<ModelPayload> {
self.pending_events.into_inner()
}
/// Bulk-delete all rows matching a column value WITHOUT recording model
/// changes or emitting events. Only use for cascades whose deletion is
@@ -80,8 +118,6 @@ impl<'a> ClientDb<'a> {
Ok(self.ctx.delete_many::<M>(col, value)?)
}
// --- Write operations (with event recording) ---
pub(crate) fn upsert<M>(&self, model: &M, source: &UpdateSource) -> Result<M>
where
M: Into<AnyModel> + UpsertModelInfo + Clone,
@@ -95,7 +131,7 @@ impl<'a> ClientDb<'a> {
};
self.record_model_change(&payload)?;
let _ = self.events_tx.send(payload);
self.pending_events.borrow_mut().push(payload);
Ok(m)
}
@@ -113,7 +149,7 @@ impl<'a> ClientDb<'a> {
};
self.record_model_change(&payload)?;
let _ = self.events_tx.send(payload);
self.pending_events.borrow_mut().push(payload);
Ok(m.clone())
}
-12
View File
@@ -9,18 +9,6 @@ pub enum Error {
#[error("SQL Pool error: {0}")]
SqlPoolError(#[from] yaak_database::PoolError),
#[error(
"Timed out after {waited_ms}ms waiting for a {what} database connection ({connections} open, {idle} idle)"
)]
PoolTimeout {
what: &'static str,
waited_ms: u128,
connections: u32,
idle: u32,
#[source]
source: yaak_database::PoolError,
},
#[error("Database error: {0}")]
Database(String),
+50 -16
View File
@@ -10,14 +10,13 @@ use yaak_database::SqlitePool;
pub mod blob_manager;
pub mod client_db;
pub mod cookies;
mod connection_or_tx;
pub mod cookies;
pub mod error;
pub mod migrate;
pub mod models;
pub mod models_ops;
pub mod path_placeholders;
mod pool;
pub mod queries;
pub mod query_manager;
pub mod render;
@@ -65,6 +64,17 @@ mod open {
.map_err(|e| Error::Database(e.to_string()))
}
/// `(readers, writer)` over one file: a pool of `max_size` readers and a
/// pool of exactly one writer.
pub fn file_pools(
path: impl Into<PathBuf>,
max_size: u32,
min_idle: u32,
) -> Result<(SqlitePool, SqlitePool)> {
let path: PathBuf = path.into();
Ok((file_pool(&path, max_size, min_idle)?, file_pool(&path, 1, 1)?))
}
pub fn memory_pool() -> Result<SqlitePool> {
let manager = SqliteConnectionManager::memory().with_init(|c| init_connection(c));
// In-memory DB doesn't support multiple connections
@@ -91,6 +101,16 @@ mod open {
Ok(SqlitePool::single(conn))
}
/// One connection is all a browser VFS allows, so it reads and writes.
pub fn file_pools(
path: impl Into<PathBuf>,
max_size: u32,
min_idle: u32,
) -> Result<(SqlitePool, SqlitePool)> {
let pool = file_pool(path, max_size, min_idle)?;
Ok((pool.clone(), pool))
}
pub fn memory_pool() -> Result<SqlitePool> {
let conn = Connection::open_in_memory()?;
init_connection(&conn)?;
@@ -109,29 +129,33 @@ pub fn init_standalone(
let db_path = db_path.as_ref();
let blob_path = blob_path.as_ref();
// Main database pool. Sized for concurrent in-flight commands, not concurrent app
// features: connections are held per-statement, but one command can nest a couple of
// them, and since every frontend command dispatches as its own task they all land at
// once. Only `min_idle` is paid up front — r2d2 opens the rest on demand — and the
// app raises its open-file limit to 10240 at startup, so the old 256 fd ceiling that
// kept this small no longer applies.
// Each database gets a reader pool and a one-connection writer pool; see
// `QueryManager` for why. Reader pools are sized for concurrent in-flight
// queries, not concurrent app features — connections are held per-statement,
// so even heavy fan-out (e.g. many gRPC streams) only needs a handful at once.
// Keep them modest: WAL connections hold ~3 file descriptors each, and macOS
// GUI apps get a 256 fd soft limit.
info!("Initializing app database {db_path:?}");
let pool = open::file_pool(db_path, 50, 2)?;
migrate_db(&pool)?;
let (readers, writer) = open::file_pools(db_path, 20, 2)?;
migrate_db(&writer)?;
info!("Initializing blobs database {blob_path:?}");
let blob_pool = open::file_pool(blob_path, 10, 1)?;
migrate_blob_db(&blob_pool)?;
let (blob_readers, blob_writer) = open::file_pools(blob_path, 10, 1)?;
migrate_blob_db(&blob_writer)?;
let (tx, rx) = mpsc::channel();
let query_manager = QueryManager::new(pool, tx);
let blob_manager = BlobManager::new(blob_pool);
let query_manager = QueryManager::new(readers, writer, tx);
let blob_manager = BlobManager::new(blob_readers, blob_writer);
bootstrap(&query_manager)?;
Ok((query_manager, blob_manager, rx))
}
/// Initialize the database managers with in-memory SQLite databases.
/// Useful for testing and CI environments.
///
/// An in-memory database is private to its connection, so the one connection
/// is both the reader pool and the writer.
pub fn init_in_memory() -> Result<(QueryManager, BlobManager, mpsc::Receiver<ModelPayload>)> {
let pool = open::memory_pool()?;
migrate_db(&pool)?;
@@ -140,8 +164,18 @@ pub fn init_in_memory() -> Result<(QueryManager, BlobManager, mpsc::Receiver<Mod
migrate_blob_db(&blob_pool)?;
let (tx, rx) = mpsc::channel();
let query_manager = QueryManager::new(pool, tx);
let blob_manager = BlobManager::new(blob_pool);
let query_manager = QueryManager::new(pool.clone(), pool, tx);
let blob_manager = BlobManager::new(blob_pool.clone(), blob_pool);
bootstrap(&query_manager)?;
Ok((query_manager, blob_manager, rx))
}
/// The rows every client assumes exist: settings and at least one workspace.
fn bootstrap(query_manager: &QueryManager) -> Result<()> {
query_manager.with_tx(|tx| {
tx.ensure_settings()?;
tx.ensure_default_workspace()?;
Ok(())
})
}
+4 -4
View File
@@ -8,14 +8,14 @@
//! caller is a desktop window or an HTTP request.
use crate::blob_manager::BlobManager;
use crate::client_db::ClientDb;
use crate::client_db::WriteDb;
use crate::error::Error::GenericError;
use crate::error::Result;
use crate::models::AnyModel;
use crate::util::UpdateSource;
pub fn upsert_model(
db: &ClientDb,
db: &WriteDb,
blobs: &BlobManager,
model: AnyModel,
source: &UpdateSource,
@@ -41,7 +41,7 @@ pub fn upsert_model(
/// Deletes cascade, so callers run this inside a transaction.
pub fn delete_model(
tx: &ClientDb,
tx: &WriteDb,
blobs: &BlobManager,
model: AnyModel,
source: &UpdateSource,
@@ -69,7 +69,7 @@ pub fn delete_model(
/// 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,
tx: &WriteDb,
model_type: &str,
model_id: &str,
source: &UpdateSource,
-48
View File
@@ -1,48 +0,0 @@
//! One way in and out of a connection pool, so every acquisition is timed the same way.
//!
//! A pool that cannot hand out a connection fails with a timeout and nothing else: r2d2
//! reports `Error(None)` whether all the connections are checked out or none could be
//! opened in the first place. Those have nothing to do with each other, so the state of
//! the pool at the moment it gave up rides along in the error and in the log line.
use crate::error::{Error, Result};
use log::warn;
use std::time::{Duration, Instant};
use yaak_database::{SqliteConn, SqlitePool, pool_status};
/// Long enough that a healthy app never logs, short enough to catch a pool going bad
/// well before the acquire timeout turns it into an error.
const SLOW_ACQUIRE: Duration = Duration::from_secs(1);
pub(crate) fn acquire(pool: &SqlitePool, what: &'static str) -> Result<SqliteConn> {
let started = Instant::now();
let result = pool.get();
let waited = started.elapsed();
match result {
Ok(conn) => {
if waited >= SLOW_ACQUIRE {
warn!(
"Waited {}ms for a {what} connection ({})",
waited.as_millis(),
pool_status(pool)
);
}
Ok(conn)
}
Err(source) => {
let status = pool_status(pool);
warn!(
"Gave up after {}ms waiting for a {what} connection ({status})",
waited.as_millis()
);
Err(Error::PoolTimeout {
what,
waited_ms: waited.as_millis(),
connections: status.connections,
idle: status.idle,
source,
})
}
}
}
+2 -2
View File
@@ -1,10 +1,10 @@
use crate::client_db::ClientDb;
use crate::client_db::WriteDb;
use crate::error::Result;
use crate::models::{Environment, Folder, GrpcRequest, HttpRequest, WebsocketRequest, Workspace};
use crate::util::{BatchUpsertResult, UpdateSource};
use log::info;
impl<'a> ClientDb<'a> {
impl<'a> WriteDb<'a> {
pub fn batch_upsert(
&self,
workspaces: Vec<Workspace>,
+10 -6
View File
@@ -1,4 +1,4 @@
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{CookieJar, CookieJarIden};
use crate::util::UpdateSource;
@@ -9,18 +9,22 @@ impl<'a> ClientDb<'a> {
}
pub fn list_cookie_jars(&self, workspace_id: &str) -> Result<Vec<CookieJar>> {
let mut cookie_jars = self.find_many(CookieJarIden::WorkspaceId, workspace_id, None)?;
self.find_many(CookieJarIden::WorkspaceId, workspace_id, None)
}
}
if cookie_jars.is_empty() {
impl<'a> WriteDb<'a> {
/// A workspace with no cookie jar gets a default one.
pub fn ensure_default_cookie_jar(&self, workspace_id: &str) -> Result<()> {
if self.list_cookie_jars(workspace_id)?.is_empty() {
let jar = CookieJar {
name: "Default".to_string(),
workspace_id: workspace_id.to_string(),
..Default::default()
};
cookie_jars.push(self.upsert_cookie_jar(&jar, &UpdateSource::Background)?);
self.upsert_cookie_jar(&jar, &UpdateSource::Background)?;
}
Ok(cookie_jars)
Ok(())
}
pub fn delete_cookie_jar(
+84 -82
View File
@@ -1,5 +1,5 @@
use super::conflict_free_name;
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Error::{MissingBaseEnvironment, MultipleBaseEnvironments};
use crate::error::Result;
use crate::models::{Environment, EnvironmentIden, EnvironmentVariable};
@@ -20,7 +20,7 @@ impl<'a> ClientDb<'a> {
}
pub fn get_base_environment(&self, workspace_id: &str) -> Result<Environment> {
let environments = self.list_environments_ensure_base(workspace_id)?;
let environments = self.list_environments(workspace_id)?;
let base_environments = environments
.into_iter()
.filter(|e| e.parent_model == "workspace")
@@ -30,38 +30,96 @@ impl<'a> ClientDb<'a> {
return Err(MultipleBaseEnvironments(workspace_id.to_string()));
}
Ok(base_environments.first().cloned().ok_or(
// Should never happen because one should be created above if it does not exist
MissingBaseEnvironment(workspace_id.to_string()),
)?)
Ok(base_environments
.first()
.cloned()
.ok_or(MissingBaseEnvironment(workspace_id.to_string()))?)
}
/// Lists environments and will create a base environment if one doesn't exist
pub fn list_environments_ensure_base(&self, workspace_id: &str) -> Result<Vec<Environment>> {
let mut environments = self.list_environments_dangerous(workspace_id)?;
pub fn list_environments(&self, workspace_id: &str) -> Result<Vec<Environment>> {
Ok(self.find_many::<Environment>(EnvironmentIden::WorkspaceId, workspace_id, None)?)
}
let base_environment = environments.iter().find(|e| e.parent_model == "workspace");
/// Find other environments with the same parent folder
fn list_duplicate_folder_environments(&self, environment: &Environment) -> Vec<Environment> {
if environment.parent_model != "folder" {
return Vec::new();
}
if let None = base_environment {
let e = self.upsert_environment(
&Environment {
workspace_id: workspace_id.to_string(),
name: "Global Variables".to_string(),
parent_model: "workspace".to_string(),
..Default::default()
},
&UpdateSource::Background,
self.list_environments(&environment.workspace_id)
.unwrap_or_default()
.into_iter()
.filter(|e| {
e.id != environment.id
&& e.parent_model == "folder"
&& e.parent_id == environment.parent_id
})
.collect()
}
pub fn resolve_environments(
&self,
workspace_id: &str,
folder_id: Option<&str>,
active_environment_id: Option<&str>,
) -> Result<Vec<Environment>> {
let mut environments = Vec::new();
if let Some(folder_id) = folder_id {
let folder = self.get_folder(folder_id)?;
// Add current folder's environment
if let Some(e) = self.get_environment_by_folder_id(folder_id)? {
environments.push(e);
};
// Recurse up
let ancestors = self.resolve_environments(
workspace_id,
folder.folder_id.as_deref(),
active_environment_id,
)?;
info!("Created base environment {} for {workspace_id}", e.id);
environments.push(e);
environments.extend(ancestors);
} else {
// Add active and base environments
if let Some(id) = active_environment_id {
if let Ok(e) = self.get_environment(&id) {
// Add active sub environment
environments.push(e);
};
};
// Add the base environment. A workspace that has never been
// opened has none yet; it simply contributes no variables.
match self.get_base_environment(workspace_id) {
Ok(e) => environments.push(e),
Err(MissingBaseEnvironment(_)) => {}
Err(e) => return Err(e),
}
}
Ok(environments)
}
}
/// List environments for a workspace. Prefer list_environments_ensure_base()
fn list_environments_dangerous(&self, workspace_id: &str) -> Result<Vec<Environment>> {
Ok(self.find_many::<Environment>(EnvironmentIden::WorkspaceId, workspace_id, None)?)
impl<'a> WriteDb<'a> {
/// The workspace's base environment, created if it does not exist.
pub fn ensure_base_environment(&self, workspace_id: &str) -> Result<Environment> {
match self.get_base_environment(workspace_id) {
Err(MissingBaseEnvironment(_)) => {}
other => return other,
}
let e = self.upsert_environment(
&Environment {
workspace_id: workspace_id.to_string(),
name: "Global Variables".to_string(),
parent_model: "workspace".to_string(),
..Default::default()
},
&UpdateSource::Background,
)?;
info!("Created base environment {} for {workspace_id}", e.id);
Ok(e)
}
pub fn delete_environment(
@@ -72,7 +130,7 @@ impl<'a> ClientDb<'a> {
let deleted_environment = self.delete(environment, source)?;
// Recreate the base environment if we happened to delete it
self.list_environments_ensure_base(&environment.workspace_id)?;
self.ensure_base_environment(&environment.workspace_id)?;
Ok(deleted_environment)
}
@@ -90,7 +148,7 @@ impl<'a> ClientDb<'a> {
let mut environment = environment.clone();
environment.id = "".to_string();
let sibling_names = self
.list_environments_dangerous(&environment.workspace_id)?
.list_environments(&environment.workspace_id)?
.into_iter()
.map(|e| e.name)
.collect::<Vec<_>>();
@@ -98,23 +156,6 @@ impl<'a> ClientDb<'a> {
self.upsert_environment(&environment, source)
}
/// Find other environments with the same parent folder
fn list_duplicate_folder_environments(&self, environment: &Environment) -> Vec<Environment> {
if environment.parent_model != "folder" {
return Vec::new();
}
self.list_environments_dangerous(&environment.workspace_id)
.unwrap_or_default()
.into_iter()
.filter(|e| {
e.id != environment.id
&& e.parent_model == "folder"
&& e.parent_id == environment.parent_id
})
.collect()
}
pub fn upsert_environment(
&self,
environment: &Environment,
@@ -154,43 +195,4 @@ impl<'a> ClientDb<'a> {
source,
)
}
pub fn resolve_environments(
&self,
workspace_id: &str,
folder_id: Option<&str>,
active_environment_id: Option<&str>,
) -> Result<Vec<Environment>> {
let mut environments = Vec::new();
if let Some(folder_id) = folder_id {
let folder = self.get_folder(folder_id)?;
// Add current folder's environment
if let Some(e) = self.get_environment_by_folder_id(folder_id)? {
environments.push(e);
};
// Recurse up
let ancestors = self.resolve_environments(
workspace_id,
folder.folder_id.as_deref(),
active_environment_id,
)?;
environments.extend(ancestors);
} else {
// Add active and base environments
if let Some(id) = active_environment_id {
if let Ok(e) = self.get_environment(&id) {
// Add active sub environment
environments.push(e);
};
};
// Add the base environment
environments.push(self.get_base_environment(workspace_id)?);
}
Ok(environments)
}
}
+96 -94
View File
@@ -1,5 +1,5 @@
use super::{conflict_free_name, merge_headers};
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::connection_or_tx::ConnectionOrTx;
use crate::error::Result;
use crate::models::{
@@ -20,99 +20,6 @@ impl<'a> ClientDb<'a> {
self.find_many(FolderIden::WorkspaceId, workspace_id, None)
}
pub fn delete_folder(&self, folder: &Folder, source: &UpdateSource) -> Result<Folder> {
match self.conn() {
ConnectionOrTx::Connection(_) => {}
ConnectionOrTx::Transaction(_) => {}
}
let fid = &folder.id;
for m in self.find_many::<HttpRequest>(HttpRequestIden::FolderId, fid, None)? {
self.delete_http_request(&m, source)?;
}
for m in self.find_many::<GrpcRequest>(GrpcRequestIden::FolderId, fid, None)? {
self.delete_grpc_request(&m, source)?;
}
for m in self.find_many::<WebsocketRequest>(WebsocketRequestIden::FolderId, fid, None)? {
self.delete_websocket_request(&m, source)?;
}
for e in self.find_many(EnvironmentIden::ParentId, fid, None)? {
self.delete_environment(&e, source)?;
}
// Recurse down into child folders
for folder in self.find_many::<Folder>(FolderIden::FolderId, fid, None)? {
self.delete_folder(&folder, source)?;
}
self.delete(folder, source)
}
pub fn delete_folder_by_id(&self, id: &str, source: &UpdateSource) -> Result<Folder> {
let folder = self.get_folder(id)?;
self.delete_folder(&folder, source)
}
pub fn upsert_folder(&self, folder: &Folder, source: &UpdateSource) -> Result<Folder> {
self.upsert(folder, source)
}
pub fn duplicate_folder(&self, src_folder: &Folder, source: &UpdateSource) -> Result<Folder> {
let fid = &src_folder.id;
let mut folder = Folder {
id: "".into(),
sort_priority: src_folder.sort_priority + 0.001,
..src_folder.clone()
};
let sibling_names = self
.list_folders(&folder.workspace_id)?
.into_iter()
.filter(|f| f.folder_id == folder.folder_id)
.map(|f| f.name)
.collect::<Vec<_>>();
folder.name = conflict_free_name(&folder.name, &sibling_names);
let new_folder = self.upsert_folder(&folder, source)?;
for m in self.find_many::<HttpRequest>(HttpRequestIden::FolderId, fid, None)? {
self.upsert_http_request(
&HttpRequest { id: "".into(), folder_id: Some(new_folder.id.clone()), ..m },
source,
)?;
}
for m in self.find_many::<WebsocketRequest>(WebsocketRequestIden::FolderId, fid, None)? {
self.upsert_websocket_request(
&WebsocketRequest { id: "".into(), folder_id: Some(new_folder.id.clone()), ..m },
source,
)?;
}
for m in self.find_many::<GrpcRequest>(GrpcRequestIden::FolderId, fid, None)? {
self.upsert_grpc_request(
&GrpcRequest { id: "".into(), folder_id: Some(new_folder.id.clone()), ..m },
source,
)?;
}
for m in self.find_many::<Environment>(EnvironmentIden::ParentId, fid, None)? {
self.upsert_environment(
&Environment { id: "".into(), parent_id: Some(new_folder.id.clone()), ..m },
source,
)?;
}
for m in self.find_many::<Folder>(FolderIden::FolderId, fid, None)? {
// Recurse down
self.duplicate_folder(&Folder { folder_id: Some(new_folder.id.clone()), ..m }, source)?;
}
Ok(new_folder)
}
pub fn resolve_auth_for_folder(
&self,
folder: &Folder,
@@ -219,3 +126,98 @@ impl<'a> ClientDb<'a> {
})
}
}
impl<'a> WriteDb<'a> {
pub fn delete_folder(&self, folder: &Folder, source: &UpdateSource) -> Result<Folder> {
match self.conn() {
ConnectionOrTx::Connection(_) => {}
ConnectionOrTx::Transaction(_) => {}
}
let fid = &folder.id;
for m in self.find_many::<HttpRequest>(HttpRequestIden::FolderId, fid, None)? {
self.delete_http_request(&m, source)?;
}
for m in self.find_many::<GrpcRequest>(GrpcRequestIden::FolderId, fid, None)? {
self.delete_grpc_request(&m, source)?;
}
for m in self.find_many::<WebsocketRequest>(WebsocketRequestIden::FolderId, fid, None)? {
self.delete_websocket_request(&m, source)?;
}
for e in self.find_many(EnvironmentIden::ParentId, fid, None)? {
self.delete_environment(&e, source)?;
}
// Recurse down into child folders
for folder in self.find_many::<Folder>(FolderIden::FolderId, fid, None)? {
self.delete_folder(&folder, source)?;
}
self.delete(folder, source)
}
pub fn delete_folder_by_id(&self, id: &str, source: &UpdateSource) -> Result<Folder> {
let folder = self.get_folder(id)?;
self.delete_folder(&folder, source)
}
pub fn upsert_folder(&self, folder: &Folder, source: &UpdateSource) -> Result<Folder> {
self.upsert(folder, source)
}
pub fn duplicate_folder(&self, src_folder: &Folder, source: &UpdateSource) -> Result<Folder> {
let fid = &src_folder.id;
let mut folder = Folder {
id: "".into(),
sort_priority: src_folder.sort_priority + 0.001,
..src_folder.clone()
};
let sibling_names = self
.list_folders(&folder.workspace_id)?
.into_iter()
.filter(|f| f.folder_id == folder.folder_id)
.map(|f| f.name)
.collect::<Vec<_>>();
folder.name = conflict_free_name(&folder.name, &sibling_names);
let new_folder = self.upsert_folder(&folder, source)?;
for m in self.find_many::<HttpRequest>(HttpRequestIden::FolderId, fid, None)? {
self.upsert_http_request(
&HttpRequest { id: "".into(), folder_id: Some(new_folder.id.clone()), ..m },
source,
)?;
}
for m in self.find_many::<WebsocketRequest>(WebsocketRequestIden::FolderId, fid, None)? {
self.upsert_websocket_request(
&WebsocketRequest { id: "".into(), folder_id: Some(new_folder.id.clone()), ..m },
source,
)?;
}
for m in self.find_many::<GrpcRequest>(GrpcRequestIden::FolderId, fid, None)? {
self.upsert_grpc_request(
&GrpcRequest { id: "".into(), folder_id: Some(new_folder.id.clone()), ..m },
source,
)?;
}
for m in self.find_many::<Environment>(EnvironmentIden::ParentId, fid, None)? {
self.upsert_environment(
&Environment { id: "".into(), parent_id: Some(new_folder.id.clone()), ..m },
source,
)?;
}
for m in self.find_many::<Folder>(FolderIden::FolderId, fid, None)? {
// Recurse down
self.duplicate_folder(&Folder { folder_id: Some(new_folder.id.clone()), ..m }, source)?;
}
Ok(new_folder)
}
}
@@ -1,4 +1,4 @@
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{GraphQlIntrospection, GraphQlIntrospectionIden};
use crate::util::UpdateSource;
@@ -11,7 +11,9 @@ impl<'a> ClientDb<'a> {
pub fn get_graphql_introspection(&self, request_id: &str) -> Option<GraphQlIntrospection> {
self.find_optional(GraphQlIntrospectionIden::RequestId, request_id)
}
}
impl<'a> WriteDb<'a> {
pub fn upsert_graphql_introspection(
&self,
workspace_id: &str,
@@ -1,4 +1,4 @@
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{GrpcConnection, GrpcConnectionIden, GrpcConnectionState};
use crate::queries::MAX_HISTORY_ITEMS;
@@ -13,6 +13,20 @@ impl<'a> ClientDb<'a> {
self.find_one(GrpcConnectionIden::Id, id)
}
pub fn list_grpc_connections_for_request(
&self,
request_id: &str,
limit: Option<u64>,
) -> Result<Vec<GrpcConnection>> {
self.find_many(GrpcConnectionIden::RequestId, request_id, limit)
}
pub fn list_grpc_connections(&self, workspace_id: &str) -> Result<Vec<GrpcConnection>> {
self.find_many(GrpcConnectionIden::WorkspaceId, workspace_id, None)
}
}
impl<'a> WriteDb<'a> {
pub fn delete_all_grpc_connections_for_request(
&self,
request_id: &str,
@@ -53,18 +67,6 @@ impl<'a> ClientDb<'a> {
self.delete_grpc_connection(&grpc_connection, source)
}
pub fn list_grpc_connections_for_request(
&self,
request_id: &str,
limit: Option<u64>,
) -> Result<Vec<GrpcConnection>> {
self.find_many(GrpcConnectionIden::RequestId, request_id, limit)
}
pub fn list_grpc_connections(&self, workspace_id: &str) -> Result<Vec<GrpcConnection>> {
self.find_many(GrpcConnectionIden::WorkspaceId, workspace_id, None)
}
pub fn cancel_pending_grpc_connections(&self) -> Result<()> {
let closed = serde_json::to_value(&GrpcConnectionState::Closed)?;
let (sql, params) = Query::update()
@@ -1,4 +1,4 @@
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{GrpcEvent, GrpcEventIden};
use crate::util::UpdateSource;
@@ -11,7 +11,9 @@ impl<'a> ClientDb<'a> {
pub fn list_grpc_events(&self, connection_id: &str) -> Result<Vec<GrpcEvent>> {
self.find_many(GrpcEventIden::ConnectionId, connection_id, None)
}
}
impl<'a> WriteDb<'a> {
pub fn upsert_grpc_event(
&self,
grpc_event: &GrpcEvent,
+47 -45
View File
@@ -1,5 +1,5 @@
use super::{conflict_free_name, merge_headers};
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{
AnyModel, Folder, FolderIden, GrpcRequest, GrpcRequestIden, HttpRequestHeader,
@@ -32,50 +32,6 @@ impl<'a> ClientDb<'a> {
Ok(children)
}
pub fn delete_grpc_request(
&self,
m: &GrpcRequest,
source: &UpdateSource,
) -> Result<GrpcRequest> {
self.delete_all_grpc_connections_for_request(m.id.as_str(), source)?;
self.delete(m, source)
}
pub fn delete_grpc_request_by_id(
&self,
id: &str,
source: &UpdateSource,
) -> Result<GrpcRequest> {
let request = self.get_grpc_request(id)?;
self.delete_grpc_request(&request, source)
}
pub fn duplicate_grpc_request(
&self,
grpc_request: &GrpcRequest,
source: &UpdateSource,
) -> Result<GrpcRequest> {
let mut request = grpc_request.clone();
request.id = "".to_string();
request.sort_priority = request.sort_priority + 0.001;
let sibling_names = self
.list_grpc_requests(&request.workspace_id)?
.into_iter()
.filter(|m| m.folder_id == request.folder_id)
.map(|m| m.name)
.collect::<Vec<_>>();
request.name = conflict_free_name(&request.name, &sibling_names);
self.upsert(&request, source)
}
pub fn upsert_grpc_request(
&self,
grpc_request: &GrpcRequest,
source: &UpdateSource,
) -> Result<GrpcRequest> {
self.upsert(grpc_request, source)
}
pub fn resolve_auth_for_grpc_request(
&self,
grpc_request: &GrpcRequest,
@@ -146,3 +102,49 @@ impl<'a> ClientDb<'a> {
})
}
}
impl<'a> WriteDb<'a> {
pub fn delete_grpc_request(
&self,
m: &GrpcRequest,
source: &UpdateSource,
) -> Result<GrpcRequest> {
self.delete_all_grpc_connections_for_request(m.id.as_str(), source)?;
self.delete(m, source)
}
pub fn delete_grpc_request_by_id(
&self,
id: &str,
source: &UpdateSource,
) -> Result<GrpcRequest> {
let request = self.get_grpc_request(id)?;
self.delete_grpc_request(&request, source)
}
pub fn duplicate_grpc_request(
&self,
grpc_request: &GrpcRequest,
source: &UpdateSource,
) -> Result<GrpcRequest> {
let mut request = grpc_request.clone();
request.id = "".to_string();
request.sort_priority = request.sort_priority + 0.001;
let sibling_names = self
.list_grpc_requests(&request.workspace_id)?
.into_iter()
.filter(|m| m.folder_id == request.folder_id)
.map(|m| m.name)
.collect::<Vec<_>>();
request.name = conflict_free_name(&request.name, &sibling_names);
self.upsert(&request, source)
}
pub fn upsert_grpc_request(
&self,
grpc_request: &GrpcRequest,
source: &UpdateSource,
) -> Result<GrpcRequest> {
self.upsert(grpc_request, source)
}
}
+112 -99
View File
@@ -1,5 +1,5 @@
use super::{conflict_free_name, merge_headers};
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{
AnyModel, Folder, FolderIden, HttpRequest, HttpRequestHeader, HttpRequestIden,
@@ -18,50 +18,6 @@ impl<'a> ClientDb<'a> {
self.find_many(HttpRequestIden::WorkspaceId, workspace_id, None)
}
pub fn delete_http_request(
&self,
m: &HttpRequest,
source: &UpdateSource,
) -> Result<HttpRequest> {
self.delete_all_http_responses_for_request(m.id.as_str(), source)?;
self.delete(m, source)
}
pub fn delete_http_request_by_id(
&self,
id: &str,
source: &UpdateSource,
) -> Result<HttpRequest> {
let http_request = self.get_http_request(id)?;
self.delete_http_request(&http_request, source)
}
pub fn duplicate_http_request(
&self,
http_request: &HttpRequest,
source: &UpdateSource,
) -> Result<HttpRequest> {
let mut http_request = http_request.clone();
http_request.id = "".to_string();
http_request.sort_priority = http_request.sort_priority + 0.001;
let sibling_names = self
.list_http_requests(&http_request.workspace_id)?
.into_iter()
.filter(|m| m.folder_id == http_request.folder_id)
.map(|m| m.name)
.collect::<Vec<_>>();
http_request.name = conflict_free_name(&http_request.name, &sibling_names);
self.upsert(&http_request, source)
}
pub fn upsert_http_request(
&self,
http_request: &HttpRequest,
source: &UpdateSource,
) -> Result<HttpRequest> {
self.upsert(http_request, source)
}
pub fn resolve_auth_for_http_request(
&self,
http_request: &HttpRequest,
@@ -179,6 +135,52 @@ impl<'a> ClientDb<'a> {
}
}
impl<'a> WriteDb<'a> {
pub fn delete_http_request(
&self,
m: &HttpRequest,
source: &UpdateSource,
) -> Result<HttpRequest> {
self.delete_all_http_responses_for_request(m.id.as_str(), source)?;
self.delete(m, source)
}
pub fn delete_http_request_by_id(
&self,
id: &str,
source: &UpdateSource,
) -> Result<HttpRequest> {
let http_request = self.get_http_request(id)?;
self.delete_http_request(&http_request, source)
}
pub fn duplicate_http_request(
&self,
http_request: &HttpRequest,
source: &UpdateSource,
) -> Result<HttpRequest> {
let mut http_request = http_request.clone();
http_request.id = "".to_string();
http_request.sort_priority = http_request.sort_priority + 0.001;
let sibling_names = self
.list_http_requests(&http_request.workspace_id)?
.into_iter()
.filter(|m| m.folder_id == http_request.folder_id)
.map(|m| m.name)
.collect::<Vec<_>>();
http_request.name = conflict_free_name(&http_request.name, &sibling_names);
self.upsert(&http_request, source)
}
pub fn upsert_http_request(
&self,
http_request: &HttpRequest,
source: &UpdateSource,
) -> Result<HttpRequest> {
self.upsert(http_request, source)
}
}
#[cfg(test)]
mod tests {
use crate::init_in_memory;
@@ -190,7 +192,7 @@ mod tests {
#[test]
fn request_resolution_preserves_duplicate_request_headers() {
let (query_manager, _blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
let db = query_manager.connect().unwrap();
let db = query_manager.connect();
let workspace = db.list_workspaces().expect("Failed to list workspaces").remove(0);
let request = HttpRequest {
workspace_id: workspace.id,
@@ -225,72 +227,83 @@ mod tests {
#[test]
fn http_version_resolves_through_the_inheritance_chain() {
let (query_manager, _blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
let db = query_manager.connect().unwrap();
let source = &UpdateSource::Background;
let workspace = db
.upsert_workspace(
&Workspace {
name: "Test".to_string(),
setting_http_version: HttpVersion::Http2,
..Default::default()
},
&UpdateSource::Background,
)
.expect("Failed to upsert workspace");
let folder = db
.upsert_folder(
&Folder { workspace_id: workspace.id.clone(), ..Default::default() },
&UpdateSource::Background,
)
.expect("Failed to upsert folder");
let request = db
.upsert_http_request(
&HttpRequest {
workspace_id: workspace.id.clone(),
folder_id: Some(folder.id.clone()),
..Default::default()
},
&UpdateSource::Background,
)
.expect("Failed to upsert request");
let (folder, request) = query_manager
.with_tx(|db| {
let workspace = db.upsert_workspace(
&Workspace {
name: "Test".to_string(),
setting_http_version: HttpVersion::Http2,
..Default::default()
},
source,
)?;
let folder = db.upsert_folder(
&Folder { workspace_id: workspace.id.clone(), ..Default::default() },
source,
)?;
let request = db.upsert_http_request(
&HttpRequest {
workspace_id: workspace.id.clone(),
folder_id: Some(folder.id.clone()),
..Default::default()
},
source,
)?;
Ok::<_, crate::error::Error>((folder, request))
})
.expect("Failed to seed");
// No overrides, so the workspace base value applies
let resolved = db.resolve_settings_for_http_request(&request).expect("Failed to resolve");
let resolved = query_manager
.connect()
.resolve_settings_for_http_request(&request)
.expect("Failed to resolve");
assert_eq!(resolved.http_version.value, HttpVersion::Http2);
assert_eq!(resolved.http_version.source_model, "workspace");
// A folder override beats the workspace base
db.upsert_folder(
&Folder {
setting_http_version: InheritedHttpVersionSetting {
enabled: true,
value: HttpVersion::Http1,
},
..folder
},
&UpdateSource::Background,
)
.expect("Failed to update folder");
let resolved = db.resolve_settings_for_http_request(&request).expect("Failed to resolve");
query_manager
.with_tx(|db| {
db.upsert_folder(
&Folder {
setting_http_version: InheritedHttpVersionSetting {
enabled: true,
value: HttpVersion::Http1,
},
..folder
},
source,
)
})
.expect("Failed to update folder");
let resolved = query_manager
.connect()
.resolve_settings_for_http_request(&request)
.expect("Failed to resolve");
assert_eq!(resolved.http_version.value, HttpVersion::Http1);
assert_eq!(resolved.http_version.source_model, "folder");
// A request override beats them both
let request = db
.upsert_http_request(
&HttpRequest {
setting_http_version: InheritedHttpVersionSetting {
enabled: true,
value: HttpVersion::Auto,
let request = query_manager
.with_tx(|db| {
db.upsert_http_request(
&HttpRequest {
setting_http_version: InheritedHttpVersionSetting {
enabled: true,
value: HttpVersion::Auto,
},
..request
},
..request
},
&UpdateSource::Background,
)
source,
)
})
.expect("Failed to update request");
let resolved = db.resolve_settings_for_http_request(&request).expect("Failed to resolve");
let resolved = query_manager
.connect()
.resolve_settings_for_http_request(&request)
.expect("Failed to resolve");
assert_eq!(resolved.http_version.value, HttpVersion::Auto);
assert_eq!(resolved.http_version.source_model, "http_request");
}
@@ -1,4 +1,4 @@
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{HttpResponseEvent, HttpResponseEventIden};
use crate::util::UpdateSource;
@@ -7,7 +7,9 @@ impl<'a> ClientDb<'a> {
pub fn list_http_response_events(&self, response_id: &str) -> Result<Vec<HttpResponseEvent>> {
self.find_many(HttpResponseEventIden::ResponseId, response_id, None)
}
}
impl<'a> WriteDb<'a> {
pub fn upsert_http_response_event(
&self,
http_response_event: &HttpResponseEvent,
@@ -1,5 +1,5 @@
use crate::blob_manager::BlobManager;
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{HttpResponse, HttpResponseIden, HttpResponseState};
use crate::queries::MAX_HISTORY_ITEMS;
@@ -31,20 +31,6 @@ impl<'a> ClientDb<'a> {
self.find_many(HttpResponseIden::WorkspaceId, workspace_id, limit)
}
/// Returns the number of responses deleted.
pub fn delete_all_http_responses_for_request(
&self,
request_id: &str,
source: &UpdateSource,
) -> Result<usize> {
let responses = self.list_http_responses_for_request(request_id, None)?;
let count = responses.len();
for m in responses {
self.delete(&m, source)?;
}
Ok(count)
}
/// Delete blob-stored response bodies whose owning HTTP response row no
/// longer exists. Blob ids are keyed by the response that owns them —
/// "{response_id}" for a response body, "{response_id}.request" for the
@@ -55,19 +41,24 @@ impl<'a> ClientDb<'a> {
///
/// Returns the number of orphaned bodies deleted.
pub fn delete_orphaned_response_body_blobs(&self, blobs: &BlobManager) -> Result<usize> {
let mut deleted = 0;
let orphaned = blobs
.connect()
.list_body_ids()?
.into_iter()
.filter(|body_id| {
let response_id = body_id.split('.').next().unwrap_or_default();
self.find_optional::<HttpResponse>(HttpResponseIden::Id, response_id).is_none()
})
.collect::<Vec<_>>();
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::<HttpResponse>(HttpResponseIden::Id, response_id).is_some() {
continue;
blobs.with_tx(|b| {
for body_id in &orphaned {
b.delete_chunks(body_id)?;
}
blob_ctx.delete_chunks(&body_id)?;
deleted += 1;
}
Ok::<_, crate::error::Error>(())
})?;
Ok(deleted)
Ok(orphaned.len())
}
/// Delete response body data (blob chunks and body files) whose owning HTTP
@@ -96,8 +87,7 @@ impl<'a> ClientDb<'a> {
let Some(response_id) = path.file_name().and_then(|n| n.to_str()) else {
continue;
};
if self.find_optional::<HttpResponse>(HttpResponseIden::Id, response_id).is_some()
{
if self.find_optional::<HttpResponse>(HttpResponseIden::Id, response_id).is_some() {
continue;
}
if fs::remove_file(&path).is_ok() {
@@ -108,6 +98,22 @@ impl<'a> ClientDb<'a> {
Ok(deleted)
}
}
impl<'a> WriteDb<'a> {
/// Returns the number of responses deleted.
pub fn delete_all_http_responses_for_request(
&self,
request_id: &str,
source: &UpdateSource,
) -> Result<usize> {
let responses = self.list_http_responses_for_request(request_id, None)?;
let count = responses.len();
for m in responses {
self.delete(&m, source)?;
}
Ok(count)
}
/// Returns the number of responses deleted.
pub fn delete_all_http_responses_for_workspace(
@@ -138,9 +144,8 @@ impl<'a> ClientDb<'a> {
}
// Delete request body blobs (pattern: {response_id}.request)
let blob_ctx = blob_manager.connect()?;
let body_id = format!("{}.request", http_response.id);
if let Err(e) = blob_ctx.delete_chunks(&body_id) {
if let Err(e) = blob_manager.with_tx(|b| b.delete_chunks(&body_id)) {
error!("Failed to delete request body blobs: {}", e);
}
@@ -187,36 +192,50 @@ impl<'a> ClientDb<'a> {
#[cfg(test)]
mod tests {
use crate::blob_manager::{BlobManager, BodyChunk};
use crate::client_db::ClientDb;
use crate::error::Error;
use crate::init_in_memory;
use crate::models::{HttpRequest, HttpResponse, Workspace};
use crate::query_manager::QueryManager;
use crate::util::UpdateSource;
/// A workspace, a request, and one response that still exists.
fn seed_live_response(db: &ClientDb, blob_manager: &BlobManager) -> HttpResponse {
fn seed_live_response(
query_manager: &QueryManager,
blob_manager: &BlobManager,
) -> HttpResponse {
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");
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")
query_manager
.with_tx(|db| {
let workspace = db.upsert_workspace(
&Workspace { name: "GC Test".to_string(), ..Default::default() },
source,
)?;
let request = db.upsert_http_request(
&HttpRequest { workspace_id: workspace.id.clone(), ..Default::default() },
source,
)?;
db.upsert_http_response(
&HttpResponse {
request_id: request.id.clone(),
workspace_id: workspace.id.clone(),
..Default::default()
},
source,
blob_manager,
)
})
.expect("Failed to seed response")
}
fn insert_bodies(blob_manager: &BlobManager, body_ids: &[&str]) {
blob_manager
.with_tx(|b| {
for id in body_ids {
b.insert_chunk(&BodyChunk::new(*id, 0, b"data".to_vec()))?;
}
Ok::<_, Error>(())
})
.expect("Failed to insert chunks");
}
/// What a browser host runs: no filesystem, so bodies exist only as blob
@@ -224,28 +243,26 @@ mod tests {
#[test]
fn deletes_orphaned_response_body_blobs() {
let (query_manager, blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
let db = query_manager.connect().unwrap();
let live = seed_live_response(&db, &blob_manager);
let live = seed_live_response(&query_manager, &blob_manager);
let live_request_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().unwrap();
blob_ctx.insert_chunk(&BodyChunk::new(&live.id, 0, b"live".to_vec())).unwrap();
blob_ctx
.insert_chunk(&BodyChunk::new(&live_request_body_id, 0, b"live".to_vec()))
.unwrap();
blob_ctx.insert_chunk(&BodyChunk::new("rs_gone", 0, b"dead".to_vec())).unwrap();
blob_ctx.insert_chunk(&BodyChunk::new("rs_gone.request", 0, b"dead".to_vec())).unwrap();
}
insert_bodies(
&blob_manager,
&[
&live.id,
&live_request_body_id,
"rs_gone",
"rs_gone.request",
],
);
let deleted = db
let deleted = query_manager
.connect()
.delete_orphaned_response_body_blobs(&blob_manager)
.expect("Failed to GC response body blobs");
assert_eq!(deleted, 2);
let blob_ctx = blob_manager.connect().unwrap();
let blob_ctx = blob_manager.connect();
assert!(blob_ctx.body_exists(&live.id).unwrap());
assert!(blob_ctx.body_exists(&live_request_body_id).unwrap());
assert!(!blob_ctx.body_exists("rs_gone").unwrap());
@@ -255,30 +272,24 @@ mod tests {
#[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().unwrap();
let live = seed_live_response(&db, &blob_manager);
let live = seed_live_response(&query_manager, &blob_manager);
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().unwrap();
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();
}
insert_bodies(&blob_manager, &[&live_body_id, "rs_gone.request"]);
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
let deleted = query_manager
.connect()
.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().unwrap();
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());
@@ -1,4 +1,4 @@
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{ImportSourceResource, ImportSourceResourceIden};
use sea_query::ExprTrait;
@@ -20,7 +20,9 @@ impl<'a> ClientDb<'a> {
let items = stmt.query_map(&*params.as_params(), |row| row.try_into())?;
Ok(items.filter_map(|v| v.ok()).collect())
}
}
impl<'a> WriteDb<'a> {
pub fn upsert_import_source_resource(
&self,
resource: &ImportSourceResource,
@@ -1,4 +1,4 @@
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{ImportSource, ImportSourceIden};
use crate::util::UpdateSource;
@@ -25,7 +25,9 @@ impl<'a> ClientDb<'a> {
let sources = self.list_import_sources(workspace_id)?;
Ok(sources.into_iter().find(|s| s.importer == importer && s.origin == origin))
}
}
impl<'a> WriteDb<'a> {
pub fn upsert_import_source(
&self,
import_source: &ImportSource,
+3 -1
View File
@@ -1,4 +1,4 @@
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{KeyValue, KeyValueIden, UpsertModelInfo};
use crate::util::UpdateSource;
@@ -89,7 +89,9 @@ impl<'a> ClientDb<'a> {
.build_rusqlite(SqliteQueryBuilder);
self.conn().resolve().query_row(sql.as_str(), &*params.as_params(), KeyValue::from_row).ok()
}
}
impl<'a> WriteDb<'a> {
pub fn set_key_value_dte(
&self,
namespace: &str,
+1 -1
View File
@@ -24,8 +24,8 @@ mod websocket_events;
mod websocket_requests;
mod workspace_metas;
pub mod workspaces;
pub use model_changes::PersistedModelChange;
pub(crate) use duplicate_name::conflict_free_name;
pub use model_changes::PersistedModelChange;
const MAX_HISTORY_ITEMS: usize = 20;
+96 -55
View File
@@ -1,4 +1,4 @@
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::util::ModelPayload;
use rusqlite::params;
@@ -69,7 +69,9 @@ impl<'a> ClientDb<'a> {
Ok(items.collect::<std::result::Result<Vec<_>, rusqlite::Error>>()?)
}
}
impl<'a> WriteDb<'a> {
pub fn prune_model_changes_older_than_days(&self, days: i64) -> Result<usize> {
let offset = format!("-{days} days");
Ok(self.conn().resolve().execute(
@@ -101,23 +103,36 @@ mod tests {
use crate::util::{ModelChangeEvent, UpdateSource};
use serde_json::json;
/// Startup bootstraps rows of its own; these tests count only their own.
fn clear_changes(query_manager: &crate::query_manager::QueryManager) {
query_manager
.with_tx(|db| {
db.conn().resolve().execute("DELETE FROM model_changes", [])?;
Ok::<_, crate::error::Error>(())
})
.expect("Failed to clear model changes");
}
#[test]
fn records_model_changes_for_upsert_and_delete() {
let (query_manager, blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
let db = query_manager.connect().unwrap();
clear_changes(&query_manager);
let workspace = db
.upsert_workspace(
&Workspace {
name: "Changes Test".to_string(),
setting_follow_redirects: true,
setting_validate_certificates: true,
..Default::default()
},
&UpdateSource::Sync,
)
let workspace = query_manager
.with_tx(|db| {
db.upsert_workspace(
&Workspace {
name: "Changes Test".to_string(),
setting_follow_redirects: true,
setting_validate_certificates: true,
..Default::default()
},
&UpdateSource::Sync,
)
})
.expect("Failed to upsert workspace");
let db = query_manager.connect();
let created_changes = db.list_model_changes_after(0, 10).expect("Failed to list changes");
assert_eq!(created_changes.len(), 1);
assert_eq!(created_changes[0].payload.model.id(), workspace.id);
@@ -128,9 +143,14 @@ mod tests {
));
assert!(matches!(created_changes[0].payload.update_source, UpdateSource::Sync));
db.delete_workspace_by_id(&workspace.id, &UpdateSource::Sync, &blob_manager)
drop(db);
query_manager
.with_tx(|db| {
db.delete_workspace_by_id(&workspace.id, &UpdateSource::Sync, &blob_manager)
})
.expect("Failed to delete workspace");
let db = query_manager.connect();
let all_changes = db.list_model_changes_after(0, 10).expect("Failed to list changes");
assert_eq!(all_changes.len(), 2);
assert!(matches!(all_changes[1].payload.change, ModelChangeEvent::Delete));
@@ -146,19 +166,23 @@ mod tests {
#[test]
fn prunes_old_model_changes() {
let (query_manager, _blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
let db = query_manager.connect().unwrap();
clear_changes(&query_manager);
db.upsert_workspace(
&Workspace {
name: "Prune Test".to_string(),
setting_follow_redirects: true,
setting_validate_certificates: true,
..Default::default()
},
&UpdateSource::Sync,
)
.expect("Failed to upsert workspace");
query_manager
.with_tx(|db| {
db.upsert_workspace(
&Workspace {
name: "Prune Test".to_string(),
setting_follow_redirects: true,
setting_validate_certificates: true,
..Default::default()
},
&UpdateSource::Sync,
)
})
.expect("Failed to upsert workspace");
let db = query_manager.connect();
let changes = db.list_model_changes_after(0, 10).expect("Failed to list changes");
assert_eq!(changes.len(), 1);
@@ -170,31 +194,41 @@ mod tests {
)
.expect("Failed to age model change row");
let pruned =
db.prune_model_changes_older_than_days(30).expect("Failed to prune model changes");
drop(db);
let pruned = query_manager
.with_tx(|db| db.prune_model_changes_older_than_days(30))
.expect("Failed to prune model changes");
assert_eq!(pruned, 1);
assert!(db.list_model_changes_after(0, 10).expect("Failed to list changes").is_empty());
assert!(
query_manager
.connect()
.list_model_changes_after(0, 10)
.expect("Failed to list changes")
.is_empty()
);
}
#[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 db = query_manager.connect().unwrap();
clear_changes(&query_manager);
let workspace = db
.upsert_workspace(
&Workspace {
name: "Cursor Test".to_string(),
setting_follow_redirects: true,
setting_validate_certificates: true,
..Default::default()
},
&UpdateSource::Sync,
)
.expect("Failed to upsert workspace");
db.delete_workspace_by_id(&workspace.id, &UpdateSource::Sync, &blob_manager)
.expect("Failed to delete workspace");
query_manager
.with_tx(|db| {
let workspace = db.upsert_workspace(
&Workspace {
name: "Cursor Test".to_string(),
setting_follow_redirects: true,
setting_validate_certificates: true,
..Default::default()
},
&UpdateSource::Sync,
)?;
db.delete_workspace_by_id(&workspace.id, &UpdateSource::Sync, &blob_manager)
})
.expect("Failed to seed changes");
let db = query_manager.connect();
let all = db.list_model_changes_after(0, 10).expect("Failed to list changes");
assert_eq!(all.len(), 2);
@@ -213,19 +247,23 @@ mod tests {
#[test]
fn prunes_old_model_changes_by_hours() {
let (query_manager, _blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
let db = query_manager.connect().unwrap();
clear_changes(&query_manager);
db.upsert_workspace(
&Workspace {
name: "Prune Hour Test".to_string(),
setting_follow_redirects: true,
setting_validate_certificates: true,
..Default::default()
},
&UpdateSource::Sync,
)
.expect("Failed to upsert workspace");
query_manager
.with_tx(|db| {
db.upsert_workspace(
&Workspace {
name: "Prune Hour Test".to_string(),
setting_follow_redirects: true,
setting_validate_certificates: true,
..Default::default()
},
&UpdateSource::Sync,
)
})
.expect("Failed to upsert workspace");
let db = query_manager.connect();
let changes = db.list_model_changes_after(0, 10).expect("Failed to list changes");
assert_eq!(changes.len(), 1);
@@ -237,15 +275,18 @@ mod tests {
)
.expect("Failed to age model change row");
let pruned =
db.prune_model_changes_older_than_hours(1).expect("Failed to prune model changes");
drop(db);
let pruned = query_manager
.with_tx(|db| db.prune_model_changes_older_than_hours(1))
.expect("Failed to prune model changes");
assert_eq!(pruned, 1);
}
#[test]
fn list_model_changes_deserializes_http_response_event_payload() {
let (query_manager, _blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
let db = query_manager.connect().unwrap();
clear_changes(&query_manager);
let db = query_manager.connect();
let payload = json!({
"model": {
@@ -1,4 +1,4 @@
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{PluginKeyValue, PluginKeyValueIden};
use sea_query::ExprTrait;
@@ -22,7 +22,9 @@ impl<'a> ClientDb<'a> {
.query_row(sql.as_str(), &*params.as_params(), |row| row.try_into())
.ok()
}
}
impl<'a> WriteDb<'a> {
pub fn set_plugin_key_value(
&self,
plugin_name: &str,
+3 -1
View File
@@ -1,4 +1,4 @@
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{Plugin, PluginIden};
use crate::util::UpdateSource;
@@ -15,7 +15,9 @@ impl<'a> ClientDb<'a> {
pub fn list_plugins(&self) -> Result<Vec<Plugin>> {
self.find_all()
}
}
impl<'a> WriteDb<'a> {
pub fn delete_plugin(&self, plugin: &Plugin, source: &UpdateSource) -> Result<Plugin> {
self.delete(plugin, source)
}
+46 -37
View File
@@ -1,52 +1,61 @@
use std::collections::HashMap;
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{EditorKeymap, Settings, SettingsIden};
use crate::util::UpdateSource;
impl<'a> ClientDb<'a> {
/// The settings row, or the defaults if it has not been written yet.
/// [`WriteDb::ensure_settings`] persists it at startup.
pub fn get_settings(&self) -> Settings {
let id = "default".to_string();
self.find_optional::<Settings>(SettingsIden::Id, "default").unwrap_or_else(default_settings)
}
}
if let Some(s) = self.find_optional::<Settings>(SettingsIden::Id, &id) {
return s;
};
let settings = Settings {
model: "settings".to_string(),
id,
created_at: Default::default(),
updated_at: Default::default(),
appearance: "system".to_string(),
client_certificates: Vec::new(),
editor_font_size: 12,
editor_font: None,
editor_keymap: EditorKeymap::Default,
editor_soft_wrap: true,
interface_font_size: 14,
interface_scale: 1.0,
interface_font: None,
hide_window_controls: false,
use_native_titlebar: false,
open_workspace_new_window: None,
proxy: None,
theme_dark: "yaak-dark".to_string(),
theme_light: "yaak-light".to_string(),
update_channel: "stable".to_string(),
autoupdate: true,
colored_methods: false,
hide_license_badge: false,
prompt_feedback: true,
auto_download_updates: true,
check_notifications: true,
hotkeys: HashMap::new(),
};
self.upsert(&settings, &UpdateSource::Background).expect("Failed to upsert settings")
impl<'a> WriteDb<'a> {
/// Create the settings row if it does not exist.
pub fn ensure_settings(&self) -> Result<Settings> {
if let Some(s) = self.find_optional::<Settings>(SettingsIden::Id, "default") {
return Ok(s);
}
self.upsert(&default_settings(), &UpdateSource::Background)
}
pub fn upsert_settings(&self, settings: &Settings, source: &UpdateSource) -> Result<Settings> {
self.upsert(settings, source)
}
}
fn default_settings() -> Settings {
Settings {
model: "settings".to_string(),
id: "default".to_string(),
created_at: Default::default(),
updated_at: Default::default(),
appearance: "system".to_string(),
client_certificates: Vec::new(),
editor_font_size: 12,
editor_font: None,
editor_keymap: EditorKeymap::Default,
editor_soft_wrap: true,
interface_font_size: 14,
interface_scale: 1.0,
interface_font: None,
hide_window_controls: false,
use_native_titlebar: false,
open_workspace_new_window: None,
proxy: None,
theme_dark: "yaak-dark".to_string(),
theme_light: "yaak-light".to_string(),
update_channel: "stable".to_string(),
autoupdate: true,
colored_methods: false,
hide_license_badge: false,
prompt_feedback: true,
auto_download_updates: true,
check_notifications: true,
hotkeys: HashMap::new(),
}
}
@@ -1,4 +1,4 @@
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{SyncState, SyncStateIden, UpsertModelInfo};
use crate::util::UpdateSource;
@@ -12,10 +12,6 @@ impl<'a> ClientDb<'a> {
self.find_one(SyncStateIden::Id, id)
}
pub fn upsert_sync_state(&self, sync_state: &SyncState) -> Result<SyncState> {
self.upsert(sync_state, &UpdateSource::Sync)
}
pub fn list_sync_states_for_workspace(
&self,
workspace_id: &str,
@@ -34,6 +30,12 @@ impl<'a> ClientDb<'a> {
let items = stmt.query_map(&*params.as_params(), SyncState::from_row)?;
Ok(items.map(|v| v.unwrap()).collect())
}
}
impl<'a> WriteDb<'a> {
pub fn upsert_sync_state(&self, sync_state: &SyncState) -> Result<SyncState> {
self.upsert(sync_state, &UpdateSource::Sync)
}
pub fn delete_sync_state(&self, sync_state: &SyncState) -> Result<SyncState> {
self.delete(sync_state, &UpdateSource::Sync)
@@ -1,4 +1,4 @@
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{WebsocketConnection, WebsocketConnectionIden, WebsocketConnectionState};
use crate::queries::MAX_HISTORY_ITEMS;
@@ -13,6 +13,22 @@ impl<'a> ClientDb<'a> {
self.find_one(WebsocketConnectionIden::Id, id)
}
pub fn list_websocket_connections(
&self,
workspace_id: &str,
) -> Result<Vec<WebsocketConnection>> {
self.find_many(WebsocketConnectionIden::WorkspaceId, workspace_id, None)
}
pub fn list_websocket_connections_for_request(
&self,
request_id: &str,
) -> Result<Vec<WebsocketConnection>> {
self.find_many(WebsocketConnectionIden::RequestId, request_id, None)
}
}
impl<'a> WriteDb<'a> {
pub fn delete_all_websocket_connections_for_request(
&self,
request_id: &str,
@@ -37,20 +53,6 @@ impl<'a> ClientDb<'a> {
Ok(())
}
pub fn list_websocket_connections(
&self,
workspace_id: &str,
) -> Result<Vec<WebsocketConnection>> {
self.find_many(WebsocketConnectionIden::WorkspaceId, workspace_id, None)
}
pub fn list_websocket_connections_for_request(
&self,
request_id: &str,
) -> Result<Vec<WebsocketConnection>> {
self.find_many(WebsocketConnectionIden::RequestId, request_id, None)
}
pub fn delete_websocket_connection(
&self,
websocket_connection: &WebsocketConnection,
@@ -1,4 +1,4 @@
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{WebsocketEvent, WebsocketEventIden};
use crate::util::UpdateSource;
@@ -11,7 +11,9 @@ impl<'a> ClientDb<'a> {
pub fn list_websocket_events(&self, connection_id: &str) -> Result<Vec<WebsocketEvent>> {
self.find_many(WebsocketEventIden::ConnectionId, connection_id, None)
}
}
impl<'a> WriteDb<'a> {
pub fn upsert_websocket_event(
&self,
websocket_event: &WebsocketEvent,
@@ -1,5 +1,5 @@
use super::{conflict_free_name, merge_headers};
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{
AnyModel, Folder, FolderIden, HttpRequestHeader, ResolvedHttpRequestSettings, ResolvedSetting,
@@ -34,50 +34,6 @@ impl<'a> ClientDb<'a> {
Ok(children)
}
pub fn delete_websocket_request(
&self,
websocket_request: &WebsocketRequest,
source: &UpdateSource,
) -> Result<WebsocketRequest> {
self.delete_all_websocket_connections_for_request(websocket_request.id.as_str(), source)?;
self.delete(websocket_request, source)
}
pub fn delete_websocket_request_by_id(
&self,
id: &str,
source: &UpdateSource,
) -> Result<WebsocketRequest> {
let request = self.get_websocket_request(id)?;
self.delete_websocket_request(&request, source)
}
pub fn duplicate_websocket_request(
&self,
websocket_request: &WebsocketRequest,
source: &UpdateSource,
) -> Result<WebsocketRequest> {
let mut websocket_request = websocket_request.clone();
websocket_request.id = "".to_string();
websocket_request.sort_priority = websocket_request.sort_priority + 0.001;
let sibling_names = self
.list_websocket_requests(&websocket_request.workspace_id)?
.into_iter()
.filter(|m| m.folder_id == websocket_request.folder_id)
.map(|m| m.name)
.collect::<Vec<_>>();
websocket_request.name = conflict_free_name(&websocket_request.name, &sibling_names);
self.upsert(&websocket_request, source)
}
pub fn upsert_websocket_request(
&self,
websocket_request: &WebsocketRequest,
source: &UpdateSource,
) -> Result<WebsocketRequest> {
self.upsert(websocket_request, source)
}
pub fn resolve_auth_for_websocket_request(
&self,
websocket_request: &WebsocketRequest,
@@ -168,3 +124,49 @@ impl<'a> ClientDb<'a> {
})
}
}
impl<'a> WriteDb<'a> {
pub fn delete_websocket_request(
&self,
websocket_request: &WebsocketRequest,
source: &UpdateSource,
) -> Result<WebsocketRequest> {
self.delete_all_websocket_connections_for_request(websocket_request.id.as_str(), source)?;
self.delete(websocket_request, source)
}
pub fn delete_websocket_request_by_id(
&self,
id: &str,
source: &UpdateSource,
) -> Result<WebsocketRequest> {
let request = self.get_websocket_request(id)?;
self.delete_websocket_request(&request, source)
}
pub fn duplicate_websocket_request(
&self,
websocket_request: &WebsocketRequest,
source: &UpdateSource,
) -> Result<WebsocketRequest> {
let mut websocket_request = websocket_request.clone();
websocket_request.id = "".to_string();
websocket_request.sort_priority = websocket_request.sort_priority + 0.001;
let sibling_names = self
.list_websocket_requests(&websocket_request.workspace_id)?
.into_iter()
.filter(|m| m.folder_id == websocket_request.folder_id)
.map(|m| m.name)
.collect::<Vec<_>>();
websocket_request.name = conflict_free_name(&websocket_request.name, &sibling_names);
self.upsert(&websocket_request, source)
}
pub fn upsert_websocket_request(
&self,
websocket_request: &WebsocketRequest,
source: &UpdateSource,
) -> Result<WebsocketRequest> {
self.upsert(websocket_request, source)
}
}
@@ -1,4 +1,4 @@
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{WorkspaceMeta, WorkspaceMetaIden};
use crate::util::UpdateSource;
@@ -10,18 +10,12 @@ impl<'a> ClientDb<'a> {
}
pub fn list_workspace_metas(&self, workspace_id: &str) -> Result<Vec<WorkspaceMeta>> {
let mut workspace_metas =
self.find_many(WorkspaceMetaIden::WorkspaceId, workspace_id, None)?;
if workspace_metas.is_empty() {
let wm = WorkspaceMeta { workspace_id: workspace_id.to_string(), ..Default::default() };
workspace_metas.push(self.upsert_workspace_meta(&wm, &UpdateSource::Background)?)
}
Ok(workspace_metas)
self.find_many(WorkspaceMetaIden::WorkspaceId, workspace_id, None)
}
}
pub fn get_or_create_workspace_meta(&self, workspace_id: &str) -> Result<WorkspaceMeta> {
impl<'a> WriteDb<'a> {
pub fn ensure_workspace_meta(&self, workspace_id: &str) -> Result<WorkspaceMeta> {
let workspace_meta = self.get_workspace_meta(workspace_id);
if let Some(workspace_meta) = workspace_meta {
return Ok(workspace_meta);
+100 -118
View File
@@ -1,6 +1,6 @@
use super::merge_headers;
use crate::blob_manager::BlobManager;
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Result;
use crate::models::{
AnyModel, CookieJar, CookieJarIden, Environment, EnvironmentIden, Folder, FolderIden,
@@ -8,9 +8,9 @@ use crate::models::{
GrpcEventIden, GrpcRequest, GrpcRequestIden, HttpRequest, HttpRequestHeader, HttpRequestIden,
HttpResponse, HttpResponseEvent, HttpResponseEventIden, HttpResponseIden, ImportSource,
ImportSourceIden, ResolvedHttpRequestSettings, ResolvedSetting, SyncState, SyncStateIden,
WebsocketConnection,
WebsocketConnectionIden, WebsocketEvent, WebsocketEventIden, WebsocketRequest,
WebsocketRequestIden, Workspace, WorkspaceIden, WorkspaceMeta, WorkspaceMetaIden,
WebsocketConnection, WebsocketConnectionIden, WebsocketEvent, WebsocketEventIden,
WebsocketRequest, WebsocketRequestIden, Workspace, WorkspaceIden, WorkspaceMeta,
WorkspaceMetaIden,
};
use crate::util::UpdateSource;
use log::warn;
@@ -23,119 +23,7 @@ impl<'a> ClientDb<'a> {
}
pub fn list_workspaces(&self) -> Result<Vec<Workspace>> {
let mut workspaces = self.find_all()?;
if workspaces.is_empty() {
workspaces.push(self.upsert_workspace(
&Workspace { name: "Yaak".to_string(), ..Default::default() },
&UpdateSource::Background,
)?)
}
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<Workspace> {
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::<HttpResponse>(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<Workspace> = (|| {
self.delete_many_untracked::<HttpResponseEvent>(
HttpResponseEventIden::WorkspaceId,
wid,
)?;
self.delete_many_untracked::<HttpResponse>(HttpResponseIden::WorkspaceId, wid)?;
self.delete_many_untracked::<HttpRequest>(HttpRequestIden::WorkspaceId, wid)?;
self.delete_many_untracked::<GrpcEvent>(GrpcEventIden::WorkspaceId, wid)?;
self.delete_many_untracked::<GrpcConnection>(GrpcConnectionIden::WorkspaceId, wid)?;
self.delete_many_untracked::<GrpcRequest>(GrpcRequestIden::WorkspaceId, wid)?;
self.delete_many_untracked::<WebsocketEvent>(WebsocketEventIden::WorkspaceId, wid)?;
self.delete_many_untracked::<WebsocketConnection>(
WebsocketConnectionIden::WorkspaceId,
wid,
)?;
self.delete_many_untracked::<WebsocketRequest>(WebsocketRequestIden::WorkspaceId, wid)?;
self.delete_many_untracked::<GraphQlIntrospection>(
GraphQlIntrospectionIden::WorkspaceId,
wid,
)?;
self.delete_many_untracked::<Folder>(FolderIden::WorkspaceId, wid)?;
self.delete_many_untracked::<Environment>(EnvironmentIden::WorkspaceId, wid)?;
self.delete_many_untracked::<CookieJar>(CookieJarIden::WorkspaceId, wid)?;
for import_source in self.list_import_sources(wid)? {
self.delete_import_source_resources(&import_source.id)?;
}
self.delete_many_untracked::<ImportSource>(ImportSourceIden::WorkspaceId, wid)?;
self.delete_many_untracked::<SyncState>(SyncStateIden::WorkspaceId, wid)?;
self.delete_many_untracked::<WorkspaceMeta>(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);
}
}
Ok(deleted)
}
pub fn delete_workspace_by_id(
&self,
id: &str,
source: &UpdateSource,
blobs: &BlobManager,
) -> Result<Workspace> {
let workspace = self.get_workspace(id)?;
self.delete_workspace(&workspace, source, blobs)
}
pub fn upsert_workspace(&self, w: &Workspace, source: &UpdateSource) -> Result<Workspace> {
self.upsert(w, source)
self.find_all()
}
pub fn resolve_auth_for_workspace(
@@ -190,6 +78,100 @@ impl<'a> ClientDb<'a> {
}
}
impl<'a> WriteDb<'a> {
/// There is always at least one workspace. Called at startup and after a
/// workspace is deleted.
pub fn ensure_default_workspace(&self) -> Result<()> {
if self.find_all::<Workspace>()?.is_empty() {
self.upsert_workspace(
&Workspace { name: "Yaak".to_string(), ..Default::default() },
&UpdateSource::Background,
)?;
}
Ok(())
}
/// 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<Workspace> {
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::<HttpResponse>(HttpResponseIden::WorkspaceId, wid, None)?;
self.delete_many_untracked::<HttpResponseEvent>(HttpResponseEventIden::WorkspaceId, wid)?;
self.delete_many_untracked::<HttpResponse>(HttpResponseIden::WorkspaceId, wid)?;
self.delete_many_untracked::<HttpRequest>(HttpRequestIden::WorkspaceId, wid)?;
self.delete_many_untracked::<GrpcEvent>(GrpcEventIden::WorkspaceId, wid)?;
self.delete_many_untracked::<GrpcConnection>(GrpcConnectionIden::WorkspaceId, wid)?;
self.delete_many_untracked::<GrpcRequest>(GrpcRequestIden::WorkspaceId, wid)?;
self.delete_many_untracked::<WebsocketEvent>(WebsocketEventIden::WorkspaceId, wid)?;
self.delete_many_untracked::<WebsocketConnection>(
WebsocketConnectionIden::WorkspaceId,
wid,
)?;
self.delete_many_untracked::<WebsocketRequest>(WebsocketRequestIden::WorkspaceId, wid)?;
self.delete_many_untracked::<GraphQlIntrospection>(
GraphQlIntrospectionIden::WorkspaceId,
wid,
)?;
self.delete_many_untracked::<Folder>(FolderIden::WorkspaceId, wid)?;
self.delete_many_untracked::<Environment>(EnvironmentIden::WorkspaceId, wid)?;
self.delete_many_untracked::<CookieJar>(CookieJarIden::WorkspaceId, wid)?;
for import_source in self.list_import_sources(wid)? {
self.delete_import_source_resources(&import_source.id)?;
}
self.delete_many_untracked::<ImportSource>(ImportSourceIden::WorkspaceId, wid)?;
self.delete_many_untracked::<SyncState>(SyncStateIden::WorkspaceId, wid)?;
self.delete_many_untracked::<WorkspaceMeta>(WorkspaceMetaIden::WorkspaceId, wid)?;
let deleted = self.delete(workspace, source)?;
self.ensure_default_workspace()?;
// Best-effort cleanup of response bodies (disk files and blob chunks).
// Failures only orphan unreferenced data, and are logged.
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}");
}
}
let pattern = format!("{}.%", m.id);
if let Err(e) = blobs.with_tx(|b| b.delete_chunks_like(&pattern)) {
warn!("Failed to delete blobs for response {}: {e}", m.id);
}
}
Ok(deleted)
}
pub fn delete_workspace_by_id(
&self,
id: &str,
source: &UpdateSource,
blobs: &BlobManager,
) -> Result<Workspace> {
let workspace = self.get_workspace(id)?;
self.delete_workspace(&workspace, source, blobs)
}
pub fn upsert_workspace(&self, w: &Workspace, source: &UpdateSource) -> Result<Workspace> {
self.upsert(w, source)
}
}
/// Global default headers that are always sent with requests unless overridden.
/// These are prepended to the inheritance chain so workspace/folder/request headers
/// can override or disable them.
@@ -215,7 +197,7 @@ mod tests {
#[test]
fn bootstraps_first_workspace_with_real_defaults() {
let (query_manager, _blob_manager, _rx) = init_in_memory().expect("Failed to init DB");
let db = query_manager.connect().unwrap();
let db = query_manager.connect();
let workspaces = db.list_workspaces().expect("Failed to list workspaces");
let workspace = workspaces.first().expect("No workspace was bootstrapped");
+45 -115
View File
@@ -1,53 +1,82 @@
use crate::client_db::ClientDb;
use crate::client_db::{ClientDb, WriteDb};
use crate::error::Error::GenericError;
use crate::util::ModelPayload;
use rusqlite::{Transaction, TransactionBehavior};
use std::sync::mpsc;
use yaak_database::{ConnectionOrTx, DbContext, SqlitePool};
// Pool is internally synchronized — don't wrap it in a Mutex. A Mutex held across the
// blocking `get()` serializes every DB access behind the slowest waiter, freezing the
// whole app whenever the pool is exhausted.
/// Reads come from a pool; writes go through one connection.
///
/// SQLite in WAL mode lets many readers run alongside a single writer, and
/// never more than one writer. A second in-process writer can only wait, and
/// while it waits in the busy handler it sleeps, retries, and keeps its pool
/// slot. Enough of those and the pool is full of writers that are all asleep,
/// and every read in the app queues behind them. Giving writes exactly one
/// connection turns that into a plain queue: the next write starts the moment
/// the previous one commits, and it never takes a slot a read could use.
///
/// The pools are internally synchronized — don't wrap them in a Mutex. A Mutex
/// held across the blocking `get()` serializes every DB access behind the
/// slowest waiter.
#[derive(Debug, Clone)]
pub struct QueryManager {
pool: SqlitePool,
readers: SqlitePool,
writer: SqlitePool,
events_tx: mpsc::Sender<ModelPayload>,
}
impl QueryManager {
pub fn new(pool: SqlitePool, events_tx: mpsc::Sender<ModelPayload>) -> Self {
QueryManager { pool, events_tx }
/// `writer` must be a pool with a single connection; see [`crate::init_standalone`].
pub fn new(
readers: SqlitePool,
writer: SqlitePool,
events_tx: mpsc::Sender<ModelPayload>,
) -> Self {
QueryManager { readers, writer, events_tx }
}
pub fn connect(&self) -> crate::error::Result<ClientDb<'_>> {
let conn = crate::pool::acquire(&self.pool, "model")?;
let ctx = DbContext::new(ConnectionOrTx::Connection(conn));
Ok(ClientDb::new(ctx, self.events_tx.clone()))
/// A read handle from the reader pool.
pub fn connect(&self) -> ClientDb<'_> {
let conn = self.readers.get().expect("Failed to get a new DB connection from the pool");
ClientDb::new(DbContext::new(ConnectionOrTx::Connection(conn)))
}
/// Run `func` in a transaction on the writer connection.
///
/// Waits for any write in progress to commit first, and fails with a pool
/// error if that takes longer than the pool's timeout. Do not call this
/// from inside another `with_tx` closure: the inner call would wait for
/// the outer transaction, which is waiting on it.
///
/// Model events for the writes are sent once the transaction commits.
pub fn with_tx<T, E>(
&self,
func: impl FnOnce(&ClientDb) -> std::result::Result<T, E>,
func: impl FnOnce(&WriteDb) -> std::result::Result<T, E>,
) -> std::result::Result<T, E>
where
E: From<crate::error::Error>,
{
let conn = crate::pool::acquire(&self.pool, "model")?;
let conn = self.writer.get().map_err(crate::error::Error::SqlPoolError)?;
// `new_unchecked` takes `&Connection`; see yaak_database::pool for why
// the pool never hands out `&mut`.
let tx = Transaction::new_unchecked(&conn, TransactionBehavior::Immediate)
.map_err(|e| GenericError(format!("Failed to start DB transaction: {e:?}")))?;
.map_err(crate::error::Error::SqlError)?;
let ctx = DbContext::new(ConnectionOrTx::Transaction(&tx));
let db = ClientDb::new(ctx, self.events_tx.clone());
let db =
WriteDb::new(DbContext::new(ConnectionOrTx::Transaction(&tx)), self.events_tx.clone());
match func(&db) {
Ok(val) => {
let events = db.into_events();
tx.commit()
.map_err(|e| GenericError(format!("Failed to commit transaction {e:?}")))?;
for payload in events {
let _ = self.events_tx.send(payload);
}
Ok(val)
}
Err(e) => {
drop(db);
tx.rollback()
.map_err(|e| GenericError(format!("Failed to rollback transaction {e:?}")))?;
Err(e)
@@ -55,102 +84,3 @@ impl QueryManager {
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::error::Error;
use crate::migrate::migrate_db;
use crate::models::Workspace;
use crate::util::UpdateSource;
use r2d2_sqlite::SqliteConnectionManager;
use std::time::Duration;
/// A one-connection pool that gives up quickly, so exhaustion is reachable in a test
/// without waiting out the app's acquire timeout.
fn one_connection_pool(
dir: &std::path::Path,
) -> (QueryManager, mpsc::Receiver<ModelPayload>) {
let manager = SqliteConnectionManager::file(dir.join("db.sqlite"));
let pool = r2d2::Pool::builder()
.max_size(1)
.connection_timeout(Duration::from_millis(100))
.build(manager)
.expect("build pool");
migrate_db(&pool).expect("migrate");
let (tx, rx) = mpsc::channel();
(QueryManager::new(pool, tx), rx)
}
fn workspace(name: &str) -> Workspace {
Workspace { name: name.to_string(), ..Default::default() }
}
#[test]
fn exhausted_pool_reports_an_error_instead_of_panicking() {
let dir = tempfile::tempdir().expect("tempdir");
let (qm, _events) = one_connection_pool(dir.path());
let held = qm.connect().expect("first connection");
match qm.connect() {
Ok(_) => panic!("pool should be exhausted"),
Err(e) => {
// The state of the pool rides along, so a report says which failure it was
assert!(
matches!(e, Error::PoolTimeout { connections: 1, idle: 0, .. }),
"got {e:?}"
);
}
}
let err = qm
.with_tx::<(), Error>(|db| {
db.upsert_workspace(&workspace("never runs"), &UpdateSource::Background)?;
Ok(())
})
.expect_err("pool is exhausted");
assert!(matches!(err, Error::PoolTimeout { .. }), "got {err:?}");
drop(held);
qm.connect().expect("pool recovers once the connection is back");
}
#[test]
fn with_tx_returns_its_connection_on_every_exit() {
let dir = tempfile::tempdir().expect("tempdir");
let (qm, _events) = one_connection_pool(dir.path());
qm.with_tx::<(), Error>(|db| {
db.upsert_workspace(&workspace("committed"), &UpdateSource::Background)?;
Ok(())
})
.expect("commit");
let rolled_back = qm.with_tx::<(), Error>(|db| {
db.upsert_workspace(&workspace("rolled back"), &UpdateSource::Background)?;
Err(Error::Unknown)
});
assert!(rolled_back.is_err());
let panicked = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
qm.with_tx::<(), Error>(|db| {
db.upsert_workspace(&workspace("panicked"), &UpdateSource::Background)?;
panic!("closure blew up");
})
}));
assert!(panicked.is_err());
// The pool holds exactly one connection, so this only succeeds if all three
// transactions above gave theirs back.
let names = qm
.connect()
.expect("connection is back in the pool")
.list_workspaces()
.expect("list")
.into_iter()
.map(|w| w.name)
.collect::<Vec<_>>();
assert_eq!(names, vec!["committed".to_string()]);
}
}
+3 -11
View File
@@ -125,19 +125,11 @@ pub enum ImportPlanWarningLevel {
impl ImportPlanWarning {
pub fn info(title: impl Into<String>, detail: impl Into<String>) -> Self {
Self {
title: title.into(),
detail: detail.into(),
level: ImportPlanWarningLevel::Info,
}
Self { title: title.into(), detail: detail.into(), level: ImportPlanWarningLevel::Info }
}
pub fn warning(title: impl Into<String>, detail: impl Into<String>) -> Self {
Self {
title: title.into(),
detail: detail.into(),
level: ImportPlanWarningLevel::Warning,
}
Self { title: title.into(), detail: detail.into(), level: ImportPlanWarningLevel::Warning }
}
}
@@ -285,7 +277,7 @@ pub fn get_workspace_export_resources(
data.resources.workspaces.push(db.find_one(WorkspaceIden::Id, workspace_id)?);
data.resources.environments.append(
&mut db
.list_environments_ensure_base(workspace_id)?
.list_environments(workspace_id)?
.into_iter()
.filter(|e| include_private_environments || e.public)
.collect(),
+4 -10
View File
@@ -24,11 +24,7 @@ pub async fn delete_and_uninstall(
Some(label) => UpdateSource::from_window_label(label),
None => UpdateSource::Background,
};
// Scope the db connection so it doesn't live across await
let plugin = {
let db = query_manager.connect()?;
db.delete_plugin_by_id(plugin_id, &update_source)?
};
let plugin = query_manager.with_tx(|db| db.delete_plugin_by_id(plugin_id, &update_source))?;
if let Err(err) = plugin_manager.uninstall(plugin_context, plugin.directory.as_str()).await {
if !matches!(err, PluginNotFoundErr(_)) {
return Err(err);
@@ -72,9 +68,7 @@ pub async fn download_and_install(
zip_extract::extract(Cursor::new(&bytes), &plugin_dir, true)?;
info!("Extracted plugin {} to {}", plugin_version.id, plugin_dir_str);
// Scope the db connection so it doesn't live across await
let plugin = {
let db = query_manager.connect()?;
let plugin = query_manager.with_tx(|db| {
db.upsert_plugin(
&Plugin {
id: plugin_version.id.clone(),
@@ -86,8 +80,8 @@ pub async fn download_and_install(
..Default::default()
},
&UpdateSource::Background,
)?
};
)
})?;
plugin_manager.add_plugin(plugin_context, &plugin).await?;
+3 -5
View File
@@ -187,9 +187,7 @@ impl PluginManager {
}
let bundled_dirs = plugin_manager.list_bundled_plugin_dirs().await?;
// Scope the db connection so the future stays Send across the await below
let plugins = {
let db = query_manager.connect()?;
let plugins = query_manager.with_tx(|db| {
for dir in &bundled_dirs {
if db.get_plugin_by_directory(dir).is_none() {
db.upsert_plugin(
@@ -204,8 +202,8 @@ impl PluginManager {
)?;
}
}
db.list_plugins()?
};
db.list_plugins()
})?;
let init_errors = plugin_manager.initialize_all_plugins(plugins, plugin_context).await;
if !init_errors.is_empty() {
+82 -51
View File
@@ -1,5 +1,4 @@
use crate::error::Result;
use yaak_models::blob_manager::BlobManager;
use crate::models::SyncModel;
use chrono::Utc;
use log::{info, warn};
@@ -11,8 +10,12 @@ use std::fs::File;
use std::io::Write;
use std::path::{Path, PathBuf};
use ts_rs::TS;
use yaak_models::client_db::ClientDb;
use yaak_models::models::{SyncState, WorkspaceMeta};
use yaak_models::blob_manager::BlobManager;
use yaak_models::client_db::{ClientDb, WriteDb};
use yaak_models::models::{
Environment, Folder, GrpcRequest, HttpRequest, SyncState, WebsocketRequest, Workspace,
WorkspaceMeta,
};
use yaak_models::util::{UpdateSource, get_workspace_export_resources};
#[derive(Debug, Clone, Serialize, Deserialize, TS)]
@@ -336,17 +339,40 @@ fn workspace_models(db: &ClientDb, version: &str, workspace_id: &str) -> Result<
Ok(sync_models)
}
/// Apply sync operations to the filesystem and database.
/// Returns a list of SyncStateOps that should be applied afterward.
pub fn apply_sync_ops(
db: &ClientDb,
blobs: &BlobManager,
/// The database half of a sync apply, ready to run once the files are on disk.
pub struct PendingDbSyncOps {
sync_state_ops: Vec<SyncStateOp>,
deletes: Vec<SyncModel>,
workspaces: Vec<Workspace>,
environments: Vec<Environment>,
folders: Vec<Folder>,
http_requests: Vec<HttpRequest>,
grpc_requests: Vec<GrpcRequest>,
websocket_requests: Vec<WebsocketRequest>,
}
/// Apply the filesystem half of the sync operations: create, rewrite and
/// delete files. Returns the database half, for [`apply_db_sync_ops`].
///
/// Split this way so the file work, which can be slow, happens before the
/// write transaction is opened rather than inside it.
pub fn apply_fs_sync_ops(
workspace_id: &str,
sync_dir: &Path,
sync_ops: Vec<SyncOp>,
) -> Result<Vec<SyncStateOp>> {
) -> Result<PendingDbSyncOps> {
let mut pending = PendingDbSyncOps {
sync_state_ops: Vec::new(),
deletes: Vec::new(),
workspaces: Vec::new(),
environments: Vec::new(),
folders: Vec::new(),
http_requests: Vec::new(),
grpc_requests: Vec::new(),
websocket_requests: Vec::new(),
};
if sync_ops.is_empty() {
return Ok(Vec::new());
return Ok(pending);
}
info!(
@@ -354,21 +380,13 @@ pub fn apply_sync_ops(
sync_ops.iter().map(|op| op.to_string()).collect::<Vec<String>>().join(", ")
);
let mut sync_state_ops = Vec::new();
let mut workspaces_to_upsert = Vec::new();
let mut environments_to_upsert = Vec::new();
let mut folders_to_upsert = Vec::new();
let mut http_requests_to_upsert = Vec::new();
let mut grpc_requests_to_upsert = Vec::new();
let mut websocket_requests_to_upsert = Vec::new();
for op in sync_ops {
// Only apply things if workspace ID matches
if op.workspace_id() != workspace_id {
continue;
}
sync_state_ops.push(match op {
let state_op = match op {
SyncOp::FsCreate { model } => {
let rel_path = derive_model_filename(&model);
let abs_path = sync_dir.join(rel_path.clone());
@@ -402,17 +420,7 @@ pub fn apply_sync_ops(
},
SyncOp::DbCreate { fs } => {
let model_id = fs.model.id();
// Push updates to arrays so we can do them all in a single
// batch upsert to make foreign keys happy
match fs.model {
SyncModel::Environment(m) => environments_to_upsert.push(m),
SyncModel::Folder(m) => folders_to_upsert.push(m),
SyncModel::GrpcRequest(m) => grpc_requests_to_upsert.push(m),
SyncModel::HttpRequest(m) => http_requests_to_upsert.push(m),
SyncModel::WebsocketRequest(m) => websocket_requests_to_upsert.push(m),
SyncModel::Workspace(m) => workspaces_to_upsert.push(m),
};
pending.push_upsert(fs.model);
SyncStateOp::Create {
model_id,
checksum: fs.checksum.to_owned(),
@@ -420,16 +428,7 @@ pub fn apply_sync_ops(
}
}
SyncOp::DbUpdate { state, fs } => {
// Push updates to arrays so we can do them all in a single
// batch upsert to make foreign keys happy
match fs.model {
SyncModel::Environment(m) => environments_to_upsert.push(m),
SyncModel::Folder(m) => folders_to_upsert.push(m),
SyncModel::GrpcRequest(m) => grpc_requests_to_upsert.push(m),
SyncModel::HttpRequest(m) => http_requests_to_upsert.push(m),
SyncModel::WebsocketRequest(m) => websocket_requests_to_upsert.push(m),
SyncModel::Workspace(m) => workspaces_to_upsert.push(m),
}
pending.push_upsert(fs.model);
SyncStateOp::Update {
state: state.to_owned(),
checksum: fs.checksum.to_owned(),
@@ -437,20 +436,52 @@ pub fn apply_sync_ops(
}
}
SyncOp::DbDelete { model, state } => {
delete_model(db, blobs, &model)?;
pending.deletes.push(model);
SyncStateOp::Delete { state: state.to_owned() }
}
SyncOp::IgnorePrivate { .. } => SyncStateOp::NoOp,
});
};
pending.sync_state_ops.push(state_op);
}
Ok(pending)
}
impl PendingDbSyncOps {
/// Upserts are collected per model type and written in one batch so
/// foreign keys are satisfied.
fn push_upsert(&mut self, model: SyncModel) {
match model {
SyncModel::Environment(m) => self.environments.push(m),
SyncModel::Folder(m) => self.folders.push(m),
SyncModel::GrpcRequest(m) => self.grpc_requests.push(m),
SyncModel::HttpRequest(m) => self.http_requests.push(m),
SyncModel::WebsocketRequest(m) => self.websocket_requests.push(m),
SyncModel::Workspace(m) => self.workspaces.push(m),
}
}
}
/// Apply the database half of the sync operations.
/// Returns a list of SyncStateOps that should be applied afterward.
pub fn apply_db_sync_ops(
db: &WriteDb,
blobs: &BlobManager,
workspace_id: &str,
sync_dir: &Path,
pending: PendingDbSyncOps,
) -> Result<Vec<SyncStateOp>> {
for model in &pending.deletes {
delete_model(db, blobs, model)?;
}
let upserted_models = db.batch_upsert(
workspaces_to_upsert,
environments_to_upsert,
folders_to_upsert,
http_requests_to_upsert,
grpc_requests_to_upsert,
websocket_requests_to_upsert,
pending.workspaces,
pending.environments,
pending.folders,
pending.http_requests,
pending.grpc_requests,
pending.websocket_requests,
&UpdateSource::Sync,
)?;
@@ -482,7 +513,7 @@ pub fn apply_sync_ops(
}?;
}
Ok(sync_state_ops)
Ok(pending.sync_state_ops)
}
#[derive(Debug)]
@@ -504,7 +535,7 @@ pub enum SyncStateOp {
}
pub fn apply_sync_state_ops(
db: &ClientDb,
db: &WriteDb,
workspace_id: &str,
sync_dir: &Path,
ops: Vec<SyncStateOp>,
@@ -549,7 +580,7 @@ fn derive_model_filename(m: &SyncModel) -> PathBuf {
Path::new(&rel).to_path_buf()
}
fn delete_model(db: &ClientDb, blobs: &BlobManager, model: &SyncModel) -> Result<()> {
fn delete_model(db: &WriteDb, blobs: &BlobManager, model: &SyncModel) -> Result<()> {
match model {
SyncModel::Workspace(m) => {
db.delete_workspace(&m, &UpdateSource::Sync, blobs)?;
Binary file not shown.
+4 -4
View File
@@ -697,22 +697,22 @@ export function __wbg_warn_b6f36cac66fc96a4(arg0, arg1) {
console.warn(arg0, arg1);
}
export function __wbindgen_cast_0000000000000001(arg0, arg1) {
// Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [Externref], shim_idx: 1117, ret: Result(Unit), inner_ret: Some(Result(Unit)) }, mutable: true }) -> Externref`.
// Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [Externref], shim_idx: 1123, ret: Result(Unit), inner_ret: Some(Result(Unit)) }, mutable: true }) -> Externref`.
const ret = makeMutClosure(arg0, arg1, wasm_bindgen__convert__closures_____invoke__ha1c2fa93df0107f3);
return ret;
}
export function __wbindgen_cast_0000000000000002(arg0, arg1) {
// Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [NamedExternref("Event")], shim_idx: 212, ret: Unit, inner_ret: Some(Unit) }, mutable: true }) -> Externref`.
// Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [NamedExternref("Event")], shim_idx: 214, ret: Unit, inner_ret: Some(Unit) }, mutable: true }) -> Externref`.
const ret = makeMutClosure(arg0, arg1, wasm_bindgen__convert__closures_____invoke__ha7903b6e296dd8f4);
return ret;
}
export function __wbindgen_cast_0000000000000003(arg0, arg1) {
// Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [NamedExternref("IDBVersionChangeEvent")], shim_idx: 83, ret: Result(Unit), inner_ret: Some(Result(Unit)) }, mutable: true }) -> Externref`.
// Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [NamedExternref("IDBVersionChangeEvent")], shim_idx: 164, ret: Result(Unit), inner_ret: Some(Result(Unit)) }, mutable: true }) -> Externref`.
const ret = makeMutClosure(arg0, arg1, wasm_bindgen__convert__closures_____invoke__h4381d8e749fe46cf);
return ret;
}
export function __wbindgen_cast_0000000000000004(arg0, arg1) {
// Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [], shim_idx: 210, ret: Unit, inner_ret: Some(Unit) }, mutable: true }) -> Externref`.
// Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [], shim_idx: 212, ret: Unit, inner_ret: Some(Unit) }, mutable: true }) -> Externref`.
const ret = makeMutClosure(arg0, arg1, wasm_bindgen__convert__closures_____invoke__ha1b480b83daa641f);
return ret;
}
Loaded 100 of 142 files, more files were not shown because too many files have changed in this diff. Show more