mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-14 13:22:37 +00:00
Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0290b0e35b | ||
|
|
07ffe243fb | ||
|
|
ae3c0119f9 | ||
|
|
5d195e7e15 |
+31
-10
@@ -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.
|
||||
|
||||
---
|
||||
|
||||
|
||||
+1
-1
@@ -1 +1 @@
|
||||
sha256-i5PbVUe2Ec+GtghV9IpCJQJ9hcUT5hFhmxneNvoD584=
|
||||
sha256-BeS2/PYAPoG4U3q7N2qOVuPC1T5sDQ+c9hv2cTn3KRU=
|
||||
|
||||
Generated
+17
-1
@@ -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",
|
||||
|
||||
@@ -46,5 +46,8 @@
|
||||
"typescript": "^5.2.2",
|
||||
"vitest": "^4.1.6",
|
||||
"zx": "^8.8.5"
|
||||
},
|
||||
"engines": {
|
||||
"node": ">=22.20.0"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
import { describe, expect, test } from "vitest";
|
||||
|
||||
import { runShutdownSequence } from "./bootstrap.js";
|
||||
|
||||
describe("runShutdownSequence", () => {
|
||||
test("runs every step before rethrowing the first failure", async () => {
|
||||
const firstFailure = new Error("first failure");
|
||||
const secondFailure = new Error("second failure");
|
||||
const completed: string[] = [];
|
||||
const reported: Array<{ name: string; error: unknown }> = [];
|
||||
|
||||
const shutdown = runShutdownSequence(
|
||||
[
|
||||
["first", () => completed.push("first")],
|
||||
[
|
||||
"second",
|
||||
() => {
|
||||
completed.push("second");
|
||||
throw firstFailure;
|
||||
},
|
||||
],
|
||||
[
|
||||
"third",
|
||||
async () => {
|
||||
completed.push("third");
|
||||
throw secondFailure;
|
||||
},
|
||||
],
|
||||
["fourth", () => completed.push("fourth")],
|
||||
],
|
||||
(name, error) => reported.push({ name, error }),
|
||||
);
|
||||
|
||||
await expect(shutdown).rejects.toBe(firstFailure);
|
||||
expect(completed).toEqual(["first", "second", "third", "fourth"]);
|
||||
expect(reported).toEqual([
|
||||
{ name: "second", error: firstFailure },
|
||||
{ name: "third", error: secondFailure },
|
||||
]);
|
||||
});
|
||||
});
|
||||
@@ -79,6 +79,32 @@ export function parseListenString(listen: string): ListenTarget {
|
||||
throw new Error(`Invalid listen string: ${listen}`);
|
||||
}
|
||||
|
||||
type ShutdownStep = readonly [name: string, run: () => void | Promise<void>];
|
||||
|
||||
export async function runShutdownSequence(
|
||||
steps: readonly ShutdownStep[],
|
||||
onFailure: (name: string, error: unknown) => void,
|
||||
): Promise<void> {
|
||||
let failed = false;
|
||||
let firstFailure: unknown;
|
||||
|
||||
for (const [name, run] of steps) {
|
||||
try {
|
||||
await run();
|
||||
} catch (error) {
|
||||
if (!failed) {
|
||||
failed = true;
|
||||
firstFailure = error;
|
||||
}
|
||||
onFailure(name, error);
|
||||
}
|
||||
}
|
||||
|
||||
if (failed) {
|
||||
throw firstFailure;
|
||||
}
|
||||
}
|
||||
|
||||
function formatListenTarget(listenTarget: ListenTarget | null): string | null {
|
||||
if (!listenTarget) {
|
||||
return null;
|
||||
@@ -216,6 +242,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 +580,21 @@ export async function createPaseoDaemon(
|
||||
): Promise<PaseoDaemon> {
|
||||
configureGitProcessPolicy(config.git ?? resolveGitProcessPolicy({ env: process.env }));
|
||||
const logger = rootLogger.child({ module: "bootstrap" });
|
||||
const database = openDatabase(config.paseoHome);
|
||||
try {
|
||||
return await createPaseoDaemonWithDatabase(config, logger, dependencies, database);
|
||||
} catch (error) {
|
||||
database.close();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function createPaseoDaemonWithDatabase(
|
||||
config: PaseoDaemonConfig,
|
||||
logger: Logger,
|
||||
dependencies: PaseoDaemonDependencies,
|
||||
database: ReturnType<typeof openDatabase>,
|
||||
): Promise<PaseoDaemon> {
|
||||
const bootstrapStart = performance.now();
|
||||
const elapsed = () => `${(performance.now() - bootstrapStart).toFixed(0)}ms`;
|
||||
const daemonVersion = config.daemonVersion ?? resolveDaemonVersion(import.meta.url);
|
||||
@@ -1205,6 +1248,7 @@ export async function createPaseoDaemon(
|
||||
};
|
||||
const scheduleService = new ScheduleService({
|
||||
paseoHome: config.paseoHome,
|
||||
database,
|
||||
logger,
|
||||
agentManager,
|
||||
agentStorage,
|
||||
@@ -1222,6 +1266,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 +1553,7 @@ export async function createPaseoDaemon(
|
||||
agentManager,
|
||||
agentStorage,
|
||||
downloadTokenStore,
|
||||
pushTokenStore,
|
||||
config.paseoHome,
|
||||
daemonConfigStore,
|
||||
mcpBaseUrl,
|
||||
@@ -1613,40 +1663,46 @@ export async function createPaseoDaemon(
|
||||
};
|
||||
|
||||
const stop = async () => {
|
||||
await hubRelationships.stop();
|
||||
workspaceReconciliation.dispose();
|
||||
scriptHealthMonitor.stop();
|
||||
// Freeze both ingress and registration before taking the agent closure snapshot.
|
||||
wsServer?.prepareForShutdown();
|
||||
agentManager.prepareForShutdown();
|
||||
await closeAllAgents(logger, agentManager);
|
||||
await agentManager.flushForShutdown().catch(() => undefined);
|
||||
detachAgentStoragePersistence();
|
||||
await agentStorage.flush().catch(() => undefined);
|
||||
await providerSnapshotManager.shutdown();
|
||||
terminalManager.killAll();
|
||||
speechService.stop();
|
||||
await scheduleService.stop().catch(() => undefined);
|
||||
await relayRuntime?.stop().catch(() => undefined);
|
||||
if (wsServer) {
|
||||
await wsServer.close();
|
||||
}
|
||||
await serviceProxy.stopStandalone();
|
||||
// Force-drop remaining sockets so httpServer.close() resolves promptly.
|
||||
// We've already closed wsServer (which sent ws-layer close frames) and
|
||||
// stopped every other service, so anything still attached is a TCP
|
||||
// socket whose higher-level shutdown hasn't fully released it (e.g.
|
||||
// upgraded WS sockets in the closing handshake, or HTTP keep-alive
|
||||
// sockets in CLOSE_WAIT). closeIdleConnections() does not catch
|
||||
// upgraded sockets, so we use closeAllConnections() here.
|
||||
httpServer.closeAllConnections();
|
||||
await new Promise<void>((resolve) => {
|
||||
httpServer.close(() => resolve());
|
||||
});
|
||||
// Clean up socket files
|
||||
if (listenTarget.type === "socket" && existsSync(listenTarget.path)) {
|
||||
unlinkSync(listenTarget.path);
|
||||
}
|
||||
await runShutdownSequence(
|
||||
[
|
||||
["hub relationships", () => hubRelationships.stop()],
|
||||
["workspace reconciliation", () => workspaceReconciliation.dispose()],
|
||||
["script health monitor", () => scriptHealthMonitor.stop()],
|
||||
// Freeze both ingress and registration before taking the agent closure snapshot.
|
||||
["WebSocket ingress", () => wsServer?.prepareForShutdown()],
|
||||
["agent ingress", () => agentManager.prepareForShutdown()],
|
||||
["agents", () => closeAllAgents(logger, agentManager)],
|
||||
["agent manager persistence", () => agentManager.flushForShutdown()],
|
||||
["agent storage persistence hook", () => detachAgentStoragePersistence()],
|
||||
["agent storage", () => agentStorage.flush()],
|
||||
["provider snapshots", () => providerSnapshotManager.shutdown()],
|
||||
["terminals", () => terminalManager.killAll()],
|
||||
["speech", () => speechService.stop()],
|
||||
["schedules", () => scheduleService.stop()],
|
||||
["relay", () => relayRuntime?.stop()],
|
||||
["WebSocket server", () => wsServer?.close()],
|
||||
["service proxy", () => serviceProxy.stopStandalone()],
|
||||
// Force-drop remaining sockets so httpServer.close() resolves promptly.
|
||||
// We've already closed wsServer (which sent ws-layer close frames) and
|
||||
// stopped every other service, so anything still attached is a TCP
|
||||
// socket whose higher-level shutdown hasn't fully released it (e.g.
|
||||
// upgraded WS sockets in the closing handshake, or HTTP keep-alive
|
||||
// sockets in CLOSE_WAIT). closeIdleConnections() does not catch
|
||||
// upgraded sockets, so we use closeAllConnections() here.
|
||||
["HTTP connections", () => httpServer.closeAllConnections()],
|
||||
["HTTP server", () => new Promise<void>((resolve) => httpServer.close(() => resolve()))],
|
||||
[
|
||||
"socket file",
|
||||
() => {
|
||||
if (listenTarget.type === "socket" && existsSync(listenTarget.path)) {
|
||||
unlinkSync(listenTarget.path);
|
||||
}
|
||||
},
|
||||
],
|
||||
["database", () => database.close()],
|
||||
],
|
||||
(name, error) => logger.warn({ err: error, step: name }, "Daemon shutdown step failed"),
|
||||
);
|
||||
};
|
||||
|
||||
return {
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
@@ -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}`);
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -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<string> {
|
||||
const home = await mkdtemp(join(tmpdir(), "paseo-db-test-"));
|
||||
homes.push(home);
|
||||
return home;
|
||||
}
|
||||
|
||||
function insertThenFail(database: ReturnType<typeof openDatabase>): 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();
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -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 });
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
import type { DatabaseSync } from "node:sqlite";
|
||||
|
||||
export function runInTransaction<T>(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;
|
||||
}
|
||||
}
|
||||
@@ -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),
|
||||
});
|
||||
|
||||
|
||||
@@ -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<PushNotifications, "send">;
|
||||
|
||||
export function createPushNotifications(options: {
|
||||
logger: pino.Logger;
|
||||
filePath: string;
|
||||
now?: () => number;
|
||||
store: PushTokenStore;
|
||||
deliver?: (tokens: string[], payload: PushPayload) => Promise<void>;
|
||||
}): 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);
|
||||
|
||||
@@ -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 });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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<string, number>();
|
||||
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<string, number>();
|
||||
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<typeof PushTokenRecordSchema> | 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<string, number>): 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,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<ScheduleExecutionResult>;
|
||||
@@ -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<ProviderSnapshotManager, "resolveCreateConfig">;
|
||||
@@ -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<void>((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: [
|
||||
|
||||
@@ -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<typeof setInterval> | 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;
|
||||
|
||||
@@ -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 });
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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<string, Promise<unknown>>();
|
||||
private readonly identityMutations = new Map<string, Promise<unknown>>();
|
||||
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<void> {
|
||||
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<StoredSchedule[]> {
|
||||
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<StoredSchedule | null> {
|
||||
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<StoredSchedule, "id">): Promise<StoredSchedule> {
|
||||
@@ -184,15 +186,39 @@ export class ScheduleStore {
|
||||
});
|
||||
}
|
||||
|
||||
private async write(schedule: StoredSchedule): Promise<void> {
|
||||
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<void> {
|
||||
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 });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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>(agentManager),
|
||||
createStub<AgentStorage>({}),
|
||||
createStub<DownloadTokenStore>({}),
|
||||
createStub<PushTokenStore>({}),
|
||||
"/tmp/paseo-browser-tools-websocket-test",
|
||||
createStub<DaemonConfigStore>(daemonConfigStore),
|
||||
null,
|
||||
|
||||
@@ -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<string, unknown>) {
|
||||
createStub<AgentManager>(agentManager),
|
||||
createStub<AgentStorage>({}),
|
||||
createStub<DownloadTokenStore>({}),
|
||||
createStub<PushTokenStore>({}),
|
||||
"/tmp/paseo-test",
|
||||
createStub<DaemonConfigStore>(daemonConfigStore),
|
||||
null,
|
||||
|
||||
@@ -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<AgentStorage>({}),
|
||||
createStub<DownloadTokenStore>({}),
|
||||
createStub<PushTokenStore>({}),
|
||||
"/tmp/paseo-test",
|
||||
createStub<DaemonConfigStore>(daemonConfigStore),
|
||||
null,
|
||||
|
||||
@@ -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>(agentManager),
|
||||
createStub<AgentStorage>({}),
|
||||
createStub<DownloadTokenStore>({}),
|
||||
createStub<PushTokenStore>({}),
|
||||
"/tmp/paseo-test",
|
||||
createStub<DaemonConfigStore>(daemonConfigStore),
|
||||
null,
|
||||
|
||||
@@ -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;
|
||||
|
||||
|
||||
Reference in New Issue
Block a user