diff --git a/package-lock.json b/package-lock.json index f8383859ea..375ba257fb 100644 --- a/package-lock.json +++ b/package-lock.json @@ -15070,6 +15070,12 @@ "url": "https://github.com/sponsors/ljharb" } }, + "node_modules/await-lock": { + "version": "2.2.2", + "resolved": "https://registry.npmjs.org/await-lock/-/await-lock-2.2.2.tgz", + "integrity": "sha512-aDczADvlvTGajTDjcjpJMqRkOF6Qdz3YbPZm/PyW6tKPkx2hlYBzxMhEywM/tU72HrVZjgl5VCdRuMlA7pZ8Gw==", + "license": "MIT" + }, "node_modules/axios": { "version": "1.13.6", "resolved": "https://registry.npmjs.org/axios/-/axios-1.13.6.tgz", @@ -21336,6 +21342,20 @@ "node": ">=8.0" } }, + "node_modules/expo-sqlite": { + "version": "16.0.10", + "resolved": "https://registry.npmjs.org/expo-sqlite/-/expo-sqlite-16.0.10.tgz", + "integrity": "sha512-tUOKxE9TpfneRG3eOfbNfhN9236SJ7IiUnP8gCqU7umd9DtgDGB/5PhYVVfl+U7KskgolgNoB9v9OZ9iwXN8Eg==", + "license": "MIT", + "dependencies": { + "await-lock": "^2.2.2" + }, + "peerDependencies": { + "expo": "*", + "react": "*", + "react-native": "*" + } + }, "node_modules/expo-system-ui": { "version": "6.0.9", "resolved": "https://registry.npmjs.org/expo-system-ui/-/expo-system-ui-6.0.9.tgz", @@ -21781,6 +21801,16 @@ "license": "MIT", "optional": true }, + "node_modules/fake-indexeddb": { + "version": "6.2.5", + "resolved": "https://registry.npmjs.org/fake-indexeddb/-/fake-indexeddb-6.2.5.tgz", + "integrity": "sha512-CGnyrvbhPlWYMngksqrSSUT1BAVP49dZocrHuK0SvtR0D5TMs5wP0o3j7jexDJW01KSadjBp1M/71o/KR3nD1w==", + "dev": true, + "license": "Apache-2.0", + "engines": { + "node": ">=18" + } + }, "node_modules/fast-copy": { "version": "4.0.2", "resolved": "https://registry.npmjs.org/fast-copy/-/fast-copy-4.0.2.tgz", @@ -36200,6 +36230,7 @@ "expo-router": "~6.0.13", "expo-sharing": "^14.0.8", "expo-splash-screen": "~31.0.10", + "expo-sqlite": "~16.0.10", "expo-system-ui": "~6.0.7", "fast-deep-equal": "^3.1.3", "htmlparser2": "^12.0.0", @@ -36241,6 +36272,7 @@ "@testing-library/dom": "^10.4.1", "@testing-library/react": "^16.3.2", "@types/markdown-it": "^14.1.2", + "@types/node": "~22.19.0", "@types/qrcode": "^1.5.6", "@types/react": "~19.2.0", "@types/turndown": "^5.0.6", @@ -36254,6 +36286,7 @@ "eslint": "^9.25.0", "eslint-config-expo": "~10.0.0", "expo-gradle-jvmargs": "^1.1.2", + "fake-indexeddb": "^6.2.5", "jsdom": "^20.0.3", "material-icon-theme": "^5.32.0", "playwright": "^1.56.1", @@ -36791,6 +36824,16 @@ "node": ">=18" } }, + "packages/app/node_modules/@types/node": { + "version": "22.19.21", + "resolved": "https://registry.npmjs.org/@types/node/-/node-22.19.21.tgz", + "integrity": "sha512-VMeFBSCKQKmm2swI2kW51SFusDqekC6q9trBCvJ/JliDchFSuoYYKN7yVNjPthP1HKZcx3U1gI/wTcEBjEFKTA==", + "dev": true, + "license": "MIT", + "dependencies": { + "undici-types": "~6.21.0" + } + }, "packages/app/node_modules/@xterm/addon-clipboard": { "version": "0.3.0-beta.213", "resolved": "https://registry.npmjs.org/@xterm/addon-clipboard/-/addon-clipboard-0.3.0-beta.213.tgz", diff --git a/packages/app/package.json b/packages/app/package.json index 319cce38bd..75a14bef65 100644 --- a/packages/app/package.json +++ b/packages/app/package.json @@ -97,6 +97,7 @@ "expo-router": "~6.0.13", "expo-sharing": "^14.0.8", "expo-splash-screen": "~31.0.10", + "expo-sqlite": "~16.0.10", "expo-system-ui": "~6.0.7", "fast-deep-equal": "^3.1.3", "htmlparser2": "^12.0.0", @@ -138,6 +139,7 @@ "@testing-library/dom": "^10.4.1", "@testing-library/react": "^16.3.2", "@types/markdown-it": "^14.1.2", + "@types/node": "~22.19.0", "@types/qrcode": "^1.5.6", "@types/react": "~19.2.0", "@types/turndown": "^5.0.6", @@ -151,6 +153,7 @@ "eslint": "^9.25.0", "eslint-config-expo": "~10.0.0", "expo-gradle-jvmargs": "^1.1.2", + "fake-indexeddb": "^6.2.5", "jsdom": "^20.0.3", "material-icon-theme": "^5.32.0", "playwright": "^1.56.1", diff --git a/packages/app/src/runtime/replica-cache/row-store-schema.ts b/packages/app/src/runtime/replica-cache/row-store-schema.ts new file mode 100644 index 0000000000..7bd7e8d197 --- /dev/null +++ b/packages/app/src/runtime/replica-cache/row-store-schema.ts @@ -0,0 +1,3 @@ +export const REPLICA_ROW_STORE_SCHEMA_VERSION = 1; + +export const REPLICA_SINGLETON_ROW_ID = "singleton"; diff --git a/packages/app/src/runtime/replica-cache/row-store-sqlite.ts b/packages/app/src/runtime/replica-cache/row-store-sqlite.ts new file mode 100644 index 0000000000..deedc2c912 --- /dev/null +++ b/packages/app/src/runtime/replica-cache/row-store-sqlite.ts @@ -0,0 +1,151 @@ +import type { ReplicaHostRows, ReplicaRow, ReplicaRowChanges, ReplicaRowStore } from "./row-store"; + +export type SqliteValue = string | number | null; + +export interface ReplicaSqliteConnection { + exec(sql: string): Promise; + run(sql: string, params?: readonly SqliteValue[]): Promise; + all(sql: string, params?: readonly SqliteValue[]): Promise; + transaction(operation: (connection: ReplicaSqliteConnection) => Promise): Promise; +} + +export interface ReplicaSqliteDriver { + open(): Promise; +} + +const CREATE_ROWS_SQL = ` + CREATE TABLE IF NOT EXISTS rows ( + server_id TEXT NOT NULL, + kind TEXT NOT NULL CHECK (kind IN ('agent', 'workspace', 'project', 'timeline', 'checkpoint')), + id TEXT NOT NULL, + payload TEXT NOT NULL, + PRIMARY KEY (server_id, kind, id) + ) +`; + +const CREATE_META_SQL = ` + CREATE TABLE IF NOT EXISTS meta ( + key TEXT PRIMARY KEY NOT NULL, + value TEXT NOT NULL + ) +`; + +interface StoredMeta { + value: string; +} + +interface StoredRow { + server_id: string; + kind: ReplicaRow["kind"]; + id: string; + payload: string; +} + +export function createSqliteReplicaRowStore( + driver: ReplicaSqliteDriver, + schemaVersion: number, +): ReplicaRowStore { + let connection: ReplicaSqliteConnection | null = null; + let opening: Promise | null = null; + + function getConnection(): ReplicaSqliteConnection { + if (!connection) { + throw new Error("Replica row store has not been opened"); + } + return connection; + } + + async function open(): Promise { + opening ??= (async () => { + const openedConnection = await driver.open(); + await openedConnection.exec(CREATE_META_SQL); + const storedMeta = await openedConnection.all( + "SELECT value FROM meta WHERE key = ?", + ["schema_version"], + ); + + if (storedMeta[0]?.value !== String(schemaVersion)) { + await openedConnection.transaction(async (transaction) => { + await transaction.exec("DROP TABLE IF EXISTS rows"); + await transaction.exec(CREATE_ROWS_SQL); + await transaction.run("DELETE FROM meta"); + await transaction.run("INSERT INTO meta (key, value) VALUES (?, ?)", [ + "schema_version", + String(schemaVersion), + ]); + }); + } else { + await openedConnection.exec(CREATE_ROWS_SQL); + } + + connection = openedConnection; + })(); + + try { + await opening; + } catch (error) { + opening = null; + throw error; + } + } + + async function readAll(): Promise { + const storedRows = await getConnection().all( + "SELECT server_id, kind, id, payload FROM rows ORDER BY server_id, kind, id", + ); + const hosts = new Map(); + for (const storedRow of storedRows) { + const rows = hosts.get(storedRow.server_id) ?? []; + rows.push({ + serverId: storedRow.server_id, + kind: storedRow.kind, + id: storedRow.id, + payload: storedRow.payload, + }); + hosts.set(storedRow.server_id, rows); + } + return Array.from(hosts, ([serverId, rows]) => ({ serverId, rows })); + } + + async function apply(changes: ReplicaRowChanges): Promise { + await getConnection().transaction(async (transaction) => { + for (const key of changes.deletes) { + await transaction.run("DELETE FROM rows WHERE server_id = ? AND kind = ? AND id = ?", [ + key.serverId, + key.kind, + key.id, + ]); + } + for (const row of changes.upserts) { + await transaction.run( + `INSERT INTO rows (server_id, kind, id, payload) VALUES (?, ?, ?, ?) + ON CONFLICT (server_id, kind, id) DO UPDATE SET payload = excluded.payload`, + [row.serverId, row.kind, row.id, row.payload], + ); + } + }); + } + + async function deleteHost(serverId: string): Promise { + await getConnection().run("DELETE FROM rows WHERE server_id = ?", [serverId]); + } + + async function renameHost(oldServerId: string, newServerId: string): Promise { + if (oldServerId === newServerId) return; + await getConnection().transaction(async (transaction) => { + await transaction.run( + `INSERT INTO rows (server_id, kind, id, payload) + SELECT ?, kind, id, payload FROM rows WHERE server_id = ? + ON CONFLICT (server_id, kind, id) DO UPDATE SET payload = excluded.payload`, + [newServerId, oldServerId], + ); + await transaction.run("DELETE FROM rows WHERE server_id = ?", [oldServerId]); + }); + } + + async function clear(): Promise { + await getConnection().run("DELETE FROM rows"); + } + + return { open, readAll, apply, deleteHost, renameHost, clear }; +} diff --git a/packages/app/src/runtime/replica-cache/row-store.contract.ts b/packages/app/src/runtime/replica-cache/row-store.contract.ts new file mode 100644 index 0000000000..ed8e57c766 --- /dev/null +++ b/packages/app/src/runtime/replica-cache/row-store.contract.ts @@ -0,0 +1,179 @@ +import { beforeEach, describe, expect, it } from "vitest"; +import type { ReplicaRow, ReplicaRowStore } from "./row-store"; + +export interface ReplicaRowStoreHarness { + store: ReplicaRowStore; + openWithSchemaVersion(schemaVersion: number): Promise; +} + +export type CreateReplicaRowStoreHarness = () => Promise; + +function row(serverId: string, kind: ReplicaRow["kind"], id: string, payload: string): ReplicaRow { + return { serverId, kind, id, payload }; +} + +export function runReplicaRowStoreContract( + engineName: string, + createHarness: CreateReplicaRowStoreHarness, +): void { + describe(`${engineName} replica row store`, () => { + let harness: ReplicaRowStoreHarness; + let store: ReplicaRowStore; + + beforeEach(async () => { + harness = await createHarness(); + store = harness.store; + await store.open(); + }); + + it("reads an empty store", async () => { + expect(await store.readAll()).toEqual([]); + }); + + it("round-trips opaque payloads grouped by server", async () => { + await store.apply({ + upserts: [ + row("server-b", "checkpoint", "singleton", '{"cursor":2}'), + row("server-a", "agent", "agent-1", '{"status":"running"}'), + row("server-a", "timeline", "singleton", "not parsed by the row store"), + ], + deletes: [], + }); + + expect(await store.readAll()).toEqual([ + { + serverId: "server-a", + rows: [ + row("server-a", "agent", "agent-1", '{"status":"running"}'), + row("server-a", "timeline", "singleton", "not parsed by the row store"), + ], + }, + { + serverId: "server-b", + rows: [row("server-b", "checkpoint", "singleton", '{"cursor":2}')], + }, + ]); + }); + + it("overwrites an existing row on upsert", async () => { + await store.apply({ + upserts: [row("server-a", "agent", "agent-1", "old")], + deletes: [], + }); + await store.apply({ + upserts: [row("server-a", "agent", "agent-1", "new")], + deletes: [], + }); + + expect(await store.readAll()).toEqual([ + { + serverId: "server-a", + rows: [row("server-a", "agent", "agent-1", "new")], + }, + ]); + }); + + it("deletes one row by its compound identity", async () => { + await store.apply({ + upserts: [ + row("server-a", "agent", "same-id", "agent"), + row("server-a", "workspace", "same-id", "workspace"), + ], + deletes: [], + }); + await store.apply({ + upserts: [], + deletes: [{ serverId: "server-a", kind: "agent", id: "same-id" }], + }); + + expect(await store.readAll()).toEqual([ + { + serverId: "server-a", + rows: [row("server-a", "workspace", "same-id", "workspace")], + }, + ]); + }); + + it("rolls back the entire apply when one write fails", async () => { + const invalidRow = { + ...row("server-a", "agent", "invalid", "invalid"), + serverId: undefined, + } as unknown as ReplicaRow; + + await expect( + store.apply({ + upserts: [row("server-a", "agent", "would-have-been-written", "payload"), invalidRow], + deletes: [], + }), + ).rejects.toBeInstanceOf(Error); + expect(await store.readAll()).toEqual([]); + }); + + it("deletes every row belonging to one host", async () => { + await store.apply({ + upserts: [ + row("server-a", "agent", "agent-1", "a"), + row("server-a", "project", "project-1", "p"), + row("server-b", "agent", "agent-2", "b"), + ], + deletes: [], + }); + + await store.deleteHost("server-a"); + + expect(await store.readAll()).toEqual([ + { serverId: "server-b", rows: [row("server-b", "agent", "agent-2", "b")] }, + ]); + }); + + it("renames a host and lets moved rows replace target collisions", async () => { + await store.apply({ + upserts: [ + row("temporary", "agent", "agent-1", "temporary-agent"), + row("temporary", "project", "project-1", "temporary-project"), + row("canonical", "agent", "agent-1", "canonical-agent"), + row("canonical", "workspace", "workspace-1", "canonical-workspace"), + ], + deletes: [], + }); + + await store.renameHost("temporary", "canonical"); + + expect(await store.readAll()).toEqual([ + { + serverId: "canonical", + rows: [ + row("canonical", "agent", "agent-1", "temporary-agent"), + row("canonical", "project", "project-1", "temporary-project"), + row("canonical", "workspace", "workspace-1", "canonical-workspace"), + ], + }, + ]); + }); + + it("clears all cached rows", async () => { + await store.apply({ + upserts: [ + row("server-a", "agent", "agent-1", "a"), + row("server-b", "agent", "agent-2", "b"), + ], + deletes: [], + }); + + await store.clear(); + + expect(await store.readAll()).toEqual([]); + }); + + it("wipes cached rows when the schema version changes on open", async () => { + await store.apply({ + upserts: [row("server-a", "agent", "agent-1", "stale")], + deletes: [], + }); + + const upgradedStore = await harness.openWithSchemaVersion(2); + + expect(await upgradedStore.readAll()).toEqual([]); + }); + }); +} diff --git a/packages/app/src/runtime/replica-cache/row-store.native.test.ts b/packages/app/src/runtime/replica-cache/row-store.native.test.ts new file mode 100644 index 0000000000..8041c05002 --- /dev/null +++ b/packages/app/src/runtime/replica-cache/row-store.native.test.ts @@ -0,0 +1,54 @@ +import { DatabaseSync, type SQLInputValue } from "node:sqlite"; +import { runReplicaRowStoreContract } from "./row-store.contract"; +import { + createSqliteReplicaRowStore, + type ReplicaSqliteConnection, + type ReplicaSqliteDriver, + type SqliteValue, +} from "./row-store-sqlite"; + +function createNodeSqliteDriver(): ReplicaSqliteDriver { + const database = new DatabaseSync(":memory:"); + + function connection(): ReplicaSqliteConnection { + return { + async exec(sql) { + database.exec(sql); + }, + async run(sql, params = []) { + database.prepare(sql).run(...(params as SQLInputValue[])); + }, + async all(sql: string, params: readonly SqliteValue[] = []) { + return database.prepare(sql).all(...(params as SQLInputValue[])) as Row[]; + }, + async transaction(operation) { + database.exec("BEGIN IMMEDIATE"); + try { + await operation(connection()); + database.exec("COMMIT"); + } catch (error) { + database.exec("ROLLBACK"); + throw error; + } + }, + }; + } + + return { + async open() { + return connection(); + }, + }; +} + +runReplicaRowStoreContract("SQLite", async () => { + const driver = createNodeSqliteDriver(); + return { + store: createSqliteReplicaRowStore(driver, 1), + async openWithSchemaVersion(schemaVersion) { + const store = createSqliteReplicaRowStore(driver, schemaVersion); + await store.open(); + return store; + }, + }; +}); diff --git a/packages/app/src/runtime/replica-cache/row-store.native.ts b/packages/app/src/runtime/replica-cache/row-store.native.ts new file mode 100644 index 0000000000..bb91eb0aa6 --- /dev/null +++ b/packages/app/src/runtime/replica-cache/row-store.native.ts @@ -0,0 +1,49 @@ +import { openDatabaseAsync, type SQLiteDatabase } from "expo-sqlite"; +import type { ReplicaRowStore } from "./row-store"; +import { REPLICA_ROW_STORE_SCHEMA_VERSION } from "./row-store-schema"; +import { + createSqliteReplicaRowStore, + type ReplicaSqliteConnection, + type ReplicaSqliteDriver, + type SqliteValue, +} from "./row-store-sqlite"; + +export type { + ReplicaHostRows, + ReplicaRow, + ReplicaRowChanges, + ReplicaRowKey, + ReplicaRowKind, + ReplicaRowStore, +} from "./row-store"; +export { REPLICA_ROW_STORE_SCHEMA_VERSION, REPLICA_SINGLETON_ROW_ID } from "./row-store-schema"; + +const DATABASE_NAME = "paseo-replica-row-store.db"; + +function bind(params: readonly SqliteValue[]): SqliteValue[] { + return [...params]; +} + +function createConnection(database: SQLiteDatabase): ReplicaSqliteConnection { + return { + exec: (sql) => database.execAsync(sql), + run: async (sql, params = []) => { + await database.runAsync(sql, bind(params)); + }, + all: (sql, params = []) => database.getAllAsync(sql, bind(params)), + transaction: (operation) => + database.withExclusiveTransactionAsync(async (transaction) => { + await operation(createConnection(transaction)); + }), + }; +} + +const expoSqliteDriver: ReplicaSqliteDriver = { + async open() { + return createConnection(await openDatabaseAsync(DATABASE_NAME)); + }, +}; + +export function createReplicaRowStore(): ReplicaRowStore { + return createSqliteReplicaRowStore(expoSqliteDriver, REPLICA_ROW_STORE_SCHEMA_VERSION); +} diff --git a/packages/app/src/runtime/replica-cache/row-store.ts b/packages/app/src/runtime/replica-cache/row-store.ts new file mode 100644 index 0000000000..7600f0c368 --- /dev/null +++ b/packages/app/src/runtime/replica-cache/row-store.ts @@ -0,0 +1,32 @@ +export { REPLICA_ROW_STORE_SCHEMA_VERSION, REPLICA_SINGLETON_ROW_ID } from "./row-store-schema"; + +export type ReplicaRowKind = "agent" | "workspace" | "project" | "timeline" | "checkpoint"; + +export interface ReplicaRowKey { + serverId: string; + kind: ReplicaRowKind; + id: string; +} + +export interface ReplicaRow extends ReplicaRowKey { + payload: string; +} + +export interface ReplicaRowChanges { + upserts: ReplicaRow[]; + deletes: ReplicaRowKey[]; +} + +export interface ReplicaHostRows { + serverId: string; + rows: ReplicaRow[]; +} + +export interface ReplicaRowStore { + open(): Promise; + readAll(): Promise; + apply(changes: ReplicaRowChanges): Promise; + deleteHost(serverId: string): Promise; + renameHost(oldServerId: string, newServerId: string): Promise; + clear(): Promise; +} diff --git a/packages/app/src/runtime/replica-cache/row-store.web.test.ts b/packages/app/src/runtime/replica-cache/row-store.web.test.ts new file mode 100644 index 0000000000..ef91a322cb --- /dev/null +++ b/packages/app/src/runtime/replica-cache/row-store.web.test.ts @@ -0,0 +1,20 @@ +import { IDBKeyRange as FakeIDBKeyRange, indexedDB as fakeIndexedDb } from "fake-indexeddb"; +import { runReplicaRowStoreContract } from "./row-store.contract"; +import { createIndexedDbReplicaRowStore } from "./row-store.web"; + +globalThis.indexedDB = fakeIndexedDb; +globalThis.IDBKeyRange = FakeIDBKeyRange; + +let databaseSequence = 0; + +runReplicaRowStoreContract("IndexedDB", async () => { + const databaseName = `replica-row-store-test-${databaseSequence++}`; + return { + store: createIndexedDbReplicaRowStore({ databaseName, schemaVersion: 1 }), + async openWithSchemaVersion(schemaVersion) { + const store = createIndexedDbReplicaRowStore({ databaseName, schemaVersion }); + await store.open(); + return store; + }, + }; +}); diff --git a/packages/app/src/runtime/replica-cache/row-store.web.ts b/packages/app/src/runtime/replica-cache/row-store.web.ts new file mode 100644 index 0000000000..c740283706 --- /dev/null +++ b/packages/app/src/runtime/replica-cache/row-store.web.ts @@ -0,0 +1,186 @@ +import type { ReplicaHostRows, ReplicaRow, ReplicaRowChanges, ReplicaRowStore } from "./row-store"; +import { REPLICA_ROW_STORE_SCHEMA_VERSION } from "./row-store-schema"; + +export type { + ReplicaHostRows, + ReplicaRow, + ReplicaRowChanges, + ReplicaRowKey, + ReplicaRowKind, + ReplicaRowStore, +} from "./row-store"; +export { REPLICA_ROW_STORE_SCHEMA_VERSION, REPLICA_SINGLETON_ROW_ID } from "./row-store-schema"; + +const DATABASE_NAME = "paseo-replica-row-store"; +const DATABASE_VERSION = 1; +const ROWS_STORE = "rows"; +const META_STORE = "meta"; +const SCHEMA_VERSION_KEY = "schema_version"; + +interface IndexedDbReplicaRowStoreOptions { + databaseName: string; + schemaVersion: number; +} + +function requestResult(request: IDBRequest): Promise { + return new Promise((resolve, reject) => { + request.addEventListener("success", () => resolve(request.result)); + request.addEventListener("error", () => + reject(request.error ?? new Error("Replica row store request failed")), + ); + }); +} + +function transactionComplete(transaction: IDBTransaction): Promise { + return new Promise((resolve, reject) => { + transaction.addEventListener("complete", () => resolve()); + transaction.addEventListener("abort", () => + reject(transaction.error ?? new Error("Replica row store transaction was aborted")), + ); + transaction.addEventListener("error", () => + reject(transaction.error ?? new Error("Replica row store transaction failed")), + ); + }); +} + +async function runTransaction( + database: IDBDatabase, + storeNames: string | string[], + operation: (transaction: IDBTransaction) => Promise, +): Promise { + const transaction = database.transaction(storeNames, "readwrite"); + const completion = transactionComplete(transaction); + try { + await operation(transaction); + await completion; + } catch (error) { + try { + transaction.abort(); + } catch { + // The transaction already aborted after the failed request. + } + await completion.catch(() => undefined); + throw error; + } +} + +async function openDatabase(databaseName: string): Promise { + return new Promise((resolve, reject) => { + const request = indexedDB.open(databaseName, DATABASE_VERSION); + request.addEventListener("upgradeneeded", () => { + if (!request.result.objectStoreNames.contains(ROWS_STORE)) { + request.result.createObjectStore(ROWS_STORE, { keyPath: ["serverId", "kind", "id"] }); + } + if (!request.result.objectStoreNames.contains(META_STORE)) { + request.result.createObjectStore(META_STORE); + } + }); + request.addEventListener("success", () => resolve(request.result)); + request.addEventListener("error", () => + reject(request.error ?? new Error("Failed to open replica row store")), + ); + request.addEventListener("blocked", () => + reject(new Error("Replica row store database open was blocked")), + ); + }); +} + +/** @package Test seam for exercising the browser engine with an isolated database. */ +export function createIndexedDbReplicaRowStore( + options: IndexedDbReplicaRowStoreOptions, +): ReplicaRowStore { + let database: IDBDatabase | null = null; + let opening: Promise | null = null; + + function getDatabase(): IDBDatabase { + if (!database) throw new Error("Replica row store has not been opened"); + return database; + } + + async function open(): Promise { + opening ??= (async () => { + const openedDatabase = await openDatabase(options.databaseName); + await runTransaction(openedDatabase, [ROWS_STORE, META_STORE], async (transaction) => { + const storedVersion = await requestResult( + transaction.objectStore(META_STORE).get(SCHEMA_VERSION_KEY), + ); + if (storedVersion !== options.schemaVersion) { + await requestResult(transaction.objectStore(ROWS_STORE).clear()); + await requestResult( + transaction.objectStore(META_STORE).put(options.schemaVersion, SCHEMA_VERSION_KEY), + ); + } + }); + database = openedDatabase; + })(); + + try { + await opening; + } catch (error) { + opening = null; + throw error; + } + } + + async function readAll(): Promise { + const transaction = getDatabase().transaction(ROWS_STORE, "readonly"); + const completion = transactionComplete(transaction); + const rows = await requestResult(transaction.objectStore(ROWS_STORE).getAll()); + await completion; + const hosts = new Map(); + for (const row of rows) { + const hostRows = hosts.get(row.serverId) ?? []; + hostRows.push(row); + hosts.set(row.serverId, hostRows); + } + return Array.from(hosts, ([serverId, hostRows]) => ({ serverId, rows: hostRows })); + } + + async function apply(changes: ReplicaRowChanges): Promise { + await runTransaction(getDatabase(), ROWS_STORE, async (transaction) => { + const rows = transaction.objectStore(ROWS_STORE); + for (const key of changes.deletes) { + await requestResult(rows.delete([key.serverId, key.kind, key.id])); + } + for (const row of changes.upserts) { + await requestResult(rows.put(row)); + } + }); + } + + async function deleteHost(serverId: string): Promise { + await runTransaction(getDatabase(), ROWS_STORE, async (transaction) => { + const rows = transaction.objectStore(ROWS_STORE); + const range = IDBKeyRange.bound([serverId], [serverId, []]); + await requestResult(rows.delete(range)); + }); + } + + async function renameHost(oldServerId: string, newServerId: string): Promise { + if (oldServerId === newServerId) return; + await runTransaction(getDatabase(), ROWS_STORE, async (transaction) => { + const rows = transaction.objectStore(ROWS_STORE); + const range = IDBKeyRange.bound([oldServerId], [oldServerId, []]); + const oldRows = await requestResult(rows.getAll(range)); + for (const row of oldRows) { + await requestResult(rows.put({ ...row, serverId: newServerId })); + await requestResult(rows.delete([row.serverId, row.kind, row.id])); + } + }); + } + + async function clear(): Promise { + await runTransaction(getDatabase(), ROWS_STORE, async (transaction) => { + await requestResult(transaction.objectStore(ROWS_STORE).clear()); + }); + } + + return { open, readAll, apply, deleteHost, renameHost, clear }; +} + +export function createReplicaRowStore(): ReplicaRowStore { + return createIndexedDbReplicaRowStore({ + databaseName: DATABASE_NAME, + schemaVersion: REPLICA_ROW_STORE_SCHEMA_VERSION, + }); +}