diff --git a/docs/architecture.md b/docs/architecture.md index a871dc348a..d35e079a89 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -104,7 +104,7 @@ Cross-platform React Native app that connects to one or more daemons. - Expo Router navigation (`/h/[serverId]/workspace/[workspaceId]`, `/h/[serverId]/agent/[agentId]`, etc.). The `workspaceId` URL segment is an opaque workspace id, not a directly meaningful filesystem path. - `HostRuntimeController` manages saved host connections, reconnection, and per-host runtime state -- `runtime/replica-cache` keeps the complete project, workspace, and active-agent directory plus one short focused timeline tail in AsyncStorage. It restores before navigation becomes ready and leaves remote hydration flags false. +- `runtime/replica-cache` keeps the complete project, workspace, and active-agent directory plus one short focused timeline tail in a row store. It restores before navigation becomes ready and leaves remote hydration flags false. - `runtime/directory-sync` owns directory reconciliation. On reconnect it passes the persisted per-entity cursor through `project.list`, `fetch_workspaces`, and `fetch_agents`; the daemon returns each entity's latest projection when its sequence is newer, plus tombstones. - `workspace-labels` owns one sequenced catalog replica per connected host, the deterministic cross-host projection that surfaces spanning hosts use (the filter page, the manager), and the per-host resolution a workspace row's chips use. Two hosts may give one name different colors, so a row resolves against its own host's catalog and a merged answer would be wrong there. Catalogs never synchronize between hosts; assignment creates a missing definition only on the target host. On the daemon, catalog and assignment rewrites share a journaled commit boundary. Startup recovery completes that commit before workspace or catalog publication. - `SessionContext` wraps the daemon client for the active session @@ -115,12 +115,14 @@ Cross-platform React Native app that connects to one or more daemons. The replica cache paints stale data immediately while the host connects. Directory cursors are reconciliation checkpoints; cached entities remain non-authoritative until the daemon answers. -Pending permission requests are not restored from it. AsyncStorage is not encrypted, so the cached +Pending permission requests are not restored from it. The row store is not encrypted, so the cached timeline tail may contain source code, prompts, and tool output; encrypted-at-rest storage is a -separate product/security decision. Its serialized payload has a 32 MiB byte budget and evicts whole -host snapshots in least-recently-written order; a single oversized host is omitted rather than -partially restored. Browser and Electron builds store it in IndexedDB. Native builds use -AsyncStorage, and Android reserves 64 MiB for that database. +separate product/security decision. Agents, workspaces, and projects have individual rows; the +focused timeline tail and directory checkpoint each have one row per host. Writes serialize only +rows whose in-memory identities changed. Restore validates every row before hydration; one invalid +row clears the cache and starts cold. The 32 MiB byte budget is tracked incrementally from row +payload lengths and evicts whole hosts in least-recently-written order. Browser and Electron builds +use IndexedDB. Native builds use expo-sqlite. The three directory entity types have independent monotonic sequences and share one daemon generation. The daemon retains only the latest projection per entity and bounded tombstones, not an diff --git a/packages/app/e2e/browser/replica-cache-performance.spec.ts b/packages/app/e2e/browser/replica-cache-performance.spec.ts index 11ac76bd73..2861e64878 100644 --- a/packages/app/e2e/browser/replica-cache-performance.spec.ts +++ b/packages/app/e2e/browser/replica-cache-performance.spec.ts @@ -14,7 +14,7 @@ import { const RUN_REPLICA_CACHE_PERF = process.env.PASEO_REPLICA_CACHE_PERF_E2E === "1"; const MAX_STREAM_WRITES = 12; -const MAX_SERIALIZED_CHARS = 190_000; +const MAX_SERIALIZED_CHARS = 45_000; const test = base.extend<{ replicaCacheAgent: MockAgentWorkspace }>({ replicaCacheAgent: async ({ page: _page }, provide) => { diff --git a/packages/app/e2e/support/helpers/replica-cache-measurement.ts b/packages/app/e2e/support/helpers/replica-cache-measurement.ts index 2042a0af0c..9876bf1d1e 100644 --- a/packages/app/e2e/support/helpers/replica-cache-measurement.ts +++ b/packages/app/e2e/support/helpers/replica-cache-measurement.ts @@ -1,11 +1,12 @@ import type { CDPSession, Page } from "@playwright/test"; -const REPLICA_CACHE_DATABASE = "paseo-replica-cache"; +const REPLICA_CACHE_DATABASE = "paseo-replica-row-store"; +const REPLICA_CACHE_ROWS_STORE = "rows"; const PERSIST_FRAME_NAMES = [ "captureSessions", "captureHost", - "buildBoundedPayload", - "serialize", + "queueReplicaChanges", + "apply", "safeParse", "stringify", "setItem", @@ -14,8 +15,7 @@ const PERSIST_FRAME_NAMES = [ const PERSIST_STACK_ANCHORS = new Set([ "captureSessions", "captureHost", - "buildBoundedPayload", - "serialize", + "queueReplicaChanges", "persist", "flushPending", "measuredPut", @@ -69,7 +69,7 @@ declare global { export async function installReplicaCacheMeasurement(page: Page): Promise { await page.addInitScript( - ({ databaseName }) => { + ({ databaseName, storeName }) => { const encoder = new TextEncoder(); const transactionWrites = new WeakMap< IDBTransaction, @@ -128,8 +128,12 @@ export async function installReplicaCacheMeasurement(page: Page): Promise IDBObjectStore.prototype.put = function measuredPut(value: unknown, key?: IDBValidKey) { const request = key === undefined ? originalPut.call(this, value) : originalPut.call(this, value, key); - if (this.transaction.db.name !== databaseName) return request; - const serialized = typeof value === "string" ? value : JSON.stringify(value); + if (this.transaction.db.name !== databaseName || this.name !== storeName) return request; + const rowPayload = + value && typeof value === "object" ? Reflect.get(value, "payload") : undefined; + let serialized = JSON.stringify(value); + if (typeof value === "string") serialized = value; + if (typeof rowPayload === "string") serialized = rowPayload; const storageKey = JSON.stringify(resolveStorageKey(this, value, key)); observeTransaction(this.transaction).push({ type: "put", @@ -143,7 +147,7 @@ export async function installReplicaCacheMeasurement(page: Page): Promise const originalDelete = IDBObjectStore.prototype.delete; IDBObjectStore.prototype.delete = function measuredDelete(query: IDBValidKey | IDBKeyRange) { const request = originalDelete.call(this, query); - if (this.transaction.db.name !== databaseName) return request; + if (this.transaction.db.name !== databaseName || this.name !== storeName) return request; observeTransaction(this.transaction).push({ type: "delete", key: `${this.name}:${JSON.stringify(query)}`, @@ -156,6 +160,7 @@ export async function installReplicaCacheMeasurement(page: Page): Promise const markRestoreRead = (store: IDBObjectStore) => { if ( store.transaction.db.name === databaseName && + store.name === storeName && store.transaction.mode === "readonly" && window.__replicaCacheRestoreReadStartedAt === undefined ) { @@ -197,7 +202,7 @@ export async function installReplicaCacheMeasurement(page: Page): Promise } } }, - { databaseName: REPLICA_CACHE_DATABASE }, + { databaseName: REPLICA_CACHE_DATABASE, storeName: REPLICA_CACHE_ROWS_STORE }, ); } diff --git a/packages/app/e2e/support/helpers/replica-cache-storage.ts b/packages/app/e2e/support/helpers/replica-cache-storage.ts index 78df9b5389..dc06de8dd0 100644 --- a/packages/app/e2e/support/helpers/replica-cache-storage.ts +++ b/packages/app/e2e/support/helpers/replica-cache-storage.ts @@ -1,26 +1,49 @@ import type { Page } from "@playwright/test"; -const DATABASE_NAME = "paseo-replica-cache"; -const STORE_NAME = "key-value"; -const STORAGE_KEY = "@paseo:replica-cache"; +const DATABASE_NAME = "paseo-replica-row-store"; +const STORE_NAME = "rows"; +const SINGLETON_ID = "singleton"; + +type ReplicaRowKind = "agent" | "workspace" | "project" | "timeline" | "checkpoint"; + +interface ReplicaRowRecord { + serverId: string; + kind: ReplicaRowKind; + id: string; + payload: string; +} + +interface ReplicaCacheHostRecord { + serverId: string; + agents: Array>; + workspaces: Array>; + projects: Array>; + timeline?: { + agentId?: string; + items?: Array>; + range?: { + endSeq?: number; + epoch?: string; + startSeq?: number; + } | null; + } | null; + directorySync?: unknown; +} export interface ReplicaCacheRecord { version?: number; - hosts?: Array<{ - timeline?: { - agentId?: string; - items?: Array>; - range?: { - endSeq?: number; - epoch?: string; - startSeq?: number; - } | null; - } | null; - }>; + hosts?: ReplicaCacheHostRecord[]; } interface ReplicaCacheWriteObserverState { - lastValue: string | null; + lastValues: Record; + redundantWrites: number; + serializedChars: number; + storageDurationMs: number; + writes: number; +} + +interface ReplicaCacheWriteObserverReport { redundantWrites: number; serializedChars: number; storageDurationMs: number; @@ -33,11 +56,10 @@ declare global { } } -async function readStoredValue(input: { +async function readRows(input: { databaseName: string; storeName: string; - storageKey: string; -}): Promise { +}): Promise { const database = await new Promise((resolve, reject) => { const request = indexedDB.open(input.databaseName, 1); request.addEventListener("success", () => resolve(request.result)); @@ -47,26 +69,103 @@ async function readStoredValue(input: { const request = database .transaction(input.storeName, "readonly") .objectStore(input.storeName) - .get(input.storageKey); - request.addEventListener("success", () => - resolve(typeof request.result === "string" ? request.result : null), - ); + .getAll(); + request.addEventListener("success", () => resolve(request.result)); request.addEventListener("error", () => reject(request.error)); }); } +function assembleCache(rows: ReplicaRowRecord[]): ReplicaCacheRecord { + const hosts = new Map(); + for (const row of rows) { + const host = hosts.get(row.serverId) ?? { + serverId: row.serverId, + agents: [], + workspaces: [], + projects: [], + timeline: null, + }; + const payload = JSON.parse(row.payload) as Record; + switch (row.kind) { + case "agent": + host.agents.push(payload); + break; + case "workspace": + host.workspaces.push(payload); + break; + case "project": + host.projects.push(payload); + break; + case "timeline": + host.timeline = payload; + break; + case "checkpoint": + host.directorySync = payload; + break; + } + hosts.set(row.serverId, host); + } + return { version: 6, hosts: [...hosts.values()] }; +} + export async function readReplicaCache(page: Page): Promise { - const raw = await page.evaluate(readStoredValue, { + const rows = await page.evaluate(readRows, { databaseName: DATABASE_NAME, storeName: STORE_NAME, - storageKey: STORAGE_KEY, }); - return raw ? (JSON.parse(raw) as ReplicaCacheRecord) : null; + return rows.length > 0 ? assembleCache(rows) : null; } export async function writeReplicaCache(page: Page, value: ReplicaCacheRecord): Promise { + const rows: ReplicaRowRecord[] = []; + for (const host of value.hosts ?? []) { + for (const agent of host.agents) { + const snapshot = agent.snapshot; + if (!snapshot || typeof snapshot !== "object") continue; + const id = Reflect.get(snapshot, "id"); + if (typeof id === "string") { + rows.push({ serverId: host.serverId, kind: "agent", id, payload: JSON.stringify(agent) }); + } + } + for (const workspace of host.workspaces) { + if (typeof workspace.id === "string") { + rows.push({ + serverId: host.serverId, + kind: "workspace", + id: workspace.id, + payload: JSON.stringify(workspace), + }); + } + } + for (const project of host.projects) { + if (typeof project.projectId === "string") { + rows.push({ + serverId: host.serverId, + kind: "project", + id: project.projectId, + payload: JSON.stringify(project), + }); + } + } + if (host.timeline) { + rows.push({ + serverId: host.serverId, + kind: "timeline", + id: SINGLETON_ID, + payload: JSON.stringify(host.timeline), + }); + } + if (host.directorySync !== undefined) { + rows.push({ + serverId: host.serverId, + kind: "checkpoint", + id: SINGLETON_ID, + payload: JSON.stringify(host.directorySync), + }); + } + } await page.evaluate( - async ({ databaseName, storeName, storageKey, raw }) => { + async ({ databaseName, storeName, nextRows }) => { const database = await new Promise((resolve, reject) => { const request = indexedDB.open(databaseName, 1); request.addEventListener("success", () => resolve(request.result)); @@ -74,26 +173,23 @@ export async function writeReplicaCache(page: Page, value: ReplicaCacheRecord): }); await new Promise((resolve, reject) => { const transaction = database.transaction(storeName, "readwrite"); - transaction.objectStore(storeName).put(raw, storageKey); + const store = transaction.objectStore(storeName); + store.clear(); + for (const row of nextRows) store.put(row); transaction.addEventListener("complete", () => resolve()); transaction.addEventListener("error", () => reject(transaction.error)); transaction.addEventListener("abort", () => reject(transaction.error)); }); }, - { - databaseName: DATABASE_NAME, - storeName: STORE_NAME, - storageKey: STORAGE_KEY, - raw: JSON.stringify(value), - }, + { databaseName: DATABASE_NAME, storeName: STORE_NAME, nextRows: rows }, ); } export async function observeReplicaCacheStorageWrites(page: Page): Promise { await page.addInitScript( - ({ databaseName, storeName, storageKey }) => { + ({ databaseName, storeName }) => { window.__replicaCacheWriteObserver = { - lastValue: null, + lastValues: {}, redundantWrites: 0, serializedChars: 0, storageDurationMs: 0, @@ -106,51 +202,71 @@ export async function observeReplicaCacheStorageWrites(page: Page): Promise { const state = window.__replicaCacheWriteObserver; if (!state) return; - if (state.lastValue === value) state.redundantWrites += 1; - state.lastValue = value; + if (state.lastValues[rowKey] === payload) state.redundantWrites += 1; + state.lastValues[rowKey] = payload; state.writes += 1; - state.serializedChars += value.length; + state.serializedChars += payload.length; state.storageDurationMs += performance.now() - startedAt; }); return request; }; }, - { databaseName: DATABASE_NAME, storeName: STORE_NAME, storageKey: STORAGE_KEY }, + { databaseName: DATABASE_NAME, storeName: STORE_NAME }, ); } export async function resetReplicaCacheStorageWriteObserver(page: Page): Promise { - const lastValue = await page.evaluate(readStoredValue, { + const rows = await page.evaluate(readRows, { databaseName: DATABASE_NAME, storeName: STORE_NAME, - storageKey: STORAGE_KEY, }); - await page.evaluate((value) => { + const lastValues = Object.fromEntries( + rows.map((row) => [`${row.serverId}:${row.kind}:${row.id}`, row.payload]), + ); + await page.evaluate((values) => { window.__replicaCacheWriteObserver = { - lastValue: value, + lastValues: values, redundantWrites: 0, serializedChars: 0, storageDurationMs: 0, writes: 0, }; - }, lastValue); + }, lastValues); } export async function readReplicaCacheStorageWriteObserver( page: Page, -): Promise { +): Promise { return page.evaluate(() => { const state = window.__replicaCacheWriteObserver; if (!state) throw new Error("Replica cache write observer is not installed"); - return state; + return { + redundantWrites: state.redundantWrites, + serializedChars: state.serializedChars, + storageDurationMs: state.storageDurationMs, + writes: state.writes, + }; }); } diff --git a/packages/app/src/projects/icon-cache-storage.ts b/packages/app/src/projects/icon-cache-storage.ts new file mode 100644 index 0000000000..bcbecf7d11 --- /dev/null +++ b/packages/app/src/projects/icon-cache-storage.ts @@ -0,0 +1,4 @@ +import AsyncStorage from "@react-native-async-storage/async-storage"; +import type { ProjectIconCacheStorage } from "./icon-cache"; + +export const projectIconCacheStorage: ProjectIconCacheStorage = AsyncStorage; diff --git a/packages/app/src/projects/icon-cache-storage.web.ts b/packages/app/src/projects/icon-cache-storage.web.ts new file mode 100644 index 0000000000..3f89026530 --- /dev/null +++ b/packages/app/src/projects/icon-cache-storage.web.ts @@ -0,0 +1,46 @@ +import type { ProjectIconCacheStorage } from "./icon-cache"; + +const DATABASE_NAME = "paseo-project-icon-cache"; +const STORE_NAME = "key-value"; + +let databasePromise: Promise | null = null; + +function openDatabase(): Promise { + databasePromise ??= new Promise((resolve, reject) => { + const request = indexedDB.open(DATABASE_NAME, 1); + request.addEventListener("upgradeneeded", () => { + if (!request.result.objectStoreNames.contains(STORE_NAME)) { + request.result.createObjectStore(STORE_NAME); + } + }); + request.addEventListener("success", () => resolve(request.result)); + request.addEventListener("error", () => + reject(request.error ?? new Error("Failed to open project icon cache")), + ); + }); + return databasePromise; +} + +async function getItem(key: string): Promise { + const database = await openDatabase(); + return new Promise((resolve, reject) => { + const request = database.transaction(STORE_NAME, "readonly").objectStore(STORE_NAME).get(key); + request.addEventListener("success", () => + resolve(typeof request.result === "string" ? request.result : null), + ); + request.addEventListener("error", () => reject(request.error)); + }); +} + +async function setItem(key: string, value: string): Promise { + const database = await openDatabase(); + return new Promise((resolve, reject) => { + const transaction = database.transaction(STORE_NAME, "readwrite"); + transaction.objectStore(STORE_NAME).put(value, key); + transaction.addEventListener("complete", () => resolve()); + transaction.addEventListener("error", () => reject(transaction.error)); + transaction.addEventListener("abort", () => reject(transaction.error)); + }); +} + +export const projectIconCacheStorage: ProjectIconCacheStorage = { getItem, setItem }; diff --git a/packages/app/src/projects/icon-cache.ts b/packages/app/src/projects/icon-cache.ts index d409a59ba4..e88704be23 100644 --- a/packages/app/src/projects/icon-cache.ts +++ b/packages/app/src/projects/icon-cache.ts @@ -1,6 +1,6 @@ import type { ProjectIcon } from "@getpaseo/protocol/messages"; import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; -import { replicaCacheStorage } from "@/runtime/replica-cache/storage"; +import { projectIconCacheStorage } from "./icon-cache-storage"; import type { ProjectIconTarget } from "./icon-target"; const STORAGE_KEY = "@paseo:project-icon-cache"; @@ -197,7 +197,7 @@ export class ProjectIconCache { } } -export const projectIconCache = new ProjectIconCache(replicaCacheStorage); +export const projectIconCache = new ProjectIconCache(projectIconCacheStorage); function resolveLookup( target: Pick, diff --git a/packages/app/src/runtime/host-runtime.test.ts b/packages/app/src/runtime/host-runtime.test.ts index 9fc711d683..0548cf853f 100644 --- a/packages/app/src/runtime/host-runtime.test.ts +++ b/packages/app/src/runtime/host-runtime.test.ts @@ -22,6 +22,7 @@ import { type HostRuntimeStorage, } from "./host-runtime"; import { ReplicaCache } from "./replica-cache"; +import type { ReplicaRow, ReplicaRowStore } from "./replica-cache/row-store"; class FakeDaemonClient { private state: ConnectionState = { status: "idle" }; @@ -451,6 +452,40 @@ function createMemoryHostRuntimeStorage(entries: Record = {}): H }; } +function createMemoryReplicaRowStore(): ReplicaRowStore { + const rows = new Map(); + const keyOf = (row: Pick) => + `${row.serverId}:${row.kind}:${row.id}`; + return { + open: async () => undefined, + readAll: async () => { + const hosts = new Map(); + for (const row of rows.values()) { + const hostRows = hosts.get(row.serverId) ?? []; + hostRows.push(row); + hosts.set(row.serverId, hostRows); + } + return [...hosts].map(([serverId, hostRows]) => ({ serverId, rows: hostRows })); + }, + apply: async (changes) => { + for (const key of changes.deletes) rows.delete(keyOf(key)); + for (const row of changes.upserts) rows.set(keyOf(row), row); + }, + deleteHost: async (serverId) => { + for (const [key, row] of rows) if (row.serverId === serverId) rows.delete(key); + }, + renameHost: async (oldServerId, newServerId) => { + for (const [key, row] of rows) { + if (row.serverId !== oldServerId) continue; + rows.delete(key); + const renamed = { ...row, serverId: newServerId }; + rows.set(keyOf(renamed), renamed); + } + }, + clear: async () => rows.clear(), + }; +} + function createAppearanceStore(storage: HostRuntimeStorage): HostRuntimeStore { return new HostRuntimeStore({ storage, @@ -1438,6 +1473,7 @@ describe("HostRuntimeStore", () => { it("restores the display replica before declaring the host registry loaded", async () => { const host = makeHost(); const storage = createMemoryHostRuntimeStorage(); + const replicaRowStore = createMemoryReplicaRowStore(); await storage.setItem("@paseo:daemon-registry", JSON.stringify([host])); await storage.setItem("@paseo:e2e", "1"); @@ -1470,13 +1506,14 @@ describe("HostRuntimeStore", () => { ], ]), ); - const cache = new ReplicaCache(storage); + const cache = new ReplicaCache(replicaRowStore, { clearLegacyCache: async () => undefined }); cache.setHosts([host.serverId]); await cache.flush(); session.clearSession(host.serverId); const store = new HostRuntimeStore({ storage, + replicaRowStore, deps: { createClient: () => { throw new Error("createClient should not be called"); diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index eda39b7e80..684b5e6227 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -60,7 +60,8 @@ import { resolveComposerAttachmentSubmitFormat } from "@/composer/attachments/su import { encodeImages } from "@/utils/encode-images"; import { DirectorySync, type RefreshAgentDirectoryResult } from "@/runtime/directory-sync"; import { ReplicaCache } from "@/runtime/replica-cache"; -import { replicaCacheStorage } from "@/runtime/replica-cache/storage"; +import type { ReplicaRowStore } from "@/runtime/replica-cache/row-store"; +import { createReplicaRowStore } from "@/runtime/replica-cache/row-store-factory"; import { projectIconCache } from "@/projects/icon-cache"; import { nativePerformanceTrace } from "@/performance/native-trace"; import { revokePushNotifications } from "@/push-notifications"; @@ -1373,11 +1374,12 @@ export class HostRuntimeStore { constructor(input?: { deps?: HostRuntimeControllerDeps; storage?: HostRuntimeStorage; + replicaRowStore?: ReplicaRowStore; revokePushNotifications?: typeof revokePushNotifications; }) { this.deps = input?.deps ?? createDefaultDeps(); this.storage = input?.storage ?? AsyncStorage; - this.replicaCache = new ReplicaCache(input?.storage ?? replicaCacheStorage); + this.replicaCache = new ReplicaCache(input?.replicaRowStore ?? createReplicaRowStore()); this.revokePushNotifications = input?.revokePushNotifications ?? revokePushNotifications; } diff --git a/packages/app/src/runtime/replica-cache/diff.test.ts b/packages/app/src/runtime/replica-cache/diff.test.ts new file mode 100644 index 0000000000..36c8f41659 --- /dev/null +++ b/packages/app/src/runtime/replica-cache/diff.test.ts @@ -0,0 +1,63 @@ +import { describe, expect, it } from "vitest"; +import type { StreamItem } from "@/types/stream"; +import { diffMap, diffReplicaInput, type ReplicaInput } from "./diff"; + +function input(overrides: Partial = {}): ReplicaInput { + return { + agents: new Map(), + workspaces: new Map(), + projects: new Map(), + focusedAgentId: undefined, + timelineItems: undefined, + timelineRange: null, + timelineHasOlder: false, + ...overrides, + }; +} + +describe("diffReplicaInput", () => { + it("upserts only values whose map identity changed and deletes missing ids", () => { + const retained = { value: "retained" }; + const previousValue = { value: "previous" }; + const changed = { value: "changed" }; + const removed = { value: "removed" }; + const previous = new Map([ + ["retained", retained], + ["changed", previousValue], + ["removed", removed], + ]); + const next = new Map([ + ["retained", retained], + ["changed", changed], + ]); + + expect(diffMap(previous, next)).toEqual({ + upserts: [changed], + deletes: ["removed"], + }); + }); + + it("does not rewrite the timeline when its inputs retain identity", () => { + const timelineItems: StreamItem[] = []; + const range = { epoch: "epoch", startSeq: 1, endSeq: 1 }; + const previous = input({ + focusedAgentId: "agent", + timelineItems, + timelineRange: range, + }); + const next = input({ + focusedAgentId: "agent", + timelineItems, + timelineRange: range, + }); + + expect(diffReplicaInput(previous, next).timelineChanged).toBe(false); + }); + + it("rewrites the timeline when one of its inputs changes identity", () => { + const previous = input({ focusedAgentId: "agent", timelineItems: [] }); + const next = input({ focusedAgentId: "agent", timelineItems: [] }); + + expect(diffReplicaInput(previous, next).timelineChanged).toBe(true); + }); +}); diff --git a/packages/app/src/runtime/replica-cache/diff.ts b/packages/app/src/runtime/replica-cache/diff.ts new file mode 100644 index 0000000000..8cfc4b0cb9 --- /dev/null +++ b/packages/app/src/runtime/replica-cache/diff.ts @@ -0,0 +1,77 @@ +import type { + Agent, + ProjectDescriptor, + SessionReplicaTimeline, + WorkspaceDescriptor, +} from "@/stores/session-store"; +import type { StreamItem } from "@/types/stream"; + +export interface ReplicaInput { + agents: ReadonlyMap; + workspaces: ReadonlyMap; + projects: ReadonlyMap; + focusedAgentId: string | undefined; + timelineItems: StreamItem[] | undefined; + timelineRange: SessionReplicaTimeline["range"]; + timelineHasOlder: boolean; +} + +interface EntityChanges { + upserts: Entity[]; + deletes: string[]; +} + +export interface ReplicaInputChanges { + agents: EntityChanges; + workspaces: EntityChanges; + projects: EntityChanges; + timelineChanged: boolean; +} + +export function diffMap( + previous: ReadonlyMap | undefined, + next: ReadonlyMap, +): EntityChanges { + if (previous === next) return { upserts: [], deletes: [] }; + const upserts: Entity[] = []; + const deletes: string[] = []; + for (const [id, value] of next) { + if (previous?.get(id) !== value) upserts.push(value); + } + if (previous) { + for (const id of previous.keys()) { + if (!next.has(id)) deletes.push(id); + } + } + return { upserts, deletes }; +} + +export function diffReplicaInput( + previous: ReplicaInput | undefined, + next: ReplicaInput, +): ReplicaInputChanges { + const timelineChanged = + previous === undefined || + previous.focusedAgentId !== next.focusedAgentId || + previous.timelineItems !== next.timelineItems || + previous.timelineRange !== next.timelineRange || + previous.timelineHasOlder !== next.timelineHasOlder; + return { + agents: diffMap(previous?.agents, next.agents), + workspaces: diffMap(previous?.workspaces, next.workspaces), + projects: diffMap(previous?.projects, next.projects), + timelineChanged, + }; +} + +export function hasReplicaInputChanges(changes: ReplicaInputChanges): boolean { + return ( + changes.agents.upserts.length > 0 || + changes.agents.deletes.length > 0 || + changes.workspaces.upserts.length > 0 || + changes.workspaces.deletes.length > 0 || + changes.projects.upserts.length > 0 || + changes.projects.deletes.length > 0 || + changes.timelineChanged + ); +} diff --git a/packages/app/src/runtime/replica-cache/index.test.ts b/packages/app/src/runtime/replica-cache/index.test.ts index 73cad3a9a2..96134fb803 100644 --- a/packages/app/src/runtime/replica-cache/index.test.ts +++ b/packages/app/src/runtime/replica-cache/index.test.ts @@ -8,27 +8,68 @@ import { useSessionStore, } from "@/stores/session-store"; import { createUserMessage, type StreamItem } from "@/types/stream"; -import { ReplicaCache, type ReplicaCacheStorage } from "."; +import { ReplicaCache } from "."; +import type { ReplicaHostRows, ReplicaRow, ReplicaRowChanges, ReplicaRowStore } from "./row-store"; const SERVER_ID = "cached-host"; const LRU_SERVER_IDS = ["host-a", "host-b", "host-c"] as const; -class MemoryStorage implements ReplicaCacheStorage { - readonly values = new Map(); +class MemoryStorage implements ReplicaRowStore { + readonly rows = new Map(); + readonly changes: ReplicaRowChanges[] = []; + readonly deletedHosts: string[] = []; + readonly renamedHosts: Array<{ oldServerId: string; newServerId: string }> = []; writes = 0; + clears = 0; - async getItem(key: string): Promise { - return this.values.get(key) ?? null; + private key(row: Pick): string { + return `${row.serverId}:${row.kind}:${row.id}`; } - async setItem(key: string, value: string): Promise { + async open(): Promise {} + + async readAll(): Promise { + const hosts = new Map(); + for (const row of this.rows.values()) { + const rows = hosts.get(row.serverId) ?? []; + rows.push(row); + hosts.set(row.serverId, rows); + } + return Array.from(hosts, ([serverId, rows]) => ({ serverId, rows })); + } + + async apply(changes: ReplicaRowChanges): Promise { this.writes += 1; - this.values.set(key, value); + this.changes.push(changes); + for (const key of changes.deletes) this.rows.delete(this.key(key)); + for (const row of changes.upserts) this.rows.set(this.key(row), row); } - async removeItem(key: string): Promise { - this.values.delete(key); + async deleteHost(serverId: string): Promise { + this.deletedHosts.push(serverId); + for (const [key, row] of this.rows) if (row.serverId === serverId) this.rows.delete(key); } + + async renameHost(oldServerId: string, newServerId: string): Promise { + this.renamedHosts.push({ oldServerId, newServerId }); + for (const [key, row] of this.rows) { + if (row.serverId !== oldServerId) continue; + this.rows.delete(key); + const renamed = { ...row, serverId: newServerId }; + this.rows.set(this.key(renamed), renamed); + } + } + + async clear(): Promise { + this.clears += 1; + this.rows.clear(); + } +} + +const NO_LEGACY_CLEANUP = { clearLegacyCache: async () => undefined }; + +function cache(storage: MemoryStorage, options: { maxBytes?: number } = {}): ReplicaCache { + return new ReplicaCache(storage, { ...options, ...NO_LEGACY_CLEANUP }); } function workspace( @@ -139,7 +180,26 @@ function toolCall(): StreamItem { function seedSession(): void { const store = useSessionStore.getState(); store.initializeSession(SERVER_ID, null); - store.setAgents(SERVER_ID, new Map([["agent-1", agent("agent-1")]])); + store.setAgents( + SERVER_ID, + new Map([ + [ + "agent-1", + { + ...agent("agent-1"), + pendingPermissions: [ + { + id: "pending-permission", + provider: "codex", + name: "shell", + kind: "tool", + title: "Approve shell command", + }, + ], + }, + ], + ]), + ); store.setWorkspaces( SERVER_ID, new Map([ @@ -212,18 +272,18 @@ describe("ReplicaCache", () => { it("persists after user inactivity even while replica changes continue", async () => { vi.useFakeTimers(); const storage = new MemoryStorage(); - const cache = new ReplicaCache(storage); - cache.setHosts([SERVER_ID]); + const replicaCache = cache(storage); + replicaCache.setHosts([SERVER_ID]); seedSession(); - await cache.flush(); - cache.start(); + await replicaCache.flush(); + replicaCache.start(); const writesBeforeChange = storage.writes; useSessionStore .getState() .setAgentStreamTail(SERVER_ID, new Map([["agent-1", [message("first", "First")]]])); await vi.advanceTimersByTimeAsync(4_000); - cache.recordUserActivity(); + replicaCache.recordUserActivity(); await vi.advanceTimersByTimeAsync(1_000); useSessionStore .getState() @@ -235,17 +295,17 @@ describe("ReplicaCache", () => { await vi.advanceTimersByTimeAsync(1); expect(storage.writes).toBe(writesBeforeChange + 1); - cache.setHosts([]); + replicaCache.setHosts([]); }); it("persists focused replica changes without writing transient stream head updates", async () => { vi.useFakeTimers(); const storage = new MemoryStorage(); - const cache = new ReplicaCache(storage); - cache.setHosts([SERVER_ID]); + const replicaCache = cache(storage); + replicaCache.setHosts([SERVER_ID]); seedSession(); - await cache.flush(); - cache.start(); + await replicaCache.flush(); + replicaCache.start(); const writesBeforeStream = storage.writes; useSessionStore @@ -261,19 +321,19 @@ describe("ReplicaCache", () => { await vi.advanceTimersByTimeAsync(5_000); expect(storage.writes).toBe(writesBeforeStream + 1); - cache.setHosts([]); + replicaCache.setHosts([]); }); it("restores the exact persisted canonical timeline window", async () => { const storage = new MemoryStorage(); - const writer = new ReplicaCache(storage); + const writer = cache(storage); writer.setHosts([SERVER_ID]); seedSession(); await writer.flush(); useSessionStore.getState().clearSession(SERVER_ID); - const reader = new ReplicaCache(storage); + const reader = cache(storage); reader.setHosts([SERVER_ID]); await reader.restore(); @@ -289,6 +349,7 @@ describe("ReplicaCache", () => { expect(Array.from(session.projects.keys())).toEqual(["project-1", "empty-project"]); expect(session.agents.get("agent-1")?.updatedAt).toBeInstanceOf(Date); expect(session.agents.get("agent-1")?.projectPlacement?.checkout.cwd).toBe("/repo/paseo"); + expect(session.agents.get("agent-1")?.pendingPermissions).toEqual([]); expect(session.workspaces.get("workspace-1")?.statusEnteredAt).toBeInstanceOf(Date); expect(session.workspaces.get("workspace-1")?.worktreeSlug).toBe("owned-worktree"); // A restored row draws its label chips. The reconnect cursor is current, so nothing re-sends @@ -313,7 +374,7 @@ describe("ReplicaCache", () => { it("restores canonical turn membership without downgrading tagged rows", async () => { const storage = new MemoryStorage(); - const writer = new ReplicaCache(storage); + const writer = cache(storage); writer.setHosts([SERVER_ID]); seedSession(); const initial: StreamItem = { @@ -340,7 +401,7 @@ describe("ReplicaCache", () => { ); await writer.flush(); useSessionStore.getState().clearSession(SERVER_ID); - const reader = new ReplicaCache(storage); + const reader = cache(storage); reader.setHosts([SERVER_ID]); await reader.restore(); const tail = @@ -350,14 +411,14 @@ describe("ReplicaCache", () => { it("restores tool calls inside an authoritative cached window", async () => { const storage = new MemoryStorage(); - const writer = new ReplicaCache(storage); + const writer = cache(storage); writer.setHosts([SERVER_ID]); seedSession(); useSessionStore.getState().setAgentStreamTail(SERVER_ID, new Map([["agent-1", [toolCall()]]])); await writer.flush(); useSessionStore.getState().clearSession(SERVER_ID); - const reader = new ReplicaCache(storage); + const reader = cache(storage); reader.setHosts([SERVER_ID]); await reader.restore(); @@ -372,7 +433,7 @@ describe("ReplicaCache", () => { it("restores display-only when retained items do not reach the stored range end", async () => { const storage = new MemoryStorage(); - const writer = new ReplicaCache(storage); + const writer = cache(storage); writer.setHosts([SERVER_ID]); seedSession(); useSessionStore @@ -384,7 +445,7 @@ describe("ReplicaCache", () => { await writer.flush(); useSessionStore.getState().clearSession(SERVER_ID); - const reader = new ReplicaCache(storage); + const reader = cache(storage); reader.setHosts([SERVER_ID]); await reader.restore(); @@ -399,8 +460,8 @@ describe("ReplicaCache", () => { it("persists the complete directory with only the focused timeline tail", async () => { const storage = new MemoryStorage(); - const cache = new ReplicaCache(storage); - cache.setHosts([SERVER_ID]); + const replicaCache = cache(storage); + replicaCache.setHosts([SERVER_ID]); seedSession(); const store = useSessionStore.getState(); @@ -431,10 +492,10 @@ describe("ReplicaCache", () => { store.setAgentTimelineHasOlder(SERVER_ID, new Map([["agent-2", true]])); store.setAgentAuthoritativeHistoryApplied(SERVER_ID, "agent-2", true); store.setFocusedAgentId(SERVER_ID, "agent-2"); - await cache.flush(); + await replicaCache.flush(); store.clearSession(SERVER_ID); - const reader = new ReplicaCache(storage); + const reader = cache(storage); reader.setHosts([SERVER_ID]); await reader.restore(); @@ -455,23 +516,17 @@ describe("ReplicaCache", () => { items: secondTimeline.slice(-50), }); - const persisted = JSON.parse(storage.values.get("@paseo:replica-cache") ?? "null") as { - version: number; - hosts: Array<{ timeline: Record | null }>; - }; - expect(persisted.version).toBe(6); - expect(Object.keys(persisted.hosts[0]?.timeline ?? {}).sort()).toEqual([ - "agentId", - "hasOlder", - "items", - "range", - ]); + const timelineRow = [...storage.rows.values()].find((row) => row.kind === "timeline"); + expect(timelineRow).toBeDefined(); + if (!timelineRow) throw new Error("Expected persisted timeline row"); + const persisted = JSON.parse(timelineRow.payload) as Record; + expect(Object.keys(persisted).sort()).toEqual(["agentId", "hasOlder", "items", "range"]); }); it("persists reconciled rows without caching unreconciled local presentations", async () => { const storage = new MemoryStorage(); - const cache = new ReplicaCache(storage); - cache.setHosts([SERVER_ID]); + const replicaCache = cache(storage); + replicaCache.setHosts([SERVER_ID]); seedSession(); const unreconciled = createUserMessage({ clientMessageId: "client-pending", @@ -489,9 +544,9 @@ describe("ReplicaCache", () => { .getState() .setAgentStreamTail(SERVER_ID, new Map([["agent-1", [unreconciled, reconciled]]])); - await cache.flush(); + await replicaCache.flush(); useSessionStore.getState().clearSession(SERVER_ID); - await cache.restore(); + await replicaCache.restore(); expect(useSessionStore.getState().sessions[SERVER_ID]?.agentStreamTail.get("agent-1")).toEqual([ reconciled, @@ -500,12 +555,12 @@ describe("ReplicaCache", () => { it("persists monotonic directory cursors with the complete host replica", async () => { const storage = new MemoryStorage(); - const cache = new ReplicaCache(storage); - cache.setHosts([SERVER_ID]); + const replicaCache = cache(storage); + replicaCache.setHosts([SERVER_ID]); seedSession(); - await cache.flush(); + await replicaCache.flush(); - cache.writeDirectoryCheckpoint(SERVER_ID, { + replicaCache.writeDirectoryCheckpoint(SERVER_ID, { agents: { generation: "daemon-generation", afterSeq: 7 }, }); useSessionStore.getState().setAgents(SERVER_ID, (agents) => { @@ -513,14 +568,75 @@ describe("ReplicaCache", () => { if (!current) throw new Error("Expected seeded agent"); return new Map(agents).set("agent-1", { ...current, title: "Updated agent" }); }); - await cache.flush(); + await replicaCache.flush(); - const reader = new ReplicaCache(storage); + const reader = cache(storage); reader.setHosts([SERVER_ID]); await reader.restore(); expect(reader.readDirectoryCheckpoint(SERVER_ID)).toEqual({ agents: { generation: "daemon-generation", afterSeq: 7 }, }); + const finalWrite = storage.changes.at(-1); + expect(finalWrite?.upserts.map((row) => row.kind).sort()).toEqual(["agent", "checkpoint"]); + }); + + it("persists one agent update as exactly one row upsert", async () => { + const storage = new MemoryStorage(); + const replicaCache = cache(storage); + replicaCache.setHosts([SERVER_ID]); + seedSession(); + await replicaCache.flush(); + storage.changes.length = 0; + + useSessionStore.getState().setAgents(SERVER_ID, (agents) => { + const current = agents.get("agent-1"); + if (!current) throw new Error("Expected seeded agent"); + return new Map(agents).set(current.id, { ...current, title: "Updated" }); + }); + await replicaCache.flush(); + + expect(storage.changes).toHaveLength(1); + expect(storage.changes[0]).toMatchObject({ + upserts: [{ serverId: SERVER_ID, kind: "agent", id: "agent-1" }], + deletes: [], + }); + }); + + it("persists a checkpoint without serializing host entity rows", async () => { + const storage = new MemoryStorage(); + const replicaCache = cache(storage); + replicaCache.setHosts([SERVER_ID]); + seedSession(); + await replicaCache.flush(); + storage.changes.length = 0; + + replicaCache.writeDirectoryCheckpoint(SERVER_ID, { + agents: { generation: "generation", afterSeq: 4 }, + }); + await replicaCache.flush(); + + expect(storage.changes).toHaveLength(1); + expect(storage.changes[0]?.upserts).toEqual([ + expect.objectContaining({ kind: "checkpoint", id: "singleton" }), + ]); + }); + + it("maps host removal and reconciliation to row-store host operations", async () => { + const storage = new MemoryStorage(); + const replicaCache = cache(storage); + replicaCache.setHosts([SERVER_ID]); + seedSession(); + await replicaCache.flush(); + + replicaCache.reconcileServerId(SERVER_ID, "reconciled-host"); + await replicaCache.flush(); + replicaCache.setHosts([]); + await replicaCache.flush(); + + expect(storage.renamedHosts).toEqual([ + { oldServerId: SERVER_ID, newServerId: "reconciled-host" }, + ]); + expect(storage.deletedHosts).toContain("reconciled-host"); }); it("restores workspace change request checks beside the directory cursor", async () => { @@ -546,7 +662,7 @@ describe("ReplicaCache", () => { error: null, }; const storage = new MemoryStorage(); - const writer = new ReplicaCache(storage); + const writer = cache(storage); writer.setHosts([SERVER_ID]); seedSession(); useSessionStore.getState().setWorkspaces( @@ -568,7 +684,7 @@ describe("ReplicaCache", () => { await writer.flush(); useSessionStore.getState().clearSession(SERVER_ID); - const reader = new ReplicaCache(storage); + const reader = cache(storage); reader.setHosts([SERVER_ID]); await reader.restore(); @@ -582,13 +698,13 @@ describe("ReplicaCache", () => { it("restores every registered host directory before any host reconnects", async () => { const storage = new MemoryStorage(); - const writer = new ReplicaCache(storage); + const writer = cache(storage); writer.setHosts(LRU_SERVER_IDS); for (const serverId of LRU_SERVER_IDS) seedTimeline(serverId, `cached-${serverId}`); await writer.flush(); for (const serverId of LRU_SERVER_IDS) useSessionStore.getState().clearSession(serverId); - const reader = new ReplicaCache(storage); + const reader = cache(storage); reader.setHosts(LRU_SERVER_IDS); await reader.restore(); @@ -604,62 +720,83 @@ describe("ReplicaCache", () => { it("evicts the least recently written host when the cache exceeds its byte budget", async () => { const storage = new MemoryStorage(); - const cache = new ReplicaCache(storage, { maxBytes: 7_000 }); - cache.setHosts(LRU_SERVER_IDS.slice(0, 2)); + const replicaCache = cache(storage, { maxBytes: 7_000 }); + replicaCache.setHosts(LRU_SERVER_IDS.slice(0, 2)); seedTimeline("host-a", "A".repeat(1_200)); seedTimeline("host-b", "B".repeat(1_200)); - await cache.flush(); + await replicaCache.flush(); seedTimeline("host-a", "A".repeat(1_201)); - await cache.flush(); + await replicaCache.flush(); - cache.setHosts(LRU_SERVER_IDS); + replicaCache.setHosts(LRU_SERVER_IDS); seedTimeline("host-c", "C".repeat(1_200)); - await cache.flush(); + await replicaCache.flush(); for (const serverId of LRU_SERVER_IDS) { useSessionStore.getState().clearSession(serverId); } - const reader = new ReplicaCache(storage, { maxBytes: 7_000 }); + const reader = cache(storage, { maxBytes: 7_000 }); reader.setHosts(LRU_SERVER_IDS); await reader.restore(); expect(Object.keys(useSessionStore.getState().sessions).sort()).toEqual(["host-a", "host-c"]); }); - it("rejects and clears version 5 cache data before overwriting it on flush", async () => { + it("clears the whole cache when one row payload is corrupt", async () => { const storage = new MemoryStorage(); - storage.values.set( - "@paseo:replica-cache", - JSON.stringify({ - version: 5, - hosts: [ - { - serverId: SERVER_ID, - agents: [], - workspaces: [], - emptyProjects: [], - timeline: { - agentId: "agent-1", - items: [], - cursor: { epoch: "poisoned", startSeq: 1, endSeq: 100 }, - hasOlder: false, - }, - }, - ], - }), - ); - const cache = new ReplicaCache(storage); - cache.setHosts([SERVER_ID]); + storage.rows.set(`${SERVER_ID}:agent:agent-1`, { + serverId: SERVER_ID, + kind: "agent", + id: "agent-1", + payload: "{not json", + }); + storage.rows.set(`${SERVER_ID}:project:project-1`, { + serverId: SERVER_ID, + kind: "project", + id: "project-1", + payload: JSON.stringify({ projectId: "project-1" }), + }); + const replicaCache = cache(storage); + replicaCache.setHosts([SERVER_ID]); - await cache.restore(); - expect(storage.values.has("@paseo:replica-cache")).toBe(false); - await cache.flush(); + await replicaCache.restore(); expect(useSessionStore.getState().sessions[SERVER_ID]).toBeUndefined(); - expect(JSON.parse(storage.values.get("@paseo:replica-cache") ?? "null")).toEqual({ - version: 6, - hosts: [], + expect(storage.rows.size).toBe(0); + expect(storage.clears).toBe(1); + }); + + it("clears the whole cache when a valid JSON row has an unknown shape", async () => { + const storage = new MemoryStorage(); + storage.rows.set(`${SERVER_ID}:project:project-1`, { + serverId: SERVER_ID, + kind: "project", + id: "project-1", + payload: JSON.stringify({ projectId: "project-1", surprise: true }), }); + const replicaCache = cache(storage); + replicaCache.setHosts([SERVER_ID]); + + await replicaCache.restore(); + + expect(storage.rows.size).toBe(0); + expect(storage.clears).toBe(1); + }); + + it("runs legacy blob cleanup when the row store is first prepared", async () => { + const storage = new MemoryStorage(); + let cleanups = 0; + const replicaCache = new ReplicaCache(storage, { + clearLegacyCache: async () => { + cleanups += 1; + }, + }); + replicaCache.setHosts([SERVER_ID]); + + await replicaCache.restore(); + await replicaCache.flush(); + + expect(cleanups).toBe(1); }); }); diff --git a/packages/app/src/runtime/replica-cache/index.ts b/packages/app/src/runtime/replica-cache/index.ts index ba758efc5d..bab3cf93f9 100644 --- a/packages/app/src/runtime/replica-cache/index.ts +++ b/packages/app/src/runtime/replica-cache/index.ts @@ -14,7 +14,6 @@ import { useSessionStore, type Agent, type SessionReplica, - type SessionReplicaTimeline, type SessionState, type ProjectDescriptor, type WorkspaceDescriptor, @@ -25,9 +24,23 @@ import { type StreamItem, } from "@/types/stream"; import { normalizeAgentSnapshot } from "@/utils/agent-snapshots"; +import { + diffReplicaInput, + hasReplicaInputChanges, + type ReplicaInput, + type ReplicaInputChanges, +} from "./diff"; +import { clearLegacyReplicaCache } from "./legacy-cleanup"; +import { + REPLICA_SINGLETON_ROW_ID, + type ReplicaHostRows, + type ReplicaRow, + type ReplicaRowChanges, + type ReplicaRowKey, + type ReplicaRowKind, + type ReplicaRowStore, +} from "./row-store"; -const STORAGE_KEY = "@paseo:replica-cache"; -const CACHE_VERSION = 6; const PERSIST_AFTER_USER_INACTIVITY_MS = 5_000; const MAX_TIMELINE_ITEMS = 50; const MAX_CACHE_BYTES = 32 * 1024 * 1024; @@ -289,36 +302,17 @@ const StoredHostSchema = z.strictObject({ directorySync: z.unknown().optional(), }); -const StoredCacheSchema = z.strictObject({ - version: z.literal(CACHE_VERSION), - hosts: z.array(StoredHostSchema), -}); - type StoredAgent = z.infer; type StoredHost = z.infer; +type StoredTimeline = z.infer; type StoredTimelineItem = z.infer; type StoredToolCall = Extract["item"]; type StoredWorkspace = z.infer; type StoredProject = z.infer; -interface ReplicaInput { - agents: ReadonlyMap; - workspaces: ReadonlyMap; - projects: ReadonlyMap; - focusedAgent: Agent | undefined; - timelineItems: StreamItem[] | undefined; - timelineRange: SessionReplicaTimeline["range"]; - timelineHasOlder: boolean; -} - -export interface ReplicaCacheStorage { - getItem: (key: string) => Promise; - setItem: (key: string, value: string) => Promise; - removeItem: (key: string) => Promise; -} - interface ReplicaCacheOptions { maxBytes?: number; + clearLegacyCache?: () => Promise; } function deserializeTimeline(stored: StoredHost["timeline"]): SessionReplica["timeline"] { @@ -628,18 +622,6 @@ function isTimelineItemStoredLosslessly(item: StreamItem): boolean { } } -function replicaInputsEqual(left: ReplicaInput, right: ReplicaInput): boolean { - return ( - left.agents === right.agents && - left.workspaces === right.workspaces && - left.projects === right.projects && - left.focusedAgent === right.focusedAgent && - left.timelineItems === right.timelineItems && - left.timelineRange === right.timelineRange && - left.timelineHasOlder === right.timelineHasOlder - ); -} - function selectReplicaInput(session: SessionState, agentId: string | null): ReplicaInput { const agent = agentId ? session.agents.get(agentId) : undefined; const timeline = agentId @@ -649,14 +631,14 @@ function selectReplicaInput(session: SessionState, agentId: string | null): Repl agents: session.agents, workspaces: session.workspaces, projects: session.projects, - focusedAgent: agent, + focusedAgentId: agent?.id, timelineItems: timeline.status === "cold" ? undefined : timeline.items, timelineRange: timeline.status === "synced" ? timeline.range : null, timelineHasOlder: timeline.status === "synced" && timeline.older === "available", }; } -function serializeHost(serverId: string, input: ReplicaInput, directorySync?: unknown): StoredHost { +function serializeTimeline(input: ReplicaInput): StoredTimeline | null { const canonicalItems = input.timelineItems?.filter( (item) => item.kind !== "user_message" || !isUnreconciledLocalUserMessage(item), ); @@ -678,24 +660,14 @@ function serializeHost(serverId: string, input: ReplicaInput, directorySync?: un item.timelineCursor.seq <= range.endSeq, ) && canonicalItems.some((item) => item.timelineCursor?.seq === range.endSeq); + if (!input.focusedAgentId || !items) return null; return { - serverId, - agents: Array.from(input.agents.values(), serializeAgent), - workspaces: Array.from(input.workspaces.values(), serializeWorkspace), - projects: Array.from(input.projects.values(), serializeProject), - emptyProjects: [], - timeline: - input.focusedAgent && items - ? { - agentId: input.focusedAgent.id, - items: items.slice(-MAX_TIMELINE_ITEMS), - range: canPersistCoverage - ? { epoch: range.epoch, startSeq: range.startSeq, endSeq: range.endSeq } - : null, - hasOlder: canPersistCoverage ? input.timelineHasOlder : false, - } - : null, - ...(directorySync ? { directorySync } : {}), + agentId: input.focusedAgentId, + items: items.slice(-MAX_TIMELINE_ITEMS), + range: canPersistCoverage + ? { epoch: range.epoch, startSeq: range.startSeq, endSeq: range.endSeq } + : null, + hasOlder: canPersistCoverage ? input.timelineHasOlder : false, }; } @@ -729,38 +701,66 @@ function legacyProjectDescriptorFromWorkspace(workspace: WorkspaceDescriptor): P }; } +function rowKey(key: Pick): string { + return `${key.kind}\u0000${key.id}`; +} + +function pendingRowKey(key: ReplicaRowKey): string { + return `${key.serverId}\u0000${rowKey(key)}`; +} + +function payloadBytes(payload: string): number { + return Buffer.byteLength(payload, "utf8"); +} + +function parseJsonPayload(payload: string): unknown { + return JSON.parse(payload); +} + +function parseStoredPayload(schema: z.ZodType, payload: string): Value { + const parsed = schema.safeParse(parseJsonPayload(payload)); + if (!parsed.success) throw new Error("Invalid replica row payload"); + return parsed.data; +} + export class ReplicaCache { private readonly activeServerIds = new Set(); - private readonly storedHosts = new Map(); + private readonly storedRows = new Map>(); + private readonly hostBytes = new Map(); + private readonly hostWriteOrder = new Map(); + private readonly evictedServerIds = new Set(); private readonly lastFocusedAgentIds = new Map(); private readonly capturedInputs = new Map(); + private readonly directoryCheckpoints = new Map(); + private readonly checkpointInputs = new Map(); + private pendingUpserts = new Map(); + private pendingDeletes = new Map(); private readonly maxBytes: number; - private needsPersist = false; + private totalBytes = 0; private unsubscribe: (() => void) | null = null; private persistTimer: ReturnType | null = null; private writeQueue: Promise = Promise.resolve(); + private preparePromise: Promise | null = null; constructor( - private readonly storage: ReplicaCacheStorage, + private readonly rowStore: ReplicaRowStore, options: ReplicaCacheOptions = {}, ) { - const emptyPayloadBytes = Buffer.byteLength( - JSON.stringify({ version: CACHE_VERSION, hosts: [] }), - "utf8", - ); - this.maxBytes = Math.max(options.maxBytes ?? MAX_CACHE_BYTES, emptyPayloadBytes); + this.maxBytes = Math.max(options.maxBytes ?? MAX_CACHE_BYTES, 0); + this.clearLegacyCache = options.clearLegacyCache ?? clearLegacyReplicaCache; } + private readonly clearLegacyCache: () => Promise; + setHosts(serverIds: Iterable): void { const next = new Set(serverIds); + const removed = [...this.activeServerIds].filter((serverId) => !next.has(serverId)); this.activeServerIds.clear(); for (const serverId of next) this.activeServerIds.add(serverId); - let removedStoredHost = false; - for (const serverId of this.storedHosts.keys()) { - if (!next.has(serverId)) { - this.storedHosts.delete(serverId); - removedStoredHost = true; - } + for (const serverId of removed) { + this.removeStoredHost(serverId); + this.dropPendingHostChanges(serverId); + this.queueOperation(() => this.rowStore.deleteHost(serverId)); } for (const serverId of this.lastFocusedAgentIds.keys()) { if (!next.has(serverId)) this.lastFocusedAgentIds.delete(serverId); @@ -768,41 +768,52 @@ export class ReplicaCache { for (const serverId of this.capturedInputs.keys()) { if (!next.has(serverId)) this.capturedInputs.delete(serverId); } - if (removedStoredHost) this.needsPersist = true; - if (this.unsubscribe && this.needsPersist) this.schedulePersist(); + for (const serverId of this.directoryCheckpoints.keys()) { + if (!next.has(serverId)) this.directoryCheckpoints.delete(serverId); + } + for (const serverId of this.checkpointInputs.keys()) { + if (!next.has(serverId)) this.checkpointInputs.delete(serverId); + } + for (const serverId of this.evictedServerIds) { + if (!next.has(serverId)) this.evictedServerIds.delete(serverId); + } } async restore(): Promise { - let raw: string | null; try { - raw = await this.storage.getItem(STORAGE_KEY); + await this.prepareStore(); } catch { return; } - if (!raw) return; - let parsed: unknown; + let hosts: ReplicaHostRows[]; try { - parsed = JSON.parse(raw); + hosts = await this.rowStore.readAll(); } catch { - await this.clearInvalidCache(); return; } - const cache = StoredCacheSchema.safeParse(parsed); - if (!cache.success) { - await this.clearInvalidCache(); - return; - } - for (const host of cache.data.hosts) { - if (!this.activeServerIds.has(host.serverId)) { - this.needsPersist = true; - continue; + this.storedRows.clear(); + this.hostBytes.clear(); + this.hostWriteOrder.clear(); + this.directoryCheckpoints.clear(); + this.checkpointInputs.clear(); + this.totalBytes = 0; + const restoredHosts: StoredHost[] = []; + try { + for (const hostRows of hosts) { + if (!this.activeServerIds.has(hostRows.serverId)) { + await this.rowStore.deleteHost(hostRows.serverId); + continue; + } + restoredHosts.push(this.restoreHostRows(hostRows)); } - this.storedHosts.set(host.serverId, host); - if (host.timeline) this.lastFocusedAgentIds.set(host.serverId, host.timeline.agentId); + } catch { + await this.clearInvalidCache(); + return; } - const bounded = this.buildBoundedPayload(); - if (bounded.evicted) this.needsPersist = true; - for (const host of this.storedHosts.values()) { + await this.evictOverBudget(); + for (const host of restoredHosts) { + if (!this.storedRows.has(host.serverId)) continue; + if (host.timeline) this.lastFocusedAgentIds.set(host.serverId, host.timeline.agentId); useSessionStore.getState().restoreSessionReplica(host.serverId, deserializeHost(host)); const session = useSessionStore.getState().sessions[host.serverId]; if (session) { @@ -828,14 +839,24 @@ export class ReplicaCache { if (this.activeServerIds.size === 0) return; this.schedulePersist(); }); - if (changedBeforeSubscription || this.needsPersist) this.schedulePersist(); + if (changedBeforeSubscription || this.hasPendingChanges()) this.schedulePersist(); } reconcileServerId(oldServerId: string, newServerId: string): void { - const stored = this.storedHosts.get(oldServerId); - if (stored) { - this.storedHosts.delete(oldServerId); - this.storedHosts.set(newServerId, { ...stored, serverId: newServerId }); + const rows = this.storedRows.get(oldServerId); + if (rows) { + const newRows = this.storedRows.get(newServerId) ?? new Map(); + this.totalBytes -= + (this.hostBytes.get(oldServerId) ?? 0) + (this.hostBytes.get(newServerId) ?? 0); + this.storedRows.delete(oldServerId); + for (const [key, row] of rows) newRows.set(key, { ...row, serverId: newServerId }); + this.storedRows.set(newServerId, newRows); + const bytes = [...newRows.values()].reduce((sum, row) => sum + payloadBytes(row.payload), 0); + this.hostBytes.delete(oldServerId); + this.hostBytes.set(newServerId, bytes); + this.totalBytes += bytes; + this.hostWriteOrder.delete(oldServerId); + this.touchHost(newServerId); } const focusedAgentId = this.lastFocusedAgentIds.get(oldServerId); if (focusedAgentId) { @@ -847,32 +868,56 @@ export class ReplicaCache { this.capturedInputs.delete(oldServerId); this.capturedInputs.set(newServerId, capturedInput); } + if (this.directoryCheckpoints.has(oldServerId)) { + const checkpoint = this.directoryCheckpoints.get(oldServerId); + this.directoryCheckpoints.delete(oldServerId); + this.directoryCheckpoints.set(newServerId, checkpoint); + } + if (this.checkpointInputs.has(oldServerId)) { + const checkpoint = this.checkpointInputs.get(oldServerId); + this.checkpointInputs.delete(oldServerId); + this.checkpointInputs.set(newServerId, checkpoint); + } + if (this.evictedServerIds.delete(oldServerId)) this.evictedServerIds.add(newServerId); + this.renamePendingHostChanges(oldServerId, newServerId); if (this.activeServerIds.delete(oldServerId)) this.activeServerIds.add(newServerId); - this.needsPersist = true; - this.schedulePersist(); + this.queueOperation(() => this.rowStore.renameHost(oldServerId, newServerId)); } readDirectoryCheckpoint(serverId: string): unknown { - return this.storedHosts.get(serverId)?.directorySync; + return this.directoryCheckpoints.get(serverId); } writeDirectoryCheckpoint(serverId: string, checkpoint: unknown): void { - let stored = this.storedHosts.get(serverId); - if (!stored) { + if (this.checkpointInputs.get(serverId) === checkpoint) return; + if (this.evictedServerIds.has(serverId)) { const session = useSessionStore.getState().sessions[serverId]; if (!session) return; - const input = selectReplicaInput(session, this.lastFocusedAgentIds.get(serverId) ?? null); - this.capturedInputs.set(serverId, input); - stored = serializeHost(serverId, input); + this.captureHost(serverId, session); + } + this.checkpointInputs.set(serverId, checkpoint); + this.directoryCheckpoints.set(serverId, checkpoint); + const payload = JSON.stringify(checkpoint); + if (payload === undefined) { + this.queueDelete({ + serverId, + kind: "checkpoint", + id: REPLICA_SINGLETON_ROW_ID, + }); + } else { + this.queueUpsert({ + serverId, + kind: "checkpoint", + id: REPLICA_SINGLETON_ROW_ID, + payload, + }); } - this.storedHosts.delete(serverId); - this.storedHosts.set(serverId, { ...stored, directorySync: checkpoint }); - this.needsPersist = true; this.schedulePersist(); } async flush(): Promise { await this.persist(false); + await this.writeQueue.catch(() => undefined); } private async flushPending(): Promise { @@ -889,14 +934,24 @@ export class ReplicaCache { this.persistTimer = null; } const changed = this.captureSessions(); - if (skipUnchanged && !changed && !this.needsPersist) return; - const bounded = this.buildBoundedPayload(); - this.needsPersist = false; + if (skipUnchanged && !changed && !this.hasPendingChanges()) return; + const changes = this.drainPendingChanges(); + if (changes.upserts.length === 0 && changes.deletes.length === 0) return; const write = this.writeQueue .catch(() => undefined) - .then(() => this.storage.setItem(STORAGE_KEY, bounded.payload)); + .then(async () => { + try { + await this.prepareStore(); + await this.rowStore.apply(changes); + this.applyStoredChanges(changes); + await this.evictOverBudget(); + } catch { + this.restorePendingChanges(changes); + } + return undefined; + }); this.writeQueue = write; - await write.catch(() => undefined); + await write; } private captureSessions(): boolean { @@ -929,41 +984,268 @@ export class ReplicaCache { } else if (hasTimelineHead) { input = { ...selected, timelineRange: null, timelineHasOlder: false }; } - if (previous && replicaInputsEqual(previous, input)) return false; - + const initialInput = this.evictedServerIds.has(serverId) ? undefined : previous; + const changes = diffReplicaInput(initialInput, input); this.capturedInputs.set(serverId, input); - const directorySync = this.storedHosts.get(serverId)?.directorySync; - this.storedHosts.delete(serverId); - this.storedHosts.set(serverId, serializeHost(serverId, input, directorySync)); + if (!hasReplicaInputChanges(changes)) return false; + this.queueReplicaChanges(serverId, input, changes); return true; } - private buildBoundedPayload(): { payload: string; evicted: boolean } { - let evicted = false; - let payload = this.serialize(); - while (Buffer.byteLength(payload, "utf8") > this.maxBytes && this.storedHosts.size > 0) { - const oldestServerId = this.storedHosts.keys().next().value; - if (oldestServerId === undefined) break; - this.storedHosts.delete(oldestServerId); - evicted = true; - payload = this.serialize(); + private queueReplicaChanges( + serverId: string, + input: ReplicaInput, + changes: ReplicaInputChanges, + ): void { + for (const agent of changes.agents.upserts) { + this.queueEntityUpsert(serverId, "agent", agent.id, serializeAgent(agent)); + } + for (const id of changes.agents.deletes) this.queueEntityDelete(serverId, "agent", id); + for (const workspace of changes.workspaces.upserts) { + this.queueEntityUpsert(serverId, "workspace", workspace.id, serializeWorkspace(workspace)); + } + for (const id of changes.workspaces.deletes) this.queueEntityDelete(serverId, "workspace", id); + for (const project of changes.projects.upserts) { + this.queueEntityUpsert(serverId, "project", project.projectId, serializeProject(project)); + } + for (const id of changes.projects.deletes) this.queueEntityDelete(serverId, "project", id); + if (!changes.timelineChanged) return; + const timeline = serializeTimeline(input); + if (timeline) { + this.queueEntityUpsert(serverId, "timeline", REPLICA_SINGLETON_ROW_ID, timeline); + } else { + this.queueEntityDelete(serverId, "timeline", REPLICA_SINGLETON_ROW_ID); } - return { payload, evicted }; } - private serialize(): string { - return JSON.stringify({ - version: CACHE_VERSION, - hosts: Array.from(this.storedHosts.values()), - }); + private queueEntityUpsert( + serverId: string, + kind: ReplicaRowKind, + id: string, + value: unknown, + ): void { + this.queueUpsert({ serverId, kind, id, payload: JSON.stringify(value) }); } private async clearInvalidCache(): Promise { try { - await this.storage.removeItem(STORAGE_KEY); + await this.rowStore.clear(); } catch { // A storage failure must not turn invalid cached data into application state. } + this.storedRows.clear(); + this.hostBytes.clear(); + this.hostWriteOrder.clear(); + this.evictedServerIds.clear(); + this.directoryCheckpoints.clear(); + this.checkpointInputs.clear(); + this.capturedInputs.clear(); + this.lastFocusedAgentIds.clear(); + this.pendingUpserts.clear(); + this.pendingDeletes.clear(); + this.totalBytes = 0; + } + + private restoreHostRows(hostRows: ReplicaHostRows): StoredHost { + const agents: StoredAgent[] = []; + const workspaces: StoredWorkspace[] = []; + const projects: StoredProject[] = []; + let timeline: StoredTimeline | null = null; + let checkpoint: unknown; + const rows = new Map(); + let bytes = 0; + for (const row of hostRows.rows) { + if (row.serverId !== hostRows.serverId) throw new Error("Replica row host key mismatch"); + switch (row.kind) { + case "agent": { + const agent = parseStoredPayload(StoredAgentSchema, row.payload); + if (agent.snapshot.id !== row.id) throw new Error("Replica agent row id mismatch"); + agents.push(agent); + break; + } + case "workspace": { + const workspace = parseStoredPayload(StoredWorkspaceSchema, row.payload); + if (workspace.id !== row.id) throw new Error("Replica workspace row id mismatch"); + workspaces.push(workspace); + break; + } + case "project": { + const project = parseStoredPayload(StoredProjectSchema, row.payload); + if (project.projectId !== row.id) throw new Error("Replica project row id mismatch"); + projects.push(project); + break; + } + case "timeline": + if (row.id !== REPLICA_SINGLETON_ROW_ID) { + throw new Error("Replica timeline row id mismatch"); + } + timeline = parseStoredPayload(StoredTimelineSchema, row.payload); + break; + case "checkpoint": + if (row.id !== REPLICA_SINGLETON_ROW_ID) { + throw new Error("Replica checkpoint row id mismatch"); + } + checkpoint = parseJsonPayload(row.payload); + break; + default: + throw new Error("Unknown replica row kind"); + } + rows.set(rowKey(row), row); + bytes += payloadBytes(row.payload); + } + this.storedRows.set(hostRows.serverId, rows); + this.hostBytes.set(hostRows.serverId, bytes); + this.totalBytes += bytes; + this.touchHost(hostRows.serverId); + if (checkpoint !== undefined) { + this.directoryCheckpoints.set(hostRows.serverId, checkpoint); + this.checkpointInputs.set(hostRows.serverId, checkpoint); + } + return { + serverId: hostRows.serverId, + agents, + workspaces, + projects, + emptyProjects: [], + timeline, + ...(checkpoint !== undefined ? { directorySync: checkpoint } : {}), + }; + } + + private queueEntityDelete(serverId: string, kind: ReplicaRowKind, id: string): void { + this.queueDelete({ serverId, kind, id }); + } + + private queueUpsert(row: ReplicaRow): void { + const key = pendingRowKey(row); + this.pendingDeletes.delete(key); + this.pendingUpserts.set(key, row); + } + + private queueDelete(key: ReplicaRowKey): void { + const pendingKey = pendingRowKey(key); + this.pendingUpserts.delete(pendingKey); + this.pendingDeletes.set(pendingKey, key); + } + + private hasPendingChanges(): boolean { + return this.pendingUpserts.size > 0 || this.pendingDeletes.size > 0; + } + + private drainPendingChanges(): ReplicaRowChanges { + const changes = { + upserts: [...this.pendingUpserts.values()], + deletes: [...this.pendingDeletes.values()], + }; + this.pendingUpserts = new Map(); + this.pendingDeletes = new Map(); + return changes; + } + + private restorePendingChanges(changes: ReplicaRowChanges): void { + const newer = this.drainPendingChanges(); + for (const key of changes.deletes) this.queueDelete(key); + for (const row of changes.upserts) this.queueUpsert(row); + for (const key of newer.deletes) this.queueDelete(key); + for (const row of newer.upserts) this.queueUpsert(row); + } + + private applyStoredChanges(changes: ReplicaRowChanges): void { + const touchedServerIds = new Set(); + for (const key of changes.deletes) { + const rows = this.storedRows.get(key.serverId); + const previous = rows?.get(rowKey(key)); + if (previous) { + rows?.delete(rowKey(key)); + this.adjustHostBytes(key.serverId, -payloadBytes(previous.payload)); + } + touchedServerIds.add(key.serverId); + } + for (const row of changes.upserts) { + const rows = this.storedRows.get(row.serverId) ?? new Map(); + const previous = rows.get(rowKey(row)); + const previousBytes = previous ? payloadBytes(previous.payload) : 0; + rows.set(rowKey(row), row); + this.storedRows.set(row.serverId, rows); + this.adjustHostBytes(row.serverId, payloadBytes(row.payload) - previousBytes); + this.evictedServerIds.delete(row.serverId); + touchedServerIds.add(row.serverId); + } + for (const serverId of touchedServerIds) { + if ((this.storedRows.get(serverId)?.size ?? 0) === 0) { + this.removeStoredHost(serverId); + } else { + this.touchHost(serverId); + } + } + } + + private adjustHostBytes(serverId: string, delta: number): void { + this.hostBytes.set(serverId, (this.hostBytes.get(serverId) ?? 0) + delta); + this.totalBytes += delta; + } + + private touchHost(serverId: string): void { + this.hostWriteOrder.delete(serverId); + this.hostWriteOrder.set(serverId, true); + } + + private async evictOverBudget(): Promise { + while (this.totalBytes > this.maxBytes) { + const oldestServerId = this.hostWriteOrder.keys().next().value; + if (oldestServerId === undefined) return; + await this.rowStore.deleteHost(oldestServerId); + this.removeStoredHost(oldestServerId); + this.evictedServerIds.add(oldestServerId); + this.directoryCheckpoints.delete(oldestServerId); + this.checkpointInputs.delete(oldestServerId); + } + } + + private removeStoredHost(serverId: string): void { + this.totalBytes -= this.hostBytes.get(serverId) ?? 0; + this.storedRows.delete(serverId); + this.hostBytes.delete(serverId); + this.hostWriteOrder.delete(serverId); + } + + private dropPendingHostChanges(serverId: string): void { + for (const [key, row] of this.pendingUpserts) { + if (row.serverId === serverId) this.pendingUpserts.delete(key); + } + for (const [key, row] of this.pendingDeletes) { + if (row.serverId === serverId) this.pendingDeletes.delete(key); + } + } + + private renamePendingHostChanges(oldServerId: string, newServerId: string): void { + const changes = this.drainPendingChanges(); + for (const key of changes.deletes) { + this.queueDelete(key.serverId === oldServerId ? { ...key, serverId: newServerId } : key); + } + for (const row of changes.upserts) { + this.queueUpsert(row.serverId === oldServerId ? { ...row, serverId: newServerId } : row); + } + } + + private prepareStore(): Promise { + this.preparePromise ??= (async () => { + await this.rowStore.open(); + // COMPAT(replica-blob-cache): remove after 2026-11 + await this.clearLegacyCache().catch(() => undefined); + })(); + return this.preparePromise; + } + + private queueOperation(operation: () => Promise): void { + this.writeQueue = this.writeQueue + .catch(() => undefined) + .then(async () => { + await this.prepareStore(); + await operation(); + return undefined; + }) + .catch(() => undefined); } private schedulePersist(): void { diff --git a/packages/app/src/runtime/replica-cache/legacy-cleanup.ts b/packages/app/src/runtime/replica-cache/legacy-cleanup.ts new file mode 100644 index 0000000000..2428b637ca --- /dev/null +++ b/packages/app/src/runtime/replica-cache/legacy-cleanup.ts @@ -0,0 +1,7 @@ +import AsyncStorage from "@react-native-async-storage/async-storage"; + +const LEGACY_STORAGE_KEY = "@paseo:replica-cache"; + +export async function clearLegacyReplicaCache(): Promise { + await AsyncStorage.removeItem(LEGACY_STORAGE_KEY); +} diff --git a/packages/app/src/runtime/replica-cache/legacy-cleanup.web.test.ts b/packages/app/src/runtime/replica-cache/legacy-cleanup.web.test.ts new file mode 100644 index 0000000000..fb1fd7c37e --- /dev/null +++ b/packages/app/src/runtime/replica-cache/legacy-cleanup.web.test.ts @@ -0,0 +1,35 @@ +import { indexedDB as fakeIndexedDb } from "fake-indexeddb"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { clearLegacyReplicaCache } from "./legacy-cleanup.web"; + +const LEGACY_DATABASE_NAME = "paseo-replica-cache"; + +beforeAll(() => vi.stubGlobal("indexedDB", fakeIndexedDb)); +afterAll(() => vi.unstubAllGlobals()); + +function openLegacyDatabase(): Promise<{ database: IDBDatabase; oldVersion: number }> { + return new Promise((resolve, reject) => { + const request = indexedDB.open(LEGACY_DATABASE_NAME, 1); + let oldVersion = 1; + request.addEventListener("upgradeneeded", (event) => { + oldVersion = event.oldVersion; + request.result.createObjectStore("key-value"); + }); + request.addEventListener("success", () => resolve({ database: request.result, oldVersion })); + request.addEventListener("error", () => reject(request.error)); + }); +} + +describe("clearLegacyReplicaCache", () => { + it("deletes the old blob database", async () => { + const seeded = await openLegacyDatabase(); + seeded.database.close(); + + await clearLegacyReplicaCache(); + + const reopened = await openLegacyDatabase(); + expect(reopened.oldVersion).toBe(0); + reopened.database.close(); + await clearLegacyReplicaCache(); + }); +}); diff --git a/packages/app/src/runtime/replica-cache/legacy-cleanup.web.ts b/packages/app/src/runtime/replica-cache/legacy-cleanup.web.ts new file mode 100644 index 0000000000..5b15704535 --- /dev/null +++ b/packages/app/src/runtime/replica-cache/legacy-cleanup.web.ts @@ -0,0 +1,12 @@ +const LEGACY_DATABASE_NAME = "paseo-replica-cache"; + +export function clearLegacyReplicaCache(): Promise { + return new Promise((resolve, reject) => { + const request = indexedDB.deleteDatabase(LEGACY_DATABASE_NAME); + request.addEventListener("success", () => resolve()); + request.addEventListener("error", () => + reject(request.error ?? new Error("Failed to delete legacy replica cache")), + ); + request.addEventListener("blocked", () => resolve()); + }); +} diff --git a/packages/app/src/runtime/replica-cache/row-store-factory.native.ts b/packages/app/src/runtime/replica-cache/row-store-factory.native.ts new file mode 100644 index 0000000000..d1c8fafad3 --- /dev/null +++ b/packages/app/src/runtime/replica-cache/row-store-factory.native.ts @@ -0,0 +1 @@ +export { createReplicaRowStore } from "./row-store.native"; diff --git a/packages/app/src/runtime/replica-cache/row-store-factory.ts b/packages/app/src/runtime/replica-cache/row-store-factory.ts new file mode 100644 index 0000000000..42c86fdb72 --- /dev/null +++ b/packages/app/src/runtime/replica-cache/row-store-factory.ts @@ -0,0 +1 @@ +export { createReplicaRowStore } from "./row-store.web"; diff --git a/packages/app/src/runtime/replica-cache/row-store-factory.web.ts b/packages/app/src/runtime/replica-cache/row-store-factory.web.ts new file mode 100644 index 0000000000..42c86fdb72 --- /dev/null +++ b/packages/app/src/runtime/replica-cache/row-store-factory.web.ts @@ -0,0 +1 @@ +export { createReplicaRowStore } from "./row-store.web"; diff --git a/packages/app/src/runtime/replica-cache/storage.ts b/packages/app/src/runtime/replica-cache/storage.ts deleted file mode 100644 index e4bc220e24..0000000000 --- a/packages/app/src/runtime/replica-cache/storage.ts +++ /dev/null @@ -1,4 +0,0 @@ -import AsyncStorage from "@react-native-async-storage/async-storage"; -import type { ReplicaCacheStorage } from "./index"; - -export const replicaCacheStorage: ReplicaCacheStorage = AsyncStorage; diff --git a/packages/app/src/runtime/replica-cache/storage.web.ts b/packages/app/src/runtime/replica-cache/storage.web.ts deleted file mode 100644 index ef2ef68960..0000000000 --- a/packages/app/src/runtime/replica-cache/storage.web.ts +++ /dev/null @@ -1,70 +0,0 @@ -import type { ReplicaCacheStorage } from "./index"; - -const DATABASE_NAME = "paseo-replica-cache"; -const STORE_NAME = "key-value"; - -let databasePromise: Promise | null = null; - -function openDatabase(): Promise { - databasePromise ??= new Promise((resolve, reject) => { - const request = indexedDB.open(DATABASE_NAME, 1); - request.addEventListener("upgradeneeded", () => { - if (!request.result.objectStoreNames.contains(STORE_NAME)) { - request.result.createObjectStore(STORE_NAME); - } - }); - request.addEventListener("success", () => resolve(request.result)); - request.addEventListener("error", () => - reject(request.error ?? new Error("Failed to open replica cache")), - ); - request.addEventListener("blocked", () => - reject(new Error("Replica cache database upgrade was blocked")), - ); - }); - return databasePromise; -} - -async function getItem(key: string): Promise { - const database = await openDatabase(); - return new Promise((resolve, reject) => { - const request = database.transaction(STORE_NAME, "readonly").objectStore(STORE_NAME).get(key); - request.addEventListener("success", () => - resolve(typeof request.result === "string" ? request.result : null), - ); - request.addEventListener("error", () => - reject(request.error ?? new Error("Failed to read replica cache")), - ); - }); -} - -async function setItem(key: string, value: string): Promise { - const database = await openDatabase(); - return new Promise((resolve, reject) => { - const transaction = database.transaction(STORE_NAME, "readwrite"); - transaction.objectStore(STORE_NAME).put(value, key); - transaction.addEventListener("complete", () => resolve()); - transaction.addEventListener("error", () => - reject(transaction.error ?? new Error("Failed to write replica cache")), - ); - transaction.addEventListener("abort", () => - reject(transaction.error ?? new Error("Replica cache write was aborted")), - ); - }); -} - -async function removeItem(key: string): Promise { - const database = await openDatabase(); - return new Promise((resolve, reject) => { - const transaction = database.transaction(STORE_NAME, "readwrite"); - transaction.objectStore(STORE_NAME).delete(key); - transaction.addEventListener("complete", () => resolve()); - transaction.addEventListener("error", () => - reject(transaction.error ?? new Error("Failed to remove replica cache")), - ); - transaction.addEventListener("abort", () => - reject(transaction.error ?? new Error("Replica cache removal was aborted")), - ); - }); -} - -export const replicaCacheStorage: ReplicaCacheStorage = { getItem, setItem, removeItem };