mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
refactor(push-web): move web push store to shared SQLite state DB (#103294)
* refactor(push-web): move state to shared SQLite Co-authored-by: hailory <hailory@xydigit.com> * fix(push-web): harden legacy migration retries * fix(push-web): format migration lock errors * fix(push-web): gate runtime during legacy migration * fix(push-web): keep store internals private --------- Co-authored-by: Peter Steinberger <steipete@gmail.com> Co-authored-by: hailory <hailory@xydigit.com>
This commit is contained in:
xydt-tanshanshan
@@ -1103,6 +1103,16 @@ sessionId})`; create, branch, continue, list, and fork flows live in their
|
||||
the complete legacy `commitments.json`, keeps newer SQLite rows, verifies the
|
||||
result, and only then removes the unchanged source. Runtime never reads or
|
||||
writes the retired file.
|
||||
- Web Push subscriptions and the generated VAPID identity now use typed shared
|
||||
`web_push_subscriptions` and `web_push_vapid_keys` rows. Runtime registration,
|
||||
expiry cleanup, and first-use key generation use row-level SQLite
|
||||
transactions. Explicit Doctor repair validates both retired JSON stores,
|
||||
claims them before the SQLite write, imports them atomically, rejects
|
||||
conflicting VAPID identities, verifies the result, and only then removes the
|
||||
claims. Doctor holds the state-directory maintenance lock for the complete
|
||||
import so an older Gateway cannot recreate the retired files. Registration,
|
||||
delivery, deletion, and key resolution fail closed until Doctor resolves
|
||||
pending legacy sources or interrupted claims.
|
||||
- Cron job definitions, schedule state, and run history no longer have runtime
|
||||
JSON writers or readers. Runtime uses `cron_jobs` rows with typed schedule,
|
||||
payload, delivery, failure-alert, session, status, and runtime-state columns plus
|
||||
@@ -2181,8 +2191,8 @@ Add a repo check that fails new runtime writes to legacy state paths:
|
||||
- `nodes/pending.json` / `nodes/paired.json` (retired 2026.7: folded into paired device records at gateway startup)
|
||||
- `identity/device.json`
|
||||
- `identity/device-auth.json`
|
||||
- `push/web-push-subscriptions.json`
|
||||
- `push/vapid-keys.json`
|
||||
- `push/web-push-subscriptions.json` (retired; Doctor-only import into `web_push_subscriptions`)
|
||||
- `push/vapid-keys.json` (retired; Doctor-only import into `web_push_vapid_keys`)
|
||||
- `push/apns-registrations.json`
|
||||
- `process-leases.json`
|
||||
- `gateway-instance-id`
|
||||
|
||||
@@ -428,12 +428,14 @@ The Control UI ships a `manifest.webmanifest` and a service worker, so modern br
|
||||
|
||||
If the page shows **Protocol mismatch** right after an OpenClaw update, first reopen the dashboard with `openclaw dashboard` and hard-refresh. If it still fails, clear site data for the dashboard origin or test in a private browser window; an old tab or browser service-worker cache can keep running a pre-update Control UI bundle against the newer Gateway.
|
||||
|
||||
| Surface | What it does |
|
||||
| ----------------------------------------------------- | ------------------------------------------------------------------ |
|
||||
| `ui/public/manifest.webmanifest` | PWA manifest. Browsers offer "Install app" once it is reachable. |
|
||||
| `ui/public/sw.js` | Service worker that handles `push` events and notification clicks. |
|
||||
| `push/vapid-keys.json` (under the OpenClaw state dir) | Auto-generated VAPID keypair used to sign Web Push payloads. |
|
||||
| `push/web-push-subscriptions.json` | Persisted browser subscription endpoints. |
|
||||
| Surface | What it does |
|
||||
| -------------------------------------------------- | ---------------------------------------------------------------------------- |
|
||||
| `ui/public/manifest.webmanifest` | PWA manifest. Browsers offer "Install app" once it is reachable. |
|
||||
| `ui/public/sw.js` | Service worker that handles `push` events and notification clicks. |
|
||||
| `state/openclaw.sqlite` → `web_push_vapid_keys` | Auto-generated VAPID keypair used to sign Web Push payloads. |
|
||||
| `state/openclaw.sqlite` → `web_push_subscriptions` | Persisted browser subscription endpoints, keys, and registration timestamps. |
|
||||
|
||||
Upgrades from the retired `push/vapid-keys.json` and `push/web-push-subscriptions.json` stores are imported by `openclaw doctor --fix`. Stop the Gateway before running that repair so an older process cannot recreate retired state during import. Run the repair before using Web Push after an upgrade; registration, delivery, deletion, and key resolution refuse to proceed while either retired source or an interrupted Doctor claim remains. The Gateway runtime reads and writes SQLite only.
|
||||
|
||||
Override the VAPID keypair through env vars on the Gateway process when you want to pin keys (multi-host deployments, secrets rotation, or tests):
|
||||
|
||||
|
||||
@@ -93,6 +93,7 @@ const legacyStorePatterns = [
|
||||
/\btui\/last-session\.json\b/u,
|
||||
/\bcommitments\/commitments\.json\b/u,
|
||||
/\bmedia\/outgoing\/records\/[^"'`]*\.json\b/u,
|
||||
/\bpush\/(?:web-push-subscriptions|vapid-keys)\.json\b/u,
|
||||
/\btmp\/skill-uploads\b/u,
|
||||
/\b(?:crestodian|openclaw)\/rescue-pending\/[^"'`]*\.json\b/u,
|
||||
/\bcron\/(?:runs\/[^"'`]+\.jsonl|jobs\.json|jobs-state\.json)\b/u,
|
||||
@@ -112,6 +113,7 @@ const allowedRuntimeMigrationPaths = [
|
||||
"src/infra/state-migrations.tui-last-session.ts",
|
||||
"src/infra/state-migrations.commitments.ts",
|
||||
"src/infra/state-migrations.managed-outgoing-images.ts",
|
||||
"src/infra/state-migrations.web-push.ts",
|
||||
"src/infra/state-migrations.rescue-pending.ts",
|
||||
"src/commands/session-state-migration.ts",
|
||||
"src/commands/doctor-state-migrations.test.ts",
|
||||
|
||||
@@ -288,6 +288,11 @@ function createLegacyStateMigrationDetectionResult(params?: {
|
||||
sourceDir: "/tmp/state/media/outgoing/records",
|
||||
hasLegacy: false,
|
||||
},
|
||||
webPush: {
|
||||
subscriptionsPath: "/tmp/state/push/web-push-subscriptions.json",
|
||||
vapidKeysPath: "/tmp/state/push/vapid-keys.json",
|
||||
hasLegacy: false,
|
||||
},
|
||||
rescuePending: {
|
||||
sourcePaths: ["/tmp/state/crestodian/rescue-pending", "/tmp/state/openclaw/rescue-pending"],
|
||||
hasLegacy: false,
|
||||
|
||||
@@ -0,0 +1,270 @@
|
||||
// Canonical shared-SQLite store for Web Push subscriptions and VAPID identity.
|
||||
import type { Insertable, Selectable } from "kysely";
|
||||
import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js";
|
||||
import {
|
||||
openOpenClawStateDatabase,
|
||||
runOpenClawStateWriteTransaction,
|
||||
type OpenClawStateDatabaseOptions,
|
||||
} from "../state/openclaw-state-db.js";
|
||||
import { sha256HexPrefix } from "./crypto-digest.js";
|
||||
import {
|
||||
executeSqliteQuerySync,
|
||||
executeSqliteQueryTakeFirstSync,
|
||||
getNodeSqliteKysely,
|
||||
} from "./kysely-sync.js";
|
||||
|
||||
export const WEB_PUSH_VAPID_KEY_ID = "default";
|
||||
export const DEFAULT_WEB_PUSH_VAPID_SUBJECT = "https://openclaw.ai";
|
||||
const WEB_PUSH_MAX_ENDPOINT_LENGTH = 2048;
|
||||
const WEB_PUSH_MAX_KEY_LENGTH = 512;
|
||||
|
||||
export type WebPushSubscription = {
|
||||
subscriptionId: string;
|
||||
endpoint: string;
|
||||
keys: { p256dh: string; auth: string };
|
||||
createdAtMs: number;
|
||||
updatedAtMs: number;
|
||||
};
|
||||
|
||||
export type VapidKeyPair = {
|
||||
publicKey: string;
|
||||
privateKey: string;
|
||||
subject: string;
|
||||
};
|
||||
|
||||
export function createWebPushVapidKeyPair(
|
||||
publicKey: string,
|
||||
privateKey: string,
|
||||
subject: string,
|
||||
): VapidKeyPair {
|
||||
return { publicKey, privateKey, subject };
|
||||
}
|
||||
|
||||
export type WebPushDatabase = Pick<
|
||||
OpenClawStateKyselyDatabase,
|
||||
"web_push_subscriptions" | "web_push_vapid_keys"
|
||||
>;
|
||||
type WebPushSubscriptionRow = Selectable<WebPushDatabase["web_push_subscriptions"]>;
|
||||
type WebPushSubscriptionInsert = Insertable<WebPushDatabase["web_push_subscriptions"]>;
|
||||
type WebPushVapidKeyInsert = Insertable<WebPushDatabase["web_push_vapid_keys"]>;
|
||||
|
||||
function webPushStateDatabaseOptions(stateDir?: string): OpenClawStateDatabaseOptions {
|
||||
return stateDir
|
||||
? { env: { ...process.env, OPENCLAW_STATE_DIR: stateDir } }
|
||||
: { env: process.env };
|
||||
}
|
||||
|
||||
export function hashWebPushEndpoint(endpoint: string): string {
|
||||
return sha256HexPrefix(endpoint, 32);
|
||||
}
|
||||
|
||||
export function isValidWebPushEndpoint(endpoint: string): boolean {
|
||||
if (!endpoint || endpoint.length > WEB_PUSH_MAX_ENDPOINT_LENGTH) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
return new URL(endpoint).protocol === "https:";
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
export function isValidWebPushKey(key: unknown): key is string {
|
||||
return typeof key === "string" && key.length > 0 && key.length <= WEB_PUSH_MAX_KEY_LENGTH;
|
||||
}
|
||||
|
||||
export function webPushSubscriptionFromRow(row: WebPushSubscriptionRow): WebPushSubscription {
|
||||
return {
|
||||
subscriptionId: row.subscription_id,
|
||||
endpoint: row.endpoint,
|
||||
keys: { p256dh: row.p256dh, auth: row.auth },
|
||||
createdAtMs: row.created_at_ms,
|
||||
updatedAtMs: row.updated_at_ms,
|
||||
};
|
||||
}
|
||||
|
||||
export function webPushSubscriptionToRow(params: {
|
||||
endpointHash: string;
|
||||
subscription: WebPushSubscription;
|
||||
}): WebPushSubscriptionInsert {
|
||||
return {
|
||||
endpoint_hash: params.endpointHash,
|
||||
subscription_id: params.subscription.subscriptionId,
|
||||
endpoint: params.subscription.endpoint,
|
||||
p256dh: params.subscription.keys.p256dh,
|
||||
auth: params.subscription.keys.auth,
|
||||
created_at_ms: params.subscription.createdAtMs,
|
||||
updated_at_ms: params.subscription.updatedAtMs,
|
||||
};
|
||||
}
|
||||
|
||||
export function webPushVapidKeyPairToRow(params: {
|
||||
keyPair: VapidKeyPair;
|
||||
nowMs: number;
|
||||
}): WebPushVapidKeyInsert {
|
||||
return {
|
||||
key_id: WEB_PUSH_VAPID_KEY_ID,
|
||||
public_key: params.keyPair.publicKey,
|
||||
private_key: params.keyPair.privateKey,
|
||||
subject: params.keyPair.subject,
|
||||
updated_at_ms: params.nowMs,
|
||||
};
|
||||
}
|
||||
|
||||
export function webPushSubscriptionsEqual(
|
||||
left: WebPushSubscription,
|
||||
right: WebPushSubscription,
|
||||
): boolean {
|
||||
return (
|
||||
left.subscriptionId === right.subscriptionId &&
|
||||
left.endpoint === right.endpoint &&
|
||||
left.keys.p256dh === right.keys.p256dh &&
|
||||
left.keys.auth === right.keys.auth &&
|
||||
left.createdAtMs === right.createdAtMs &&
|
||||
left.updatedAtMs === right.updatedAtMs
|
||||
);
|
||||
}
|
||||
|
||||
export function listWebPushSubscriptions(stateDir?: string): WebPushSubscription[] {
|
||||
const database = openOpenClawStateDatabase(webPushStateDatabaseOptions(stateDir));
|
||||
const stateDb = getNodeSqliteKysely<WebPushDatabase>(database.db);
|
||||
return executeSqliteQuerySync(
|
||||
database.db,
|
||||
stateDb
|
||||
.selectFrom("web_push_subscriptions")
|
||||
.selectAll()
|
||||
.orderBy("created_at_ms", "asc")
|
||||
.orderBy("subscription_id", "asc"),
|
||||
).rows.map(webPushSubscriptionFromRow);
|
||||
}
|
||||
|
||||
/** Reread the endpoint row inside the write transaction before creating or updating it. */
|
||||
export function upsertWebPushSubscription(params: {
|
||||
endpointHash: string;
|
||||
endpoint: string;
|
||||
keys: { p256dh: string; auth: string };
|
||||
candidateSubscriptionId: string;
|
||||
nowMs: number;
|
||||
stateDir?: string;
|
||||
}): WebPushSubscription {
|
||||
return runOpenClawStateWriteTransaction(({ db }) => {
|
||||
const stateDb = getNodeSqliteKysely<WebPushDatabase>(db);
|
||||
const existingRow = executeSqliteQueryTakeFirstSync(
|
||||
db,
|
||||
stateDb
|
||||
.selectFrom("web_push_subscriptions")
|
||||
.selectAll()
|
||||
.where("endpoint_hash", "=", params.endpointHash),
|
||||
);
|
||||
if (existingRow && existingRow.endpoint !== params.endpoint) {
|
||||
throw new Error("web push endpoint hash collision");
|
||||
}
|
||||
const subscription: WebPushSubscription = {
|
||||
subscriptionId: existingRow?.subscription_id ?? params.candidateSubscriptionId,
|
||||
endpoint: params.endpoint,
|
||||
keys: { ...params.keys },
|
||||
createdAtMs: existingRow?.created_at_ms ?? params.nowMs,
|
||||
updatedAtMs: params.nowMs,
|
||||
};
|
||||
const row = webPushSubscriptionToRow({
|
||||
endpointHash: params.endpointHash,
|
||||
subscription,
|
||||
});
|
||||
executeSqliteQuerySync(
|
||||
db,
|
||||
stateDb
|
||||
.insertInto("web_push_subscriptions")
|
||||
.values(row)
|
||||
.onConflict((conflict) =>
|
||||
conflict.column("endpoint_hash").doUpdateSet({
|
||||
subscription_id: row.subscription_id,
|
||||
endpoint: row.endpoint,
|
||||
p256dh: row.p256dh,
|
||||
auth: row.auth,
|
||||
updated_at_ms: row.updated_at_ms,
|
||||
}),
|
||||
),
|
||||
);
|
||||
return subscription;
|
||||
}, webPushStateDatabaseOptions(params.stateDir));
|
||||
}
|
||||
|
||||
export function deleteWebPushSubscriptionByEndpoint(params: {
|
||||
endpointHash: string;
|
||||
endpoint: string;
|
||||
stateDir?: string;
|
||||
}): boolean {
|
||||
return runOpenClawStateWriteTransaction(({ db }) => {
|
||||
const result = executeSqliteQuerySync(
|
||||
db,
|
||||
getNodeSqliteKysely<WebPushDatabase>(db)
|
||||
.deleteFrom("web_push_subscriptions")
|
||||
.where("endpoint_hash", "=", params.endpointHash)
|
||||
.where("endpoint", "=", params.endpoint),
|
||||
);
|
||||
return Number(result.numAffectedRows ?? 0) > 0;
|
||||
}, webPushStateDatabaseOptions(params.stateDir));
|
||||
}
|
||||
|
||||
/** Delete an expired send target only if no newer registration replaced it in flight. */
|
||||
export function deleteWebPushSubscriptionIfCurrent(params: {
|
||||
endpointHash: string;
|
||||
subscription: WebPushSubscription;
|
||||
stateDir?: string;
|
||||
}): boolean {
|
||||
const subscription = params.subscription;
|
||||
return runOpenClawStateWriteTransaction(({ db }) => {
|
||||
const result = executeSqliteQuerySync(
|
||||
db,
|
||||
getNodeSqliteKysely<WebPushDatabase>(db)
|
||||
.deleteFrom("web_push_subscriptions")
|
||||
.where("endpoint_hash", "=", params.endpointHash)
|
||||
.where("subscription_id", "=", subscription.subscriptionId)
|
||||
.where("endpoint", "=", subscription.endpoint)
|
||||
.where("p256dh", "=", subscription.keys.p256dh)
|
||||
.where("auth", "=", subscription.keys.auth)
|
||||
.where("updated_at_ms", "=", subscription.updatedAtMs),
|
||||
);
|
||||
return Number(result.numAffectedRows ?? 0) > 0;
|
||||
}, webPushStateDatabaseOptions(params.stateDir));
|
||||
}
|
||||
|
||||
export function readPersistedVapidKeyPair(stateDir?: string): VapidKeyPair | null {
|
||||
const database = openOpenClawStateDatabase(webPushStateDatabaseOptions(stateDir));
|
||||
const row = executeSqliteQueryTakeFirstSync(
|
||||
database.db,
|
||||
getNodeSqliteKysely<WebPushDatabase>(database.db)
|
||||
.selectFrom("web_push_vapid_keys")
|
||||
.selectAll()
|
||||
.where("key_id", "=", WEB_PUSH_VAPID_KEY_ID),
|
||||
);
|
||||
return row ? createWebPushVapidKeyPair(row.public_key, row.private_key, row.subject) : null;
|
||||
}
|
||||
|
||||
/** First committed keypair wins so concurrent gateway bootstraps share one signing identity. */
|
||||
export function insertVapidKeyPairIfAbsent(params: {
|
||||
candidate: VapidKeyPair;
|
||||
nowMs: number;
|
||||
stateDir?: string;
|
||||
}): VapidKeyPair {
|
||||
return runOpenClawStateWriteTransaction(({ db }) => {
|
||||
const stateDb = getNodeSqliteKysely<WebPushDatabase>(db);
|
||||
const existing = executeSqliteQueryTakeFirstSync(
|
||||
db,
|
||||
stateDb
|
||||
.selectFrom("web_push_vapid_keys")
|
||||
.selectAll()
|
||||
.where("key_id", "=", WEB_PUSH_VAPID_KEY_ID),
|
||||
);
|
||||
if (existing) {
|
||||
return createWebPushVapidKeyPair(existing.public_key, existing.private_key, existing.subject);
|
||||
}
|
||||
executeSqliteQuerySync(
|
||||
db,
|
||||
stateDb
|
||||
.insertInto("web_push_vapid_keys")
|
||||
.values(webPushVapidKeyPairToRow({ keyPair: params.candidate, nowMs: params.nowMs })),
|
||||
);
|
||||
return params.candidate;
|
||||
}, webPushStateDatabaseOptions(params.stateDir));
|
||||
}
|
||||
+255
-64
@@ -1,9 +1,16 @@
|
||||
// Tests web push subscription storage and delivery helpers.
|
||||
// Tests SQLite-backed Web Push subscription storage and delivery helpers.
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import webPush from "web-push";
|
||||
import { closeOpenClawStateDatabase } from "../state/openclaw-state-db.js";
|
||||
import { captureEnv, setTestEnvValue } from "../test-utils/env.js";
|
||||
import {
|
||||
createWebPushVapidKeyPair,
|
||||
listWebPushSubscriptions,
|
||||
readPersistedVapidKeyPair,
|
||||
} from "./push-web-store.js";
|
||||
import {
|
||||
broadcastWebPush,
|
||||
clearWebPushSubscriptionByEndpoint,
|
||||
@@ -11,19 +18,21 @@ import {
|
||||
resolveVapidKeys,
|
||||
} from "./push-web.js";
|
||||
|
||||
// Stub resolveStateDir so tests use a temp directory.
|
||||
let tmpDir: string;
|
||||
const generatedVapidKeys = vi.hoisted(
|
||||
() =>
|
||||
Object.fromEntries([
|
||||
["publicKey", "test-public-key-base64url"],
|
||||
["privateKey", "test-private-key-base64url"],
|
||||
]) as { publicKey: string; privateKey: string },
|
||||
);
|
||||
vi.mock("../config/paths.js", () => ({
|
||||
resolveStateDir: () => tmpDir,
|
||||
}));
|
||||
|
||||
// Stub web-push so we don't make real HTTP requests.
|
||||
vi.mock("web-push", () => ({
|
||||
default: {
|
||||
generateVAPIDKeys: vi.fn(() => ({
|
||||
publicKey: "test-public-key-base64url",
|
||||
privateKey: "test-private-key-base64url",
|
||||
})),
|
||||
generateVAPIDKeys: vi.fn(() => generatedVapidKeys),
|
||||
setVapidDetails: vi.fn(),
|
||||
sendNotification: vi.fn().mockResolvedValue({ statusCode: 201 }),
|
||||
},
|
||||
@@ -32,47 +41,99 @@ vi.mock("web-push", () => ({
|
||||
beforeEach(async () => {
|
||||
tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), "push-web-test-"));
|
||||
vi.clearAllMocks();
|
||||
vi.mocked(webPush.sendNotification).mockResolvedValue({ statusCode: 201 } as never);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
closeOpenClawStateDatabase();
|
||||
await fs.rm(tmpDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
describe("resolveVapidKeys", () => {
|
||||
it("generates and persists VAPID keys on first call", async () => {
|
||||
it("generates one durable SQLite VAPID identity", async () => {
|
||||
const keys = await resolveVapidKeys(tmpDir);
|
||||
expect(keys.publicKey).toBe("test-public-key-base64url");
|
||||
expect(keys.privateKey).toBe("test-private-key-base64url");
|
||||
expect(keys.subject).toBe("https://openclaw.ai");
|
||||
const persistedKeys = JSON.parse(
|
||||
await fs.readFile(path.join(tmpDir, "push", "vapid-keys.json"), "utf8"),
|
||||
) as { subject?: string };
|
||||
expect(persistedKeys.subject).toBe("https://openclaw.ai");
|
||||
expect(keys).toEqual(
|
||||
createWebPushVapidKeyPair(
|
||||
"test-public-key-base64url",
|
||||
"test-private-key-base64url",
|
||||
"https://openclaw.ai",
|
||||
),
|
||||
);
|
||||
expect(readPersistedVapidKeyPair(tmpDir)).toEqual(keys);
|
||||
|
||||
// Second call returns same keys.
|
||||
const keys2 = await resolveVapidKeys(tmpDir);
|
||||
expect(keys2.publicKey).toBe(keys.publicKey);
|
||||
expect(keys2.privateKey).toBe(keys.privateKey);
|
||||
closeOpenClawStateDatabase();
|
||||
await expect(resolveVapidKeys(tmpDir)).resolves.toEqual(keys);
|
||||
expect(vi.mocked(webPush.generateVAPIDKeys)).toHaveBeenCalledTimes(1);
|
||||
await expect(fs.stat(path.join(tmpDir, "push", "vapid-keys.json"))).rejects.toMatchObject({
|
||||
code: "ENOENT",
|
||||
});
|
||||
});
|
||||
|
||||
it("prefers env vars over persisted keys", async () => {
|
||||
// Persist keys first.
|
||||
await resolveVapidKeys(tmpDir);
|
||||
it("requires Doctor before creating an identity beside retired state", async () => {
|
||||
const pushDir = path.join(tmpDir, "push");
|
||||
const legacyPath = path.join(pushDir, "vapid-keys.json");
|
||||
await fs.mkdir(pushDir, { recursive: true });
|
||||
await fs.writeFile(legacyPath, "{}", "utf8");
|
||||
|
||||
// Set env overrides.
|
||||
process.env.OPENCLAW_VAPID_PUBLIC_KEY = "env-public";
|
||||
process.env.OPENCLAW_VAPID_PRIVATE_KEY = "env-private";
|
||||
process.env.OPENCLAW_VAPID_SUBJECT = "mailto:env@test.com";
|
||||
await expect(resolveVapidKeys(tmpDir)).rejects.toThrow("openclaw doctor --fix");
|
||||
expect(readPersistedVapidKeyPair(tmpDir)).toBeNull();
|
||||
expect(vi.mocked(webPush.generateVAPIDKeys)).not.toHaveBeenCalled();
|
||||
|
||||
await fs.rename(legacyPath, `${legacyPath}.doctor-importing`);
|
||||
await expect(resolveVapidKeys(tmpDir)).rejects.toThrow("openclaw doctor --fix");
|
||||
expect(vi.mocked(webPush.generateVAPIDKeys)).not.toHaveBeenCalled();
|
||||
|
||||
await fs.rm(`${legacyPath}.doctor-importing`);
|
||||
await fs.symlink(path.join(tmpDir, "missing-vapid-keys.json"), legacyPath);
|
||||
await expect(resolveVapidKeys(tmpDir)).rejects.toThrow("openclaw doctor --fix");
|
||||
expect(vi.mocked(webPush.generateVAPIDKeys)).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("converges concurrent first-use generation on the first committed identity", async () => {
|
||||
vi.mocked(webPush.generateVAPIDKeys)
|
||||
.mockReturnValueOnce(createWebPushVapidKeyPair("public-a", "private-a", "ignored"))
|
||||
.mockReturnValueOnce(createWebPushVapidKeyPair("public-b", "private-b", "ignored"));
|
||||
|
||||
const [first, second] = await Promise.all([resolveVapidKeys(tmpDir), resolveVapidKeys(tmpDir)]);
|
||||
|
||||
expect(first).toEqual(second);
|
||||
expect(readPersistedVapidKeyPair(tmpDir)).toEqual(first);
|
||||
expect(vi.mocked(webPush.generateVAPIDKeys)).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("prefers a complete environment override without persisting it", async () => {
|
||||
const environmentKeys = createWebPushVapidKeyPair(
|
||||
"env-public",
|
||||
"env-private",
|
||||
"mailto:env@test.com",
|
||||
);
|
||||
const envSnapshot = captureEnv([
|
||||
"OPENCLAW_VAPID_PUBLIC_KEY",
|
||||
"OPENCLAW_VAPID_PRIVATE_KEY",
|
||||
"OPENCLAW_VAPID_SUBJECT",
|
||||
]);
|
||||
setTestEnvValue("OPENCLAW_VAPID_PUBLIC_KEY", environmentKeys.publicKey);
|
||||
setTestEnvValue("OPENCLAW_VAPID_PRIVATE_KEY", environmentKeys.privateKey);
|
||||
setTestEnvValue("OPENCLAW_VAPID_SUBJECT", environmentKeys.subject);
|
||||
try {
|
||||
const keys = await resolveVapidKeys(tmpDir);
|
||||
expect(keys.publicKey).toBe("env-public");
|
||||
expect(keys.privateKey).toBe("env-private");
|
||||
expect(keys.subject).toBe("mailto:env@test.com");
|
||||
expect(vi.mocked(webPush.generateVAPIDKeys)).toHaveBeenCalledTimes(1);
|
||||
await expect(resolveVapidKeys(tmpDir)).resolves.toEqual(environmentKeys);
|
||||
expect(readPersistedVapidKeyPair(tmpDir)).toBeNull();
|
||||
expect(vi.mocked(webPush.generateVAPIDKeys)).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
envSnapshot.restore();
|
||||
}
|
||||
});
|
||||
|
||||
it("applies the current subject to a persisted identity", async () => {
|
||||
const initial = await resolveVapidKeys(tmpDir);
|
||||
process.env.OPENCLAW_VAPID_SUBJECT = "mailto:changed@test.com";
|
||||
try {
|
||||
await expect(resolveVapidKeys(tmpDir)).resolves.toEqual({
|
||||
...initial,
|
||||
subject: "mailto:changed@test.com",
|
||||
});
|
||||
expect(readPersistedVapidKeyPair(tmpDir)?.subject).toBe("https://openclaw.ai");
|
||||
} finally {
|
||||
delete process.env.OPENCLAW_VAPID_PUBLIC_KEY;
|
||||
delete process.env.OPENCLAW_VAPID_PRIVATE_KEY;
|
||||
delete process.env.OPENCLAW_VAPID_SUBJECT;
|
||||
}
|
||||
});
|
||||
@@ -82,43 +143,53 @@ describe("subscription CRUD", () => {
|
||||
const endpoint = "https://push.example.com/send/abc123";
|
||||
const keys = { p256dh: "p256dh-key", auth: "auth-key" };
|
||||
|
||||
it("registers a new subscription", async () => {
|
||||
const sub = await registerWebPushSubscription({
|
||||
endpoint,
|
||||
keys,
|
||||
baseDir: tmpDir,
|
||||
});
|
||||
expect(sub.subscriptionId).toMatch(/^[0-9a-f-]{36}$/);
|
||||
expect(sub.endpoint).toBe(endpoint);
|
||||
expect(sub.keys.p256dh).toBe("p256dh-key");
|
||||
expect(sub.keys.auth).toBe("auth-key");
|
||||
expect(sub.createdAtMs).toBeGreaterThan(0);
|
||||
});
|
||||
|
||||
it("updates an existing subscription with the same endpoint", async () => {
|
||||
const sub1 = await registerWebPushSubscription({
|
||||
endpoint,
|
||||
keys,
|
||||
baseDir: tmpDir,
|
||||
});
|
||||
const sub2 = await registerWebPushSubscription({
|
||||
it("registers, updates, and reopens a durable subscription", async () => {
|
||||
const first = await registerWebPushSubscription({ endpoint, keys, baseDir: tmpDir });
|
||||
const updated = await registerWebPushSubscription({
|
||||
endpoint,
|
||||
keys: { p256dh: "new-p256dh", auth: "new-auth" },
|
||||
baseDir: tmpDir,
|
||||
});
|
||||
// Same subscription ID, same created time, updated keys.
|
||||
expect(sub2.subscriptionId).toBe(sub1.subscriptionId);
|
||||
expect(sub2.createdAtMs).toBe(sub1.createdAtMs);
|
||||
expect(sub2.keys.p256dh).toBe("new-p256dh");
|
||||
expect(updated).toMatchObject({
|
||||
subscriptionId: first.subscriptionId,
|
||||
createdAtMs: first.createdAtMs,
|
||||
endpoint,
|
||||
keys: { p256dh: "new-p256dh", auth: "new-auth" },
|
||||
});
|
||||
|
||||
closeOpenClawStateDatabase();
|
||||
expect(listWebPushSubscriptions(tmpDir)).toEqual([updated]);
|
||||
await expect(fs.stat(path.join(tmpDir, "push"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
|
||||
it("clears a subscription by endpoint", async () => {
|
||||
it("preserves unrelated concurrent registrations", async () => {
|
||||
await Promise.all(
|
||||
["a", "b", "c"].map((suffix) =>
|
||||
registerWebPushSubscription({
|
||||
endpoint: `https://push.example.com/${suffix}`,
|
||||
keys,
|
||||
baseDir: tmpDir,
|
||||
}),
|
||||
),
|
||||
);
|
||||
expect(
|
||||
listWebPushSubscriptions(tmpDir)
|
||||
.map((entry) => entry.endpoint)
|
||||
.toSorted(),
|
||||
).toEqual([
|
||||
"https://push.example.com/a",
|
||||
"https://push.example.com/b",
|
||||
"https://push.example.com/c",
|
||||
]);
|
||||
});
|
||||
|
||||
it("clears only the matching endpoint", async () => {
|
||||
await registerWebPushSubscription({ endpoint, keys, baseDir: tmpDir });
|
||||
await expect(clearWebPushSubscriptionByEndpoint(endpoint, tmpDir)).resolves.toBe(true);
|
||||
await expect(clearWebPushSubscriptionByEndpoint(endpoint, tmpDir)).resolves.toBe(false);
|
||||
});
|
||||
|
||||
it("rejects invalid endpoint", async () => {
|
||||
it("rejects invalid registration data", async () => {
|
||||
await expect(
|
||||
registerWebPushSubscription({
|
||||
endpoint: "http://insecure.example.com",
|
||||
@@ -126,23 +197,66 @@ describe("subscription CRUD", () => {
|
||||
baseDir: tmpDir,
|
||||
}),
|
||||
).rejects.toThrow("invalid push subscription endpoint");
|
||||
});
|
||||
|
||||
it("rejects empty keys", async () => {
|
||||
await expect(
|
||||
registerWebPushSubscription({
|
||||
endpoint,
|
||||
keys: { p256dh: "", auth: "auth-key" },
|
||||
keys: { p256dh: "", auth: "auth" },
|
||||
baseDir: tmpDir,
|
||||
}),
|
||||
).rejects.toThrow("invalid push subscription keys");
|
||||
});
|
||||
|
||||
it("blocks an empty broadcast while retired subscriptions await Doctor", async () => {
|
||||
const pushDir = path.join(tmpDir, "push");
|
||||
const legacyPath = path.join(pushDir, "web-push-subscriptions.json");
|
||||
await fs.mkdir(pushDir, { recursive: true });
|
||||
await fs.writeFile(
|
||||
legacyPath,
|
||||
JSON.stringify({
|
||||
subscriptionsByEndpointHash: {
|
||||
legacy: {
|
||||
subscriptionId: "c0a80101-0000-4000-8000-000000000001",
|
||||
endpoint: "https://push.example.com/legacy",
|
||||
keys,
|
||||
createdAtMs: 1,
|
||||
updatedAtMs: 1,
|
||||
},
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
expect(listWebPushSubscriptions(tmpDir)).toEqual([]);
|
||||
await expect(broadcastWebPush({ title: "Blocked" }, tmpDir)).rejects.toThrow(
|
||||
"openclaw doctor --fix",
|
||||
);
|
||||
expect(vi.mocked(webPush.sendNotification)).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("blocks mutations while a Doctor claim is pending", async () => {
|
||||
const existing = await registerWebPushSubscription({ endpoint, keys, baseDir: tmpDir });
|
||||
const pushDir = path.join(tmpDir, "push");
|
||||
const claimPath = path.join(pushDir, "web-push-subscriptions.json.doctor-importing");
|
||||
await fs.mkdir(pushDir, { recursive: true });
|
||||
await fs.writeFile(claimPath, "{}", "utf8");
|
||||
|
||||
await expect(clearWebPushSubscriptionByEndpoint(endpoint, tmpDir)).rejects.toThrow(
|
||||
"openclaw doctor --fix",
|
||||
);
|
||||
await expect(
|
||||
registerWebPushSubscription({
|
||||
endpoint: "https://push.example.com/new",
|
||||
keys,
|
||||
baseDir: tmpDir,
|
||||
}),
|
||||
).rejects.toThrow("openclaw doctor --fix");
|
||||
expect(listWebPushSubscriptions(tmpDir)).toEqual([existing]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("sending", () => {
|
||||
const keys = { p256dh: "p256dh-key", auth: "auth-key" };
|
||||
|
||||
it("configures VAPID details once before broadcasting to subscribers", async () => {
|
||||
it("configures VAPID details once before broadcasting", async () => {
|
||||
await registerWebPushSubscription({
|
||||
endpoint: "https://push.example.com/a",
|
||||
keys,
|
||||
@@ -161,4 +275,81 @@ describe("sending", () => {
|
||||
expect(vi.mocked(webPush.setVapidDetails)).toHaveBeenCalledTimes(1);
|
||||
expect(vi.mocked(webPush.sendNotification)).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("does not delete a subscription re-registered during an expired send", async () => {
|
||||
const endpoint = "https://push.example.com/reregistered";
|
||||
await registerWebPushSubscription({ endpoint, keys, baseDir: tmpDir });
|
||||
let rejectSend: ((error: unknown) => void) | undefined;
|
||||
vi.mocked(webPush.sendNotification).mockImplementationOnce(
|
||||
() =>
|
||||
new Promise((_, reject) => {
|
||||
rejectSend = reject;
|
||||
}),
|
||||
);
|
||||
|
||||
const broadcast = broadcastWebPush({ title: "Race" }, tmpDir);
|
||||
await vi.waitFor(() => expect(rejectSend).toBeTypeOf("function"));
|
||||
const replacement = await registerWebPushSubscription({
|
||||
endpoint,
|
||||
keys: { p256dh: "replacement-p256dh", auth: "replacement-auth" },
|
||||
baseDir: tmpDir,
|
||||
});
|
||||
rejectSend?.(Object.assign(new Error("gone"), { statusCode: 410 }));
|
||||
await broadcast;
|
||||
|
||||
expect(listWebPushSubscriptions(tmpDir)).toEqual([replacement]);
|
||||
});
|
||||
|
||||
it("does not delete an expired subscription after a legacy claim appears", async () => {
|
||||
const endpoint = "https://push.example.com/pending-claim";
|
||||
const subscription = await registerWebPushSubscription({ endpoint, keys, baseDir: tmpDir });
|
||||
let rejectSend: ((error: unknown) => void) | undefined;
|
||||
vi.mocked(webPush.sendNotification).mockImplementationOnce(
|
||||
() =>
|
||||
new Promise((_, reject) => {
|
||||
rejectSend = reject;
|
||||
}),
|
||||
);
|
||||
|
||||
const broadcast = broadcastWebPush({ title: "Race" }, tmpDir);
|
||||
await vi.waitFor(() => expect(rejectSend).toBeTypeOf("function"));
|
||||
const pushDir = path.join(tmpDir, "push");
|
||||
await fs.mkdir(pushDir, { recursive: true });
|
||||
await fs.writeFile(
|
||||
path.join(pushDir, "web-push-subscriptions.json.doctor-importing"),
|
||||
"{}",
|
||||
"utf8",
|
||||
);
|
||||
rejectSend?.(Object.assign(new Error("gone"), { statusCode: 410 }));
|
||||
|
||||
await expect(broadcast).resolves.toEqual([
|
||||
expect.objectContaining({ ok: false, statusCode: 410 }),
|
||||
]);
|
||||
expect(listWebPushSubscriptions(tmpDir)).toEqual([subscription]);
|
||||
});
|
||||
|
||||
it("keeps completed delivery results when expired-subscription cleanup fails", async () => {
|
||||
const endpoint = "https://push.example.com/expired";
|
||||
await registerWebPushSubscription({ endpoint, keys, baseDir: tmpDir });
|
||||
await resolveVapidKeys(tmpDir);
|
||||
let rejectSend: ((error: unknown) => void) | undefined;
|
||||
vi.mocked(webPush.sendNotification).mockImplementationOnce(
|
||||
() =>
|
||||
new Promise((_, reject) => {
|
||||
rejectSend = reject;
|
||||
}),
|
||||
);
|
||||
|
||||
const broadcast = broadcastWebPush({ title: "Expired" }, tmpDir);
|
||||
await vi.waitFor(() => expect(rejectSend).toBeTypeOf("function"));
|
||||
closeOpenClawStateDatabase();
|
||||
const databasePath = path.join(tmpDir, "state", "openclaw.sqlite");
|
||||
await fs.rename(databasePath, `${databasePath}.backup`);
|
||||
await fs.mkdir(databasePath);
|
||||
rejectSend?.(Object.assign(new Error("gone"), { statusCode: 410 }));
|
||||
|
||||
await expect(broadcast).resolves.toEqual([
|
||||
expect.objectContaining({ ok: false, statusCode: 410 }),
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
+90
-124
@@ -1,27 +1,28 @@
|
||||
// Stores and verifies web push subscriptions and delivery payloads.
|
||||
import { randomUUID } from "node:crypto";
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import { expectDefined } from "@openclaw/normalization-core";
|
||||
import type { PushSubscription, VapidKeys } from "web-push";
|
||||
import { resolveStateDir } from "../config/paths.js";
|
||||
import { createLazyRuntimeModule } from "../shared/lazy-runtime.js";
|
||||
import { sha256HexPrefix } from "./crypto-digest.js";
|
||||
import { createAsyncLock, tryReadJson, writeJson } from "./json-files.js";
|
||||
import {
|
||||
createWebPushVapidKeyPair,
|
||||
deleteWebPushSubscriptionByEndpoint,
|
||||
deleteWebPushSubscriptionIfCurrent,
|
||||
hashWebPushEndpoint,
|
||||
insertVapidKeyPairIfAbsent,
|
||||
isValidWebPushEndpoint,
|
||||
isValidWebPushKey,
|
||||
listWebPushSubscriptions,
|
||||
readPersistedVapidKeyPair,
|
||||
upsertWebPushSubscription,
|
||||
DEFAULT_WEB_PUSH_VAPID_SUBJECT,
|
||||
type VapidKeyPair,
|
||||
type WebPushSubscription,
|
||||
} from "./push-web-store.js";
|
||||
|
||||
// --- Types ---
|
||||
|
||||
type WebPushSubscription = PushSubscription & {
|
||||
subscriptionId: string;
|
||||
createdAtMs: number;
|
||||
updatedAtMs: number;
|
||||
};
|
||||
|
||||
type WebPushRegistrationState = {
|
||||
subscriptionsByEndpointHash: Record<string, WebPushSubscription>;
|
||||
};
|
||||
|
||||
type VapidKeyPair = VapidKeys & { subject: string };
|
||||
|
||||
type WebPushSendResult = {
|
||||
ok: boolean;
|
||||
subscriptionId: string;
|
||||
@@ -31,13 +32,7 @@ type WebPushSendResult = {
|
||||
|
||||
// --- Constants ---
|
||||
|
||||
const WEB_PUSH_STATE_FILENAME = "push/web-push-subscriptions.json";
|
||||
const VAPID_KEYS_FILENAME = "push/vapid-keys.json";
|
||||
const MAX_ENDPOINT_LENGTH = 2048;
|
||||
const MAX_KEY_LENGTH = 512;
|
||||
const DEFAULT_VAPID_SUBJECT = "https://openclaw.ai";
|
||||
|
||||
const withLock = createAsyncLock();
|
||||
const LEGACY_WEB_PUSH_PATHS = ["push/web-push-subscriptions.json", "push/vapid-keys.json"] as const;
|
||||
|
||||
type WebPushRuntime = typeof import("web-push");
|
||||
type WebPushRuntimeModule = WebPushRuntime & { default?: WebPushRuntime };
|
||||
@@ -46,54 +41,40 @@ const loadWebPushRuntime = createLazyRuntimeModule(() =>
|
||||
import("web-push").then((mod: WebPushRuntimeModule) => mod.default ?? mod),
|
||||
);
|
||||
|
||||
// --- Helpers ---
|
||||
|
||||
function resolveWebPushStatePath(baseDir?: string): string {
|
||||
const root = baseDir ?? resolveStateDir();
|
||||
return path.join(root, WEB_PUSH_STATE_FILENAME);
|
||||
}
|
||||
|
||||
function resolveVapidKeysPath(baseDir?: string): string {
|
||||
const root = baseDir ?? resolveStateDir();
|
||||
return path.join(root, VAPID_KEYS_FILENAME);
|
||||
}
|
||||
|
||||
function hashEndpoint(endpoint: string): string {
|
||||
return sha256HexPrefix(endpoint, 32);
|
||||
}
|
||||
|
||||
function isValidEndpoint(endpoint: string): boolean {
|
||||
if (!endpoint || endpoint.length > MAX_ENDPOINT_LENGTH) {
|
||||
return false;
|
||||
}
|
||||
function legacyWebPushPathMayExist(filePath: string): boolean {
|
||||
try {
|
||||
const url = new URL(endpoint);
|
||||
return url.protocol === "https:";
|
||||
} catch {
|
||||
return false;
|
||||
fs.lstatSync(filePath);
|
||||
return true;
|
||||
} catch (error) {
|
||||
// Only a definite absence permits creating a new signing identity.
|
||||
return (error as NodeJS.ErrnoException).code !== "ENOENT";
|
||||
}
|
||||
}
|
||||
|
||||
function isValidKey(key: string): boolean {
|
||||
return typeof key === "string" && key.length > 0 && key.length <= MAX_KEY_LENGTH;
|
||||
}
|
||||
|
||||
// --- State persistence ---
|
||||
|
||||
async function loadState(baseDir?: string): Promise<WebPushRegistrationState> {
|
||||
const filePath = resolveWebPushStatePath(baseDir);
|
||||
const state = await tryReadJson<WebPushRegistrationState>(filePath);
|
||||
return state ?? { subscriptionsByEndpointHash: {} };
|
||||
}
|
||||
|
||||
async function persistState(state: WebPushRegistrationState, baseDir?: string): Promise<void> {
|
||||
const filePath = resolveWebPushStatePath(baseDir);
|
||||
await writeJson(filePath, state, { trailingNewline: true });
|
||||
// Production callers run under the Gateway's lifetime state/config lock. Doctor must
|
||||
// acquire those same locks before claiming legacy files, so this check remains stable
|
||||
// through the following SQLite operation or asynchronous delivery fan-out.
|
||||
function assertLegacyWebPushMigrationComplete(baseDir?: string): void {
|
||||
const stateDir = baseDir ?? resolveStateDir();
|
||||
const pendingLegacyPath = LEGACY_WEB_PUSH_PATHS.find((relativePath) => {
|
||||
const sourcePath = path.join(stateDir, relativePath);
|
||||
return (
|
||||
legacyWebPushPathMayExist(sourcePath) ||
|
||||
legacyWebPushPathMayExist(`${sourcePath}.doctor-importing`)
|
||||
);
|
||||
});
|
||||
if (pendingLegacyPath) {
|
||||
throw new Error(
|
||||
`legacy Web Push state requires migration; run \`openclaw doctor --fix\` before using Web Push`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// --- VAPID keys ---
|
||||
|
||||
export async function resolveVapidKeys(baseDir?: string): Promise<VapidKeyPair> {
|
||||
assertLegacyWebPushMigrationComplete(baseDir);
|
||||
|
||||
// Env vars take precedence — allows operators to share a stable VAPID
|
||||
// identity across multiple gateway instances.
|
||||
const envPublic = resolveVapidPublicKeyFromEnv();
|
||||
@@ -106,34 +87,29 @@ export async function resolveVapidKeys(baseDir?: string): Promise<VapidKeyPair>
|
||||
};
|
||||
}
|
||||
|
||||
// Fall back to persisted keys, generating on first use under a lock to
|
||||
// prevent concurrent bootstraps from writing different keypairs.
|
||||
return await withLock(async () => {
|
||||
const filePath = resolveVapidKeysPath(baseDir);
|
||||
const existing = await tryReadJson<VapidKeyPair>(filePath);
|
||||
if (existing?.publicKey && existing?.privateKey) {
|
||||
return {
|
||||
publicKey: existing.publicKey,
|
||||
privateKey: existing.privateKey,
|
||||
// Env var always wins so operators can change subject without deleting vapid-keys.json.
|
||||
subject: resolveVapidSubjectFromEnv(),
|
||||
};
|
||||
}
|
||||
const existing = readPersistedVapidKeyPair(baseDir);
|
||||
if (existing) {
|
||||
return { ...existing, subject: resolveVapidSubjectFromEnv() };
|
||||
}
|
||||
|
||||
const webPush = await loadWebPushRuntime();
|
||||
const keys = webPush.generateVAPIDKeys();
|
||||
const pair: VapidKeyPair = {
|
||||
publicKey: keys.publicKey,
|
||||
privateKey: keys.privateKey,
|
||||
subject: resolveVapidSubjectFromEnv(),
|
||||
};
|
||||
await writeJson(filePath, pair, { trailingNewline: true });
|
||||
return pair;
|
||||
// Generation can race across gateway processes. SQLite selects one durable
|
||||
// identity, then every contender returns that committed keypair.
|
||||
const webPush = await loadWebPushRuntime();
|
||||
const keys = webPush.generateVAPIDKeys();
|
||||
const pair = insertVapidKeyPairIfAbsent({
|
||||
candidate: createWebPushVapidKeyPair(
|
||||
keys.publicKey,
|
||||
keys.privateKey,
|
||||
resolveVapidSubjectFromEnv(),
|
||||
),
|
||||
nowMs: Date.now(),
|
||||
stateDir: baseDir,
|
||||
});
|
||||
return { ...pair, subject: resolveVapidSubjectFromEnv() };
|
||||
}
|
||||
|
||||
function resolveVapidSubjectFromEnv(): string {
|
||||
return process.env.OPENCLAW_VAPID_SUBJECT || DEFAULT_VAPID_SUBJECT;
|
||||
return process.env.OPENCLAW_VAPID_SUBJECT || DEFAULT_WEB_PUSH_VAPID_SUBJECT;
|
||||
}
|
||||
|
||||
function resolveVapidPublicKeyFromEnv(): string | undefined {
|
||||
@@ -157,51 +133,33 @@ export async function registerWebPushSubscription(
|
||||
): Promise<WebPushSubscription> {
|
||||
const { endpoint, keys, baseDir } = params;
|
||||
|
||||
if (!isValidEndpoint(endpoint)) {
|
||||
if (!isValidWebPushEndpoint(endpoint)) {
|
||||
throw new Error("invalid push subscription endpoint: must be an HTTPS URL under 2048 chars");
|
||||
}
|
||||
if (!isValidKey(keys.p256dh) || !isValidKey(keys.auth)) {
|
||||
if (!isValidWebPushKey(keys.p256dh) || !isValidWebPushKey(keys.auth)) {
|
||||
throw new Error("invalid push subscription keys: must be non-empty strings under 512 chars");
|
||||
}
|
||||
assertLegacyWebPushMigrationComplete(baseDir);
|
||||
|
||||
return await withLock(async () => {
|
||||
const state = await loadState(baseDir);
|
||||
const hash = hashEndpoint(endpoint);
|
||||
const now = Date.now();
|
||||
|
||||
const existing = state.subscriptionsByEndpointHash[hash];
|
||||
const subscription: WebPushSubscription = {
|
||||
subscriptionId: existing?.subscriptionId ?? randomUUID(),
|
||||
endpoint,
|
||||
keys: { p256dh: keys.p256dh, auth: keys.auth },
|
||||
createdAtMs: existing?.createdAtMs ?? now,
|
||||
updatedAtMs: now,
|
||||
};
|
||||
|
||||
state.subscriptionsByEndpointHash[hash] = subscription;
|
||||
await persistState(state, baseDir);
|
||||
return subscription;
|
||||
return upsertWebPushSubscription({
|
||||
endpointHash: hashWebPushEndpoint(endpoint),
|
||||
endpoint,
|
||||
keys: { p256dh: keys.p256dh, auth: keys.auth },
|
||||
candidateSubscriptionId: randomUUID(),
|
||||
nowMs: Date.now(),
|
||||
stateDir: baseDir,
|
||||
});
|
||||
}
|
||||
|
||||
async function listWebPushSubscriptions(baseDir?: string): Promise<WebPushSubscription[]> {
|
||||
const state = await loadState(baseDir);
|
||||
return Object.values(state.subscriptionsByEndpointHash);
|
||||
}
|
||||
|
||||
export async function clearWebPushSubscriptionByEndpoint(
|
||||
endpoint: string,
|
||||
baseDir?: string,
|
||||
): Promise<boolean> {
|
||||
return await withLock(async () => {
|
||||
const state = await loadState(baseDir);
|
||||
const hash = hashEndpoint(endpoint);
|
||||
if (state.subscriptionsByEndpointHash[hash]) {
|
||||
delete state.subscriptionsByEndpointHash[hash];
|
||||
await persistState(state, baseDir);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
assertLegacyWebPushMigrationComplete(baseDir);
|
||||
return deleteWebPushSubscriptionByEndpoint({
|
||||
endpointHash: hashWebPushEndpoint(endpoint),
|
||||
endpoint,
|
||||
stateDir: baseDir,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -260,7 +218,8 @@ export async function broadcastWebPush(
|
||||
payload: WebPushPayload,
|
||||
baseDir?: string,
|
||||
): Promise<WebPushSendResult[]> {
|
||||
const subscriptions = await listWebPushSubscriptions(baseDir);
|
||||
assertLegacyWebPushMigrationComplete(baseDir);
|
||||
const subscriptions = listWebPushSubscriptions(baseDir);
|
||||
if (subscriptions.length === 0) {
|
||||
return [];
|
||||
}
|
||||
@@ -287,15 +246,22 @@ export async function broadcastWebPush(
|
||||
);
|
||||
|
||||
// Clean up expired subscriptions (HTTP 410 Gone or 404 Not Found) per Web Push spec.
|
||||
const expiredEndpoints = mapped
|
||||
const expiredSubscriptions = mapped
|
||||
.map((result, i) => ({ result, sub: subscriptions[i] }))
|
||||
.filter(({ result }) => !result.ok && (result.statusCode === 410 || result.statusCode === 404))
|
||||
.map(({ sub }) => expectDefined(sub, "push web sub").endpoint);
|
||||
.map(({ sub }) => expectDefined(sub, "push web sub"));
|
||||
|
||||
if (expiredEndpoints.length > 0) {
|
||||
await Promise.allSettled(
|
||||
expiredEndpoints.map((endpoint) => clearWebPushSubscriptionByEndpoint(endpoint, baseDir)),
|
||||
);
|
||||
for (const subscription of expiredSubscriptions) {
|
||||
try {
|
||||
assertLegacyWebPushMigrationComplete(baseDir);
|
||||
deleteWebPushSubscriptionIfCurrent({
|
||||
endpointHash: hashWebPushEndpoint(subscription.endpoint),
|
||||
subscription,
|
||||
stateDir: baseDir,
|
||||
});
|
||||
} catch {
|
||||
// Delivery already completed. Cleanup stays best-effort so callers do not retry valid sends.
|
||||
}
|
||||
}
|
||||
|
||||
return mapped;
|
||||
|
||||
@@ -119,6 +119,7 @@ import {
|
||||
migrateLegacyUpdateCheckState,
|
||||
resolveLegacyUpdateCheckPath,
|
||||
} from "./state-migrations.update-check.js";
|
||||
import { detectLegacyWebPush, migrateLegacyWebPush } from "./state-migrations.web-push.js";
|
||||
|
||||
let autoMigrateChecked = false;
|
||||
|
||||
@@ -397,6 +398,10 @@ export async function detectLegacyStateMigrations(params: {
|
||||
stateDir,
|
||||
doctorOnlyStateMigrations: params.doctorOnlyStateMigrations,
|
||||
});
|
||||
const webPush = detectLegacyWebPush({
|
||||
stateDir,
|
||||
doctorOnlyStateMigrations: params.doctorOnlyStateMigrations,
|
||||
});
|
||||
const rescuePending = detectLegacyRescuePending({
|
||||
stateDir,
|
||||
doctorOnlyStateMigrations: params.doctorOnlyStateMigrations,
|
||||
@@ -555,6 +560,9 @@ export async function detectLegacyStateMigrations(params: {
|
||||
if (managedOutgoingImages.hasLegacy) {
|
||||
preview.push("- Managed outgoing images: legacy record JSON → shared SQLite state");
|
||||
}
|
||||
if (webPush.hasLegacy) {
|
||||
preview.push("- Web Push subscriptions and VAPID identity: legacy JSON → shared SQLite state");
|
||||
}
|
||||
if (rescuePending.hasLegacy) {
|
||||
preview.push("- System-agent rescue approvals: discard retired pending JSON capabilities");
|
||||
}
|
||||
@@ -646,6 +654,7 @@ export async function detectLegacyStateMigrations(params: {
|
||||
tuiLastSessions,
|
||||
commitments,
|
||||
managedOutgoingImages,
|
||||
webPush,
|
||||
rescuePending,
|
||||
channelPairing,
|
||||
execApprovals,
|
||||
@@ -837,6 +846,11 @@ export async function runLegacyStateMigrations(params: {
|
||||
detected: detected.managedOutgoingImages,
|
||||
stateDir: detected.stateDir,
|
||||
});
|
||||
const webPush = await migrateLegacyWebPush({
|
||||
detected: detected.webPush,
|
||||
env,
|
||||
stateDir: detected.stateDir,
|
||||
});
|
||||
const rescuePending = discardLegacyRescuePending({
|
||||
detected: detected.rescuePending,
|
||||
stateDir: detected.stateDir,
|
||||
@@ -874,6 +888,7 @@ export async function runLegacyStateMigrations(params: {
|
||||
tuiLastSessions,
|
||||
commitments,
|
||||
managedOutgoingImages,
|
||||
webPush,
|
||||
pluginPlans,
|
||||
]);
|
||||
return {
|
||||
@@ -892,6 +907,7 @@ export async function runLegacyStateMigrations(params: {
|
||||
...tuiLastSessions.changes,
|
||||
...commitments.changes,
|
||||
...managedOutgoingImages.changes,
|
||||
...webPush.changes,
|
||||
...rescuePending.changes,
|
||||
...channelPairing.changes,
|
||||
...execApprovals.changes,
|
||||
@@ -918,6 +934,7 @@ export async function runLegacyStateMigrations(params: {
|
||||
...tuiLastSessions.warnings,
|
||||
...commitments.warnings,
|
||||
...managedOutgoingImages.warnings,
|
||||
...webPush.warnings,
|
||||
...rescuePending.warnings,
|
||||
...channelPairing.warnings,
|
||||
...execApprovals.warnings,
|
||||
|
||||
@@ -20,6 +20,12 @@ import {
|
||||
executeSqliteQueryTakeFirstSync,
|
||||
getNodeSqliteKysely,
|
||||
} from "./kysely-sync.js";
|
||||
import {
|
||||
createWebPushVapidKeyPair,
|
||||
hashWebPushEndpoint,
|
||||
listWebPushSubscriptions,
|
||||
readPersistedVapidKeyPair,
|
||||
} from "./push-web-store.js";
|
||||
import {
|
||||
autoMigrateLegacyState,
|
||||
autoMigrateLegacyPluginDoctorState,
|
||||
@@ -2068,6 +2074,62 @@ describe("state migrations", () => {
|
||||
await expect(fs.access(path.join(canonicalStateDir, "legacy.txt"))).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
it("routes explicit Doctor repair through the Web Push SQLite importer", async () => {
|
||||
const root = await createTempDir();
|
||||
const stateDir = path.join(root, ".openclaw");
|
||||
const env = createEnv(stateDir);
|
||||
const cfg = createConfig();
|
||||
const endpoint = "https://push.example.com/doctor-integration";
|
||||
const subscription = {
|
||||
subscriptionId: "c0a80101-0000-4000-8000-000000000001",
|
||||
endpoint,
|
||||
keys: { p256dh: "doctor-p256dh", auth: "doctor-auth" },
|
||||
createdAtMs: 1,
|
||||
updatedAtMs: 2,
|
||||
};
|
||||
const pushDir = path.join(stateDir, "push");
|
||||
const subscriptionsPath = path.join(pushDir, "web-push-subscriptions.json");
|
||||
const vapidKeysPath = path.join(pushDir, "vapid-keys.json");
|
||||
await fs.mkdir(pushDir, { recursive: true });
|
||||
await fs.writeFile(
|
||||
subscriptionsPath,
|
||||
JSON.stringify({
|
||||
subscriptionsByEndpointHash: {
|
||||
[hashWebPushEndpoint(endpoint)]: subscription,
|
||||
},
|
||||
}),
|
||||
"utf8",
|
||||
);
|
||||
await fs.writeFile(
|
||||
vapidKeysPath,
|
||||
JSON.stringify(
|
||||
createWebPushVapidKeyPair("doctor-public", "doctor-private", "https://openclaw.ai"),
|
||||
),
|
||||
"utf8",
|
||||
);
|
||||
|
||||
const detected = await detectLegacyStateMigrations({
|
||||
cfg,
|
||||
env,
|
||||
homedir: () => root,
|
||||
doctorOnlyStateMigrations: true,
|
||||
});
|
||||
expect(detected.webPush.hasLegacy).toBe(true);
|
||||
expect(detected.preview).toContain(
|
||||
"- Web Push subscriptions and VAPID identity: legacy JSON → shared SQLite state",
|
||||
);
|
||||
|
||||
const result = await runLegacyStateMigrations({ detected, config: cfg, env });
|
||||
|
||||
expect(result.warnings).toStrictEqual([]);
|
||||
expect(listWebPushSubscriptions(stateDir)).toStrictEqual([subscription]);
|
||||
expect(readPersistedVapidKeyPair(stateDir)).toStrictEqual(
|
||||
createWebPushVapidKeyPair("doctor-public", "doctor-private", "https://openclaw.ai"),
|
||||
);
|
||||
await expectMissingPath(subscriptionsPath);
|
||||
await expectMissingPath(vapidKeysPath);
|
||||
});
|
||||
|
||||
it("migrates legacy update-check JSON into shared SQLite state", async () => {
|
||||
const root = await createTempDir();
|
||||
const stateDir = path.join(root, ".openclaw");
|
||||
|
||||
@@ -104,6 +104,11 @@ export type LegacyStateDetection = {
|
||||
sourceDir: string;
|
||||
hasLegacy: boolean;
|
||||
};
|
||||
webPush: {
|
||||
subscriptionsPath: string;
|
||||
vapidKeysPath: string;
|
||||
hasLegacy: boolean;
|
||||
};
|
||||
rescuePending: LegacyRescuePendingDetection;
|
||||
channelPairing: LegacyChannelPairingStateDetection;
|
||||
execApprovals: {
|
||||
|
||||
@@ -0,0 +1,543 @@
|
||||
// Covers fail-closed Doctor import of the retired Web Push JSON stores.
|
||||
import fs from "node:fs";
|
||||
import fsp from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js";
|
||||
import {
|
||||
closeOpenClawStateDatabaseForTest,
|
||||
openOpenClawStateDatabase,
|
||||
} from "../state/openclaw-state-db.js";
|
||||
import { captureEnv, setTestEnvValue } from "../test-utils/env.js";
|
||||
import { acquireGatewayLock } from "./gateway-lock.js";
|
||||
import { executeSqliteQuerySync, getNodeSqliteKysely } from "./kysely-sync.js";
|
||||
import {
|
||||
createWebPushVapidKeyPair,
|
||||
hashWebPushEndpoint,
|
||||
listWebPushSubscriptions,
|
||||
readPersistedVapidKeyPair,
|
||||
webPushSubscriptionToRow,
|
||||
webPushVapidKeyPairToRow,
|
||||
DEFAULT_WEB_PUSH_VAPID_SUBJECT,
|
||||
type VapidKeyPair,
|
||||
type WebPushDatabase,
|
||||
type WebPushSubscription,
|
||||
} from "./push-web-store.js";
|
||||
import { detectLegacyWebPush, migrateLegacyWebPush } from "./state-migrations.web-push.js";
|
||||
|
||||
describe("legacy Web Push Doctor migration", () => {
|
||||
let envSnapshot: ReturnType<typeof captureEnv> | undefined;
|
||||
const tempDirs = useAutoCleanupTempDirTracker((cleanup) => {
|
||||
afterEach(() => {
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
envSnapshot?.restore();
|
||||
envSnapshot = undefined;
|
||||
cleanup();
|
||||
});
|
||||
});
|
||||
|
||||
function useStateDir(): string {
|
||||
const stateDir = tempDirs.make("openclaw-web-push-migration-");
|
||||
envSnapshot ??= captureEnv(["OPENCLAW_STATE_DIR"]);
|
||||
setTestEnvValue("OPENCLAW_STATE_DIR", stateDir);
|
||||
return stateDir;
|
||||
}
|
||||
|
||||
function subscription(overrides: Partial<WebPushSubscription> = {}): WebPushSubscription {
|
||||
return {
|
||||
subscriptionId: "c0a80101-0000-4000-8000-000000000001",
|
||||
endpoint: "https://push.example.com/send/legacy",
|
||||
keys: { p256dh: "legacy-p256dh", auth: "legacy-auth" },
|
||||
createdAtMs: 1_000,
|
||||
updatedAtMs: 2_000,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function vapidKeys(overrides: Partial<VapidKeyPair> = {}): VapidKeyPair {
|
||||
return {
|
||||
...createWebPushVapidKeyPair(
|
||||
"legacy-public-key",
|
||||
"legacy-private-key",
|
||||
"https://openclaw.ai",
|
||||
),
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
async function writeLegacyState(params: {
|
||||
stateDir: string;
|
||||
subscriptions?: unknown;
|
||||
vapid?: unknown;
|
||||
}): Promise<{ subscriptionsPath?: string; vapidKeysPath?: string }> {
|
||||
const pushDir = path.join(params.stateDir, "push");
|
||||
await fsp.mkdir(pushDir, { recursive: true });
|
||||
const result: { subscriptionsPath?: string; vapidKeysPath?: string } = {};
|
||||
if (params.subscriptions !== undefined) {
|
||||
result.subscriptionsPath = path.join(pushDir, "web-push-subscriptions.json");
|
||||
const subscriptionsByEndpointHash = Array.isArray(params.subscriptions)
|
||||
? Object.fromEntries(
|
||||
(params.subscriptions as readonly WebPushSubscription[]).map((entry) => [
|
||||
hashWebPushEndpoint(entry.endpoint),
|
||||
entry,
|
||||
]),
|
||||
)
|
||||
: params.subscriptions;
|
||||
await fsp.writeFile(
|
||||
result.subscriptionsPath,
|
||||
JSON.stringify({ subscriptionsByEndpointHash }, null, 2),
|
||||
"utf8",
|
||||
);
|
||||
}
|
||||
if (params.vapid !== undefined) {
|
||||
result.vapidKeysPath = path.join(pushDir, "vapid-keys.json");
|
||||
await fsp.writeFile(result.vapidKeysPath, JSON.stringify(params.vapid, null, 2), "utf8");
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
function seedSubscription(endpointHash: string, value: WebPushSubscription): void {
|
||||
const database = openOpenClawStateDatabase();
|
||||
executeSqliteQuerySync(
|
||||
database.db,
|
||||
getNodeSqliteKysely<WebPushDatabase>(database.db)
|
||||
.insertInto("web_push_subscriptions")
|
||||
.values(webPushSubscriptionToRow({ endpointHash, subscription: value })),
|
||||
);
|
||||
}
|
||||
|
||||
function seedVapid(value: VapidKeyPair): void {
|
||||
const database = openOpenClawStateDatabase();
|
||||
executeSqliteQuerySync(
|
||||
database.db,
|
||||
getNodeSqliteKysely<WebPushDatabase>(database.db)
|
||||
.insertInto("web_push_vapid_keys")
|
||||
.values(webPushVapidKeyPairToRow({ keyPair: value, nowMs: 1 })),
|
||||
);
|
||||
}
|
||||
|
||||
it("detects original and interrupted-claim files only for explicit Doctor repair", async () => {
|
||||
const stateDir = useStateDir();
|
||||
const { subscriptionsPath } = await writeLegacyState({
|
||||
stateDir,
|
||||
subscriptions: [],
|
||||
});
|
||||
expect(detectLegacyWebPush({ stateDir }).hasLegacy).toBe(false);
|
||||
expect(detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }).hasLegacy).toBe(true);
|
||||
|
||||
await fsp.rename(subscriptionsPath!, `${subscriptionsPath}.doctor-importing`);
|
||||
expect(detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }).hasLegacy).toBe(true);
|
||||
});
|
||||
|
||||
it("requires exclusive state ownership before reading or committing legacy state", async () => {
|
||||
const stateDir = useStateDir();
|
||||
const { subscriptionsPath } = await writeLegacyState({
|
||||
stateDir,
|
||||
subscriptions: [subscription()],
|
||||
});
|
||||
const env = { ...process.env, OPENCLAW_STATE_DIR: stateDir };
|
||||
const gatewayLock = await acquireGatewayLock({
|
||||
allowInTests: true,
|
||||
env,
|
||||
pollIntervalMs: 10,
|
||||
port: 18_789,
|
||||
timeoutMs: 100,
|
||||
});
|
||||
if (!gatewayLock) {
|
||||
throw new Error("expected test Gateway lock");
|
||||
}
|
||||
|
||||
let blocked: Awaited<ReturnType<typeof migrateLegacyWebPush>>;
|
||||
try {
|
||||
blocked = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
env,
|
||||
stateDir,
|
||||
});
|
||||
} finally {
|
||||
await gatewayLock.release();
|
||||
}
|
||||
|
||||
expect(blocked.warnings[0]).toContain("Gateway or another SQLite maintenance command");
|
||||
expect(fs.existsSync(subscriptionsPath!)).toBe(true);
|
||||
expect(listWebPushSubscriptions(stateDir)).toEqual([]);
|
||||
|
||||
const retry = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
env,
|
||||
stateDir,
|
||||
});
|
||||
expect(retry.warnings).toEqual([]);
|
||||
expect(listWebPushSubscriptions(stateDir)).toEqual([subscription()]);
|
||||
expect(fs.existsSync(subscriptionsPath!)).toBe(false);
|
||||
});
|
||||
|
||||
it("imports subscriptions and VAPID identity in one verified operation", async () => {
|
||||
const stateDir = useStateDir();
|
||||
const first = subscription();
|
||||
const second = subscription({
|
||||
subscriptionId: "c0a80101-0000-4000-8000-000000000002",
|
||||
endpoint: "https://push.example.com/send/second",
|
||||
});
|
||||
const paths = await writeLegacyState({
|
||||
stateDir,
|
||||
subscriptions: [first, second],
|
||||
vapid: vapidKeys(),
|
||||
});
|
||||
|
||||
const result = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
});
|
||||
|
||||
expect(result.warnings).toEqual([]);
|
||||
expect(listWebPushSubscriptions(stateDir)).toEqual([first, second]);
|
||||
expect(readPersistedVapidKeyPair(stateDir)).toEqual(vapidKeys());
|
||||
expect(fs.existsSync(paths.subscriptionsPath!)).toBe(false);
|
||||
expect(fs.existsSync(paths.vapidKeysPath!)).toBe(false);
|
||||
});
|
||||
|
||||
it.each([
|
||||
["missing", undefined],
|
||||
["empty", ""],
|
||||
])("normalizes a %s legacy VAPID subject", async (_label, subject) => {
|
||||
const stateDir = useStateDir();
|
||||
const legacyKeys = vapidKeys({ subject: subject ?? "" });
|
||||
if (subject === undefined) {
|
||||
delete (legacyKeys as Partial<VapidKeyPair>).subject;
|
||||
}
|
||||
await writeLegacyState({ stateDir, vapid: legacyKeys });
|
||||
|
||||
const result = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
});
|
||||
|
||||
expect(result.warnings).toEqual([]);
|
||||
expect(readPersistedVapidKeyPair(stateDir)?.subject).toBe(DEFAULT_WEB_PUSH_VAPID_SUBJECT);
|
||||
});
|
||||
|
||||
it("removes an empty valid store only after opening SQLite", async () => {
|
||||
const stateDir = useStateDir();
|
||||
const { subscriptionsPath } = await writeLegacyState({ stateDir, subscriptions: [] });
|
||||
|
||||
const result = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
});
|
||||
|
||||
expect(result.warnings).toEqual([]);
|
||||
expect(fs.existsSync(path.join(stateDir, "state", "openclaw.sqlite"))).toBe(true);
|
||||
expect(fs.existsSync(subscriptionsPath!)).toBe(false);
|
||||
});
|
||||
|
||||
it("rejects either malformed file without importing its valid pair", async () => {
|
||||
const stateDir = useStateDir();
|
||||
const paths = await writeLegacyState({
|
||||
stateDir,
|
||||
subscriptions: [subscription()],
|
||||
vapid: { publicKey: "incomplete" },
|
||||
});
|
||||
|
||||
const result = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
});
|
||||
|
||||
expect(result.warnings[0]).toContain("VAPID keys are invalid");
|
||||
expect(listWebPushSubscriptions(stateDir)).toEqual([]);
|
||||
expect(readPersistedVapidKeyPair(stateDir)).toBeNull();
|
||||
expect(fs.existsSync(paths.subscriptionsPath!)).toBe(true);
|
||||
expect(fs.existsSync(paths.vapidKeysPath!)).toBe(true);
|
||||
});
|
||||
|
||||
it("rejects a forged endpoint hash and duplicate subscription ids", async () => {
|
||||
const stateDir = useStateDir();
|
||||
const pushDir = path.join(stateDir, "push");
|
||||
await fsp.mkdir(pushDir, { recursive: true });
|
||||
const sourcePath = path.join(pushDir, "web-push-subscriptions.json");
|
||||
await fsp.writeFile(
|
||||
sourcePath,
|
||||
JSON.stringify({ subscriptionsByEndpointHash: { forged: subscription() } }),
|
||||
"utf8",
|
||||
);
|
||||
let result = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
});
|
||||
expect(result.warnings[0]).toContain("subscription is invalid");
|
||||
|
||||
const first = subscription();
|
||||
const second = subscription({ endpoint: "https://push.example.com/send/second" });
|
||||
await fsp.writeFile(
|
||||
sourcePath,
|
||||
JSON.stringify({
|
||||
subscriptionsByEndpointHash: {
|
||||
[hashWebPushEndpoint(first.endpoint)]: first,
|
||||
[hashWebPushEndpoint(second.endpoint)]: second,
|
||||
},
|
||||
}),
|
||||
"utf8",
|
||||
);
|
||||
result = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
});
|
||||
expect(result.warnings[0]).toContain("duplicate subscription id");
|
||||
expect(listWebPushSubscriptions(stateDir)).toEqual([]);
|
||||
});
|
||||
|
||||
it("keeps newer SQLite fields while preserving the earliest creation time", async () => {
|
||||
const stateDir = useStateDir();
|
||||
const legacy = subscription({ createdAtMs: 100, updatedAtMs: 200 });
|
||||
const canonical = subscription({
|
||||
keys: { p256dh: "canonical-p256dh", auth: "canonical-auth" },
|
||||
createdAtMs: 150,
|
||||
updatedAtMs: 300,
|
||||
});
|
||||
seedSubscription(hashWebPushEndpoint(canonical.endpoint), canonical);
|
||||
await writeLegacyState({ stateDir, subscriptions: [legacy] });
|
||||
|
||||
const result = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
});
|
||||
|
||||
expect(result.warnings).toEqual([]);
|
||||
expect(listWebPushSubscriptions(stateDir)).toEqual([{ ...canonical, createdAtMs: 100 }]);
|
||||
});
|
||||
|
||||
it("updates an older SQLite row from newer legacy state", async () => {
|
||||
const stateDir = useStateDir();
|
||||
const canonical = subscription({ updatedAtMs: 200 });
|
||||
const legacy = subscription({
|
||||
keys: { p256dh: "newer-p256dh", auth: "newer-auth" },
|
||||
createdAtMs: 500,
|
||||
updatedAtMs: 600,
|
||||
});
|
||||
seedSubscription(hashWebPushEndpoint(canonical.endpoint), canonical);
|
||||
await writeLegacyState({ stateDir, subscriptions: [legacy] });
|
||||
|
||||
const result = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
});
|
||||
|
||||
expect(result.warnings).toEqual([]);
|
||||
expect(listWebPushSubscriptions(stateDir)).toEqual([legacy]);
|
||||
});
|
||||
|
||||
it("retries a committed newer-row merge after normalizing its creation time", async () => {
|
||||
const stateDir = useStateDir();
|
||||
const canonical = subscription({ createdAtMs: 100, updatedAtMs: 200 });
|
||||
const legacy = subscription({
|
||||
keys: { p256dh: "newer-p256dh", auth: "newer-auth" },
|
||||
createdAtMs: 500,
|
||||
updatedAtMs: 600,
|
||||
});
|
||||
seedSubscription(hashWebPushEndpoint(canonical.endpoint), canonical);
|
||||
const { subscriptionsPath } = await writeLegacyState({
|
||||
stateDir,
|
||||
subscriptions: [legacy],
|
||||
});
|
||||
|
||||
const first = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
removeSource: () => {
|
||||
throw new Error("simulated unlink failure");
|
||||
},
|
||||
});
|
||||
expect(first.warnings[0]).toContain("legacy cleanup failed");
|
||||
expect(listWebPushSubscriptions(stateDir)).toEqual([{ ...legacy, createdAtMs: 100 }]);
|
||||
|
||||
const retry = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
});
|
||||
expect(retry.warnings).toEqual([]);
|
||||
expect(listWebPushSubscriptions(stateDir)).toEqual([{ ...legacy, createdAtMs: 100 }]);
|
||||
expect(fs.existsSync(`${subscriptionsPath}.doctor-importing`)).toBe(false);
|
||||
});
|
||||
|
||||
it("rolls back equal-timestamp divergence and VAPID identity conflicts", async () => {
|
||||
const stateDir = useStateDir();
|
||||
const canonical = subscription({ keys: { p256dh: "canonical", auth: "canonical" } });
|
||||
seedSubscription(hashWebPushEndpoint(canonical.endpoint), canonical);
|
||||
seedVapid(
|
||||
createWebPushVapidKeyPair("canonical-public", "canonical-private", "https://openclaw.ai"),
|
||||
);
|
||||
const paths = await writeLegacyState({
|
||||
stateDir,
|
||||
subscriptions: [subscription()],
|
||||
vapid: vapidKeys(),
|
||||
});
|
||||
|
||||
const result = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
});
|
||||
|
||||
expect(result.warnings[0]).toContain("diverges at the same timestamp");
|
||||
expect(listWebPushSubscriptions(stateDir)).toEqual([canonical]);
|
||||
expect(readPersistedVapidKeyPair(stateDir)?.publicKey).toBe("canonical-public");
|
||||
expect(fs.existsSync(paths.subscriptionsPath!)).toBe(true);
|
||||
expect(fs.existsSync(paths.vapidKeysPath!)).toBe(true);
|
||||
expect(fs.existsSync(`${paths.subscriptionsPath}.doctor-importing`)).toBe(false);
|
||||
expect(fs.existsSync(`${paths.vapidKeysPath}.doctor-importing`)).toBe(false);
|
||||
});
|
||||
|
||||
it("rolls back subscription changes when only VAPID conflicts", async () => {
|
||||
const stateDir = useStateDir();
|
||||
seedVapid(
|
||||
createWebPushVapidKeyPair("canonical-public", "canonical-private", "https://openclaw.ai"),
|
||||
);
|
||||
await writeLegacyState({
|
||||
stateDir,
|
||||
subscriptions: [subscription()],
|
||||
vapid: vapidKeys(),
|
||||
});
|
||||
|
||||
const result = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
});
|
||||
|
||||
expect(result.warnings[0]).toContain("VAPID identity conflicts");
|
||||
expect(listWebPushSubscriptions(stateDir)).toEqual([]);
|
||||
});
|
||||
|
||||
it("rejects a subscription id already owned by another endpoint", async () => {
|
||||
const stateDir = useStateDir();
|
||||
const canonical = subscription({ endpoint: "https://push.example.com/canonical" });
|
||||
const legacy = subscription({ endpoint: "https://push.example.com/legacy" });
|
||||
seedSubscription(hashWebPushEndpoint(canonical.endpoint), canonical);
|
||||
const { subscriptionsPath } = await writeLegacyState({
|
||||
stateDir,
|
||||
subscriptions: [legacy],
|
||||
});
|
||||
|
||||
const result = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
});
|
||||
|
||||
expect(result.warnings[0]).toContain("subscription id conflicts");
|
||||
expect(listWebPushSubscriptions(stateDir)).toEqual([canonical]);
|
||||
expect(fs.existsSync(subscriptionsPath!)).toBe(true);
|
||||
});
|
||||
|
||||
it("fails before database mutation when a source changes after parsing", async () => {
|
||||
const stateDir = useStateDir();
|
||||
const { subscriptionsPath } = await writeLegacyState({
|
||||
stateDir,
|
||||
subscriptions: [subscription()],
|
||||
});
|
||||
|
||||
const result = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
beforeVerify: () => fs.appendFileSync(subscriptionsPath!, "\n"),
|
||||
});
|
||||
|
||||
expect(result.warnings[0]).toContain("source changed");
|
||||
expect(listWebPushSubscriptions(stateDir)).toEqual([]);
|
||||
expect(fs.existsSync(subscriptionsPath!)).toBe(true);
|
||||
});
|
||||
|
||||
it("restores claimed sources without database mutation when claim verification fails", async () => {
|
||||
const stateDir = useStateDir();
|
||||
const { subscriptionsPath } = await writeLegacyState({
|
||||
stateDir,
|
||||
subscriptions: [subscription()],
|
||||
});
|
||||
|
||||
const result = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
beforeClaim: () => fs.appendFileSync(subscriptionsPath!, "\n"),
|
||||
});
|
||||
|
||||
expect(result.warnings[0]).toContain("source changed before doctor could claim it");
|
||||
expect(listWebPushSubscriptions(stateDir)).toEqual([]);
|
||||
expect(fs.existsSync(subscriptionsPath!)).toBe(true);
|
||||
expect(fs.existsSync(`${subscriptionsPath}.doctor-importing`)).toBe(false);
|
||||
});
|
||||
|
||||
it("retains fixed claims on cleanup failure and retries idempotently", async () => {
|
||||
const stateDir = useStateDir();
|
||||
const paths = await writeLegacyState({
|
||||
stateDir,
|
||||
subscriptions: [subscription()],
|
||||
vapid: vapidKeys(),
|
||||
});
|
||||
const first = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
removeSource: () => {
|
||||
throw new Error("simulated unlink failure");
|
||||
},
|
||||
});
|
||||
expect(first.warnings[0]).toContain("legacy cleanup failed");
|
||||
expect(fs.existsSync(`${paths.subscriptionsPath}.doctor-importing`)).toBe(true);
|
||||
expect(fs.existsSync(`${paths.vapidKeysPath}.doctor-importing`)).toBe(true);
|
||||
expect(listWebPushSubscriptions(stateDir)).toEqual([subscription()]);
|
||||
|
||||
const retry = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
});
|
||||
expect(retry.warnings).toEqual([]);
|
||||
expect(fs.existsSync(`${paths.subscriptionsPath}.doctor-importing`)).toBe(false);
|
||||
expect(fs.existsSync(`${paths.vapidKeysPath}.doctor-importing`)).toBe(false);
|
||||
expect(listWebPushSubscriptions(stateDir)).toEqual([subscription()]);
|
||||
});
|
||||
|
||||
it("refuses symlinked sources", async () => {
|
||||
const stateDir = useStateDir();
|
||||
const outside = path.join(stateDir, "outside.json");
|
||||
await fsp.writeFile(outside, JSON.stringify({ subscriptionsByEndpointHash: {} }), "utf8");
|
||||
const sourcePath = path.join(stateDir, "push", "web-push-subscriptions.json");
|
||||
await fsp.mkdir(path.dirname(sourcePath), { recursive: true });
|
||||
await fsp.symlink(outside, sourcePath);
|
||||
|
||||
const result = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
});
|
||||
|
||||
expect(result.warnings[0]).toContain("Failed reading legacy Web Push state");
|
||||
expect(fs.lstatSync(sourcePath).isSymbolicLink()).toBe(true);
|
||||
expect(listWebPushSubscriptions(stateDir)).toEqual([]);
|
||||
});
|
||||
|
||||
it("refuses a legacy store reached through a symlinked state-directory ancestor", async () => {
|
||||
if (process.platform === "win32") {
|
||||
return;
|
||||
}
|
||||
const stateDir = useStateDir();
|
||||
const outside = tempDirs.make("openclaw-web-push-outside-");
|
||||
const legacy = subscription();
|
||||
const sourcePath = path.join(outside, "web-push-subscriptions.json");
|
||||
await fsp.writeFile(
|
||||
sourcePath,
|
||||
JSON.stringify({
|
||||
subscriptionsByEndpointHash: {
|
||||
[hashWebPushEndpoint(legacy.endpoint)]: legacy,
|
||||
},
|
||||
}),
|
||||
"utf8",
|
||||
);
|
||||
await fsp.mkdir(stateDir, { recursive: true });
|
||||
await fsp.symlink(outside, path.join(stateDir, "push"));
|
||||
|
||||
const result = await migrateLegacyWebPush({
|
||||
detected: detectLegacyWebPush({ stateDir, doctorOnlyStateMigrations: true }),
|
||||
stateDir,
|
||||
});
|
||||
|
||||
expect(result.warnings[0]).toContain("Failed reading legacy Web Push state");
|
||||
expect(fs.existsSync(sourcePath)).toBe(true);
|
||||
expect(listWebPushSubscriptions(stateDir)).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,741 @@
|
||||
// Doctor-only import for the retired Web Push JSON stores.
|
||||
import { createHash } from "node:crypto";
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import type { DatabaseSync } from "node:sqlite";
|
||||
import { root, type Root } from "@openclaw/fs-safe";
|
||||
import { isRecord } from "@openclaw/normalization-core/record-coerce";
|
||||
import { runOpenClawStateWriteTransaction } from "../state/openclaw-state-db.js";
|
||||
import { formatErrorMessage } from "./errors.js";
|
||||
import { acquireGatewayLock, GatewayLockError } from "./gateway-lock.js";
|
||||
import {
|
||||
executeSqliteQuerySync,
|
||||
executeSqliteQueryTakeFirstSync,
|
||||
getNodeSqliteKysely,
|
||||
} from "./kysely-sync.js";
|
||||
import {
|
||||
createWebPushVapidKeyPair,
|
||||
hashWebPushEndpoint,
|
||||
isValidWebPushEndpoint,
|
||||
isValidWebPushKey,
|
||||
webPushSubscriptionFromRow,
|
||||
webPushSubscriptionToRow,
|
||||
webPushSubscriptionsEqual,
|
||||
webPushVapidKeyPairToRow,
|
||||
DEFAULT_WEB_PUSH_VAPID_SUBJECT,
|
||||
WEB_PUSH_VAPID_KEY_ID,
|
||||
type VapidKeyPair,
|
||||
type WebPushDatabase,
|
||||
type WebPushSubscription,
|
||||
} from "./push-web-store.js";
|
||||
import type { LegacyStateDetection, MigrationMessages } from "./state-migrations.types.js";
|
||||
|
||||
const LEGACY_SUBSCRIPTIONS_MAX_BYTES = 4 * 1024 * 1024;
|
||||
const LEGACY_VAPID_KEYS_MAX_BYTES = 64 * 1024;
|
||||
const MIGRATION_LOCK_TIMEOUT_MS = 250;
|
||||
const MIGRATION_LOCK_POLL_INTERVAL_MS = 25;
|
||||
const DOCTOR_CLAIM_SUFFIX = ".doctor-importing";
|
||||
const SUBSCRIPTION_STORE_KEYS = new Set(["subscriptionsByEndpointHash"]);
|
||||
const SUBSCRIPTION_KEYS = new Set([
|
||||
"subscriptionId",
|
||||
"endpoint",
|
||||
"keys",
|
||||
"createdAtMs",
|
||||
"updatedAtMs",
|
||||
]);
|
||||
const PUSH_KEYS = new Set(["p256dh", "auth"]);
|
||||
const VAPID_KEYS = new Set(["publicKey", "privateKey", "subject"]);
|
||||
const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
|
||||
|
||||
type LegacySourceSnapshot = {
|
||||
sourcePath: string;
|
||||
dev: number;
|
||||
ino: number;
|
||||
mtimeMs: number;
|
||||
raw: string;
|
||||
sha256: string;
|
||||
size: number;
|
||||
};
|
||||
|
||||
type ParsedLegacyState = {
|
||||
subscriptions: Map<string, WebPushSubscription>;
|
||||
vapidKeys: VapidKeyPair | null;
|
||||
snapshots: LegacySourceSnapshot[];
|
||||
};
|
||||
|
||||
function resolveLegacyWebPushPaths(stateDir: string): {
|
||||
subscriptionsPath: string;
|
||||
vapidKeysPath: string;
|
||||
} {
|
||||
const pushDir = path.join(stateDir, "push");
|
||||
return {
|
||||
subscriptionsPath: path.join(pushDir, "web-push-subscriptions.json"),
|
||||
vapidKeysPath: path.join(pushDir, "vapid-keys.json"),
|
||||
};
|
||||
}
|
||||
|
||||
function legacyPathMayExist(filePath: string): boolean {
|
||||
try {
|
||||
fs.lstatSync(filePath);
|
||||
return true;
|
||||
} catch (error) {
|
||||
return (error as NodeJS.ErrnoException).code !== "ENOENT";
|
||||
}
|
||||
}
|
||||
|
||||
function relativeLegacyPath(stateDir: string, filePath: string): string {
|
||||
const relativePath = path.relative(path.resolve(stateDir), path.resolve(filePath));
|
||||
if (
|
||||
!relativePath ||
|
||||
relativePath === ".." ||
|
||||
relativePath.startsWith(`..${path.sep}`) ||
|
||||
path.isAbsolute(relativePath)
|
||||
) {
|
||||
throw new Error(`legacy Web Push path is outside the state directory: ${filePath}`);
|
||||
}
|
||||
return relativePath;
|
||||
}
|
||||
|
||||
function sourceOrClaimMayExist(sourcePath: string): boolean {
|
||||
return (
|
||||
legacyPathMayExist(sourcePath) || legacyPathMayExist(`${sourcePath}${DOCTOR_CLAIM_SUFFIX}`)
|
||||
);
|
||||
}
|
||||
|
||||
/** Detect retired Web Push state only when an explicit doctor flow opts in. */
|
||||
export function detectLegacyWebPush(params: {
|
||||
stateDir: string;
|
||||
doctorOnlyStateMigrations?: boolean;
|
||||
}): LegacyStateDetection["webPush"] {
|
||||
const paths = resolveLegacyWebPushPaths(params.stateDir);
|
||||
return {
|
||||
...paths,
|
||||
hasLegacy:
|
||||
params.doctorOnlyStateMigrations === true &&
|
||||
(sourceOrClaimMayExist(paths.subscriptionsPath) ||
|
||||
sourceOrClaimMayExist(paths.vapidKeysPath)),
|
||||
};
|
||||
}
|
||||
|
||||
async function readLegacySourceSnapshot(
|
||||
stateRoot: Root,
|
||||
stateDir: string,
|
||||
sourcePath: string,
|
||||
maxBytes: number,
|
||||
): Promise<LegacySourceSnapshot> {
|
||||
const opened = await stateRoot.read(relativeLegacyPath(stateDir, sourcePath), {
|
||||
hardlinks: "reject",
|
||||
maxBytes,
|
||||
symlinks: "reject",
|
||||
});
|
||||
const raw = opened.buffer.toString("utf8");
|
||||
return {
|
||||
sourcePath,
|
||||
dev: opened.stat.dev,
|
||||
ino: opened.stat.ino,
|
||||
mtimeMs: opened.stat.mtimeMs,
|
||||
raw,
|
||||
sha256: createHash("sha256").update(raw).digest("hex"),
|
||||
size: opened.stat.size,
|
||||
};
|
||||
}
|
||||
|
||||
function sourceSnapshotsMatch(left: LegacySourceSnapshot, right: LegacySourceSnapshot): boolean {
|
||||
return (
|
||||
left.dev === right.dev &&
|
||||
left.ino === right.ino &&
|
||||
left.mtimeMs === right.mtimeMs &&
|
||||
left.sha256 === right.sha256 &&
|
||||
left.size === right.size
|
||||
);
|
||||
}
|
||||
|
||||
function contentSnapshotsMatch(left: LegacySourceSnapshot, right: LegacySourceSnapshot): boolean {
|
||||
return left.sha256 === right.sha256 && left.size === right.size;
|
||||
}
|
||||
|
||||
function maxBytesForSource(sourcePath: string, subscriptionsPath: string): number {
|
||||
return sourcePath === subscriptionsPath
|
||||
? LEGACY_SUBSCRIPTIONS_MAX_BYTES
|
||||
: LEGACY_VAPID_KEYS_MAX_BYTES;
|
||||
}
|
||||
|
||||
async function recoverInterruptedClaim(
|
||||
stateRoot: Root,
|
||||
stateDir: string,
|
||||
sourcePath: string,
|
||||
maxBytes: number,
|
||||
): Promise<void> {
|
||||
const claimPath = `${sourcePath}${DOCTOR_CLAIM_SUFFIX}`;
|
||||
const claimRelativePath = relativeLegacyPath(stateDir, claimPath);
|
||||
const sourceRelativePath = relativeLegacyPath(stateDir, sourcePath);
|
||||
if (!(await stateRoot.exists(claimRelativePath))) {
|
||||
return;
|
||||
}
|
||||
const claim = await readLegacySourceSnapshot(stateRoot, stateDir, claimPath, maxBytes);
|
||||
if (!(await stateRoot.exists(sourceRelativePath))) {
|
||||
await stateRoot.move(claimRelativePath, sourceRelativePath);
|
||||
return;
|
||||
}
|
||||
const source = await readLegacySourceSnapshot(stateRoot, stateDir, sourcePath, maxBytes);
|
||||
if (!contentSnapshotsMatch(claim, source)) {
|
||||
throw new Error("interrupted Web Push doctor claim conflicts with its source");
|
||||
}
|
||||
await stateRoot.remove(claimRelativePath);
|
||||
}
|
||||
|
||||
function assertOnlyKeys(
|
||||
value: Record<string, unknown>,
|
||||
allowed: ReadonlySet<string>,
|
||||
label: string,
|
||||
) {
|
||||
const unexpected = Object.keys(value).find((key) => !allowed.has(key));
|
||||
if (unexpected) {
|
||||
throw new Error(`${label} has unexpected field ${unexpected}`);
|
||||
}
|
||||
}
|
||||
|
||||
function parseLegacySubscriptions(raw: string): Map<string, WebPushSubscription> {
|
||||
const parsed = JSON.parse(raw) as unknown;
|
||||
if (!isRecord(parsed) || !isRecord(parsed.subscriptionsByEndpointHash)) {
|
||||
throw new Error("legacy Web Push subscriptions must be an object");
|
||||
}
|
||||
assertOnlyKeys(parsed, SUBSCRIPTION_STORE_KEYS, "legacy Web Push subscriptions store");
|
||||
|
||||
const subscriptions = new Map<string, WebPushSubscription>();
|
||||
const subscriptionIds = new Set<string>();
|
||||
for (const [endpointHash, rawSubscription] of Object.entries(
|
||||
parsed.subscriptionsByEndpointHash,
|
||||
)) {
|
||||
if (!isRecord(rawSubscription) || !isRecord(rawSubscription.keys)) {
|
||||
throw new Error("legacy Web Push subscription is not an object");
|
||||
}
|
||||
assertOnlyKeys(rawSubscription, SUBSCRIPTION_KEYS, "legacy Web Push subscription");
|
||||
assertOnlyKeys(rawSubscription.keys, PUSH_KEYS, "legacy Web Push subscription keys");
|
||||
const { subscriptionId, endpoint, createdAtMs, updatedAtMs } = rawSubscription;
|
||||
const p256dh = rawSubscription.keys.p256dh;
|
||||
const auth = rawSubscription.keys.auth;
|
||||
if (
|
||||
typeof subscriptionId !== "string" ||
|
||||
!UUID_RE.test(subscriptionId) ||
|
||||
typeof endpoint !== "string" ||
|
||||
!isValidWebPushEndpoint(endpoint) ||
|
||||
hashWebPushEndpoint(endpoint) !== endpointHash ||
|
||||
!isValidWebPushKey(p256dh) ||
|
||||
!isValidWebPushKey(auth) ||
|
||||
typeof createdAtMs !== "number" ||
|
||||
!Number.isSafeInteger(createdAtMs) ||
|
||||
createdAtMs < 0 ||
|
||||
typeof updatedAtMs !== "number" ||
|
||||
!Number.isSafeInteger(updatedAtMs) ||
|
||||
updatedAtMs < createdAtMs
|
||||
) {
|
||||
throw new Error("legacy Web Push subscription is invalid");
|
||||
}
|
||||
if (subscriptionIds.has(subscriptionId)) {
|
||||
throw new Error("legacy Web Push subscriptions contain a duplicate subscription id");
|
||||
}
|
||||
subscriptionIds.add(subscriptionId);
|
||||
subscriptions.set(endpointHash, {
|
||||
subscriptionId,
|
||||
endpoint,
|
||||
keys: { p256dh, auth },
|
||||
createdAtMs,
|
||||
updatedAtMs,
|
||||
});
|
||||
}
|
||||
return subscriptions;
|
||||
}
|
||||
|
||||
function parseLegacyVapidKeys(raw: string): VapidKeyPair {
|
||||
const parsed = JSON.parse(raw) as unknown;
|
||||
if (!isRecord(parsed)) {
|
||||
throw new Error("legacy Web Push VAPID keys must be an object");
|
||||
}
|
||||
assertOnlyKeys(parsed, VAPID_KEYS, "legacy Web Push VAPID keys");
|
||||
const subject =
|
||||
parsed.subject === undefined || parsed.subject === ""
|
||||
? process.env.OPENCLAW_VAPID_SUBJECT || DEFAULT_WEB_PUSH_VAPID_SUBJECT
|
||||
: parsed.subject;
|
||||
if (
|
||||
!isValidWebPushKey(parsed.publicKey) ||
|
||||
!isValidWebPushKey(parsed.privateKey) ||
|
||||
typeof subject !== "string" ||
|
||||
subject.length > 512
|
||||
) {
|
||||
throw new Error("legacy Web Push VAPID keys are invalid");
|
||||
}
|
||||
return createWebPushVapidKeyPair(parsed.publicKey, parsed.privateKey, subject);
|
||||
}
|
||||
|
||||
async function readLegacyState(
|
||||
stateRoot: Root,
|
||||
stateDir: string,
|
||||
detected: LegacyStateDetection["webPush"],
|
||||
): Promise<ParsedLegacyState> {
|
||||
await recoverInterruptedClaim(
|
||||
stateRoot,
|
||||
stateDir,
|
||||
detected.subscriptionsPath,
|
||||
LEGACY_SUBSCRIPTIONS_MAX_BYTES,
|
||||
);
|
||||
await recoverInterruptedClaim(
|
||||
stateRoot,
|
||||
stateDir,
|
||||
detected.vapidKeysPath,
|
||||
LEGACY_VAPID_KEYS_MAX_BYTES,
|
||||
);
|
||||
const snapshots: LegacySourceSnapshot[] = [];
|
||||
let subscriptions = new Map<string, WebPushSubscription>();
|
||||
let vapidKeys: VapidKeyPair | null = null;
|
||||
if (await stateRoot.exists(relativeLegacyPath(stateDir, detected.subscriptionsPath))) {
|
||||
const snapshot = await readLegacySourceSnapshot(
|
||||
stateRoot,
|
||||
stateDir,
|
||||
detected.subscriptionsPath,
|
||||
LEGACY_SUBSCRIPTIONS_MAX_BYTES,
|
||||
);
|
||||
subscriptions = parseLegacySubscriptions(snapshot.raw);
|
||||
snapshots.push(snapshot);
|
||||
}
|
||||
if (await stateRoot.exists(relativeLegacyPath(stateDir, detected.vapidKeysPath))) {
|
||||
const snapshot = await readLegacySourceSnapshot(
|
||||
stateRoot,
|
||||
stateDir,
|
||||
detected.vapidKeysPath,
|
||||
LEGACY_VAPID_KEYS_MAX_BYTES,
|
||||
);
|
||||
vapidKeys = parseLegacyVapidKeys(snapshot.raw);
|
||||
snapshots.push(snapshot);
|
||||
}
|
||||
return { subscriptions, vapidKeys, snapshots };
|
||||
}
|
||||
|
||||
async function assertSourcesUnchanged(
|
||||
stateRoot: Root,
|
||||
stateDir: string,
|
||||
snapshots: readonly LegacySourceSnapshot[],
|
||||
subscriptionsPath: string,
|
||||
): Promise<void> {
|
||||
for (const snapshot of snapshots) {
|
||||
const current = await readLegacySourceSnapshot(
|
||||
stateRoot,
|
||||
stateDir,
|
||||
snapshot.sourcePath,
|
||||
maxBytesForSource(snapshot.sourcePath, subscriptionsPath),
|
||||
);
|
||||
if (!sourceSnapshotsMatch(current, snapshot)) {
|
||||
throw new Error("legacy Web Push source changed after doctor loaded it");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function mergedSubscription(params: {
|
||||
existing: WebPushSubscription;
|
||||
legacy: WebPushSubscription;
|
||||
}): WebPushSubscription {
|
||||
const { existing, legacy } = params;
|
||||
const createdAtMs = Math.min(existing.createdAtMs, legacy.createdAtMs);
|
||||
if (existing.updatedAtMs === legacy.updatedAtMs) {
|
||||
const normalizedExisting = { ...existing, createdAtMs };
|
||||
const normalizedLegacy = { ...legacy, createdAtMs };
|
||||
if (!webPushSubscriptionsEqual(normalizedExisting, normalizedLegacy)) {
|
||||
throw new Error("Web Push subscription diverges at the same timestamp");
|
||||
}
|
||||
return normalizedExisting;
|
||||
}
|
||||
const winner = existing.updatedAtMs > legacy.updatedAtMs ? existing : legacy;
|
||||
return { ...winner, createdAtMs };
|
||||
}
|
||||
|
||||
function findSubscriptionById(db: DatabaseSync, subscriptionId: string) {
|
||||
return executeSqliteQueryTakeFirstSync(
|
||||
db,
|
||||
getNodeSqliteKysely<WebPushDatabase>(db)
|
||||
.selectFrom("web_push_subscriptions")
|
||||
.selectAll()
|
||||
.where("subscription_id", "=", subscriptionId),
|
||||
);
|
||||
}
|
||||
|
||||
function writeSubscription(
|
||||
db: DatabaseSync,
|
||||
endpointHash: string,
|
||||
subscription: WebPushSubscription,
|
||||
): void {
|
||||
const row = webPushSubscriptionToRow({ endpointHash, subscription });
|
||||
executeSqliteQuerySync(
|
||||
db,
|
||||
getNodeSqliteKysely<WebPushDatabase>(db)
|
||||
.insertInto("web_push_subscriptions")
|
||||
.values(row)
|
||||
.onConflict((conflict) =>
|
||||
conflict.column("endpoint_hash").doUpdateSet({
|
||||
subscription_id: row.subscription_id,
|
||||
endpoint: row.endpoint,
|
||||
p256dh: row.p256dh,
|
||||
auth: row.auth,
|
||||
created_at_ms: row.created_at_ms,
|
||||
updated_at_ms: row.updated_at_ms,
|
||||
}),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
function migrateIntoDatabase(params: {
|
||||
stateDir: string;
|
||||
legacy: ParsedLegacyState;
|
||||
nowMs: number;
|
||||
}): { importedSubscriptions: number; importedVapidKeys: boolean } {
|
||||
let importedSubscriptions = 0;
|
||||
let importedVapidKeys = false;
|
||||
runOpenClawStateWriteTransaction(
|
||||
({ db }) => {
|
||||
const webPushDb = getNodeSqliteKysely<WebPushDatabase>(db);
|
||||
const expectedSubscriptions = new Map<string, WebPushSubscription>();
|
||||
for (const [endpointHash, legacySubscription] of params.legacy.subscriptions) {
|
||||
const existingRow = executeSqliteQueryTakeFirstSync(
|
||||
db,
|
||||
webPushDb
|
||||
.selectFrom("web_push_subscriptions")
|
||||
.selectAll()
|
||||
.where("endpoint_hash", "=", endpointHash),
|
||||
);
|
||||
if (existingRow && existingRow.endpoint !== legacySubscription.endpoint) {
|
||||
throw new Error("Web Push endpoint hash collision during legacy import");
|
||||
}
|
||||
const existing = existingRow ? webPushSubscriptionFromRow(existingRow) : null;
|
||||
const expected = existing
|
||||
? mergedSubscription({ existing, legacy: legacySubscription })
|
||||
: legacySubscription;
|
||||
const conflictingIdRow = findSubscriptionById(db, expected.subscriptionId);
|
||||
if (conflictingIdRow && conflictingIdRow.endpoint_hash !== endpointHash) {
|
||||
throw new Error("Web Push subscription id conflicts with another endpoint");
|
||||
}
|
||||
if (!existing || !webPushSubscriptionsEqual(existing, expected)) {
|
||||
writeSubscription(db, endpointHash, expected);
|
||||
importedSubscriptions += 1;
|
||||
}
|
||||
expectedSubscriptions.set(endpointHash, expected);
|
||||
}
|
||||
|
||||
let expectedVapidKeys: VapidKeyPair | null = null;
|
||||
if (params.legacy.vapidKeys) {
|
||||
const existingVapidRow = executeSqliteQueryTakeFirstSync(
|
||||
db,
|
||||
webPushDb
|
||||
.selectFrom("web_push_vapid_keys")
|
||||
.selectAll()
|
||||
.where("key_id", "=", WEB_PUSH_VAPID_KEY_ID),
|
||||
);
|
||||
if (existingVapidRow) {
|
||||
if (
|
||||
existingVapidRow.public_key !== params.legacy.vapidKeys.publicKey ||
|
||||
existingVapidRow.private_key !== params.legacy.vapidKeys.privateKey
|
||||
) {
|
||||
throw new Error("legacy Web Push VAPID identity conflicts with SQLite");
|
||||
}
|
||||
expectedVapidKeys = createWebPushVapidKeyPair(
|
||||
existingVapidRow.public_key,
|
||||
existingVapidRow.private_key,
|
||||
existingVapidRow.subject,
|
||||
);
|
||||
} else {
|
||||
executeSqliteQuerySync(
|
||||
db,
|
||||
webPushDb
|
||||
.insertInto("web_push_vapid_keys")
|
||||
.values(
|
||||
webPushVapidKeyPairToRow({ keyPair: params.legacy.vapidKeys, nowMs: params.nowMs }),
|
||||
),
|
||||
);
|
||||
expectedVapidKeys = params.legacy.vapidKeys;
|
||||
importedVapidKeys = true;
|
||||
}
|
||||
}
|
||||
|
||||
for (const [endpointHash, expected] of expectedSubscriptions) {
|
||||
const row = executeSqliteQueryTakeFirstSync(
|
||||
db,
|
||||
webPushDb
|
||||
.selectFrom("web_push_subscriptions")
|
||||
.selectAll()
|
||||
.where("endpoint_hash", "=", endpointHash),
|
||||
);
|
||||
if (!row || !webPushSubscriptionsEqual(webPushSubscriptionFromRow(row), expected)) {
|
||||
throw new Error("SQLite verification failed for a Web Push subscription");
|
||||
}
|
||||
}
|
||||
if (expectedVapidKeys) {
|
||||
const row = executeSqliteQueryTakeFirstSync(
|
||||
db,
|
||||
webPushDb
|
||||
.selectFrom("web_push_vapid_keys")
|
||||
.selectAll()
|
||||
.where("key_id", "=", WEB_PUSH_VAPID_KEY_ID),
|
||||
);
|
||||
if (
|
||||
!row ||
|
||||
row.public_key !== expectedVapidKeys.publicKey ||
|
||||
row.private_key !== expectedVapidKeys.privateKey ||
|
||||
row.subject !== expectedVapidKeys.subject
|
||||
) {
|
||||
throw new Error("SQLite verification failed for the Web Push VAPID identity");
|
||||
}
|
||||
}
|
||||
},
|
||||
{ env: { ...process.env, OPENCLAW_STATE_DIR: params.stateDir } },
|
||||
);
|
||||
return { importedSubscriptions, importedVapidKeys };
|
||||
}
|
||||
|
||||
async function restoreClaims(params: {
|
||||
stateRoot: Root;
|
||||
stateDir: string;
|
||||
claimed: readonly LegacySourceSnapshot[];
|
||||
}): Promise<string[]> {
|
||||
const errors: string[] = [];
|
||||
for (const snapshot of params.claimed.toReversed()) {
|
||||
const claimPath = `${snapshot.sourcePath}${DOCTOR_CLAIM_SUFFIX}`;
|
||||
const claimRelativePath = relativeLegacyPath(params.stateDir, claimPath);
|
||||
const sourceRelativePath = relativeLegacyPath(params.stateDir, snapshot.sourcePath);
|
||||
try {
|
||||
if (!(await params.stateRoot.exists(claimRelativePath))) {
|
||||
continue;
|
||||
}
|
||||
if (await params.stateRoot.exists(sourceRelativePath)) {
|
||||
errors.push(`source path already exists: ${snapshot.sourcePath}`);
|
||||
continue;
|
||||
}
|
||||
await params.stateRoot.move(claimRelativePath, sourceRelativePath);
|
||||
} catch (error) {
|
||||
errors.push(String(error));
|
||||
}
|
||||
}
|
||||
return errors;
|
||||
}
|
||||
|
||||
async function claimLegacySources(params: {
|
||||
stateRoot: Root;
|
||||
stateDir: string;
|
||||
snapshots: readonly LegacySourceSnapshot[];
|
||||
subscriptionsPath: string;
|
||||
beforeClaim?: () => void;
|
||||
}): Promise<LegacySourceSnapshot[]> {
|
||||
params.beforeClaim?.();
|
||||
const claimed: LegacySourceSnapshot[] = [];
|
||||
try {
|
||||
for (const snapshot of params.snapshots) {
|
||||
const claimPath = `${snapshot.sourcePath}${DOCTOR_CLAIM_SUFFIX}`;
|
||||
await params.stateRoot.move(
|
||||
relativeLegacyPath(params.stateDir, snapshot.sourcePath),
|
||||
relativeLegacyPath(params.stateDir, claimPath),
|
||||
);
|
||||
claimed.push(snapshot);
|
||||
const current = await readLegacySourceSnapshot(
|
||||
params.stateRoot,
|
||||
params.stateDir,
|
||||
claimPath,
|
||||
maxBytesForSource(snapshot.sourcePath, params.subscriptionsPath),
|
||||
);
|
||||
if (!sourceSnapshotsMatch(current, snapshot)) {
|
||||
throw new Error("legacy Web Push source changed before doctor could claim it");
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
const restoreErrors = await restoreClaims({
|
||||
stateRoot: params.stateRoot,
|
||||
stateDir: params.stateDir,
|
||||
claimed,
|
||||
});
|
||||
throw new Error(
|
||||
`${String(error)}${restoreErrors.length > 0 ? `; restore failures: ${restoreErrors.join("; ")}` : ""}`,
|
||||
{ cause: error },
|
||||
);
|
||||
}
|
||||
|
||||
return claimed;
|
||||
}
|
||||
|
||||
async function removeClaimedSources(params: {
|
||||
stateRoot: Root;
|
||||
stateDir: string;
|
||||
claimed: readonly LegacySourceSnapshot[];
|
||||
removeSource?: (sourcePath: string) => Promise<void> | void;
|
||||
}): Promise<void> {
|
||||
for (const snapshot of params.claimed) {
|
||||
if (await params.stateRoot.exists(relativeLegacyPath(params.stateDir, snapshot.sourcePath))) {
|
||||
throw new Error(`legacy Web Push source reappeared during import: ${snapshot.sourcePath}`);
|
||||
}
|
||||
}
|
||||
for (const snapshot of params.claimed) {
|
||||
const claimPath = `${snapshot.sourcePath}${DOCTOR_CLAIM_SUFFIX}`;
|
||||
if (params.removeSource) {
|
||||
await params.removeSource(claimPath);
|
||||
} else {
|
||||
await params.stateRoot.remove(relativeLegacyPath(params.stateDir, claimPath));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function migrateLegacyWebPushWithExclusiveStateOwnership(params: {
|
||||
stateRoot: Root;
|
||||
detected: LegacyStateDetection["webPush"];
|
||||
stateDir: string;
|
||||
beforeClaim?: () => void;
|
||||
beforeVerify?: () => void;
|
||||
removeSource?: (sourcePath: string) => Promise<void> | void;
|
||||
}): Promise<MigrationMessages> {
|
||||
const changes: string[] = [];
|
||||
const warnings: string[] = [];
|
||||
const notices: string[] = [];
|
||||
if (!params.detected.hasLegacy) {
|
||||
return { changes, warnings };
|
||||
}
|
||||
|
||||
let legacy: ParsedLegacyState;
|
||||
try {
|
||||
legacy = await readLegacyState(params.stateRoot, params.stateDir, params.detected);
|
||||
} catch (error) {
|
||||
warnings.push(`Failed reading legacy Web Push state: ${String(error)}`);
|
||||
return { changes, warnings };
|
||||
}
|
||||
|
||||
let claimed: LegacySourceSnapshot[];
|
||||
try {
|
||||
params.beforeVerify?.();
|
||||
await assertSourcesUnchanged(
|
||||
params.stateRoot,
|
||||
params.stateDir,
|
||||
legacy.snapshots,
|
||||
params.detected.subscriptionsPath,
|
||||
);
|
||||
// Claim both sources before the database transaction. A legacy writer can no longer
|
||||
// overwrite the retired paths after SQLite becomes canonical.
|
||||
claimed = await claimLegacySources({
|
||||
stateRoot: params.stateRoot,
|
||||
stateDir: params.stateDir,
|
||||
snapshots: legacy.snapshots,
|
||||
subscriptionsPath: params.detected.subscriptionsPath,
|
||||
beforeClaim: params.beforeClaim,
|
||||
});
|
||||
} catch (error) {
|
||||
warnings.push(`Failed migrating legacy Web Push state: ${String(error)}`);
|
||||
return { changes, warnings };
|
||||
}
|
||||
|
||||
let result: { importedSubscriptions: number; importedVapidKeys: boolean };
|
||||
try {
|
||||
result = migrateIntoDatabase({
|
||||
stateDir: params.stateDir,
|
||||
legacy,
|
||||
nowMs: Date.now(),
|
||||
});
|
||||
} catch (error) {
|
||||
const restoreErrors = await restoreClaims({
|
||||
stateRoot: params.stateRoot,
|
||||
stateDir: params.stateDir,
|
||||
claimed,
|
||||
});
|
||||
warnings.push(
|
||||
`Failed migrating legacy Web Push state: ${String(error)}${
|
||||
restoreErrors.length > 0 ? `; restore failures: ${restoreErrors.join("; ")}` : ""
|
||||
}`,
|
||||
);
|
||||
return { changes, warnings };
|
||||
}
|
||||
|
||||
try {
|
||||
await removeClaimedSources({
|
||||
stateRoot: params.stateRoot,
|
||||
stateDir: params.stateDir,
|
||||
claimed,
|
||||
removeSource: params.removeSource,
|
||||
});
|
||||
} catch (error) {
|
||||
warnings.push(`Web Push state is in SQLite, but legacy cleanup failed: ${String(error)}`);
|
||||
return { changes, warnings };
|
||||
}
|
||||
|
||||
changes.push(
|
||||
`Migrated ${result.importedSubscriptions} Web Push subscription${result.importedSubscriptions === 1 ? "" : "s"} to SQLite.`,
|
||||
);
|
||||
if (result.importedVapidKeys) {
|
||||
changes.push("Migrated the Web Push VAPID identity to SQLite.");
|
||||
}
|
||||
notices.push("Removed retired Web Push JSON state after verified SQLite import.");
|
||||
return { changes, warnings, notices };
|
||||
}
|
||||
|
||||
/** Import both retired stores while excluding old Gateways that can recreate them. */
|
||||
export async function migrateLegacyWebPush(params: {
|
||||
detected: LegacyStateDetection["webPush"];
|
||||
stateDir: string;
|
||||
env?: NodeJS.ProcessEnv;
|
||||
beforeClaim?: () => void;
|
||||
beforeVerify?: () => void;
|
||||
removeSource?: (sourcePath: string) => Promise<void> | void;
|
||||
}): Promise<MigrationMessages> {
|
||||
if (!params.detected.hasLegacy) {
|
||||
return { changes: [], warnings: [] };
|
||||
}
|
||||
|
||||
const env = { ...(params.env ?? process.env), OPENCLAW_STATE_DIR: params.stateDir };
|
||||
let lock: Awaited<ReturnType<typeof acquireGatewayLock>>;
|
||||
try {
|
||||
lock = await acquireGatewayLock({
|
||||
allowInTests: true,
|
||||
env,
|
||||
pollIntervalMs: MIGRATION_LOCK_POLL_INTERVAL_MS,
|
||||
role: "sqlite-maintenance",
|
||||
timeoutMs: MIGRATION_LOCK_TIMEOUT_MS,
|
||||
});
|
||||
} catch (error) {
|
||||
const detail =
|
||||
error instanceof GatewayLockError
|
||||
? "the Gateway or another SQLite maintenance command owns this state directory"
|
||||
: String(error);
|
||||
return {
|
||||
changes: [],
|
||||
warnings: [
|
||||
`Failed migrating legacy Web Push state: ${detail}. Stop the Gateway and run \`openclaw doctor --fix\` again.`,
|
||||
],
|
||||
};
|
||||
}
|
||||
if (!lock) {
|
||||
return {
|
||||
changes: [],
|
||||
warnings: ["Failed migrating legacy Web Push state: exclusive state ownership unavailable."],
|
||||
};
|
||||
}
|
||||
|
||||
let result: MigrationMessages = { changes: [], warnings: [] };
|
||||
let releaseError: unknown;
|
||||
try {
|
||||
try {
|
||||
const stateRoot = await root(params.stateDir, {
|
||||
hardlinks: "reject",
|
||||
maxBytes: LEGACY_SUBSCRIPTIONS_MAX_BYTES,
|
||||
symlinks: "reject",
|
||||
});
|
||||
result = await migrateLegacyWebPushWithExclusiveStateOwnership({
|
||||
...params,
|
||||
stateRoot,
|
||||
});
|
||||
} catch (error) {
|
||||
result.warnings.push(`Failed reading legacy Web Push state: ${String(error)}`);
|
||||
}
|
||||
} finally {
|
||||
try {
|
||||
await lock.release();
|
||||
} catch (error) {
|
||||
releaseError = error;
|
||||
}
|
||||
}
|
||||
if (releaseError) {
|
||||
result.warnings.push(
|
||||
`Web Push migration lock release failed: ${formatErrorMessage(releaseError)}`,
|
||||
);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
@@ -214,6 +214,23 @@ describe("check-database-first-legacy-stores", () => {
|
||||
expect(violations).toEqual([{ kind: "legacy store filesystem write", line: 4 }]);
|
||||
});
|
||||
|
||||
it("flags runtime writes to retired Web Push JSON stores", () => {
|
||||
const violations = collectDatabaseFirstLegacyStoreViolations(
|
||||
`
|
||||
import { promises as fs } from "node:fs";
|
||||
import path from "node:path";
|
||||
await fs.writeFile(path.join(stateDir, "push", "web-push-subscriptions.json"), "{}\n");
|
||||
await fs.writeFile(path.join(stateDir, "push", "vapid-keys.json"), "{}\n");
|
||||
`,
|
||||
"src/infra/push-web-file-store.ts",
|
||||
);
|
||||
|
||||
expect(violations).toEqual([
|
||||
{ kind: "legacy store filesystem write", line: 4 },
|
||||
{ kind: "legacy store filesystem write", line: 6 },
|
||||
]);
|
||||
});
|
||||
|
||||
it("flags runtime writes to retired skill-upload staging", () => {
|
||||
const violations = collectDatabaseFirstLegacyStoreViolations(
|
||||
`
|
||||
|
||||
Reference in New Issue
Block a user