From 5d195e7e1511085736d9f3b9c6783be6beb6f437 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 9 Aug 2026 01:32:35 +0200 Subject: [PATCH] feat(storage): add SQLite persistence foundation Begin the storage migration with a bootstrap-owned database and one-way legacy imports for schedules and push tokens. Legacy JSON remains untouched for downgrade compatibility while SQLite becomes the sole authority after first open. --- docs/data-model.md | 41 ++- package-lock.json | 18 +- packages/cli/package.json | 3 + packages/server/package.json | 6 +- packages/server/scripts/db-query.ts | 20 ++ packages/server/src/server/bootstrap.ts | 11 + .../server/src/server/db/legacy-imports.ts | 26 ++ packages/server/src/server/db/migrations.ts | 69 +++++ packages/server/src/server/db/open.test.ts | 72 ++++++ packages/server/src/server/db/open.ts | 70 +++++ packages/server/src/server/db/transaction.ts | 13 + packages/server/src/server/push/index.test.ts | 34 ++- packages/server/src/server/push/index.ts | 17 +- .../src/server/push/token-store.test.ts | 190 ++++++++++---- .../server/src/server/push/token-store.ts | 243 ++++++++++-------- .../src/server/schedule/service.test.ts | 30 ++- .../server/src/server/schedule/service.ts | 8 +- .../server/src/server/schedule/store.test.ts | 137 +++++++++- packages/server/src/server/schedule/store.ts | 154 ++++++++--- .../websocket-server.browser-tools.test.ts | 2 + .../websocket-server.notifications.test.ts | 2 + .../websocket-server.relay-reconnect.test.ts | 2 + ...cket-server.terminal-notifications.test.ts | 2 + .../server/src/server/websocket-server.ts | 5 +- 24 files changed, 936 insertions(+), 239 deletions(-) create mode 100644 packages/server/scripts/db-query.ts create mode 100644 packages/server/src/server/db/legacy-imports.ts create mode 100644 packages/server/src/server/db/migrations.ts create mode 100644 packages/server/src/server/db/open.test.ts create mode 100644 packages/server/src/server/db/open.ts create mode 100644 packages/server/src/server/db/transaction.ts diff --git a/docs/data-model.md b/docs/data-model.md index bc1d8bd17b..89c48e3e10 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -30,7 +30,20 @@ Workspace archive runs lifecycle teardown from the exact `cwd` but removes only `worktreeRoot` after its last active reference disappears. Worktree recovery recreates that backing checkout from `mainRepoRoot`, then restores the relative path from `worktreeRoot` to `cwd`. -Paseo uses **file-based JSON persistence** instead of a traditional database. All data is validated at runtime with Zod schemas. Most stores write atomically (write to temp file, then rename); a few still use plain `writeFile` — see each section. There is no schema-versioning/migration framework — schemas rely on optional fields with defaults for forward compatibility, with a small amount of inline normalization in `persisted-config.ts` for legacy provider/speech entries. +Paseo stores schedules and push tokens in `$PASEO_HOME/paseo.db`. The remaining record stores use +JSON files. SQLite migrations are ordered SQL scripts tracked by `PRAGMA user_version`. Record +tables keep identity and queried fields in columns and the complete record in a JSON `payload` +column. Reads validate that payload with the same Zod schema used at the store boundary. + +Each SQLite-backed store owns a one-way import from its legacy JSON path. On first open, an empty +table imports valid legacy records in one transaction and writes a marker to `legacy_imports`. +Invalid records are skipped and logged. Once the marker exists, SQLite is the only authority for +that store: the daemon does not read from or write to the legacy path. Legacy files remain untouched +so downgrading resumes from the state captured when the import ran; there is no automatic +down-migration. + +JSON stores rely on optional fields with defaults for forward compatibility. Most write atomically +with a temporary file and rename; see each section for exceptions. All server-side stores live under `$PASEO_HOME` (defaults to `~/.paseo`). @@ -49,11 +62,12 @@ $PASEO_HOME/ ├── daemon-keypair.json # E2EE keypair for relay (mode 0600) ├── paseo.pid # Daemon PID lock file ├── daemon.log # Default log file (path configurable) +├── paseo.db # SQLite schedules, push tokens, and import markers ├── agents/ │ └── {sanitized-cwd}/ │ └── {agentId}.json # One file per agent -├── schedules/ -│ └── {scheduleId}.json # One file per schedule +├── schedules/ # Legacy schedule import source, left untouched +│ └── {scheduleId}.json ├── projects/ │ ├── projects.json # Project registry │ ├── workspaces.json # Workspace registry @@ -61,10 +75,10 @@ $PASEO_HOME/ ├── runtime/ │ └── managed-processes/ │ └── {recordId}.json # Helper processes owned by Paseo; reconciled on daemon bootstrap -└── push-tokens.json # Expo push notification tokens +└── push-tokens.json # Legacy push token import source, left untouched ``` -The `agents/{sanitized-cwd}/` directory name is derived from the agent's `cwd` by stripping the filesystem root and replacing path separators with `-` (Windows drive letters become a `C-` style prefix). Persistent server stores write atomically by writing a temp file in the target directory and then renaming it into place. +The `agents/{sanitized-cwd}/` directory name is derived from the agent's `cwd` by stripping the filesystem root and replacing path separators with `-` (Windows drive letters become a `C-` style prefix). File-backed record stores write atomically by writing a temp file in the target directory and then renaming it into place. --- @@ -325,9 +339,11 @@ Paseo uses these paths under the configured OpenAI base URL: ## 3. Schedule -**Path:** `$PASEO_HOME/schedules/{id}.json` +**Table:** `paseo.db:schedules` -One file per schedule. ID is 8 hex characters. +Each row stores `id`, `created_at`, the active name-and-target `identity_key`, and the validated +schedule JSON in `payload`. ID is 8 hex characters. The legacy +`$PASEO_HOME/schedules/{id}.json` files are import sources only. | Field | Type | Description | | ----------- | ------------------------------------- | -------------------------------- | @@ -442,15 +458,20 @@ than treating it as valid. ## 6. Push Token Store -**Path:** `$PASEO_HOME/push-tokens.json` +**Table:** `paseo.db:push_tokens` + +Legacy import source: ```json { - "tokens": ["ExponentPushToken[...]", ...] + "subscriptions": [{ "token": "ExponentPushToken[...]", "expiresAt": "2026-08-12T00:00:00.000Z" }] } ``` -Simple set of Expo push notification tokens. Loaded with permissive parsing (filters non-string entries). Persisted with atomic temp-file rename. +The older `{ "tokens": [...] }` shape is imported with a fresh lease. Each token is a row keyed by +its normalized string, with the validated subscription record in `payload`. +`$PASEO_HOME/push-tokens.json` is read once as an import source. Non-string and empty legacy entries +are skipped and logged. --- diff --git a/package-lock.json b/package-lock.json index f6a403bfed..fee98653ff 100644 --- a/package-lock.json +++ b/package-lock.json @@ -37163,6 +37163,9 @@ "typescript": "^5.2.2", "vitest": "^4.1.6", "zx": "^8.8.5" + }, + "engines": { + "node": ">=22.20.0" } }, "packages/cli/node_modules/@vitest/expect": { @@ -38845,7 +38848,7 @@ "devDependencies": { "@playwright/test": "^1.56.1", "@types/express": "^4.17.20", - "@types/node": "^20.9.0", + "@types/node": "^22.20.1", "@types/qrcode": "^1.5.6", "@types/uuid": "^9.0.7", "@types/ws": "^8.5.8", @@ -38854,6 +38857,9 @@ "tsx": "^4.6.0", "typescript": "^5.2.2", "vitest": "^4.1.6" + }, + "engines": { + "node": ">=22.20.0" } }, "packages/server/node_modules/@anthropic-ai/claude-agent-sdk": { @@ -39002,6 +39008,16 @@ "cross-spawn": "7.0.6" } }, + "packages/server/node_modules/@types/node": { + "version": "22.20.1", + "resolved": "https://registry.npmjs.org/@types/node/-/node-22.20.1.tgz", + "integrity": "sha512-EANqOCF9QFyra+4pfxUcX9STKJpCLjMbObVzljIJomAWSnuSIEAvyzEU53GaajbXJEgdh0iEcPL+DGvpUd4k1Q==", + "dev": true, + "license": "MIT", + "dependencies": { + "undici-types": "~6.21.0" + } + }, "packages/server/node_modules/@vitest/expect": { "version": "4.1.6", "resolved": "https://registry.npmjs.org/@vitest/expect/-/expect-4.1.6.tgz", diff --git a/packages/cli/package.json b/packages/cli/package.json index 3e4bea361a..59bb32c157 100644 --- a/packages/cli/package.json +++ b/packages/cli/package.json @@ -46,5 +46,8 @@ "typescript": "^5.2.2", "vitest": "^4.1.6", "zx": "^8.8.5" + }, + "engines": { + "node": ">=22.20.0" } } diff --git a/packages/server/package.json b/packages/server/package.json index 3fc98085c5..0f734865f1 100644 --- a/packages/server/package.json +++ b/packages/server/package.json @@ -51,6 +51,7 @@ "speech:models": "tsx scripts/list-speech-models.ts", "speech:download": "tsx scripts/download-speech-models.ts", "speech:transcribe:local": "tsx scripts/transcribe-local-wav.ts", + "db:query": "../../scripts/dev-home.sh tsx scripts/db-query.ts", "test": "npm run test:unit && npm run test:integration", "test:unit": "vitest run --fileParallelism --exclude \"**/*.e2e.test.ts\"", "test:integration": "vitest run --maxWorkers=1 src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts && vitest run --maxWorkers=1 src/server/session.create-agent-worktree-autoarchive.e2e.test.ts -t \"creates a worktree and auto-archives both\"", @@ -107,7 +108,7 @@ "devDependencies": { "@playwright/test": "^1.56.1", "@types/express": "^4.17.20", - "@types/node": "^20.9.0", + "@types/node": "^22.20.1", "@types/qrcode": "^1.5.6", "@types/uuid": "^9.0.7", "@types/ws": "^8.5.8", @@ -119,5 +120,8 @@ }, "overrides": { "lightningcss": "1.30.1" + }, + "engines": { + "node": ">=22.20.0" } } diff --git a/packages/server/scripts/db-query.ts b/packages/server/scripts/db-query.ts new file mode 100644 index 0000000000..91d3a00f7c --- /dev/null +++ b/packages/server/scripts/db-query.ts @@ -0,0 +1,20 @@ +import { resolvePaseoHome } from "../src/server/paseo-home.js"; +import { openDatabase } from "../src/server/db/open.js"; + +const sql = process.argv.slice(2).join(" ").trim(); +if (!sql) { + throw new Error('Usage: npm run db:query -- "SELECT ..."'); +} + +const database = openDatabase(resolvePaseoHome()); +try { + const statement = database.prepare(sql); + if (statement.columns().length > 0) { + console.table(statement.all()); + } else { + const result = statement.run(); + console.log(`changes: ${result.changes}`); + } +} finally { + database.close(); +} diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index e270216744..30c8d26e3c 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -216,6 +216,8 @@ import { type HubRelationshipRemote, } from "./hub/relationship-remote.js"; import { DaemonExecutions } from "./hub/daemon-executions.js"; +import { openDatabase } from "./db/open.js"; +import { PushTokenStore } from "./push/token-store.js"; const MAX_MCP_DEBUG_BATCH_ITEMS = 10; const REDACTED_LOG_VALUE = "[redacted]"; @@ -552,6 +554,7 @@ export async function createPaseoDaemon( ): Promise { configureGitProcessPolicy(config.git ?? resolveGitProcessPolicy({ env: process.env })); const logger = rootLogger.child({ module: "bootstrap" }); + const database = openDatabase(config.paseoHome); const bootstrapStart = performance.now(); const elapsed = () => `${(performance.now() - bootstrapStart).toFixed(0)}ms`; const daemonVersion = config.daemonVersion ?? resolveDaemonVersion(import.meta.url); @@ -1205,6 +1208,7 @@ export async function createPaseoDaemon( }; const scheduleService = new ScheduleService({ paseoHome: config.paseoHome, + database, logger, agentManager, agentStorage, @@ -1222,6 +1226,11 @@ export async function createPaseoDaemon( } }); logger.info({ elapsed: elapsed() }, "Schedule service initialized"); + const pushTokenStore = new PushTokenStore({ + database, + legacyFilePath: path.join(config.paseoHome, "push-tokens.json"), + logger, + }); logger.info({ elapsed: elapsed() }, "Loading persisted agent registry"); const persistedRecords = await agentStorage.list(); logger.info( @@ -1504,6 +1513,7 @@ export async function createPaseoDaemon( agentManager, agentStorage, downloadTokenStore, + pushTokenStore, config.paseoHome, daemonConfigStore, mcpBaseUrl, @@ -1647,6 +1657,7 @@ export async function createPaseoDaemon( if (listenTarget.type === "socket" && existsSync(listenTarget.path)) { unlinkSync(listenTarget.path); } + database.close(); }; return { diff --git a/packages/server/src/server/db/legacy-imports.ts b/packages/server/src/server/db/legacy-imports.ts new file mode 100644 index 0000000000..1022c6a02b --- /dev/null +++ b/packages/server/src/server/db/legacy-imports.ts @@ -0,0 +1,26 @@ +import type { DatabaseSync } from "node:sqlite"; + +export interface LegacyImportResult { + importedCount: number; + skippedCount: number; +} + +export function hasLegacyImportMarker(database: DatabaseSync, store: string): boolean { + const row = database.prepare("SELECT 1 FROM legacy_imports WHERE store = ?").get(store) as + | { 1: number } + | undefined; + return row !== undefined; +} + +export function recordLegacyImportMarker( + database: DatabaseSync, + store: string, + result: LegacyImportResult, +): void { + database + .prepare( + `INSERT INTO legacy_imports (store, imported_at, imported_count, skipped_count) + VALUES (?, ?, ?, ?)`, + ) + .run(store, new Date().toISOString(), result.importedCount, result.skippedCount); +} diff --git a/packages/server/src/server/db/migrations.ts b/packages/server/src/server/db/migrations.ts new file mode 100644 index 0000000000..77a9e910f4 --- /dev/null +++ b/packages/server/src/server/db/migrations.ts @@ -0,0 +1,69 @@ +import type { DatabaseSync } from "node:sqlite"; +import { runInTransaction } from "./transaction.js"; + +interface Migration { + version: number; + sql: string; +} + +const MIGRATIONS: readonly Migration[] = [ + { + version: 1, + sql: ` + CREATE TABLE legacy_imports ( + store TEXT PRIMARY KEY, + imported_at TEXT NOT NULL, + imported_count INTEGER NOT NULL, + skipped_count INTEGER NOT NULL + ) STRICT; + `, + }, + { + version: 2, + sql: ` + CREATE TABLE schedules ( + id TEXT PRIMARY KEY, + created_at TEXT NOT NULL, + identity_key TEXT, + payload TEXT NOT NULL + ) STRICT; + + CREATE INDEX schedules_created_at_idx ON schedules(created_at); + CREATE INDEX schedules_identity_key_idx ON schedules(identity_key); + `, + }, + { + version: 3, + sql: ` + CREATE TABLE push_tokens ( + token TEXT PRIMARY KEY, + payload TEXT NOT NULL + ) STRICT; + `, + }, +]; + +function readUserVersion(database: DatabaseSync): number { + const row = database.prepare("PRAGMA user_version").get() as { user_version: number }; + return row.user_version; +} + +export function runMigrations(database: DatabaseSync): void { + const currentVersion = readUserVersion(database); + const latestVersion = MIGRATIONS.at(-1)?.version ?? 0; + if (currentVersion > latestVersion) { + throw new Error( + `paseo.db schema version ${currentVersion} is newer than this daemon supports (${latestVersion})`, + ); + } + + for (const migration of MIGRATIONS) { + if (migration.version <= currentVersion) { + continue; + } + runInTransaction(database, () => { + database.exec(migration.sql); + database.exec(`PRAGMA user_version = ${migration.version}`); + }); + } +} diff --git a/packages/server/src/server/db/open.test.ts b/packages/server/src/server/db/open.test.ts new file mode 100644 index 0000000000..5cc28f74eb --- /dev/null +++ b/packages/server/src/server/db/open.test.ts @@ -0,0 +1,72 @@ +import { existsSync } from "node:fs"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, test } from "vitest"; +import { assertSupportedNodeVersion, openDatabase } from "./open.js"; +import { runInTransaction } from "./transaction.js"; + +const homes: string[] = []; + +afterEach(async () => { + await Promise.all(homes.splice(0).map((home) => rm(home, { recursive: true, force: true }))); +}); + +async function createHome(): Promise { + const home = await mkdtemp(join(tmpdir(), "paseo-db-test-")); + homes.push(home); + return home; +} + +function insertThenFail(database: ReturnType): void { + runInTransaction(database, () => { + database + .prepare("INSERT INTO push_tokens (token, payload) VALUES (?, ?)") + .run("token", JSON.stringify({ token: "token" })); + throw new Error("stop"); + }); +} + +describe("openDatabase", () => { + test("creates and configures the migrated database", async () => { + const home = await createHome(); + const database = openDatabase(home); + try { + expect(existsSync(join(home, "paseo.db"))).toBe(true); + expect(database.prepare("PRAGMA journal_mode").get()).toMatchObject({ journal_mode: "wal" }); + expect(database.prepare("PRAGMA foreign_keys").get()).toMatchObject({ foreign_keys: 1 }); + expect(database.prepare("PRAGMA busy_timeout").get()).toMatchObject({ timeout: 5_000 }); + expect(database.prepare("PRAGMA user_version").get()).toMatchObject({ user_version: 3 }); + expect( + database.prepare("SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name").all(), + ).toEqual( + expect.arrayContaining([ + expect.objectContaining({ name: "legacy_imports" }), + expect.objectContaining({ name: "push_tokens" }), + expect.objectContaining({ name: "schedules" }), + ]), + ); + } finally { + database.close(); + } + }); + + test("rejects runtimes older than Node 22.20", () => { + expect(() => assertSupportedNodeVersion("22.19.9")).toThrow("Paseo requires Node.js >=22.20.0"); + expect(() => assertSupportedNodeVersion("22.20.0")).not.toThrow(); + expect(() => assertSupportedNodeVersion("23.0.0")).not.toThrow(); + }); + + test("rolls back failed transactions", async () => { + const home = await createHome(); + const database = openDatabase(home); + try { + expect(() => insertThenFail(database)).toThrow("stop"); + expect(database.prepare("SELECT count(*) AS count FROM push_tokens").get()).toMatchObject({ + count: 0, + }); + } finally { + database.close(); + } + }); +}); diff --git a/packages/server/src/server/db/open.ts b/packages/server/src/server/db/open.ts new file mode 100644 index 0000000000..39a59623af --- /dev/null +++ b/packages/server/src/server/db/open.ts @@ -0,0 +1,70 @@ +import { chmodSync, mkdirSync } from "node:fs"; +import { join } from "node:path"; +// node:sqlite emits an ExperimentalWarning on Node 22. We deliberately accept it instead of +// changing process-wide warning behavior; the startup smoke checks below pin the features we use. +import { DatabaseSync } from "node:sqlite"; +import { PRIVATE_FILE_MODE } from "../private-files.js"; +import { runMigrations } from "./migrations.js"; + +const MINIMUM_NODE_VERSION = [22, 20, 0] as const; +const BUSY_TIMEOUT_MS = 5_000; + +function parseNodeVersion(version: string): readonly number[] { + const match = /^(\d+)\.(\d+)\.(\d+)/.exec(version); + if (!match) { + throw new Error(`Unable to determine Node.js version from ${JSON.stringify(version)}`); + } + return [Number(match[1]), Number(match[2]), Number(match[3])]; +} + +export function assertSupportedNodeVersion(version = process.versions.node): void { + const actual = parseNodeVersion(version); + for (let index = 0; index < MINIMUM_NODE_VERSION.length; index += 1) { + if (actual[index] > MINIMUM_NODE_VERSION[index]) { + return; + } + if (actual[index] < MINIMUM_NODE_VERSION[index]) { + throw new Error(`Paseo requires Node.js >=22.20.0 for SQLite support; running ${version}`); + } + } +} + +function readJournalMode(database: DatabaseSync): string { + const row = database.prepare("PRAGMA journal_mode").get() as { journal_mode: string }; + return row.journal_mode.toLowerCase(); +} + +function verifyFts5(database: DatabaseSync): void { + try { + database.exec("CREATE VIRTUAL TABLE temp.paseo_fts5_smoke USING fts5(content)"); + database.exec("DROP TABLE temp.paseo_fts5_smoke"); + } catch (error) { + throw new Error("This Node.js runtime does not provide the required SQLite FTS5 support", { + cause: error, + }); + } +} + +export function openDatabase(paseoHome: string): DatabaseSync { + assertSupportedNodeVersion(); + mkdirSync(paseoHome, { recursive: true }); + const databasePath = join(paseoHome, "paseo.db"); + const database = new DatabaseSync(databasePath); + try { + chmodSync(databasePath, PRIVATE_FILE_MODE); + database.exec(` + PRAGMA journal_mode = WAL; + PRAGMA foreign_keys = ON; + PRAGMA busy_timeout = ${BUSY_TIMEOUT_MS}; + `); + if (readJournalMode(database) !== "wal") { + throw new Error("paseo.db could not enable required WAL journal mode"); + } + verifyFts5(database); + runMigrations(database); + return database; + } catch (error) { + database.close(); + throw new Error(`Failed to initialize SQLite database at ${databasePath}`, { cause: error }); + } +} diff --git a/packages/server/src/server/db/transaction.ts b/packages/server/src/server/db/transaction.ts new file mode 100644 index 0000000000..b5050a40bf --- /dev/null +++ b/packages/server/src/server/db/transaction.ts @@ -0,0 +1,13 @@ +import type { DatabaseSync } from "node:sqlite"; + +export function runInTransaction(database: DatabaseSync, operation: () => T): T { + database.exec("BEGIN IMMEDIATE"); + try { + const result = operation(); + database.exec("COMMIT"); + return result; + } catch (error) { + database.exec("ROLLBACK"); + throw error; + } +} diff --git a/packages/server/src/server/push/index.test.ts b/packages/server/src/server/push/index.test.ts index 1cd4c1baa6..0e4c4899c5 100644 --- a/packages/server/src/server/push/index.test.ts +++ b/packages/server/src/server/push/index.test.ts @@ -1,10 +1,13 @@ -import { mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; +import type { DatabaseSync } from "node:sqlite"; import type pino from "pino"; import { afterEach, describe, expect, test } from "vitest"; +import { openDatabase } from "../db/open.js"; import { createPushNotifications } from "./index.js"; +import { PushTokenStore } from "./token-store.js"; function createLogger(): pino.Logger { const logger = { @@ -19,8 +22,10 @@ function createLogger(): pino.Logger { describe("push notifications", () => { const homes: string[] = []; + const databases: DatabaseSync[] = []; afterEach(() => { + for (const database of databases.splice(0)) database.close(); for (const home of homes.splice(0)) { rmSync(home, { recursive: true, force: true }); } @@ -29,13 +34,19 @@ describe("push notifications", () => { test("an offline device stops receiving notifications after 48 hours", async () => { const home = mkdtempSync(path.join(tmpdir(), "paseo-push-notifications-")); homes.push(home); - const filePath = path.join(home, "push-tokens.json"); + const database = openDatabase(home); + databases.push(database); let now = Date.parse("2026-08-10T00:00:00.000Z"); const deliveries: string[][] = []; + const store = new PushTokenStore({ + database, + legacyFilePath: path.join(home, "push-tokens.json"), + logger: createLogger(), + now: () => now, + }); const pushNotifications = createPushNotifications({ logger: createLogger(), - filePath, - now: () => now, + store, deliver: async (tokens) => deliveries.push(tokens), }); @@ -44,17 +55,26 @@ describe("push notifications", () => { await pushNotifications.send({ title: "Agent finished", body: "Done" }); expect(deliveries).toEqual([]); - expect(JSON.parse(readFileSync(filePath, "utf8"))).toEqual({ subscriptions: [] }); + expect(database.prepare("SELECT count(*) AS count FROM push_tokens").get()).toMatchObject({ + count: 0, + }); }); test("online revocation stops notifications immediately", async () => { const home = mkdtempSync(path.join(tmpdir(), "paseo-push-notifications-")); homes.push(home); + const database = openDatabase(home); + databases.push(database); const deliveries: string[][] = []; + const store = new PushTokenStore({ + database, + legacyFilePath: path.join(home, "push-tokens.json"), + logger: createLogger(), + now: () => Date.parse("2026-08-10T00:00:00.000Z"), + }); const pushNotifications = createPushNotifications({ logger: createLogger(), - filePath: path.join(home, "push-tokens.json"), - now: () => Date.parse("2026-08-10T00:00:00.000Z"), + store, deliver: async (tokens) => deliveries.push(tokens), }); diff --git a/packages/server/src/server/push/index.ts b/packages/server/src/server/push/index.ts index 3f1a8e0900..a67b123b48 100644 --- a/packages/server/src/server/push/index.ts +++ b/packages/server/src/server/push/index.ts @@ -1,12 +1,10 @@ import type pino from "pino"; import { PushService, type PushPayload } from "./push-service.js"; -import { PushTokenStore } from "./token-store.js"; +import type { PushTokenStore } from "./token-store.js"; export type { PushPayload }; -const PUSH_TOKEN_LEASE_MS = 48 * 60 * 60 * 1000; - export interface PushNotifications { renew(token: string): void; revoke(token: string): void; @@ -17,26 +15,23 @@ export type PushNotificationSender = Pick; export function createPushNotifications(options: { logger: pino.Logger; - filePath: string; - now?: () => number; + store: PushTokenStore; deliver?: (tokens: string[], payload: PushPayload) => Promise; }): PushNotifications { - const now = options.now ?? Date.now; - const store = new PushTokenStore(options.logger, options.filePath, now, PUSH_TOKEN_LEASE_MS); - const service = new PushService(options.logger, (token) => store.revokeToken(token)); + const service = new PushService(options.logger, (token) => options.store.revokeToken(token)); const deliver = options.deliver ?? ((tokens: string[], payload: PushPayload) => service.sendPush(tokens, payload)); return { renew(token) { - store.renewToken(token); + options.store.renewToken(token); }, revoke(token) { - store.revokeToken(token); + options.store.revokeToken(token); }, async send(payload) { - const tokens = store.getActiveTokens(); + const tokens = options.store.getActiveTokens(); options.logger.info({ tokenCount: tokens.length }, "Sending push notification"); if (tokens.length === 0) return; await deliver(tokens, payload); diff --git a/packages/server/src/server/push/token-store.test.ts b/packages/server/src/server/push/token-store.test.ts index afd7d601c3..977ddb2f31 100644 --- a/packages/server/src/server/push/token-store.test.ts +++ b/packages/server/src/server/push/token-store.test.ts @@ -1,22 +1,22 @@ import { chmodSync, mkdtempSync, rmSync, statSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; +import type { DatabaseSync } from "node:sqlite"; import type pino from "pino"; -import { describe, expect, test } from "vitest"; - +import { describe, expect, test, vi } from "vitest"; +import { openDatabase } from "../db/open.js"; import { PRIVATE_FILE_MODE } from "../private-files.js"; -import { createPushNotifications } from "./index.js"; import { PushTokenStore } from "./token-store.js"; const MODE_MASK = 0o777; const PERMISSIVE_FILE_MODE = 0o644; -function createLogger(): pino.Logger { +function createLogger(warn: pino.Logger["warn"] = () => undefined): pino.Logger { const logger = { child: () => logger, debug: () => undefined, info: () => undefined, - warn: () => undefined, + warn, error: () => undefined, }; return logger as unknown as pino.Logger; @@ -26,42 +26,128 @@ function modeOf(filePath: string): number { return statSync(filePath).mode & MODE_MASK; } -describe.skipIf(process.platform === "win32")("PushTokenStore file permissions", () => { - test("persists push tokens with private permissions", () => { - const home = mkdtempSync(path.join(tmpdir(), "paseo-push-tokens-")); - const tokenPath = path.join(home, "push-tokens.json"); - try { - const pushNotifications = createPushNotifications({ - logger: createLogger(), - filePath: tokenPath, - }); +function createStore( + database: DatabaseSync, + tokenPath: string, + logger = createLogger(), +): PushTokenStore { + return new PushTokenStore({ database, legacyFilePath: tokenPath, logger }); +} - pushNotifications.renew("ExponentPushToken[test]"); +describe.skipIf(process.platform === "win32")("PushTokenStore file permissions", () => { + test("persists push tokens in a private database", () => { + const home = mkdtempSync(path.join(tmpdir(), "paseo-push-tokens-")); + const tokenPath = path.join(home, "paseo.db"); + const database = openDatabase(home); + try { + const store = createStore(database, path.join(home, "push-tokens.json")); + + store.renewToken("ExponentPushToken[test]"); expect(modeOf(tokenPath)).toBe(PRIVATE_FILE_MODE); } finally { + database.close(); rmSync(home, { recursive: true, force: true }); } }); - test("repairs existing push token file permissions when loading", async () => { + test("repairs existing push token file permissions when importing", () => { const home = mkdtempSync(path.join(tmpdir(), "paseo-push-tokens-")); const tokenPath = path.join(home, "push-tokens.json"); + writeFileSync(tokenPath, JSON.stringify({ tokens: ["ExponentPushToken[test]"] })); + chmodSync(tokenPath, PERMISSIVE_FILE_MODE); + const database = openDatabase(home); try { - writeFileSync(tokenPath, JSON.stringify({ tokens: ["ExponentPushToken[test]"] })); - chmodSync(tokenPath, PERMISSIVE_FILE_MODE); + const store = createStore(database, tokenPath); - const deliveries: string[][] = []; - const pushNotifications = createPushNotifications({ - logger: createLogger(), - filePath: tokenPath, - deliver: async (tokens) => deliveries.push(tokens), - }); - await pushNotifications.send({ title: "Agent finished", body: "Done" }); - - expect(deliveries).toEqual([["ExponentPushToken[test]"]]); + expect(store.getActiveTokens()).toEqual(["ExponentPushToken[test]"]); expect(modeOf(tokenPath)).toBe(PRIVATE_FILE_MODE); } finally { + database.close(); + rmSync(home, { recursive: true, force: true }); + } + }); +}); + +describe("PushTokenStore legacy import", () => { + test("imports once and records the marker", () => { + const home = mkdtempSync(path.join(tmpdir(), "paseo-push-tokens-")); + const tokenPath = path.join(home, "push-tokens.json"); + writeFileSync( + tokenPath, + JSON.stringify({ + subscriptions: [ + { + token: "ExponentPushToken[legacy]", + expiresAt: "2099-08-12T00:00:00.000Z", + }, + ], + }), + ); + const database = openDatabase(home); + try { + const store = createStore(database, tokenPath); + expect(store.getActiveTokens()).toEqual(["ExponentPushToken[legacy]"]); + store.revokeToken("ExponentPushToken[legacy]"); + + const reloaded = createStore(database, tokenPath); + + expect(reloaded.getActiveTokens()).toEqual([]); + expect( + database + .prepare("SELECT imported_count FROM legacy_imports WHERE store = ?") + .get("push_tokens"), + ).toMatchObject({ imported_count: 1 }); + } finally { + database.close(); + rmSync(home, { recursive: true, force: true }); + } + }); + + test("skips and logs invalid legacy tokens", () => { + const home = mkdtempSync(path.join(tmpdir(), "paseo-push-tokens-")); + const tokenPath = path.join(home, "push-tokens.json"); + writeFileSync(tokenPath, JSON.stringify({ tokens: ["ExponentPushToken[valid]", " ", 42] })); + const database = openDatabase(home); + const warn = vi.fn(); + try { + const store = createStore(database, tokenPath, createLogger(warn)); + + expect(store.getActiveTokens()).toEqual(["ExponentPushToken[valid]"]); + expect(warn).toHaveBeenCalledTimes(2); + expect( + database + .prepare("SELECT skipped_count FROM legacy_imports WHERE store = ?") + .get("push_tokens"), + ).toMatchObject({ skipped_count: 2 }); + } finally { + database.close(); + rmSync(home, { recursive: true, force: true }); + } + }); + + test("rolls back the whole legacy import when its marker cannot be written", () => { + const home = mkdtempSync(path.join(tmpdir(), "paseo-push-tokens-")); + const tokenPath = path.join(home, "push-tokens.json"); + writeFileSync(tokenPath, JSON.stringify({ tokens: ["ExponentPushToken[legacy]"] })); + const database = openDatabase(home); + database.exec(` + CREATE TRIGGER reject_legacy_import_marker + BEFORE INSERT ON legacy_imports + BEGIN + SELECT RAISE(ABORT, 'marker rejected'); + END; + `); + try { + expect(() => createStore(database, tokenPath)).toThrow("Failed to import legacy push tokens"); + expect(database.prepare("SELECT count(*) AS count FROM push_tokens").get()).toMatchObject({ + count: 0, + }); + expect(database.prepare("SELECT count(*) AS count FROM legacy_imports").get()).toMatchObject({ + count: 0, + }); + } finally { + database.close(); rmSync(home, { recursive: true, force: true }); } }); @@ -69,36 +155,28 @@ describe.skipIf(process.platform === "win32")("PushTokenStore file permissions", describe("PushTokenStore persistence", () => { test("does not apply a revocation until the updated subscriptions are durable", () => { - let rejectWrites = false; - let persisted = ""; - const store = new PushTokenStore( - createLogger(), - path.join(tmpdir(), `missing-push-token-store-${process.pid}`, "push-tokens.json"), - () => Date.parse("2026-08-10T00:00:00.000Z"), - 48 * 60 * 60 * 1000, - (_filePath, payload) => { - if (rejectWrites) throw new Error("disk full"); - persisted = String(payload); - }, - ); + const home = mkdtempSync(path.join(tmpdir(), "paseo-push-tokens-")); + const database = openDatabase(home); + try { + const store = createStore(database, path.join(home, "push-tokens.json")); + store.renewToken("ExponentPushToken[test]"); + database.exec(` + CREATE TRIGGER reject_push_token_delete + BEFORE DELETE ON push_tokens + BEGIN + SELECT RAISE(ABORT, 'delete rejected'); + END; + `); - store.renewToken("ExponentPushToken[test]"); - rejectWrites = true; + expect(() => store.revokeToken("ExponentPushToken[test]")).toThrow("delete rejected"); + expect(store.getActiveTokens()).toEqual(["ExponentPushToken[test]"]); - expect(() => store.revokeToken("ExponentPushToken[test]")).toThrow("disk full"); - expect(store.getActiveTokens()).toEqual(["ExponentPushToken[test]"]); - expect(JSON.parse(persisted)).toEqual({ - subscriptions: [ - { - token: "ExponentPushToken[test]", - expiresAt: "2026-08-12T00:00:00.000Z", - }, - ], - }); - - rejectWrites = false; - store.revokeToken("ExponentPushToken[test]"); - expect(store.getActiveTokens()).toEqual([]); - expect(JSON.parse(persisted)).toEqual({ subscriptions: [] }); + database.exec("DROP TRIGGER reject_push_token_delete"); + store.revokeToken("ExponentPushToken[test]"); + expect(store.getActiveTokens()).toEqual([]); + } finally { + database.close(); + rmSync(home, { recursive: true, force: true }); + } }); }); diff --git a/packages/server/src/server/push/token-store.ts b/packages/server/src/server/push/token-store.ts index 540f053138..6193405726 100644 --- a/packages/server/src/server/push/token-store.ts +++ b/packages/server/src/server/push/token-store.ts @@ -1,141 +1,178 @@ -import type pino from "pino"; import { existsSync, readFileSync } from "node:fs"; +import type { DatabaseSync } from "node:sqlite"; +import type pino from "pino"; +import { z } from "zod"; -import { ensurePrivateFile, writePrivateFileAtomicSync } from "../private-files.js"; +import { hasLegacyImportMarker, recordLegacyImportMarker } from "../db/legacy-imports.js"; +import { runInTransaction } from "../db/transaction.js"; +import { ensurePrivateFile } from "../private-files.js"; -/** - * Store for Expo push tokens. - * - * Tokens are persisted to disk so pushes still work after daemon restarts. - */ +const LEGACY_IMPORT_STORE = "push_tokens"; +const DEFAULT_LEASE_MS = 48 * 60 * 60 * 1000; +const PushTokenRecordSchema = z.object({ + token: z.string().trim().min(1), + expiresAt: z.string().refine((value) => Number.isFinite(Date.parse(value))), +}); + +interface PushTokenStoreOptions { + database: DatabaseSync; + legacyFilePath: string; + logger: pino.Logger; + now?: () => number; + leaseMs?: number; +} + +interface PushTokenRow { + payload: string; +} + +/** Store for leased Expo push-token subscriptions. */ export class PushTokenStore { + private readonly database: DatabaseSync; private readonly logger: pino.Logger; - private subscriptions = new Map(); - private readonly filePath: string; + private readonly legacyFilePath: string; private readonly now: () => number; private readonly leaseMs: number; - private readonly write: typeof writePrivateFileAtomicSync; - constructor( - logger: pino.Logger, - filePath: string, - now: () => number, - leaseMs: number, - write: typeof writePrivateFileAtomicSync = writePrivateFileAtomicSync, - ) { - this.logger = logger.child({ component: "token-store" }); - this.filePath = filePath; - this.now = now; - this.leaseMs = leaseMs; - this.write = write; - this.loadFromDisk(); + constructor(options: PushTokenStoreOptions) { + this.database = options.database; + this.logger = options.logger.child({ component: "token-store" }); + this.legacyFilePath = options.legacyFilePath; + this.now = options.now ?? Date.now; + this.leaseMs = options.leaseMs ?? DEFAULT_LEASE_MS; + this.importLegacyTokens(); } renewToken(token: string): void { const normalized = token.trim(); if (!normalized) return; + + const current = this.readToken(normalized); const now = this.now(); - const currentExpiry = this.subscriptions.get(normalized); - if (currentExpiry !== undefined && currentExpiry - now > this.leaseMs / 2) return; - const next = new Map(this.subscriptions); - next.set(normalized, now + this.leaseMs); - this.persist(next); - this.subscriptions = next; - this.logger.debug({ total: this.subscriptions.size }, "Renewed token"); + if (current && Date.parse(current.expiresAt) - now > this.leaseMs / 2) return; + + const record = { + token: normalized, + expiresAt: new Date(now + this.leaseMs).toISOString(), + }; + this.database + .prepare( + "INSERT INTO push_tokens (token, payload) VALUES (?, ?) " + + "ON CONFLICT(token) DO UPDATE SET payload = excluded.payload", + ) + .run(record.token, JSON.stringify(record)); + this.logger.debug({ total: this.count() }, "Renewed token"); } revokeToken(token: string): void { const normalized = token.trim(); if (!normalized) return; - if (!this.subscriptions.has(normalized)) return; - const next = new Map(this.subscriptions); - next.delete(normalized); - this.persist(next); - this.subscriptions = next; - this.logger.debug({ total: this.subscriptions.size }, "Revoked token"); + const result = this.database.prepare("DELETE FROM push_tokens WHERE token = ?").run(normalized); + if (result.changes > 0) { + this.logger.debug({ total: this.count() }, "Revoked token"); + } } getActiveTokens(): string[] { const now = this.now(); - const active = new Map(this.subscriptions); - for (const [token, expiresAt] of this.subscriptions) { - if (expiresAt <= now) { - active.delete(token); + const rows = this.database + .prepare("SELECT payload FROM push_tokens ORDER BY rowid") + .all() as unknown as PushTokenRow[]; + const active: string[] = []; + const expired: string[] = []; + for (const row of rows) { + const record = PushTokenRecordSchema.parse(JSON.parse(row.payload)); + if (Date.parse(record.expiresAt) <= now) { + expired.push(record.token); + } else { + active.push(record.token); } } - if (active.size !== this.subscriptions.size) { + if (expired.length > 0) { try { - this.persist(active); - this.subscriptions = active; + runInTransaction(this.database, () => { + const remove = this.database.prepare("DELETE FROM push_tokens WHERE token = ?"); + for (const token of expired) remove.run(token); + }); } catch { - // Keep the previous state so a later send retries pruning. Expired tokens - // are still excluded from this delivery. + // Expired tokens stay excluded and a later delivery retries pruning them. } } - return Array.from(active.keys()); + return active; } - private loadFromDisk(): void { - try { - if (!existsSync(this.filePath)) { - return; - } - ensurePrivateFile(this.filePath); - const raw = readFileSync(this.filePath, "utf-8"); - const parsed = JSON.parse(raw) as { subscriptions?: unknown; tokens?: unknown }; - const loaded = new Map(); - const subscriptions = Array.isArray(parsed.subscriptions) ? parsed.subscriptions : []; - for (const value of subscriptions) { - if (!value || typeof value !== "object") continue; - const candidate = value as { token?: unknown; expiresAt?: unknown }; - if (typeof candidate.token !== "string" || typeof candidate.expiresAt !== "string") - continue; - const token = candidate.token.trim(); - const expiresAt = Date.parse(candidate.expiresAt); - if (token && Number.isFinite(expiresAt)) { - loaded.set(token, expiresAt); - } - } - this.subscriptions = loaded; - - const legacyTokens = Array.isArray(parsed.tokens) - ? parsed.tokens.filter((token): token is string => typeof token === "string") - : []; - if (legacyTokens.length > 0) { - const migrated = new Map(loaded); - const expiresAt = this.now() + this.leaseMs; - for (const token of legacyTokens) { - const normalized = token.trim(); - if (normalized) migrated.set(normalized, expiresAt); - } - this.persist(migrated); - this.subscriptions = migrated; - } - this.logger.info({ total: this.subscriptions.size }, "Loaded push tokens"); - } catch (error) { - const err = error instanceof Error ? error : new Error(String(error)); - this.logger.warn({ err }, "Failed to load push tokens"); - } + private readToken(token: string): z.infer | null { + const row = this.database + .prepare("SELECT payload FROM push_tokens WHERE token = ?") + .get(token) as PushTokenRow | undefined; + return row ? PushTokenRecordSchema.parse(JSON.parse(row.payload)) : null; } - private persist(subscriptions: ReadonlyMap): void { + private count(): number { + const row = this.database.prepare("SELECT count(*) AS count FROM push_tokens").get() as { + count: number; + }; + return row.count; + } + + private importLegacyTokens(): void { + if (hasLegacyImportMarker(this.database, LEGACY_IMPORT_STORE)) return; + try { - const payload = - JSON.stringify( - { - subscriptions: Array.from(subscriptions, ([token, expiresAt]) => ({ - token, - expiresAt: new Date(expiresAt).toISOString(), - })), - }, - null, - 2, - ) + "\n"; - this.write(this.filePath, payload); + runInTransaction(this.database, () => { + let importedCount = 0; + let skippedCount = 0; + if (this.count() === 0 && existsSync(this.legacyFilePath)) { + ensurePrivateFile(this.legacyFilePath); + const raw = readFileSync(this.legacyFilePath, "utf-8"); + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch (error) { + this.logger.warn( + { err: error, filePath: this.legacyFilePath }, + "Skipping invalid legacy push token file", + ); + skippedCount += 1; + } + + const source = parsed && typeof parsed === "object" ? parsed : {}; + const subscriptions = Array.isArray((source as { subscriptions?: unknown }).subscriptions) + ? (source as { subscriptions: unknown[] }).subscriptions + : []; + const tokens = Array.isArray((source as { tokens?: unknown }).tokens) + ? (source as { tokens: unknown[] }).tokens.map((token) => ({ + token, + expiresAt: new Date(this.now() + this.leaseMs).toISOString(), + })) + : []; + + for (const candidate of [...subscriptions, ...tokens]) { + const record = PushTokenRecordSchema.safeParse(candidate); + if (!record.success) { + skippedCount += 1; + this.logger.warn( + { filePath: this.legacyFilePath, err: record.error }, + "Skipping invalid legacy push token", + ); + continue; + } + const result = this.database + .prepare("INSERT OR IGNORE INTO push_tokens (token, payload) VALUES (?, ?)") + .run(record.data.token, JSON.stringify(record.data)); + importedCount += Number(result.changes); + } + } + recordLegacyImportMarker(this.database, LEGACY_IMPORT_STORE, { + importedCount, + skippedCount, + }); + this.logger.info({ importedCount, skippedCount }, "Legacy push token import complete"); + }); } catch (error) { - const err = error instanceof Error ? error : new Error(String(error)); - this.logger.warn({ err }, "Failed to persist push tokens"); - throw err; + throw new Error(`Failed to import legacy push tokens from ${this.legacyFilePath}`, { + cause: error, + }); } } } diff --git a/packages/server/src/server/schedule/service.test.ts b/packages/server/src/server/schedule/service.test.ts index 89b0171597..dbdfc4c37a 100644 --- a/packages/server/src/server/schedule/service.test.ts +++ b/packages/server/src/server/schedule/service.test.ts @@ -1,6 +1,7 @@ import { mkdir, mkdtemp, rm } from "node:fs/promises"; import { join } from "node:path"; import { tmpdir } from "node:os"; +import type { DatabaseSync } from "node:sqlite"; import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import { AgentManager } from "../agent/agent-manager.js"; import { AgentStorage } from "../agent/agent-storage.js"; @@ -41,6 +42,7 @@ import { } from "./service.js"; import { ScheduleStore } from "./store.js"; import type { ScheduleExecutionResult, StoredSchedule } from "@getpaseo/protocol/schedule/types"; +import { openDatabase } from "../db/open.js"; interface ScheduleServiceInternals { executeSchedule(schedule: StoredSchedule, runId: string): Promise; @@ -75,10 +77,15 @@ const TEST_CLAUDE_PROVIDER_DEFINITION = { }; let workspaceArchiveInProgress = false; +let database: DatabaseSync; type TestScheduleServiceOptions = Omit< ScheduleServiceOptions, - "createAgent" | "createDirectoryWorkspace" | "createPaseoWorktreeWorkspace" | "archiveWorkspace" + | "database" + | "createAgent" + | "createDirectoryWorkspace" + | "createPaseoWorktreeWorkspace" + | "archiveWorkspace" > & { agentManager: AgentManager; providerSnapshotManager: Pick; @@ -159,6 +166,7 @@ function createScheduleService(options: TestScheduleServiceOptions): ScheduleSer }; return new ScheduleService({ ...options, + database, createAgent: options.createAgent ?? ((input) => @@ -300,6 +308,7 @@ describe("ScheduleService", () => { beforeEach(async () => { tempDir = await mkdtemp(join(tmpdir(), "schedule-service-test-")); + database = openDatabase(tempDir); await mkdir(join(tempDir, "agents"), { recursive: true }); agentStorage = new AgentStorage(join(tempDir, "agents"), createTestLogger()); await agentStorage.initialize(); @@ -310,6 +319,7 @@ describe("ScheduleService", () => { // Drain pending background persists before deleting the dir to avoid // ENOTEMPTY races when AgentManager flushes a snapshot mid-cleanup. await agentStorage.flush(); + database.close(); await rm(tempDir, { recursive: true, force: true }); }); @@ -768,7 +778,11 @@ describe("ScheduleService", () => { const runStarted = new Promise((resolve) => { releaseRun = resolve; }); - const store = new ScheduleStore(join(tempDir, "schedules")); + const store = new ScheduleStore({ + database, + legacyDir: join(tempDir, "schedules"), + logger: createTestLogger(), + }); const legacy = await store.create({ name: null, prompt: "finish/update race", @@ -1891,7 +1905,11 @@ describe("ScheduleService", () => { const interruptedAt = now.toISOString(); const associatedAgentId = "11111111-1111-4111-8111-111111111111"; const workspaceId = "wks_interrupted_with_agent"; - const store = new ScheduleStore(join(tempDir, "schedules")); + const store = new ScheduleStore({ + database, + legacyDir: join(tempDir, "schedules"), + logger: createTestLogger(), + }); await store.update(created.id, (schedule) => ({ ...schedule, runs: [ @@ -1959,7 +1977,11 @@ describe("ScheduleService", () => { const interruptedAt = now.toISOString(); const workspaceId = "wks_interrupted_without_agent"; - const store = new ScheduleStore(join(tempDir, "schedules")); + const store = new ScheduleStore({ + database, + legacyDir: join(tempDir, "schedules"), + logger: createTestLogger(), + }); await store.update(created.id, (schedule) => ({ ...schedule, runs: [ diff --git a/packages/server/src/server/schedule/service.ts b/packages/server/src/server/schedule/service.ts index 9b740a4a57..48c251157b 100644 --- a/packages/server/src/server/schedule/service.ts +++ b/packages/server/src/server/schedule/service.ts @@ -1,6 +1,7 @@ import { randomUUID } from "node:crypto"; import { stat } from "node:fs/promises"; import { join } from "node:path"; +import type { DatabaseSync } from "node:sqlite"; import type { Logger } from "pino"; import type { AgentManager } from "../agent/agent-manager.js"; import type { AgentSessionConfig } from "../agent/agent-sdk-types.js"; @@ -215,6 +216,7 @@ interface ScheduleWorkspaceCreateInput { export interface ScheduleServiceOptions { paseoHome: string; + database: DatabaseSync; logger: Logger; agentManager: ScheduleAgentManager; agentStorage: AgentStorage; @@ -252,7 +254,11 @@ export class ScheduleService { private tickTimer: ReturnType | null = null; constructor(options: ScheduleServiceOptions) { - this.store = new ScheduleStore(join(options.paseoHome, "schedules")); + this.store = new ScheduleStore({ + database: options.database, + legacyDir: join(options.paseoHome, "schedules"), + logger: options.logger, + }); this.logger = options.logger.child({ module: "schedule-service" }); this.agentManager = options.agentManager; this.agentStorage = options.agentStorage; diff --git a/packages/server/src/server/schedule/store.test.ts b/packages/server/src/server/schedule/store.test.ts index 7e19bbfd92..0acb1a86ea 100644 --- a/packages/server/src/server/schedule/store.test.ts +++ b/packages/server/src/server/schedule/store.test.ts @@ -1,19 +1,43 @@ -import { mkdtemp, rm } from "node:fs/promises"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import type { DatabaseSync } from "node:sqlite"; +import type pino from "pino"; import { join } from "node:path"; import { tmpdir } from "node:os"; -import { afterEach, beforeEach, describe, expect, test } from "vitest"; +import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; +import { openDatabase } from "../db/open.js"; +import { createTestLogger } from "../../test-utils/test-logger.js"; import { ScheduleStore } from "./store.js"; describe("ScheduleStore", () => { let tempDir: string; + let database: DatabaseSync; let store: ScheduleStore; + function createStore(logger = createTestLogger()): ScheduleStore { + return new ScheduleStore({ + database, + legacyDir: join(tempDir, "schedules"), + logger, + }); + } + + function createRecordingLogger(warn: pino.Logger["warn"]): pino.Logger { + const logger = { + child: () => logger, + info: () => undefined, + warn, + }; + return logger as unknown as pino.Logger; + } + beforeEach(async () => { tempDir = await mkdtemp(join(tmpdir(), "schedule-store-test-")); - store = new ScheduleStore(tempDir); + database = openDatabase(tempDir); + store = createStore(); }); afterEach(async () => { + database.close(); await rm(tempDir, { recursive: true, force: true }); }); @@ -40,7 +64,7 @@ describe("ScheduleStore", () => { runs: [], }); - const reloaded = new ScheduleStore(tempDir); + const reloaded = createStore(); const listed = await reloaded.list(); expect(created.id).toHaveLength(8); @@ -81,7 +105,7 @@ describe("ScheduleStore", () => { }; await store.update(created.id, () => updated); - const reloaded = await new ScheduleStore(tempDir).get(created.id); + const reloaded = await createStore().get(created.id); expect(reloaded).toEqual(updated); }); @@ -182,7 +206,7 @@ describe("ScheduleStore", () => { prompt: "after", runs: [{ id: "run-1" }], }); - await expect(new ScheduleStore(tempDir).get(created.id)).resolves.toMatchObject({ + await expect(createStore().get(created.id)).resolves.toMatchObject({ prompt: "after", runs: [{ id: "run-1" }], }); @@ -211,7 +235,11 @@ describe("ScheduleStore", () => { } } - const gatedStore = new GatedListScheduleStore(tempDir); + const gatedStore = new GatedListScheduleStore({ + database, + legacyDir: join(tempDir, "schedules"), + logger: createTestLogger(), + }); const target = { type: "new-agent" as const, config: { provider: "claude" as const, cwd: tempDir }, @@ -304,4 +332,99 @@ describe("ScheduleStore", () => { }); expect(await gatedStore.list()).toHaveLength(2); }); + + test("imports legacy schedules once and records the marker", async () => { + const legacyDir = join(tempDir, "schedules"); + await mkdir(legacyDir, { recursive: true }); + const legacy = { + id: "legacy01", + name: "Legacy", + prompt: "Imported", + cadence: { type: "every" as const, everyMs: 60_000 }, + target: { + type: "new-agent" as const, + config: { provider: "claude" as const, cwd: tempDir }, + }, + status: "active" as const, + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + nextRunAt: "2026-01-01T00:01:00.000Z", + lastRunAt: null, + pausedAt: null, + expiresAt: null, + maxRuns: null, + runs: [], + }; + await writeFile(join(legacyDir, "legacy01.json"), JSON.stringify(legacy)); + database.prepare("DELETE FROM legacy_imports WHERE store = ?").run("schedules"); + + const imported = createStore(); + await expect(imported.get(legacy.id)).resolves.toEqual(legacy); + await imported.delete(legacy.id); + + createStore(); + await expect(imported.get(legacy.id)).resolves.toBeNull(); + expect( + database + .prepare("SELECT imported_count FROM legacy_imports WHERE store = ?") + .get("schedules"), + ).toMatchObject({ imported_count: 1 }); + }); + + test("skips and logs invalid legacy schedule records", async () => { + const legacyDir = join(tempDir, "schedules"); + await mkdir(legacyDir, { recursive: true }); + await writeFile(join(legacyDir, "invalid.json"), "{not-json"); + database.prepare("DELETE FROM legacy_imports WHERE store = ?").run("schedules"); + const warn = vi.fn(); + + createStore(createRecordingLogger(warn)); + + expect(warn).toHaveBeenCalledWith( + expect.objectContaining({ filePath: join(legacyDir, "invalid.json") }), + "Skipping invalid legacy schedule record", + ); + expect( + database.prepare("SELECT skipped_count FROM legacy_imports WHERE store = ?").get("schedules"), + ).toMatchObject({ skipped_count: 1 }); + }); + + test("rolls back the whole legacy schedule import on failure", async () => { + const legacyDir = join(tempDir, "schedules"); + await mkdir(legacyDir, { recursive: true }); + const legacy = { + id: "duplicate", + name: null, + prompt: "Imported", + cadence: { type: "every" as const, everyMs: 60_000 }, + target: { + type: "new-agent" as const, + config: { provider: "claude" as const, cwd: tempDir }, + }, + status: "active" as const, + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + nextRunAt: "2026-01-01T00:01:00.000Z", + lastRunAt: null, + pausedAt: null, + expiresAt: null, + maxRuns: null, + runs: [], + }; + await Promise.all([ + writeFile(join(legacyDir, "one.json"), JSON.stringify(legacy)), + writeFile(join(legacyDir, "two.json"), JSON.stringify(legacy)), + ]); + database.prepare("DELETE FROM legacy_imports WHERE store = ?").run("schedules"); + + expect(() => createStore()).toThrow("Failed to import legacy schedules"); + expect(database.prepare("SELECT count(*) AS count FROM schedules").get()).toMatchObject({ + count: 0, + }); + expect( + database + .prepare("SELECT count(*) AS count FROM legacy_imports WHERE store = ?") + .get("schedules"), + ).toMatchObject({ count: 0 }); + }); }); diff --git a/packages/server/src/server/schedule/store.ts b/packages/server/src/server/schedule/store.ts index f3583d5f70..16ced3932b 100644 --- a/packages/server/src/server/schedule/store.ts +++ b/packages/server/src/server/schedule/store.ts @@ -1,12 +1,17 @@ import { randomBytes } from "node:crypto"; -import { mkdir, readFile, readdir, rm } from "node:fs/promises"; +import { existsSync, readFileSync, readdirSync } from "node:fs"; import { join } from "node:path"; +import type { DatabaseSync } from "node:sqlite"; import { StoredScheduleSchema, type ScheduleTarget, type StoredSchedule, } from "@getpaseo/protocol/schedule/types"; -import { writeJsonFileAtomic } from "../atomic-file.js"; +import type { Logger } from "pino"; +import { hasLegacyImportMarker, recordLegacyImportMarker } from "../db/legacy-imports.js"; +import { runInTransaction } from "../db/transaction.js"; + +const LEGACY_IMPORT_STORE = "schedules"; function generateScheduleId(): string { return randomBytes(4).toString("hex"); @@ -19,6 +24,16 @@ interface ScheduleNameTargetUpsert { update: ScheduleUpdater; } +interface ScheduleStoreOptions { + database: DatabaseSync; + legacyDir: string; + logger: Logger; +} + +interface ScheduleRow { + payload: string; +} + function canonicalize(value: unknown): unknown { if (Array.isArray(value)) { return value.map(canonicalize); @@ -90,42 +105,29 @@ function matchesNameAndTarget( export class ScheduleStore { private readonly scheduleMutations = new Map>(); private readonly identityMutations = new Map>(); + private readonly database: DatabaseSync; + private readonly legacyDir: string; + private readonly logger: Logger; - constructor(private readonly dir: string) {} - - private filePath(id: string): string { - return join(this.dir, `${id}.json`); - } - - private async ensureDir(): Promise { - await mkdir(this.dir, { recursive: true }); + constructor(options: ScheduleStoreOptions) { + this.database = options.database; + this.legacyDir = options.legacyDir; + this.logger = options.logger.child({ component: "schedule-store" }); + this.importLegacySchedules(); } async list(): Promise { - await this.ensureDir(); - const entries = await readdir(this.dir, { withFileTypes: true }); - const schedules = await Promise.all( - entries - .filter((entry) => entry.isFile() && entry.name.endsWith(".json")) - .map(async (entry) => { - const content = await readFile(join(this.dir, entry.name), "utf-8"); - return StoredScheduleSchema.parse(JSON.parse(content)); - }), - ); - return schedules.sort((left, right) => left.createdAt.localeCompare(right.createdAt)); + const rows = this.database + .prepare("SELECT payload FROM schedules ORDER BY created_at, id") + .all() as unknown as ScheduleRow[]; + return rows.map((row) => StoredScheduleSchema.parse(JSON.parse(row.payload))); } async get(id: string): Promise { - await this.ensureDir(); - try { - const content = await readFile(this.filePath(id), "utf-8"); - return StoredScheduleSchema.parse(JSON.parse(content)); - } catch (error) { - if ((error as NodeJS.ErrnoException).code === "ENOENT") { - return null; - } - throw error; - } + const row = this.database.prepare("SELECT payload FROM schedules WHERE id = ?").get(id) as + | ScheduleRow + | undefined; + return row ? StoredScheduleSchema.parse(JSON.parse(row.payload)) : null; } async create(schedule: Omit): Promise { @@ -184,15 +186,39 @@ export class ScheduleStore { }); } - private async write(schedule: StoredSchedule): Promise { - await this.ensureDir(); - await writeJsonFileAtomic(this.filePath(schedule.id), schedule); + private write(schedule: StoredSchedule): void { + const identityKey = + schedule.status !== "completed" && normalizeOptionalScheduleName(schedule.name) !== null + ? nameTargetIdentityKey(schedule.name!, schedule.target) + : null; + this.database + .prepare( + `INSERT INTO schedules (id, created_at, identity_key, payload) + VALUES (?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + created_at = excluded.created_at, + identity_key = excluded.identity_key, + payload = excluded.payload`, + ) + .run(schedule.id, schedule.createdAt, identityKey, JSON.stringify(schedule)); + } + + private insert(schedule: StoredSchedule): void { + const identityKey = + schedule.status !== "completed" && normalizeOptionalScheduleName(schedule.name) !== null + ? nameTargetIdentityKey(schedule.name!, schedule.target) + : null; + this.database + .prepare( + `INSERT INTO schedules (id, created_at, identity_key, payload) + VALUES (?, ?, ?, ?)`, + ) + .run(schedule.id, schedule.createdAt, identityKey, JSON.stringify(schedule)); } async delete(id: string): Promise { await this.serializeScheduleMutation(id, async () => { - await this.ensureDir(); - await rm(this.filePath(id), { force: true }); + this.database.prepare("DELETE FROM schedules WHERE id = ?").run(id); }); } @@ -250,4 +276,60 @@ export class ScheduleStore { return updated; }); } + + private count(): number { + const row = this.database.prepare("SELECT count(*) AS count FROM schedules").get() as { + count: number; + }; + return row.count; + } + + private importLegacySchedules(): void { + if (hasLegacyImportMarker(this.database, LEGACY_IMPORT_STORE)) { + return; + } + + try { + runInTransaction(this.database, () => { + let importedCount = 0; + let skippedCount = 0; + if (this.count() === 0 && existsSync(this.legacyDir)) { + const entries = readdirSync(this.legacyDir, { withFileTypes: true }); + for (const entry of entries) { + if (!entry.isFile() || !entry.name.endsWith(".json")) { + continue; + } + const filePath = join(this.legacyDir, entry.name); + const content = readFileSync(filePath, "utf-8"); + let json: unknown; + try { + json = JSON.parse(content); + } catch (error) { + skippedCount += 1; + this.logger.warn({ err: error, filePath }, "Skipping invalid legacy schedule record"); + continue; + } + const parsed = StoredScheduleSchema.safeParse(json); + if (!parsed.success) { + skippedCount += 1; + this.logger.warn( + { err: parsed.error, filePath }, + "Skipping invalid legacy schedule record", + ); + continue; + } + this.insert(parsed.data); + importedCount += 1; + } + } + recordLegacyImportMarker(this.database, LEGACY_IMPORT_STORE, { + importedCount, + skippedCount, + }); + this.logger.info({ importedCount, skippedCount }, "Legacy schedule import complete"); + }); + } catch (error) { + throw new Error(`Failed to import legacy schedules from ${this.legacyDir}`, { cause: error }); + } + } } diff --git a/packages/server/src/server/websocket-server.browser-tools.test.ts b/packages/server/src/server/websocket-server.browser-tools.test.ts index 7eb238cc66..8fee6190c0 100644 --- a/packages/server/src/server/websocket-server.browser-tools.test.ts +++ b/packages/server/src/server/websocket-server.browser-tools.test.ts @@ -17,6 +17,7 @@ import { BrowserToolsBroker } from "./browser-tools/broker.js"; import type { CheckoutDiffManager } from "./checkout-diff-manager.js"; import type { DaemonConfigStore } from "./daemon-config-store.js"; import type { DownloadTokenStore } from "./file-download/token-store.js"; +import type { PushTokenStore } from "./push/token-store.js"; import type { ScheduleService } from "./schedule/service.js"; import { createStub } from "./test-utils/class-mocks.js"; import { DaemonClient } from "./test-utils/daemon-client.js"; @@ -292,6 +293,7 @@ function createVoiceAssistantWebSocketServer(params: { createStub(agentManager), createStub({}), createStub({}), + createStub({}), "/tmp/paseo-browser-tools-websocket-test", createStub(daemonConfigStore), null, diff --git a/packages/server/src/server/websocket-server.notifications.test.ts b/packages/server/src/server/websocket-server.notifications.test.ts index e2bec9c710..07541ad595 100644 --- a/packages/server/src/server/websocket-server.notifications.test.ts +++ b/packages/server/src/server/websocket-server.notifications.test.ts @@ -4,6 +4,7 @@ import type pino from "pino"; import type { AgentManager } from "./agent/agent-manager.js"; import type { AgentStorage } from "./agent/agent-storage.js"; import type { DownloadTokenStore } from "./file-download/token-store.js"; +import type { PushTokenStore } from "./push/token-store.js"; import type { DaemonConfigStore } from "./daemon-config-store.js"; import type { ScheduleService } from "./schedule/service.js"; import type { CheckoutDiffManager } from "./checkout-diff-manager.js"; @@ -110,6 +111,7 @@ function createServer(agentManagerOverrides?: Record) { createStub(agentManager), createStub({}), createStub({}), + createStub({}), "/tmp/paseo-test", createStub(daemonConfigStore), null, diff --git a/packages/server/src/server/websocket-server.relay-reconnect.test.ts b/packages/server/src/server/websocket-server.relay-reconnect.test.ts index b14e90c7a0..d786ff298a 100644 --- a/packages/server/src/server/websocket-server.relay-reconnect.test.ts +++ b/packages/server/src/server/websocket-server.relay-reconnect.test.ts @@ -4,6 +4,7 @@ import type pino from "pino"; import type { AgentManager } from "./agent/agent-manager.js"; import type { AgentStorage } from "./agent/agent-storage.js"; import type { DownloadTokenStore } from "./file-download/token-store.js"; +import type { PushTokenStore } from "./push/token-store.js"; import type { DaemonConfigStore } from "./daemon-config-store.js"; import type { ScheduleService } from "./schedule/service.js"; import type { CheckoutDiffManager } from "./checkout-diff-manager.js"; @@ -244,6 +245,7 @@ function createServer(options?: { }), createStub({}), createStub({}), + createStub({}), "/tmp/paseo-test", createStub(daemonConfigStore), null, diff --git a/packages/server/src/server/websocket-server.terminal-notifications.test.ts b/packages/server/src/server/websocket-server.terminal-notifications.test.ts index 1d7160ed30..7b488009c5 100644 --- a/packages/server/src/server/websocket-server.terminal-notifications.test.ts +++ b/packages/server/src/server/websocket-server.terminal-notifications.test.ts @@ -4,6 +4,7 @@ import type pino from "pino"; import type { AgentManager } from "./agent/agent-manager.js"; import type { AgentStorage } from "./agent/agent-storage.js"; import type { DownloadTokenStore } from "./file-download/token-store.js"; +import type { PushTokenStore } from "./push/token-store.js"; import type { DaemonConfigStore } from "./daemon-config-store.js"; import type { ScheduleService } from "./schedule/service.js"; import type { CheckoutDiffManager } from "./checkout-diff-manager.js"; @@ -141,6 +142,7 @@ function createServer(terminalManager: TerminalManager, workspaceRegistry?: Work createStub(agentManager), createStub({}), createStub({}), + createStub({}), "/tmp/paseo-test", createStub(daemonConfigStore), null, diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index d14681b531..06e68bcf44 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -1,6 +1,5 @@ import { WebSocket, WebSocketServer } from "ws"; import type { IncomingMessage, Server as HTTPServer } from "http"; -import { join } from "path"; import { hostname as getHostname } from "node:os"; import { randomUUID } from "node:crypto"; import { monitorEventLoopDelay } from "node:perf_hooks"; @@ -50,6 +49,7 @@ import { type PushNotifications, type PushNotificationSender, } from "./push/index.js"; +import type { PushTokenStore } from "./push/token-store.js"; import type { ScriptHealthState } from "./script-health-monitor.js"; import type { ServiceProxySubsystem } from "./service-proxy.js"; import type { WorkspaceScriptRuntimeStore } from "./workspace-script-runtime-store.js"; @@ -585,6 +585,7 @@ export class VoiceAssistantWebSocketServer { agentManager: AgentManager, agentStorage: AgentStorage, downloadTokenStore: DownloadTokenStore, + pushTokenStore: PushTokenStore, paseoHome: string, daemonConfigStore: DaemonConfigStore, mcpBaseUrl: string | null, @@ -688,7 +689,7 @@ export class VoiceAssistantWebSocketServer { const pushLogger = this.logger.child({ module: "push" }); this.pushNotifications = createPushNotifications({ logger: pushLogger, - filePath: join(paseoHome, "push-tokens.json"), + store: pushTokenStore, }); this.pushNotificationSender = pushNotificationSender ?? this.pushNotifications;