mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-28 04:03:05 +00:00
feat(app): add replica cache row store
Ship the platform storage engines behind a dark interface so the cache migration can wire them separately. Both engines share a contract, including atomic writes and destructive schema-version resets.
This commit is contained in:
Generated
+43
@@ -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",
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
export const REPLICA_ROW_STORE_SCHEMA_VERSION = 1;
|
||||
|
||||
export const REPLICA_SINGLETON_ROW_ID = "singleton";
|
||||
@@ -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<void>;
|
||||
run(sql: string, params?: readonly SqliteValue[]): Promise<void>;
|
||||
all<Row>(sql: string, params?: readonly SqliteValue[]): Promise<Row[]>;
|
||||
transaction(operation: (connection: ReplicaSqliteConnection) => Promise<void>): Promise<void>;
|
||||
}
|
||||
|
||||
export interface ReplicaSqliteDriver {
|
||||
open(): Promise<ReplicaSqliteConnection>;
|
||||
}
|
||||
|
||||
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<void> | null = null;
|
||||
|
||||
function getConnection(): ReplicaSqliteConnection {
|
||||
if (!connection) {
|
||||
throw new Error("Replica row store has not been opened");
|
||||
}
|
||||
return connection;
|
||||
}
|
||||
|
||||
async function open(): Promise<void> {
|
||||
opening ??= (async () => {
|
||||
const openedConnection = await driver.open();
|
||||
await openedConnection.exec(CREATE_META_SQL);
|
||||
const storedMeta = await openedConnection.all<StoredMeta>(
|
||||
"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<ReplicaHostRows[]> {
|
||||
const storedRows = await getConnection().all<StoredRow>(
|
||||
"SELECT server_id, kind, id, payload FROM rows ORDER BY server_id, kind, id",
|
||||
);
|
||||
const hosts = new Map<string, ReplicaRow[]>();
|
||||
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<void> {
|
||||
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<void> {
|
||||
await getConnection().run("DELETE FROM rows WHERE server_id = ?", [serverId]);
|
||||
}
|
||||
|
||||
async function renameHost(oldServerId: string, newServerId: string): Promise<void> {
|
||||
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<void> {
|
||||
await getConnection().run("DELETE FROM rows");
|
||||
}
|
||||
|
||||
return { open, readAll, apply, deleteHost, renameHost, clear };
|
||||
}
|
||||
@@ -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<ReplicaRowStore>;
|
||||
}
|
||||
|
||||
export type CreateReplicaRowStoreHarness = () => Promise<ReplicaRowStoreHarness>;
|
||||
|
||||
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([]);
|
||||
});
|
||||
});
|
||||
}
|
||||
@@ -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<Row>(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;
|
||||
},
|
||||
};
|
||||
});
|
||||
@@ -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);
|
||||
}
|
||||
@@ -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<void>;
|
||||
readAll(): Promise<ReplicaHostRows[]>;
|
||||
apply(changes: ReplicaRowChanges): Promise<void>;
|
||||
deleteHost(serverId: string): Promise<void>;
|
||||
renameHost(oldServerId: string, newServerId: string): Promise<void>;
|
||||
clear(): Promise<void>;
|
||||
}
|
||||
@@ -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;
|
||||
},
|
||||
};
|
||||
});
|
||||
@@ -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<Result>(request: IDBRequest<Result>): Promise<Result> {
|
||||
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<void> {
|
||||
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<void>,
|
||||
): Promise<void> {
|
||||
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<IDBDatabase> {
|
||||
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<void> | null = null;
|
||||
|
||||
function getDatabase(): IDBDatabase {
|
||||
if (!database) throw new Error("Replica row store has not been opened");
|
||||
return database;
|
||||
}
|
||||
|
||||
async function open(): Promise<void> {
|
||||
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<ReplicaHostRows[]> {
|
||||
const transaction = getDatabase().transaction(ROWS_STORE, "readonly");
|
||||
const completion = transactionComplete(transaction);
|
||||
const rows = await requestResult<ReplicaRow[]>(transaction.objectStore(ROWS_STORE).getAll());
|
||||
await completion;
|
||||
const hosts = new Map<string, ReplicaRow[]>();
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<ReplicaRow[]>(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<void> {
|
||||
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,
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user