Keep the proxy app fast with many flows

The app held every flow in memory and redid all of its work for each new
one: the sidebar mounted every domain and path node (its tree scrolled
inside a container that never bounded it), every tree row looked up its
atoms by scanning every atom the family had made, and the layout, detail
pane and list rows re-rendered per flow. At 3,000 flows each new one cost
about 90 ms of main thread, more than the traffic left time for.

The app now holds a window of the newest 2,000 flows, applies new ones in
batches, and loads older ones in pages from the database on scrolling to
the top, through a new list_flows RPC that pages in SQL. The database
keeps every flow.

Also adds clear_flows: an RPC, POST /flows/clear on the control API, and
a Clear button in the header that Cmd+K opens, confirming first.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Gregory SchierandClaude Opus 5.5 authored and Stetta task tk_mvsazf4rsaotrdpf committed 2026-10-28 09:52:00 -07:00
1 parent 4abc882e13
commit 1a994f7532
19 files changed
+855 -216

No files matched your search

@@ -0,0 +1,49 @@
import { Dropdown, type DropdownItem, IconButton } from "@yaakapp-internal/ui";
import { useAtomValue } from "jotai";
import { useRpcMutation } from "../hooks/useRpcMutation";
import { flowTotalAtom } from "../lib/store";
/** Deletes every flow, once confirmed in its menu, which its hotkey opens too */
export function ClearFlowsButton() {
const total = useAtomValue(flowTotalAtom);
const clear = useRpcMutation("clear_flows");
const items: DropdownItem[] = [
{
type: "content",
label: (
<div className="px-1.5 py-1 max-w-[17rem]">
<div className="text-sm text-text">
{total === 1 ? "Clear 1 flow?" : `Clear all ${total} flows?`}
</div>
<div className="text-xs text-text-subtle mt-0.5">
{clear.error != null ? (
<span className="text-danger">{clear.error.message}</span>
) : (
"They are deleted from disk with their bodies, for good."
)}
</div>
</div>
),
},
{ type: "separator" },
{
label: "Clear",
icon: "trash",
color: "danger",
onSelect: () => clear.mutate({}),
},
];
return (
<Dropdown items={items} hotKeyAction="flows_clear">
<IconButton
className="pointer-events-auto"
icon="trash"
title="Clear flows"
size="sm"
iconColor="secondary"
/>
</Dropdown>
);
}
+4 -8
View File
@@ -17,21 +17,15 @@ import {
Tabs,
} from "@yaakapp-internal/ui";
import classNames from "classnames";
import { atom, useAtomValue } from "jotai";
import { useAtomValue } from "jotai";
import type { CSSProperties } from "react";
import { useMemo, useState } from "react";
import { selectedFlowIdAtom } from "../lib/atoms";
import { selectedFlowAtom } from "../lib/atoms";
import { findHeader, mimeEssence } from "../lib/body";
import { statusReason } from "../lib/statusReason";
import { flowsAtom } from "../lib/store";
import { EmptyState, FlowBodyViewer, useFlowBody } from "./FlowBodyViewer";
import { FlowHeaders, FlowQuery, queryParams } from "./FlowHeaders";
const selectedFlowAtom = atom((get) => {
const id = get(selectedFlowIdAtom);
return id == null ? null : (get(flowsAtom).find((f) => f.id === id) ?? null);
});
const TAB_RESPONSE = "response";
const TAB_REQUEST = "request";
const TAB_HEADERS = "headers";
@@ -60,6 +54,8 @@ export function FlowDetail({ style, sideBySide }: { style?: CSSProperties; sideB
<Keys keys={["⌘", "⇧", "P"]} />
<span>Stop the proxy</span>
<Keys keys={["⌘", "⇧", "S"]} />
<span>Clear flows</span>
<Keys keys={["⌘", "K"]} />
</div>
</EmptyState>
) : (
+63 -21
View File
@@ -3,34 +3,48 @@ import { formatSize } from "@yaakapp-internal/lib/formatSize";
import type { Flow } from "@yaakapp-internal/proxy-lib";
import { formatMillis, HttpMethodTagRaw, HttpStatusTagRaw, Icon } from "@yaakapp-internal/ui";
import classNames from "classnames";
import { useAtom, useAtomValue } from "jotai";
import type { CSSProperties, KeyboardEvent } from "react";
import { useAtom, useAtomValue, useSetAtom } from "jotai";
import type { CSSProperties, KeyboardEvent, WheelEvent } from "react";
import { memo, useCallback, useLayoutEffect, useRef, useState } from "react";
import { selectedFlowIdAtom } from "../lib/atoms";
import { selectedFlowAtom, selectedFlowIdAtom } from "../lib/atoms";
import { shortContentType } from "../lib/body";
import { compareFlows } from "../lib/flowWindow";
import { countNewFlows } from "../lib/follow";
import { flowsAtom } from "../lib/store";
import { followAtom, loadOlderFlowsAtom, newestStartedAtAtom, oldestFlowAtom } from "../lib/store";
import { filteredExchangesAtom } from "./Sidebar";
const ROW_HEIGHT = 24;
const BOTTOM_PADDING = 6;
/** Rows from the top at which older flows start loading */
const LOAD_OLDER_ROWS = 20;
interface Props {
flows: Flow[];
sideBySide: boolean;
style?: CSSProperties;
className?: string;
}
/** Follows new traffic until a flow is selected or the list is scrolled, then holds */
export function FlowList({ flows, sideBySide, style, className }: Props) {
const [selectedId, setSelectedId] = useAtom(selectedFlowIdAtom);
/**
* Follows new traffic until a flow is selected or the list is scrolled, then holds. Scrolling to
* the top loads older flows from the database.
*/
export function FlowList({ sideBySide, style, className }: Props) {
const flows = useAtomValue(filteredExchangesAtom);
const selectedId = useAtomValue(selectedFlowIdAtom);
const setSelectedFlow = useSetAtom(selectedFlowAtom);
const loadOlder = useSetAtom(loadOlderFlowsAtom);
const oldestFlow = useAtomValue(oldestFlowAtom);
const scrollRef = useRef<HTMLDivElement>(null);
const ignoreScroll = useRef(false);
const [following, setFollowing] = useState(true);
const [following, setFollowing] = useAtom(followAtom);
const [heldAt, setHeldAt] = useState<string | null>(null);
// The newest flow across all traffic, not just this list, so a filter change keeps the hold
const allFlows = useAtomValue(flowsAtom);
const newestStartedAt = allFlows[allFlows.length - 1]?.startedAt ?? null;
const newestStartedAt = useAtomValue(newestStartedAtAtom);
// Read by the callbacks below, which stay the same across new traffic so rows don't re-render
const latest = useRef({ flows, following, newestStartedAt });
useLayoutEffect(() => {
latest.current = { flows, following, newestStartedAt };
});
const virtualizer = useVirtualizer({
count: flows.length,
@@ -52,20 +66,37 @@ export function FlowList({ flows, sideBySide, style, className }: Props) {
}, []);
const hold = useCallback(() => {
if (!following) return;
if (!latest.current.following) return;
setFollowing(false);
setHeldAt(newestStartedAt);
}, [following, newestStartedAt]);
setHeldAt(latest.current.newestStartedAt);
}, [setFollowing]);
const follow = useCallback(() => {
setFollowing(true);
setHeldAt(null);
setScrollTop(Number.MAX_SAFE_INTEGER);
}, [setScrollTop]);
}, [setFollowing, setScrollTop]);
useLayoutEffect(() => {
if (following) setScrollTop(Number.MAX_SAFE_INTEGER);
}, [following, flows.length, setScrollTop]);
}, [following, flows, setScrollTop]);
// Older flows that loaded in above push the rows down; scroll by as much, so they stay put.
const anchor = useRef<Flow | null>(null);
useLayoutEffect(() => {
const previous = anchor.current;
anchor.current = oldestFlow;
if (following || previous == null || oldestFlow == null) return;
if (compareFlows(oldestFlow, previous) >= 0) return;
const added = flows.findIndex((f) => compareFlows(f, previous) >= 0);
const el = scrollRef.current;
if (added > 0 && el != null) setScrollTop(el.scrollTop + added * ROW_HEIGHT);
}, [oldestFlow, flows, following, setScrollTop]);
const loadOlderHeld = useCallback(() => {
hold();
void loadOlder();
}, [hold, loadOlder]);
const handleScroll = useCallback(() => {
if (ignoreScroll.current) {
@@ -80,14 +111,23 @@ export function FlowList({ flows, sideBySide, style, className }: Props) {
setHeldAt(null);
} else {
hold();
if (el.scrollTop < LOAD_OLDER_ROWS * ROW_HEIGHT) void loadOlder();
}
}, [hold]);
}, [hold, loadOlder, setFollowing]);
// A list too short to scroll still loads older flows on scrolling up.
const handleWheel = useCallback(
(e: WheelEvent<HTMLDivElement>) => {
if (e.deltaY < 0 && e.currentTarget.scrollTop <= 0) loadOlderHeld();
},
[loadOlderHeld],
);
const select = useCallback(
(index: number) => {
const flow = flows[index];
const flow = latest.current.flows[index];
if (flow == null) return;
setSelectedId(flow.id);
setSelectedFlow(flow);
hold();
const el = scrollRef.current;
if (el == null) return;
@@ -97,7 +137,7 @@ export function FlowList({ flows, sideBySide, style, className }: Props) {
setScrollTop(top + ROW_HEIGHT - el.clientHeight);
}
},
[flows, hold, setSelectedId, setScrollTop],
[hold, setSelectedFlow, setScrollTop],
);
const handleKeyDown = useCallback(
@@ -107,9 +147,10 @@ export function FlowList({ flows, sideBySide, style, className }: Props) {
e.preventDefault();
const index = flows.findIndex((f) => f.id === selectedId);
if (index === -1) select(e.key === "ArrowDown" ? 0 : flows.length - 1);
else if (index === 0 && e.key === "ArrowUp") loadOlderHeld();
else select(e.key === "ArrowDown" ? index + 1 : index - 1);
},
[flows, select, selectedId],
[flows, loadOlderHeld, select, selectedId],
);
const newCount = following ? 0 : countNewFlows(flows, heldAt);
@@ -146,6 +187,7 @@ export function FlowList({ flows, sideBySide, style, className }: Props) {
aria-label="Flows"
tabIndex={0}
onScroll={handleScroll}
onWheel={handleWheel}
onKeyDown={handleKeyDown}
className="group/flows h-full overflow-y-auto px-1.5 outline-hidden"
>
+13 -11
View File
@@ -5,14 +5,14 @@ import { useState } from "react";
import { useLocalStorage } from "react-use";
import { flowLayoutAtom } from "../lib/atoms";
import { getOsType } from "../lib/tauri";
import { ClearFlowsButton } from "./ClearFlowsButton";
import { FlowDetail } from "./FlowDetail";
import { FlowList } from "./FlowList";
import { ProxyStatus } from "./ProxyStatus";
import { filteredExchangesAtom, selectedSidebarLabelAtom, Sidebar } from "./Sidebar";
import { filteredCountAtom, selectedSidebarLabelAtom, Sidebar } from "./Sidebar";
export function ProxyLayout() {
const os = getOsType();
const exchanges = useAtomValue(filteredExchangesAtom);
const title = useAtomValue(selectedSidebarLabelAtom);
const [flowLayout, setFlowLayout] = useAtom(flowLayoutAtom);
const [sidebarWidth, setSidebarWidth] = useLocalStorage("sidebar_width", 250);
@@ -61,12 +61,11 @@ export function ProxyLayout() {
className="pointer-events-none flex items-center justify-center gap-2 text-sm px-2 min-w-0"
>
<span className="truncate">{title}</span>
<span className="text-text-subtlest shrink-0">
{exchanges.length === 1 ? "1 flow" : `${exchanges.length} flows`}
</span>
<FlowCount />
</div>
<div className="flex items-center gap-1 pr-1">
<ProxyStatus />
<ClearFlowsButton />
<IconButton
className="pointer-events-auto"
icon={flowLayout === "horizontal" ? "columns_2" : "rows_2"}
@@ -127,12 +126,7 @@ export function ProxyLayout() {
layout={flowLayout}
defaultRatio={0.55}
firstSlot={({ style, orientation }) => (
<FlowList
flows={exchanges}
sideBySide={orientation === "horizontal"}
style={style}
className="pt-1"
/>
<FlowList sideBySide={orientation === "horizontal"} style={style} className="pt-1" />
)}
secondSlot={({ style, orientation }) => (
<FlowDetail style={style} sideBySide={orientation === "horizontal"} />
@@ -142,3 +136,11 @@ export function ProxyLayout() {
</div>
);
}
/** Its own component, so new traffic re-renders the count rather than the whole layout */
function FlowCount() {
const count = useAtomValue(filteredCountAtom);
return (
<span className="text-text-subtlest shrink-0">{count === 1 ? "1 flow" : `${count} flows`}</span>
);
}
+82 -94
View File
@@ -4,13 +4,14 @@ import { selectedIdsFamily, Tree } from "@yaakapp-internal/ui";
import { atom, useAtomValue } from "jotai";
import { atomFamily } from "jotai-family";
import { useCallback } from "react";
import { flowsAtom } from "../lib/store";
import { flowsAtom, flowTotalAtom } from "../lib/store";
/** A node in the sidebar tree — either a domain or a path segment. */
export type SidebarItem = {
id: string;
label: string;
exchangeIds: string[];
/** Flows held under this node; for the Domains folder, every flow in the database */
count: number;
};
const collapsedAtom = atomFamily((_treeId: string) => atom<Record<string, boolean>>({}));
@@ -19,37 +20,32 @@ export const SIDEBAR_TREE_ID = "proxy-sidebar";
const sidebarTreeAtom = atom<TreeNode<SidebarItem>>((get) => {
const exchanges = get(flowsAtom);
return buildTree(exchanges);
return buildTree(exchanges, get(flowTotalAtom));
});
/** The sidebar nodes selected, or null when the list shows all traffic */
const selectedNodeIdsAtom = atom((get) => {
const selectedIds = get(selectedIdsFamily(SIDEBAR_TREE_ID));
// Nothing selected, or a node holding everything → show all
if (selectedIds.length === 0 || selectedIds.includes("root") || selectedIds.includes("domains")) {
return null;
}
return new Set(selectedIds);
});
/** Exchanges filtered by the currently selected sidebar node(s). */
export const filteredExchangesAtom = atom((get) => {
const exchanges = get(flowsAtom);
const tree = get(sidebarTreeAtom);
const selectedIds = get(selectedIdsFamily(SIDEBAR_TREE_ID));
// Nothing selected or root selected → show all
if (selectedIds.length === 0 || selectedIds.includes("root")) {
return exchanges;
}
// Collect exchange IDs from all selected nodes
const allowedIds = new Set<string>();
const nodeMap = new Map<string, SidebarItem>();
collectNodes(tree, nodeMap);
for (const selectedId of selectedIds) {
const node = nodeMap.get(selectedId);
if (node) {
for (const id of node.exchangeIds) {
allowedIds.add(id);
}
}
}
return exchanges.filter((ex) => allowedIds.has(ex.id));
const selected = get(selectedNodeIdsAtom);
if (selected == null) return exchanges;
return exchanges.filter((ex) => nodeIdsOf(ex).some((id) => selected.has(id)));
});
/** Flows in the database for all traffic, or those held under the selected nodes */
export const filteredCountAtom = atom((get) =>
get(selectedNodeIdsAtom) == null ? get(flowTotalAtom) : get(filteredExchangesAtom).length,
);
/** The selected sidebar node's name, for the header */
export const selectedSidebarLabelAtom = atom((get) => {
const selectedId = get(selectedIdsFamily(SIDEBAR_TREE_ID))[0];
@@ -58,13 +54,37 @@ export const selectedSidebarLabelAtom = atom((get) => {
return "All traffic";
});
function collectNodes(node: TreeNode<SidebarItem>, map: Map<string, SidebarItem>) {
map.set(node.item.id, node.item);
for (const child of node.children ?? []) {
collectNodes(child, map);
const nodeIdsByFlow = new WeakMap<Flow, string[]>();
/**
* The ids of the nodes a flow sits under, its domain first and then each path segment's, parsed
* once per flow rather than on every change to the list.
*/
function nodeIdsOf(ex: Flow): string[] {
let ids = nodeIdsByFlow.get(ex);
if (ids != null) return ids;
let hostname: string;
let segments: string[];
try {
const url = new URL(ex.url);
hostname = url.host;
segments = url.pathname.split("/").filter(Boolean);
} catch {
hostname = ex.url;
segments = [];
}
ids = [`domain:${hostname}`];
let path = `path:${hostname}`;
for (const seg of segments) {
path += `/${seg}`;
ids.push(path);
}
nodeIdsByFlow.set(ex, ids);
return ids;
}
const collator = new Intl.Collator();
/**
* Build a domain → path-segment trie from a flat list of exchanges.
*
@@ -80,8 +100,8 @@ function collectNodes(node: TreeNode<SidebarItem>, map: Map<string, SidebarItem>
* /123
* /orders
*/
function buildTree(exchanges: Flow[]): TreeNode<SidebarItem> {
const root: SidebarItem = { id: "root", label: "All Traffic", exchangeIds: [] };
function buildTree(exchanges: Flow[], total: number): TreeNode<SidebarItem> {
const root: SidebarItem = { id: "root", label: "All Traffic", count: 0 };
const rootNode: TreeNode<SidebarItem> = {
item: root,
parent: null,
@@ -94,54 +114,26 @@ function buildTree(exchanges: Flow[]): TreeNode<SidebarItem> {
type TrieNode = {
id: string;
label: string;
exchangeIds: string[];
count: number;
children: Map<string, TrieNode>;
};
const domainMap = new Map<string, TrieNode>();
for (const ex of exchanges) {
let hostname: string;
let segments: string[];
try {
const url = new URL(ex.url);
hostname = url.host;
segments = url.pathname.split("/").filter(Boolean);
} catch {
hostname = ex.url;
segments = [];
}
// Get or create domain node
let domainNode = domainMap.get(hostname);
if (!domainNode) {
domainNode = {
id: `domain:${hostname}`,
label: hostname,
exchangeIds: [],
children: new Map(),
};
domainMap.set(hostname, domainNode);
}
domainNode.exchangeIds.push(ex.id);
// Walk path segments
let current = domainNode;
const pathSoFar: string[] = [];
for (const seg of segments) {
pathSoFar.push(seg);
let child = current.children.get(seg);
if (!child) {
child = {
id: `path:${hostname}/${pathSoFar.join("/")}`,
label: `/${seg}`,
exchangeIds: [],
children: new Map(),
};
current.children.set(seg, child);
const ids = nodeIdsOf(ex);
let current: TrieNode | undefined;
for (const id of ids) {
const siblings = current == null ? domainMap : current.children;
let node = siblings.get(id);
if (node == null) {
// The domain, or the path's last segment
const label = current == null ? id.slice("domain:".length) : id.slice(id.lastIndexOf("/"));
node = { id, label, count: 0, children: new Map() };
siblings.set(id, node);
}
child.exchangeIds.push(ex.id);
current = child;
node.count++;
current = node;
}
}
@@ -155,7 +147,7 @@ function buildTree(exchanges: Flow[]): TreeNode<SidebarItem> {
item: {
id: trie.id,
label: trie.label,
exchangeIds: trie.exchangeIds,
count: trie.count,
},
parent,
depth,
@@ -163,7 +155,7 @@ function buildTree(exchanges: Flow[]): TreeNode<SidebarItem> {
draggable: false,
};
const sortedChildren = [...trie.children.values()].sort((a, b) =>
a.label.localeCompare(b.label),
collator.compare(a.label, b.label),
);
for (const child of sortedChildren) {
node.children?.push(toTreeNode(child, node, depth + 1));
@@ -172,16 +164,15 @@ function buildTree(exchanges: Flow[]): TreeNode<SidebarItem> {
}
// Add a "Domains" folder between root and domain nodes
const allExchangeIds = exchanges.map((ex) => ex.id);
const domainsFolder: TreeNode<SidebarItem> = {
item: { id: "domains", label: "Domains", exchangeIds: allExchangeIds },
item: { id: "domains", label: "Domains", count: total },
parent: rootNode,
depth: 1,
children: [],
draggable: false,
};
const sortedDomains = [...domainMap.values()].sort((a, b) => a.label.localeCompare(b.label));
const sortedDomains = [...domainMap.values()].sort((a, b) => collator.compare(a.label, b.label));
for (const domain of sortedDomains) {
domainsFolder.children?.push(toTreeNode(domain, domainsFolder, 2));
}
@@ -192,7 +183,7 @@ function buildTree(exchanges: Flow[]): TreeNode<SidebarItem> {
}
function ItemInner({ item }: { item: SidebarItem }) {
const count = item.exchangeIds.length;
const count = item.count;
return (
<div className="flex items-center gap-2 w-full min-w-0">
<span className="truncate">{item.label}</span>
@@ -205,23 +196,20 @@ export function Sidebar() {
const tree = useAtomValue(sidebarTreeAtom);
const treeId = SIDEBAR_TREE_ID;
const getItemKey = useCallback(
(item: SidebarItem) => `${item.id}:${item.exchangeIds.length}`,
[],
);
const getItemKey = useCallback((item: SidebarItem) => `${item.id}:${item.count}`, []);
return (
<aside className="x-theme-sidebar bg-surface h-full w-full min-w-0 overflow-y-auto border-r border-border-subtle">
<div className="pt-2">
<Tree
treeId={treeId}
collapsedAtom={collapsedAtom(treeId)}
className="px-2 pb-10"
root={tree}
getItemKey={getItemKey}
ItemInner={ItemInner}
/>
</div>
// The tree scrolls itself, in a row of bounded height, so its virtualizer mounts only the rows
// in view rather than every domain and path.
<aside className="x-theme-sidebar bg-surface h-full w-full min-w-0 grid grid-rows-[minmax(0,1fr)] border-r border-border-subtle">
<Tree
treeId={treeId}
collapsedAtom={collapsedAtom(treeId)}
className="px-2 pt-2 pb-10"
root={tree}
getItemKey={getItemKey}
ItemInner={ItemInner}
/>
</aside>
);
}
+8 -1
View File
@@ -1,8 +1,15 @@
import type { Flow } from "@yaakapp-internal/proxy-lib";
import type { SplitLayoutLayout } from "@yaakapp-internal/ui";
import { atom } from "jotai";
import { atomWithStorage } from "jotai/utils";
export const selectedFlowIdAtom = atom<string | null>(null);
/** The selected flow as last seen, kept once the window has dropped it so the detail stays */
export const selectedFlowAtom = atom<Flow | null>(null);
export const selectedFlowIdAtom = atom((get) => get(selectedFlowAtom)?.id ?? null);
/** The list follows new traffic, rather than being held where it was scrolled or selected */
export const followingAtom = atom(true);
/** Stacked by default; the header's toggle switches to side by side */
export const flowLayoutAtom = atomWithStorage<SplitLayoutLayout>("flow_layout", "vertical");
+131
View File
@@ -0,0 +1,131 @@
import type { Flow, ModelPayload } from "@yaakapp-internal/proxy-lib";
import { describe, expect, test } from "vite-plus/test";
import {
applyChanges,
emptyWindow,
type FlowWindow,
prependOlder,
trimWindow,
windowFromPage,
} from "./flowWindow";
/** A flow started at `second` past 10:00, with an id that sorts the same way */
function flow(second: number, extra: Partial<Flow> = {}): Flow {
const s = String(second).padStart(2, "0");
return { id: `fl_${s}`, startedAt: `2026-10-28T10:00:${s}`, ...extra } as Flow;
}
const created = (f: Flow): ModelPayload => ({
model: f,
change: { type: "upsert", created: true },
});
const updated = (f: Flow): ModelPayload => ({
model: f,
change: { type: "upsert", created: false },
});
const ids = (win: FlowWindow) => win.flows.map((f) => f.id);
describe("windowFromPage", () => {
test("holds the newest page oldest first", () => {
const win = windowFromPage([flow(3), flow(2), flow(1)], 3, 10);
expect(ids(win)).toEqual(["fl_01", "fl_02", "fl_03"]);
expect(win).toMatchObject({ hasOlder: true, total: 10 });
});
test("knows there is nothing older when the page came back short", () => {
expect(windowFromPage([flow(1)], 3, 1).hasOlder).toBe(false);
});
});
describe("applyChanges", () => {
test("appends new traffic and counts it", () => {
const win = applyChanges(windowFromPage([flow(2), flow(1)], 10, 2), [
created(flow(3)),
created(flow(4)),
]);
expect(ids(win)).toEqual(["fl_01", "fl_02", "fl_03", "fl_04"]);
expect(win.total).toBe(4);
});
test("puts a slow flow where it started", () => {
const win = applyChanges(windowFromPage([flow(5), flow(1)], 10, 2), [created(flow(3))]);
expect(ids(win)).toEqual(["fl_01", "fl_03", "fl_05"]);
});
test("orders flows that started together by id", () => {
const a = flow(2, { id: "fl_a" });
const b = flow(2, { id: "fl_b" });
const win = applyChanges(emptyWindow, [created(b), created(a)]);
expect(ids(win)).toEqual(["fl_a", "fl_b"]);
});
test("replaces a changed flow in place without counting it again", () => {
const saved = flow(2, { savedRequestId: "rq_1" });
const win = applyChanges(windowFromPage([flow(3), flow(2), flow(1)], 10, 3), [updated(saved)]);
expect(win.flows[1]).toBe(saved);
expect(win.total).toBe(3);
});
test("replaces a flow the first page already had, without counting it again", () => {
const win = applyChanges(windowFromPage([flow(2), flow(1)], 10, 2), [created(flow(2))]);
expect(ids(win)).toEqual(["fl_01", "fl_02"]);
expect(win.total).toBe(2);
});
test("leaves a flow older than the window to paging, but counts it", () => {
const win = applyChanges(windowFromPage([flow(5), flow(4)], 2, 9), [created(flow(1))]);
expect(ids(win)).toEqual(["fl_04", "fl_05"]);
expect(win.total).toBe(10);
});
test("keeps an older flow when the window already holds everything", () => {
const win = applyChanges(windowFromPage([flow(5), flow(4)], 10, 2), [created(flow(1))]);
expect(ids(win)).toEqual(["fl_01", "fl_04", "fl_05"]);
});
test("returns the same window for no changes", () => {
const win = windowFromPage([flow(1)], 10, 1);
expect(applyChanges(win, [])).toBe(win);
});
});
describe("trimWindow", () => {
const many = (n: number) =>
windowFromPage(
Array.from({ length: n }, (_, i) => flow(n - i)),
1000,
n,
);
test("leaves the window alone within the slack", () => {
const win = many(7);
expect(trimWindow(win, 4, 3)).toBe(win);
});
test("drops the oldest back to the size, an even number of them", () => {
const even = trimWindow(many(8), 4, 3);
expect(ids(even)).toEqual(["fl_05", "fl_06", "fl_07", "fl_08"]);
expect(even).toMatchObject({ hasOlder: true, total: 8 });
// Five over rounds up to six, so the list's stripes stay on the same flows.
expect(ids(trimWindow(many(9), 4, 3))).toEqual(["fl_07", "fl_08", "fl_09"]);
});
});
describe("prependOlder", () => {
test("adds an older page before the oldest flow held", () => {
const win = prependOlder(windowFromPage([flow(5), flow(4)], 2, 5), [flow(3), flow(2)], 2);
expect(ids(win)).toEqual(["fl_02", "fl_03", "fl_04", "fl_05"]);
expect(win.hasOlder).toBe(true);
});
test("knows it reached the start when a page comes back short", () => {
const win = prependOlder(windowFromPage([flow(5), flow(4)], 2, 3), [flow(1)], 2);
expect(ids(win)).toEqual(["fl_01", "fl_04", "fl_05"]);
expect(win.hasOlder).toBe(false);
});
test("skips flows the window already has", () => {
const win = prependOlder(windowFromPage([flow(5), flow(4)], 2, 5), [flow(4), flow(3)], 2);
expect(ids(win)).toEqual(["fl_03", "fl_04", "fl_05"]);
});
});
+100
View File
@@ -0,0 +1,100 @@
import type { Flow, ModelPayload } from "@yaakapp-internal/proxy-lib";
/** Flows the window keeps while it follows new traffic */
export const WINDOW_SIZE = 2000;
/**
* Flows past `WINDOW_SIZE` before the oldest are dropped, so they go a few hundred at a time
* rather than one per new flow, which would shift every row of the list each time.
*/
export const WINDOW_SLACK = 500;
/** Older flows loaded per page on scrolling to the top */
export const PAGE_SIZE = 500;
/**
* The flows the app holds in memory: the newest in the database, oldest first, ordered by start
* and then id as `list_flows` orders them. The database keeps every flow; older ones load back
* in pages.
*/
export interface FlowWindow {
flows: Flow[];
/** The database has flows older than the first one here */
hasOlder: boolean;
/** Flows in the database */
total: number;
}
export const emptyWindow: FlowWindow = { flows: [], hasOlder: false, total: 0 };
type FlowKey = Pick<Flow, "startedAt" | "id">;
export function compareFlows(a: FlowKey, b: FlowKey): number {
if (a.startedAt !== b.startedAt) return a.startedAt < b.startedAt ? -1 : 1;
if (a.id !== b.id) return a.id < b.id ? -1 : 1;
return 0;
}
/** The window from the newest page of `list_flows`, which comes newest first */
export function windowFromPage(page: Flow[], limit: number, total: number): FlowWindow {
return { flows: [...page].reverse(), hasOlder: page.length >= limit, total };
}
/**
* Adds new flows where they belong and replaces changed ones. New traffic almost always lands at
* the end, so this is an append. A flow that started before the oldest one held, while older ones
* are still in the database, is left for paging to load, so the window never has a gap.
*/
export function applyChanges(win: FlowWindow, changes: ModelPayload[]): FlowWindow {
if (changes.length === 0) return win;
const flows = win.flows.slice();
let total = win.total;
for (const { model, change } of changes) {
const i = insertionIndex(flows, model);
const existing = flows[i]?.id === model.id;
if (change.type === "delete") {
if (existing) flows.splice(i, 1);
total = Math.max(0, total - 1);
} else if (existing) {
flows[i] = model;
} else {
if (change.created) total++;
const first = flows[0];
if (win.hasOlder && first != null && compareFlows(model, first) < 0) continue;
flows.splice(i, 0, model);
}
}
return { ...win, flows, total };
}
/**
* Drops the oldest flows once there are `slack` more than `size`, back down to `size`. Always an
* even number of them, so the list's alternating rows keep their shade.
*/
export function trimWindow(win: FlowWindow, size = WINDOW_SIZE, slack = WINDOW_SLACK): FlowWindow {
if (win.flows.length <= size + slack) return win;
let drop = win.flows.length - size;
drop += drop % 2;
return { ...win, flows: win.flows.slice(drop), hasOlder: true };
}
/** Adds a page of older flows from `list_flows`, newest first, before the oldest one held */
export function prependOlder(win: FlowWindow, page: Flow[], limit: number): FlowWindow {
const first = win.flows[0];
const older = page.filter((f) => first == null || compareFlows(f, first) < 0).reverse();
return { ...win, flows: [...older, ...win.flows], hasOlder: page.length >= limit };
}
/** Where `flow` is, or would go, in `flows` */
function insertionIndex(flows: FlowKey[], flow: FlowKey): number {
// New traffic goes at the end; skip the search for it.
const last = flows[flows.length - 1];
if (last == null || compareFlows(last, flow) < 0) return flows.length;
let lo = 0;
let hi = flows.length;
while (lo < hi) {
const mid = (lo + hi) >>> 1;
const item = flows[mid];
if (item != null && compareFlows(item, flow) < 0) lo = mid + 1;
else hi = mid;
}
return lo;
}
+116 -8
View File
@@ -1,12 +1,120 @@
import { createModelStore } from "@yaakapp-internal/model-store";
import type { Flow } from "@yaakapp-internal/proxy-lib";
import type { ModelPayload } from "@yaakapp-internal/proxy-lib";
import { atom, type createStore } from "jotai";
import { followingAtom, selectedFlowAtom } from "./atoms";
import {
applyChanges,
emptyWindow,
type FlowWindow,
PAGE_SIZE,
prependOlder,
trimWindow,
WINDOW_SIZE,
windowFromPage,
} from "./flowWindow";
import { listen, rpc } from "./rpc";
type ProxyModels = {
flow: Flow;
};
/** How long new flows gather before they reach the list, so a burst renders once */
const FLUSH_MS = 100;
export const { dataAtom, applyChange, replaceAll, listAtom, orderedListAtom } =
createModelStore<ProxyModels>(["flow"]);
/** The flows in memory: a window onto the database, which keeps them all */
export const flowWindowAtom = atom<FlowWindow>(emptyWindow);
/** Oldest first, so new traffic arrives at the end of the list */
export const flowsAtom = orderedListAtom("flow", "startedAt", "asc");
export const flowsAtom = atom((get) => get(flowWindowAtom).flows);
/** Start of the newest flow held, across all traffic */
export const newestStartedAtAtom = atom((get) => get(flowsAtom).at(-1)?.startedAt ?? null);
/** Start and id of the oldest flow held, which loading older flows moves back */
export const oldestFlowAtom = atom((get) => get(flowsAtom)[0] ?? null);
export const hasOlderFlowsAtom = atom((get) => get(flowWindowAtom).hasOlder);
/** Flows in the database */
export const flowTotalAtom = atom((get) => get(flowWindowAtom).total);
/** Follows new traffic, dropping the older flows scrolling up loaded, or holds the list */
export const followAtom = atom(
(get) => get(followingAtom),
(_get, set, following: boolean) => {
set(followingAtom, following);
if (following) set(flowWindowAtom, (win) => trimWindow(win));
},
);
let loadingOlder = false;
/** Loads the page of flows before the oldest one held */
export const loadOlderFlowsAtom = atom(null, async (get, set) => {
const { flows, hasOlder } = get(flowWindowAtom);
const first = flows[0];
if (loadingOlder || !hasOlder || first == null) return;
loadingOlder = true;
try {
const res = await rpc("list_flows", { before: first.id, limit: PAGE_SIZE });
// Dropped when the window moved on meanwhile, trimmed or cleared.
set(flowWindowAtom, (win) =>
win.flows[0]?.id === first.id ? prependOlder(win, res.flows, PAGE_SIZE) : win,
);
} catch (err) {
console.warn("Failed to load older flows", err);
} finally {
loadingOlder = false;
}
});
/**
* Loads the newest flows, then keeps them current with the engine's events. Events gather for
* `FLUSH_MS` and apply together, and the window drops its oldest flows while the list follows,
* so a new flow costs the same however many the database holds.
*/
export function syncFlows(store: ReturnType<typeof createStore>) {
let pending: ModelPayload[] = [];
let timer: ReturnType<typeof setTimeout> | null = null;
let loaded = false;
function flush() {
timer = null;
if (!loaded || pending.length === 0) return;
const changes = pending;
pending = [];
store.set(flowWindowAtom, (win) => {
const next = applyChanges(win, changes);
return store.get(followingAtom) ? trimWindow(next) : next;
});
const selected = store.get(selectedFlowAtom);
const change = changes.findLast((c) => c.model.id === selected?.id);
if (change != null) {
store.set(selectedFlowAtom, change.change.type === "delete" ? null : change.model);
}
}
listen("model_write", (payload) => {
pending.push(payload);
if (timer == null) timer = setTimeout(flush, FLUSH_MS);
});
// Loads again rather than only emptying, for a flow written between the delete and the event.
listen("flows_cleared", () => {
pending = [];
store.set(flowWindowAtom, emptyWindow);
store.set(selectedFlowAtom, null);
store.set(followingAtom, true);
load();
});
let loads = 0;
function load() {
loaded = false;
const thisLoad = ++loads;
void rpc("list_flows", { limit: WINDOW_SIZE }).then((res) => {
// A clear since asked for a newer page.
if (thisLoad !== loads) return;
store.set(flowWindowAtom, windowFromPage(res.flows, WINDOW_SIZE, res.total));
loaded = true;
flush();
});
}
load();
}
+2 -11
View File
@@ -6,22 +6,13 @@ import { type ReactNode, StrictMode } from "react";
import { createRoot } from "react-dom/client";
import { ProxyLayout } from "./components/ProxyLayout";
import { useHotkeyContextValue } from "./lib/hotkeys";
import { listen, rpc } from "./lib/rpc";
import { applyChange, dataAtom, replaceAll } from "./lib/store";
import { syncFlows } from "./lib/store";
import "./main.css";
const queryClient = new QueryClient();
const jotaiStore = createStore();
// Load initial models from the database
void rpc("list_models", {}).then((res) => {
jotaiStore.set(dataAtom, (prev) => replaceAll(prev, "flow", res.flows));
});
// Subscribe to model change events from the backend
void listen("model_write", (payload) => {
jotaiStore.set(dataAtom, (prev) => applyChange(prev, "flow", payload.model, payload.change));
});
syncFlows(jotaiStore);
function Hotkeys({ children }: { children: ReactNode }) {
return <HotkeyProvider value={useHotkeyContextValue()}>{children}</HotkeyProvider>;
-1
View File
@@ -13,7 +13,6 @@
"@tanstack/react-virtual": "^3.14.17",
"@tauri-apps/api": "^2.12.3",
"@tauri-apps/plugin-os": "^2.4.0",
"@yaakapp-internal/model-store": "^1.0.0",
"@yaakapp-internal/proxy-lib": "^1.0.0",
"@yaakapp-internal/theme": "^1.0.0",
"@yaakapp-internal/ui": "^1.0.0",
+47 -6
View File
@@ -19,6 +19,8 @@ export type CaStatusRequest = Record<string, never>;
export type CaTrustRequest = Record<string, never>;
export type ClearFlowsRequest = Record<string, never>;
export type FilterBodyRequest = { flowId: string, which: FlowBody,
/**
* JSONPath, or XPath for an XML or HTML body.
@@ -33,6 +35,39 @@ export type FilterBodyResponse = { content: string, error: string | null, };
export type FilterLanguage = "jsonpath" | "xpath";
/**
* Which flows a listing or a HAR export covers. Every field narrows it; none is all of them.
*/
export type FlowFilter = {
/**
* Part of the host, any case: `stripe` matches `api.stripe.com`.
*/
host?: string,
/**
* The method, any case.
*/
method?: string,
/**
* A status (`404`), a class of them (`4xx`), or `error` for flows that failed.
*/
status?: string,
/**
* At most this many flows, the newest.
*/
limit?: number,
/**
* Only flows recorded after the flow with this id, for following new traffic. Flows are
* recorded when they end, so a slow one that started earlier still comes through.
*/
after?: string,
/**
* Only flows older than the flow with this id, by start and then id, for paging back
* through older traffic.
*/
before?: string, };
export type FlowsClearedPayload = Record<string, never>;
export type FormatBodyRequest = { text: string, };
export type GetFlowBodyRequest = { flowId: string, which: FlowBody, };
@@ -59,15 +94,21 @@ export type GetProxyStateResponse = { state: ProxyState,
*/
port: number | null, };
export type GlobalAction = "proxy_start" | "proxy_stop";
export type GlobalAction = "proxy_start" | "proxy_stop" | "flows_clear";
export type ListActionsRequest = Record<string, never>;
export type ListActionsResponse = { actions: Array<[ActionInvocation, ActionMetadata]>, };
export type ListModelsRequest = Record<string, never>;
export type ListModelsResponse = { flows: Array<Flow>, };
export type ListFlowsResponse = {
/**
* Newest first.
*/
flows: Array<Flow>,
/**
* Flows in the database, whatever the filter.
*/
total: number, };
/**
* The process that runs the engine.
@@ -80,9 +121,9 @@ export type ProxyState = "running" | "stopped";
export type ProxyStatePayload = { state: ProxyState, };
export type RpcEventSchema = { model_write: ModelPayload, proxy_state_changed: ProxyStatePayload, };
export type RpcEventSchema = { flows_cleared: FlowsClearedPayload, model_write: ModelPayload, proxy_state_changed: ProxyStatePayload, };
export type RpcSchema = { execute_action: [ActionInvocation, boolean], format_body: [FormatBodyRequest, string], get_proxy_state: [GetProxyStateRequest, GetProxyStateResponse], list_actions: [ListActionsRequest, ListActionsResponse], list_models: [ListModelsRequest, ListModelsResponse], ca_status: [CaStatusRequest, CaStatus], ca_trust: [CaTrustRequest, CaStatus], filter_body: [FilterBodyRequest, FilterBodyResponse], get_flow_body: [GetFlowBodyRequest, GetFlowBodyResponse], system_proxy_set: [SystemProxySetRequest, SystemProxyStatus], system_proxy_status: [SystemProxyStatusRequest, SystemProxyStatus], };
export type RpcSchema = { execute_action: [ActionInvocation, boolean], format_body: [FormatBodyRequest, string], get_proxy_state: [GetProxyStateRequest, GetProxyStateResponse], list_actions: [ListActionsRequest, ListActionsResponse], list_flows: [FlowFilter, ListFlowsResponse], ca_status: [CaStatusRequest, CaStatus], ca_trust: [CaTrustRequest, CaStatus], clear_flows: [ClearFlowsRequest, number], filter_body: [FilterBodyRequest, FilterBodyResponse], get_flow_body: [GetFlowBodyRequest, GetFlowBodyResponse], system_proxy_set: [SystemProxySetRequest, SystemProxyStatus], system_proxy_status: [SystemProxyStatusRequest, SystemProxyStatus], };
export type ServiceProxy = {
/**
@@ -7,6 +7,7 @@ use ts_rs::TS;
pub enum GlobalAction {
ProxyStart,
ProxyStop,
FlowsClear,
}
#[derive(Debug, Clone, Serialize, Deserialize, TS)]
@@ -45,5 +46,12 @@ pub fn all_global_actions() -> Vec<(ActionInvocation, ActionMetadata)> {
default_hotkey: default_hotkey("Meta+Shift+S", "Ctrl+Shift+S"),
},
),
(
ActionInvocation::Global { action: GlobalAction::FlowsClear },
ActionMetadata {
label: "Clear Flows".into(),
default_hotkey: default_hotkey("Meta+K", "Ctrl+K"),
},
),
]
}
@@ -222,6 +222,12 @@ pub struct SystemProxyRequest {
pub enabled: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ClearResponse {
/// How many flows were deleted.
pub cleared: usize,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ErrorBody {
pub error: String,
@@ -470,6 +476,10 @@ async fn route(state: &ServerState, req: Request<Incoming>) -> Result<ControlRes
let filter = FlowFilter::from_query(&query).map_err(bad_request)?;
json(&flows::list(&ctx.db, &filter)?)
}
(&Method::POST, ["flows", "clear"]) => {
let ctx = ctx.clone();
json(&ClearResponse { cleared: blocking(move || ctx.clear_flows()).await? })
}
(&Method::GET, ["flows", id]) => {
found(ctx, id)?;
let limit = body_limit(&query)?;
+143 -26
View File
@@ -4,27 +4,40 @@ use crate::models::{Flow, FlowBody, FlowIden};
use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64;
use serde::{Deserialize, Serialize};
use ts_rs::TS;
use yaak_database::UpsertModelInfo;
use yaak_models::blob_manager::BlobManager;
/// Which flows a listing or a HAR export covers. Every field narrows it; none is all of them.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, TS)]
#[serde(default, rename_all = "camelCase")]
#[ts(export, export_to = "gen_rpc.ts")]
pub struct FlowFilter {
/// Part of the host, any case: `stripe` matches `api.stripe.com`.
#[ts(optional)]
pub host: Option<String>,
/// The method, any case.
#[ts(optional)]
pub method: Option<String>,
/// A status (`404`), a class of them (`4xx`), or `error` for flows that failed.
#[ts(optional)]
pub status: Option<String>,
/// At most this many flows, the newest.
#[ts(optional, type = "number")]
pub limit: Option<usize>,
/// Only flows recorded after the flow with this id, for following new traffic. Flows are
/// recorded when they end, so a slow one that started earlier still comes through.
#[ts(optional)]
pub after: Option<String>,
/// Only flows older than the flow with this id, by start and then id, for paging back
/// through older traffic.
#[ts(optional)]
pub before: Option<String>,
}
impl FlowFilter {
/// The filter from a query string's `host`, `method`, `status`, `limit` and `after`.
/// The filter from a query string's `host`, `method`, `status`, `limit`, `after` and
/// `before`.
pub fn from_query(query: &str) -> Result<Self, String> {
let mut filter = Self::default();
for (key, value) in url::form_urlencoded::parse(query.as_bytes()) {
@@ -34,6 +47,7 @@ impl FlowFilter {
"method" => filter.method = value,
"status" => filter.status = value,
"after" => filter.after = value,
"before" => filter.before = value,
"limit" => {
filter.limit = value
.map(|v| v.parse().map_err(|_| format!("Invalid limit {v:?}")))
@@ -54,6 +68,7 @@ impl FlowFilter {
("status", &self.status),
("limit", &limit),
("after", &self.after),
("before", &self.before),
] {
if let Some(value) = value {
query.append_pair(key, value);
@@ -91,35 +106,70 @@ fn host_of(url: &str) -> &str {
rest.split(['/', '?', '#']).next().unwrap_or_default()
}
/// The flows `filter` covers, newest first.
/// The flows `filter` covers, newest first: by start, then by id. Reads only as far into the
/// table as the limit needs, so a page costs the same however many flows there are.
pub fn list(db: &ProxyQueryManager, filter: &FlowFilter) -> Result<Vec<Flow>, String> {
db.with_conn(|c| {
let flows = c.find_all::<Flow>().map_err(|e| e.to_string())?;
let recorded_after = match &filter.after {
Some(id) => Some(recorded_after(c.conn().resolve(), id)?),
None => None,
};
Ok(flows
.into_iter()
.filter(|f| recorded_after.as_ref().is_none_or(|ids| ids.contains(&f.id)))
.filter(|f| filter.matches(f))
.take(filter.limit.unwrap_or(usize::MAX))
.collect())
let conn = c.conn().resolve();
let mut sql = String::from("SELECT * FROM flows WHERE 1");
let mut params: Vec<rusqlite::types::Value> = Vec::new();
if let Some(id) = &filter.after {
// By rowid, which only grows as flows are written.
let rowid: i64 = conn
.query_row("SELECT rowid FROM flows WHERE id = ?", [id], |r| r.get(0))
.map_err(|_| format!("No flow {id}"))?;
sql.push_str(" AND rowid > ?");
params.push(rowid.into());
}
if let Some(id) = &filter.before {
let started_at: rusqlite::types::Value = conn
.query_row("SELECT started_at FROM flows WHERE id = ?", [id], |r| r.get(0))
.map_err(|_| format!("No flow {id}"))?;
sql.push_str(" AND (started_at, id) < (?, ?)");
params.push(started_at);
params.push(id.clone().into());
}
sql.push_str(" ORDER BY started_at DESC, id DESC");
let mut stmt = conn.prepare(&sql).map_err(|e| e.to_string())?;
let rows = stmt
.query_map(rusqlite::params_from_iter(params), Flow::from_row)
.map_err(|e| e.to_string())?;
let mut flows = Vec::new();
for flow in rows {
if flows.len() >= filter.limit.unwrap_or(usize::MAX) {
break;
}
let flow = flow.map_err(|e| e.to_string())?;
if filter.matches(&flow) {
flows.push(flow);
}
}
Ok(flows)
})
}
/// Ids of the flows inserted after flow `id`, by rowid, which only grows as flows are written.
fn recorded_after(
conn: &rusqlite::Connection,
id: &str,
) -> Result<std::collections::HashSet<String>, String> {
let rowid: i64 = conn
.query_row("SELECT rowid FROM flows WHERE id = ?", [id], |r| r.get(0))
.map_err(|_| format!("No flow {id}"))?;
let mut stmt =
conn.prepare("SELECT id FROM flows WHERE rowid > ?").map_err(|e| e.to_string())?;
let ids = stmt.query_map([rowid], |r| r.get(0)).map_err(|e| e.to_string())?;
ids.collect::<Result<_, _>>().map_err(|e| e.to_string())
/// Deletes every flow, then their bodies, each in a transaction. Returns how many there were.
pub fn clear(db: &ProxyQueryManager, blobs: &BlobManager) -> Result<usize, String> {
let ids = db
.with_tx(|c| {
let conn = c.conn().resolve();
let mut stmt = conn.prepare("SELECT id FROM flows")?;
let ids = stmt.query_map([], |r| r.get(0))?.collect::<Result<Vec<String>, _>>()?;
conn.execute("DELETE FROM flows", [])?;
Ok::<_, yaak_database::Error>(ids)
})
.map_err(|e| format!("Failed to clear flows: {e}"))?;
// Only the deleted flows' bodies, so a flow still in flight keeps what it has written.
blobs
.with_tx(|b| {
for id in &ids {
b.delete_chunks(&Flow::body_id(id, FlowBody::Request))?;
b.delete_chunks(&Flow::body_id(id, FlowBody::Response))?;
}
Ok::<_, yaak_models::error::Error>(())
})
.map_err(|e| format!("Failed to clear flow bodies: {e}"))?;
Ok(ids.len())
}
pub fn find(db: &ProxyQueryManager, flow_id: &str) -> Option<Flow> {
@@ -267,6 +317,73 @@ mod tests {
assert_eq!(list(&db, &missing).unwrap_err(), "No flow fl_gone");
}
#[test]
fn before_pages_back_through_older_flows() {
use yaak_database::UpdateSource;
let dir = tempfile::tempdir().unwrap();
let db = ProxyQueryManager::new(&dir.path().join("proxy.db")).unwrap();
let at = |s| chrono::DateTime::from_timestamp(s, 0).unwrap().naive_utc();
// Two started together, which their ids order.
for (id, started) in [("fl_a", 10), ("fl_b", 20), ("fl_c", 20), ("fl_d", 30), ("fl_e", 40)]
{
let flow = Flow {
id: id.into(),
started_at: at(started),
status: Some(if id == "fl_c" { 404 } else { 200 }),
..Default::default()
};
db.with_tx(|tx| tx.upsert(&flow, &UpdateSource::Background)).unwrap();
}
let page = |filter: FlowFilter| {
list(&db, &filter).unwrap().into_iter().map(|f| f.id).collect::<Vec<_>>()
};
let newest = FlowFilter { limit: Some(2), ..Default::default() };
assert_eq!(page(newest), vec!["fl_e", "fl_d"]);
let older = FlowFilter { before: Some("fl_d".into()), limit: Some(2), ..Default::default() };
assert_eq!(page(older), vec!["fl_c", "fl_b"]);
let oldest = FlowFilter { before: Some("fl_b".into()), limit: Some(2), ..Default::default() };
assert_eq!(page(oldest), vec!["fl_a"]);
// The limit counts flows the filter keeps.
let ok = FlowFilter {
before: Some("fl_e".into()),
status: Some("2xx".into()),
limit: Some(2),
..Default::default()
};
assert_eq!(page(ok), vec!["fl_d", "fl_b"]);
let missing = FlowFilter { before: Some("fl_gone".into()), ..Default::default() };
assert_eq!(list(&db, &missing).unwrap_err(), "No flow fl_gone");
}
#[test]
fn clear_deletes_flows_and_their_bodies() {
use yaak_database::UpdateSource;
use yaak_models::blob_manager::BodyChunkWriter;
let dir = tempfile::tempdir().unwrap();
let db = ProxyQueryManager::new(&dir.path().join("proxy.db")).unwrap();
let blobs = yaak_models::init_blob_store(dir.path().join("blobs.sqlite")).unwrap();
let write_body = |id: &str, which| {
let mut writer = BodyChunkWriter::new(blobs.clone(), Flow::body_id(id, which));
writer.write(b"body").unwrap();
writer.finish().unwrap();
};
for id in ["fl_a", "fl_b"] {
write_body(id, FlowBody::Request);
write_body(id, FlowBody::Response);
let flow = Flow { id: id.into(), ..Default::default() };
db.with_tx(|tx| tx.upsert(&flow, &UpdateSource::Background)).unwrap();
}
// A flow still in flight has written a body but not its row.
write_body("fl_live", FlowBody::Request);
assert_eq!(clear(&db, &blobs).unwrap(), 2);
assert_eq!(count(&db).unwrap(), 0);
let body_ids = blobs.connect().list_body_ids().unwrap();
assert_eq!(body_ids, vec![Flow::body_id("fl_live", FlowBody::Request)]);
assert_eq!(clear(&db, &blobs).unwrap(), 0);
}
#[test]
fn query_round_trip() {
let filter = FlowFilter {
+35 -12
View File
@@ -329,6 +329,14 @@ impl ProxyCtx {
);
Ok(saved)
}
/// Deletes every flow and its bodies, and tells the app so its list empties. Returns how
/// many there were.
pub fn clear_flows(&self) -> Result<usize, String> {
let cleared = flows::clear(&self.db, &self.blobs)?;
self.events.emit("flows_cleared", &FlowsClearedPayload {});
Ok(cleared)
}
}
/// Something accepts connections on `127.0.0.1:<port>`.
@@ -365,17 +373,25 @@ pub struct ListActionsResponse {
pub actions: Vec<(ActionInvocation, ActionMetadata)>,
}
#[derive(Deserialize, TS)]
#[ts(export, export_to = "gen_rpc.ts")]
pub struct ListModelsRequest {}
#[derive(Serialize, TS)]
#[ts(export, export_to = "gen_rpc.ts")]
#[serde(rename_all = "camelCase")]
pub struct ListModelsResponse {
pub struct ListFlowsResponse {
/// Newest first.
pub flows: Vec<Flow>,
/// Flows in the database, whatever the filter.
#[ts(type = "number")]
pub total: usize,
}
#[derive(Deserialize, TS)]
#[ts(export, export_to = "gen_rpc.ts")]
pub struct ClearFlowsRequest {}
#[derive(Serialize, TS)]
#[ts(export, export_to = "gen_rpc.ts")]
pub struct FlowsClearedPayload {}
#[derive(Deserialize, TS)]
#[ts(export, export_to = "gen_rpc.ts")]
pub struct GetProxyStateRequest {}
@@ -469,6 +485,9 @@ fn execute_action(ctx: &ProxyCtx, invocation: ActionInvocation) -> Result<bool,
ActionInvocation::Global { action } => match action {
GlobalAction::ProxyStart => ctx.start().map_err(|message| RpcError { message })?,
GlobalAction::ProxyStop => ctx.stop().map_err(|message| RpcError { message })?,
GlobalAction::FlowsClear => {
ctx.clear_flows().map_err(|message| RpcError { message })?;
}
},
}
Ok(true)
@@ -587,12 +606,14 @@ fn list_actions(
Ok(ListActionsResponse { actions: crate::actions::all_global_actions() })
}
fn list_models(ctx: &ProxyCtx, _req: ListModelsRequest) -> Result<ListModelsResponse, RpcError> {
ctx.db.with_conn(|db| {
Ok(ListModelsResponse {
flows: db.find_all::<Flow>().map_err(|e| RpcError { message: e.to_string() })?,
})
})
fn list_flows(ctx: &ProxyCtx, req: flows::FlowFilter) -> Result<ListFlowsResponse, RpcError> {
let flows = flows::list(&ctx.db, &req).map_err(rpc_error)?;
let total = flows::count(&ctx.db).map_err(rpc_error)?;
Ok(ListFlowsResponse { flows, total })
}
async fn clear_flows(ctx: ProxyCtx, _req: ClearFlowsRequest) -> Result<usize, RpcError> {
blocking(move || ctx.clear_flows()).await
}
// -- Event loop --
@@ -746,17 +767,19 @@ define_rpc! {
format_body(FormatBodyRequest) -> String,
get_proxy_state(GetProxyStateRequest) -> GetProxyStateResponse,
list_actions(ListActionsRequest) -> ListActionsResponse,
list_models(ListModelsRequest) -> ListModelsResponse,
list_flows(flows::FlowFilter) -> ListFlowsResponse,
}
async_commands {
ca_status(CaStatusRequest) -> CaStatus,
ca_trust(CaTrustRequest) -> CaStatus,
clear_flows(ClearFlowsRequest) -> usize,
filter_body(FilterBodyRequest) -> FilterBodyResponse,
get_flow_body(GetFlowBodyRequest) -> GetFlowBodyResponse,
system_proxy_set(SystemProxySetRequest) -> SystemProxyStatus,
system_proxy_status(SystemProxyStatusRequest) -> SystemProxyStatus,
}
events {
flows_cleared(FlowsClearedPayload),
model_write(ModelPayload),
proxy_state_changed(ProxyStatePayload),
}
@@ -164,6 +164,16 @@ async fn round_trip() {
);
assert_eq!(control::read(dir.path()).unwrap().proxy_port, None);
let clear = http
.post(format!("{}/flows/clear", info.base_url()))
.bearer_auth(&info.token)
.send()
.await
.unwrap();
assert_eq!(clear.json::<control::ClearResponse>().await.unwrap().cleared, 1);
assert!(client.flows(&FlowFilter::default()).await.unwrap().is_empty());
assert!(client.body(&flow.id, FlowBody::Response).await.is_err());
drop(server);
assert!(!dir.path().join("control.json").exists());
assert!(control::live_owner(dir.path()).is_none());
-1
View File
@@ -249,7 +249,6 @@
"@tanstack/react-virtual": "^3.14.17",
"@tauri-apps/api": "^2.12.3",
"@tauri-apps/plugin-os": "^2.4.0",
"@yaakapp-internal/model-store": "^1.0.0",
"@yaakapp-internal/proxy-lib": "^1.0.0",
"@yaakapp-internal/theme": "^1.0.0",
"@yaakapp-internal/ui": "^1.0.0",
+34 -16
View File
@@ -2,16 +2,33 @@ import { atom } from "jotai";
import { atomFamily } from "jotai-family";
import { selectAtom } from "jotai/utils";
/**
* An atom family for object params, looked up by a string key. `atomFamily` with a comparator
* scans every atom it has made on each lookup, which every rendered row does several times a
* render, so a large tree slowed down in proportion to the rows it had ever shown.
*/
function keyedFamily<P, A>(key: (param: P) => string, create: (param: P) => A): (param: P) => A {
const atoms = new Map<string, A>();
return (param) => {
const k = key(param);
let a = atoms.get(k);
if (a === undefined) {
a = create(param);
atoms.set(k, a);
}
return a;
};
}
// eslint-disable-next-line @typescript-eslint/no-unused-vars
export const selectedIdsFamily = atomFamily((_treeId: string) => {
return atom<string[]>([]);
});
export const isSelectedFamily = atomFamily(
({ treeId, itemId }: { treeId: string; itemId: string }) => {
return selectAtom(selectedIdsFamily(treeId), (ids) => ids.includes(itemId), Object.is);
},
(a, b) => a.treeId === b.treeId && a.itemId === b.itemId,
export const isSelectedFamily = keyedFamily(
({ treeId, itemId }: { treeId: string; itemId: string }) => `${treeId}\0${itemId}`,
({ treeId, itemId }) =>
selectAtom(selectedIdsFamily(treeId), (ids) => ids.includes(itemId), Object.is),
);
// eslint-disable-next-line @typescript-eslint/no-unused-vars
@@ -19,10 +36,9 @@ export const focusIdsFamily = atomFamily((_treeId: string) => {
return atom<{ lastId: string | null; anchorId: string | null }>({ lastId: null, anchorId: null });
});
export const isLastFocusedFamily = atomFamily(
({ treeId, itemId }: { treeId: string; itemId: string }) =>
selectAtom(focusIdsFamily(treeId), (v) => v.lastId === itemId, Object.is),
(a, b) => a.treeId === b.treeId && a.itemId === b.itemId,
export const isLastFocusedFamily = keyedFamily(
({ treeId, itemId }: { treeId: string; itemId: string }) => `${treeId}\0${itemId}`,
({ treeId, itemId }) => selectAtom(focusIdsFamily(treeId), (v) => v.lastId === itemId, Object.is),
);
// eslint-disable-next-line @typescript-eslint/no-unused-vars
@@ -45,26 +61,28 @@ export const hoveredParentFamily = atomFamily((_treeId: string) => {
});
});
export const isParentHoveredFamily = atomFamily(
export const isParentHoveredFamily = keyedFamily(
({ treeId, parentId }: { treeId: string; parentId: string | null }) =>
`${treeId}\0${parentId ?? ""}`,
({ treeId, parentId }) =>
selectAtom(hoveredParentFamily(treeId), (v) => v.parentId === parentId, Object.is),
(a, b) => a.treeId === b.treeId && a.parentId === b.parentId,
);
export const isAncestorHoveredFamily = atomFamily(
export const isAncestorHoveredFamily = keyedFamily(
({ treeId, ancestorIds }: { treeId: string; ancestorIds: string[] }) =>
`${treeId}\0${ancestorIds.join(",")}`,
({ treeId, ancestorIds }) =>
selectAtom(
hoveredParentFamily(treeId),
(v) => v.parentId && ancestorIds.includes(v.parentId),
Object.is,
),
(a, b) => a.treeId === b.treeId && a.ancestorIds.join(",") === b.ancestorIds.join(","),
);
export const isIndexHoveredFamily = atomFamily(
({ treeId, index }: { treeId: string; index: number }) =>
export const isIndexHoveredFamily = keyedFamily(
({ treeId, index }: { treeId: string; index: number }) => `${treeId}\0${index}`,
({ treeId, index }) =>
selectAtom(hoveredParentFamily(treeId), (v) => v.index === index, Object.is),
(a, b) => a.treeId === b.treeId && a.index === b.index,
);
export const hoveredParentDepthFamily = atomFamily((treeId: string) =>