diff --git a/extensions/device-pair/doctor-contract-api.test.ts b/extensions/device-pair/doctor-contract-api.test.ts new file mode 100644 index 000000000000..651abf3d2fe3 --- /dev/null +++ b/extensions/device-pair/doctor-contract-api.test.ts @@ -0,0 +1,120 @@ +// Device Pair tests cover doctor migration of legacy notify state. +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { + createPluginStateKeyedStoreForTests, + resetPluginStateStoreForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; +import type { + OpenKeyedStoreOptions, + PluginDoctorStateMigrationContext, +} from "openclaw/plugin-sdk/runtime-doctor"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { stateMigrations } from "./doctor-contract-api.js"; +import { + DEVICE_PAIR_NOTIFY_LEGACY_STATE_FILE, + DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES, + DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE, + notifySubscriberStoreKey, + type NotifySubscription, +} from "./notify-state.js"; + +function createDoctorContext(env: NodeJS.ProcessEnv): PluginDoctorStateMigrationContext { + return { + openPluginStateKeyedStore(options: OpenKeyedStoreOptions) { + return createPluginStateKeyedStoreForTests("device-pair", { + ...options, + env: options.env ?? env, + }); + }, + }; +} + +describe("device-pair doctor notify migration", () => { + let stateDir = ""; + let env: NodeJS.ProcessEnv; + + beforeEach(async () => { + resetPluginStateStoreForTests(); + stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-device-pair-doctor-")); + env = { ...process.env, OPENCLAW_STATE_DIR: stateDir }; + }); + + afterEach(async () => { + await fs.rm(stateDir, { recursive: true, force: true }); + }); + + function migrationParams() { + return { + config: {}, + env, + stateDir, + oauthDir: path.join(stateDir, "oauth"), + context: createDoctorContext(env), + }; + } + + it("imports legacy notify subscribers into plugin state", async () => { + const sourcePath = path.join(stateDir, DEVICE_PAIR_NOTIFY_LEGACY_STATE_FILE); + const subscriber: NotifySubscription = { + to: "chat-123", + accountId: "telegram-default", + messageThreadId: 271, + mode: "persistent", + addedAtMs: 1, + }; + await fs.writeFile( + sourcePath, + JSON.stringify({ + subscribers: [subscriber], + notifiedRequestIds: { stale: Date.now() }, + }), + "utf8", + ); + + const migration = stateMigrations[0]; + await expect(migration.detectLegacyState(migrationParams())).resolves.toMatchObject({ + preview: [expect.stringContaining("Device Pair notify subscribers")], + }); + + const result = await migration.migrateLegacyState(migrationParams()); + + expect(result.warnings).toEqual([]); + expect(result.changes).toEqual([ + "Migrated Device Pair notify subscribers -> plugin state (1 imported, 0 already present)", + expect.stringContaining("Archived Device Pair notify-state legacy source"), + ]); + await expect(fs.access(sourcePath)).rejects.toThrow(); + await expect(fs.access(`${sourcePath}.migrated`)).resolves.toBeUndefined(); + await expect( + createDoctorContext(env) + .openPluginStateKeyedStore({ + namespace: DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE, + maxEntries: DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES, + }) + .lookup(notifySubscriberStoreKey(subscriber)), + ).resolves.toEqual(subscriber); + }); + + it("ignores legacy notify files that only contain cache state", async () => { + const sourcePath = path.join(stateDir, DEVICE_PAIR_NOTIFY_LEGACY_STATE_FILE); + await fs.writeFile( + sourcePath, + JSON.stringify({ + subscribers: [], + notifiedRequestIds: { cached: Date.now() }, + }), + "utf8", + ); + + const migration = stateMigrations[0]; + + await expect(migration.detectLegacyState(migrationParams())).resolves.toBeNull(); + await expect(migration.migrateLegacyState(migrationParams())).resolves.toEqual({ + changes: [], + warnings: [], + }); + await expect(fs.access(sourcePath)).resolves.toBeUndefined(); + }); +}); diff --git a/extensions/device-pair/doctor-contract-api.ts b/extensions/device-pair/doctor-contract-api.ts new file mode 100644 index 000000000000..78a4da5baca2 --- /dev/null +++ b/extensions/device-pair/doctor-contract-api.ts @@ -0,0 +1,106 @@ +// Device Pair doctor contract migrates shipped plugin-owned state. +import fs from "node:fs/promises"; +import path from "node:path"; +import type { PluginDoctorStateMigration } from "openclaw/plugin-sdk/runtime-doctor"; +import { + DEVICE_PAIR_NOTIFY_LEGACY_STATE_FILE, + DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES, + DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE, + normalizeLegacyNotifyState, + notifySubscriberStoreKey, + type LegacyNotifyStateFile, + type NotifySubscription, +} from "./notify-state.js"; + +function resolveLegacyNotifyStatePath(stateDir: string): string { + return path.join(stateDir, DEVICE_PAIR_NOTIFY_LEGACY_STATE_FILE); +} + +async function fileExists(filePath: string): Promise { + try { + const stat = await fs.stat(filePath); + return stat.isFile(); + } catch { + return false; + } +} + +async function readLegacyNotifyState(filePath: string): Promise { + try { + return normalizeLegacyNotifyState(JSON.parse(await fs.readFile(filePath, "utf8")) as unknown); + } catch { + return null; + } +} + +async function archiveLegacySource(params: { + filePath: string; + changes: string[]; + warnings: string[]; +}): Promise { + const archivedPath = `${params.filePath}.migrated`; + if (await fileExists(archivedPath)) { + params.warnings.push( + `Left migrated Device Pair notify-state source in place because ${archivedPath} already exists`, + ); + return; + } + try { + await fs.rename(params.filePath, archivedPath); + params.changes.push(`Archived Device Pair notify-state legacy source -> ${archivedPath}`); + } catch (err) { + params.warnings.push(`Failed archiving Device Pair notify-state legacy source: ${String(err)}`); + } +} + +export const stateMigrations: PluginDoctorStateMigration[] = [ + { + id: "device-pair-notify-json-to-plugin-state", + label: "Device Pair notify subscribers", + async detectLegacyState(params) { + const filePath = resolveLegacyNotifyStatePath(params.stateDir); + const state = await readLegacyNotifyState(filePath); + if (!state || state.subscribers.length === 0) { + return null; + } + return { + preview: [ + `- Device Pair notify subscribers: ${filePath} -> plugin state (${DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE}, ${state.subscribers.length} subscriber(s))`, + ], + }; + }, + async migrateLegacyState(params) { + const changes: string[] = []; + const warnings: string[] = []; + const filePath = resolveLegacyNotifyStatePath(params.stateDir); + const state = await readLegacyNotifyState(filePath); + if (!state || state.subscribers.length === 0) { + return { changes, warnings }; + } + + const store = params.context.openPluginStateKeyedStore({ + namespace: DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE, + maxEntries: DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES, + }); + let imported = 0; + let alreadyPresent = 0; + for (const subscriber of state.subscribers) { + const inserted = await store.registerIfAbsent( + notifySubscriberStoreKey(subscriber), + subscriber, + ); + if (inserted) { + imported++; + } else { + alreadyPresent++; + } + } + + changes.push( + `Migrated Device Pair notify subscribers -> plugin state (${imported} imported, ${alreadyPresent} already present)`, + ); + await archiveLegacySource({ filePath, changes, warnings }); + return { changes, warnings }; + }, + }, +]; diff --git a/extensions/device-pair/notify-state.ts b/extensions/device-pair/notify-state.ts new file mode 100644 index 000000000000..3ba450ff490a --- /dev/null +++ b/extensions/device-pair/notify-state.ts @@ -0,0 +1,131 @@ +// Device Pair notify state helpers keep runtime and doctor migration in sync. +import { createHash } from "node:crypto"; +import { normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runtime"; + +export const DEVICE_PAIR_NOTIFY_LEGACY_STATE_FILE = "device-pair-notify.json"; +export const DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE = "notify-subscribers"; +export const DEVICE_PAIR_NOTIFY_SEEN_REQUEST_NAMESPACE = "notify-seen-requests"; +export const DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES = 1024; +export const DEVICE_PAIR_NOTIFY_SEEN_REQUEST_MAX_ENTRIES = 4096; +export const DEVICE_PAIR_NOTIFY_MAX_SEEN_AGE_MS = 24 * 60 * 60 * 1000; + +export type NotifySubscription = { + to: string; + accountId?: string; + messageThreadId?: string | number; + mode: "persistent" | "once"; + addedAtMs: number; +}; + +export type NotifySeenRequest = { + requestId: string; + notifiedAtMs: number; +}; + +export type LegacyNotifyStateFile = { + subscribers: NotifySubscription[]; + notifiedRequestIds: Record; +}; + +export function normalizeLegacyNotifyState(raw: unknown): LegacyNotifyStateFile { + const root = typeof raw === "object" && raw !== null ? (raw as Record) : {}; + const subscribersRaw = Array.isArray(root.subscribers) ? root.subscribers : []; + const notifiedRaw = + typeof root.notifiedRequestIds === "object" && root.notifiedRequestIds !== null + ? (root.notifiedRequestIds as Record) + : {}; + + const subscribers: NotifySubscription[] = []; + for (const item of subscribersRaw) { + if (typeof item !== "object" || item === null) { + continue; + } + const record = item as Record; + const to = normalizeOptionalString(record.to) ?? ""; + if (!to) { + continue; + } + const accountId = normalizeOptionalString(record.accountId) ?? undefined; + const messageThreadId = + typeof record.messageThreadId === "string" + ? normalizeOptionalString(record.messageThreadId) || undefined + : typeof record.messageThreadId === "number" && Number.isFinite(record.messageThreadId) + ? Math.trunc(record.messageThreadId) + : undefined; + const mode = record.mode === "once" ? "once" : "persistent"; + const addedAtMs = + typeof record.addedAtMs === "number" && Number.isFinite(record.addedAtMs) + ? Math.trunc(record.addedAtMs) + : Date.now(); + subscribers.push({ + to, + accountId, + messageThreadId, + mode, + addedAtMs, + }); + } + + const notifiedRequestIds: Record = {}; + for (const [requestId, ts] of Object.entries(notifiedRaw)) { + const normalizedRequestId = normalizeOptionalString(requestId); + if (!normalizedRequestId) { + continue; + } + if (typeof ts !== "number" || !Number.isFinite(ts) || ts <= 0) { + continue; + } + notifiedRequestIds[normalizedRequestId] = Math.trunc(ts); + } + + return { subscribers, notifiedRequestIds }; +} + +export function normalizeNotifyThreadKey(messageThreadId?: string | number): string { + if (typeof messageThreadId === "number" && Number.isFinite(messageThreadId)) { + return String(Math.trunc(messageThreadId)); + } + if (typeof messageThreadId !== "string") { + return ""; + } + const normalized = normalizeOptionalString(messageThreadId); + if (!normalized) { + return ""; + } + if (!/^-?\d+$/u.test(normalized)) { + return normalized; + } + try { + return BigInt(normalized).toString(); + } catch { + return normalized; + } +} + +export function notifySubscriberKey(subscriber: { + to: string; + accountId?: string; + messageThreadId?: string | number; +}): string { + return JSON.stringify([ + subscriber.to, + subscriber.accountId ?? "", + normalizeNotifyThreadKey(subscriber.messageThreadId), + ]); +} + +function hashStoreKey(value: string): string { + return createHash("sha256").update(value).digest("hex"); +} + +export function notifySubscriberStoreKey(subscriber: { + to: string; + accountId?: string; + messageThreadId?: string | number; +}): string { + return hashStoreKey(notifySubscriberKey(subscriber)); +} + +export function notifyRequestStoreKey(requestId: string): string { + return hashStoreKey(requestId); +} diff --git a/extensions/device-pair/notify.test.ts b/extensions/device-pair/notify.test.ts index 2a6747138889..4c8aeaedaaa9 100644 --- a/extensions/device-pair/notify.test.ts +++ b/extensions/device-pair/notify.test.ts @@ -2,8 +2,19 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; +import type { OpenKeyedStoreOptions } from "openclaw/plugin-sdk/plugin-state-runtime"; +import { + createPluginStateKeyedStoreForTests, + resetPluginStateStoreForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; import { createTestPluginApi } from "openclaw/plugin-sdk/plugin-test-api"; import { afterAll, afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES, + DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE, + notifySubscriberStoreKey, + type NotifySubscription, +} from "./notify-state.js"; const listDevicePairingMock = vi.hoisted(() => vi.fn(async () => ({ pending: [] }))); @@ -20,46 +31,55 @@ afterAll(() => { describe("device-pair notify persistence", () => { let stateDir: string; + let env: NodeJS.ProcessEnv; beforeEach(async () => { + resetPluginStateStoreForTests(); vi.clearAllMocks(); listDevicePairingMock.mockResolvedValue({ pending: [] }); stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "device-pair-notify-")); + env = { ...process.env, OPENCLAW_STATE_DIR: stateDir }; }); afterEach(async () => { await fs.rm(stateDir, { recursive: true, force: true }); }); - it("matches persisted telegram thread ids across number and string roundtrips", async () => { - await fs.writeFile( - path.join(stateDir, "device-pair-notify.json"), - JSON.stringify( - { - subscribers: [ - { - to: "chat-123", - accountId: "telegram-default", - messageThreadId: 271, - mode: "persistent", - addedAtMs: 1, - }, - ], - notifiedRequestIds: {}, - }, - null, - 2, - ), - "utf8", - ); + function openStore(options: OpenKeyedStoreOptions) { + return createPluginStateKeyedStoreForTests("device-pair", { + ...options, + env: options.env ?? env, + }); + } - const api = createTestPluginApi({ + function createApi() { + return createTestPluginApi({ runtime: { state: { resolveStateDir: () => stateDir, + openKeyedStore: openStore, }, } as never, }); + } + + function openSubscriberStore() { + return openStore({ + namespace: DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE, + maxEntries: DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES, + }); + } + + it("matches persisted telegram thread ids across number and string roundtrips", async () => { + const subscriber: NotifySubscription = { + to: "chat-123", + accountId: "telegram-default", + messageThreadId: 271, + mode: "persistent", + addedAtMs: 1, + }; + await openSubscriberStore().register(notifySubscriberStoreKey(subscriber), subscriber); + const api = createApi(); const status = await handleNotifyCommand({ api, @@ -86,46 +106,26 @@ describe("device-pair notify persistence", () => { action: "off", }); - const persisted = JSON.parse( - await fs.readFile(path.join(stateDir, "device-pair-notify.json"), "utf8"), - ) as { subscribers: unknown[] }; - expect(persisted.subscribers).toStrictEqual([]); + await expect(openSubscriberStore().entries()).resolves.toStrictEqual([]); }); it("does not remove a different persisted subscriber when notify fields contain pipes", async () => { - await fs.writeFile( - path.join(stateDir, "device-pair-notify.json"), - JSON.stringify( - { - subscribers: [ - { - to: "chat|123", - accountId: "acct", - mode: "persistent", - addedAtMs: 1, - }, - { - to: "chat", - accountId: "123|acct", - mode: "persistent", - addedAtMs: 2, - }, - ], - notifiedRequestIds: {}, - }, - null, - 2, - ), - "utf8", - ); - - const api = createTestPluginApi({ - runtime: { - state: { - resolveStateDir: () => stateDir, - }, - } as never, - }); + const firstSubscriber: NotifySubscription = { + to: "chat|123", + accountId: "acct", + mode: "persistent", + addedAtMs: 1, + }; + const secondSubscriber: NotifySubscription = { + to: "chat", + accountId: "123|acct", + mode: "persistent", + addedAtMs: 2, + }; + const store = openSubscriberStore(); + await store.register(notifySubscriberStoreKey(firstSubscriber), firstSubscriber); + await store.register(notifySubscriberStoreKey(secondSubscriber), secondSubscriber); + const api = createApi(); await handleNotifyCommand({ api, @@ -148,19 +148,11 @@ describe("device-pair notify persistence", () => { }); expect(status.text).toContain("Pair request notifications: disabled for this chat."); - const persisted = JSON.parse( - await fs.readFile(path.join(stateDir, "device-pair-notify.json"), "utf8"), - ) as unknown; - expect(persisted).toStrictEqual({ - subscribers: [ - { - to: "chat|123", - accountId: "acct", - mode: "persistent", - addedAtMs: 1, - }, - ], - notifiedRequestIds: {}, - }); + await expect(openSubscriberStore().entries()).resolves.toMatchObject([ + { + key: notifySubscriberStoreKey(firstSubscriber), + value: firstSubscriber, + }, + ]); }); }); diff --git a/extensions/device-pair/notify.ts b/extensions/device-pair/notify.ts index 54ca3781720d..8ceeb8b56ea7 100644 --- a/extensions/device-pair/notify.ts +++ b/extensions/device-pair/notify.ts @@ -1,24 +1,24 @@ // Device Pair plugin module implements notify behavior. -import { promises as fs } from "node:fs"; -import path from "node:path"; import type { OpenClawPluginService } from "openclaw/plugin-sdk/core"; import { listDevicePairing } from "openclaw/plugin-sdk/device-bootstrap"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import type { OpenClawPluginApi } from "openclaw/plugin-sdk/plugin-entry"; -import { replaceFileAtomic } from "openclaw/plugin-sdk/security-runtime"; +import type { PluginStateKeyedStore } from "openclaw/plugin-sdk/plugin-state-runtime"; import { normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runtime"; +import { + DEVICE_PAIR_NOTIFY_MAX_SEEN_AGE_MS, + DEVICE_PAIR_NOTIFY_SEEN_REQUEST_MAX_ENTRIES, + DEVICE_PAIR_NOTIFY_SEEN_REQUEST_NAMESPACE, + DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES, + DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE, + notifyRequestStoreKey, + notifySubscriberKey, + notifySubscriberStoreKey, + type NotifySeenRequest, + type NotifySubscription, +} from "./notify-state.js"; -const NOTIFY_STATE_FILE = "device-pair-notify.json"; const NOTIFY_POLL_INTERVAL_MS = 10_000; -const NOTIFY_MAX_SEEN_AGE_MS = 24 * 60 * 60 * 1000; - -type NotifySubscription = { - to: string; - accountId?: string; - messageThreadId?: string | number; - mode: "persistent" | "once"; - addedAtMs: number; -}; type NotifyStateFile = { subscribers: NotifySubscription[]; @@ -79,112 +79,79 @@ export function formatPendingRequests(pending: PendingPairingRequest[]): string return lines.join("\n"); } -function resolveNotifyStatePath(stateDir: string): string { - return path.join(stateDir, NOTIFY_STATE_FILE); +function openNotifySubscriberStore( + api: OpenClawPluginApi, +): PluginStateKeyedStore { + return api.runtime.state.openKeyedStore({ + namespace: DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE, + maxEntries: DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES, + }); } -function normalizeNotifyState(raw: unknown): NotifyStateFile { - const root = typeof raw === "object" && raw !== null ? (raw as Record) : {}; - const subscribersRaw = Array.isArray(root.subscribers) ? root.subscribers : []; - const notifiedRaw = - typeof root.notifiedRequestIds === "object" && root.notifiedRequestIds !== null - ? (root.notifiedRequestIds as Record) - : {}; +function openNotifySeenRequestStore( + api: OpenClawPluginApi, +): PluginStateKeyedStore { + return api.runtime.state.openKeyedStore({ + namespace: DEVICE_PAIR_NOTIFY_SEEN_REQUEST_NAMESPACE, + maxEntries: DEVICE_PAIR_NOTIFY_SEEN_REQUEST_MAX_ENTRIES, + defaultTtlMs: DEVICE_PAIR_NOTIFY_MAX_SEEN_AGE_MS, + }); +} - const subscribers: NotifySubscription[] = []; - for (const item of subscribersRaw) { - if (typeof item !== "object" || item === null) { - continue; - } - const record = item as Record; - const to = normalizeOptionalString(record.to) ?? ""; - if (!to) { - continue; - } - const accountId = normalizeOptionalString(record.accountId) ?? undefined; - const messageThreadId = - typeof record.messageThreadId === "string" - ? normalizeOptionalString(record.messageThreadId) || undefined - : typeof record.messageThreadId === "number" && Number.isFinite(record.messageThreadId) - ? Math.trunc(record.messageThreadId) - : undefined; - const mode = record.mode === "once" ? "once" : "persistent"; - const addedAtMs = - typeof record.addedAtMs === "number" && Number.isFinite(record.addedAtMs) - ? Math.trunc(record.addedAtMs) - : Date.now(); - subscribers.push({ - to, - accountId, - messageThreadId, - mode, - addedAtMs, - }); - } +async function readNotifyState(api: OpenClawPluginApi): Promise { + const subscriberStore = openNotifySubscriberStore(api); + const seenRequestStore = openNotifySeenRequestStore(api); + const [subscriberEntries, seenRequestEntries] = await Promise.all([ + subscriberStore.entries(), + seenRequestStore.entries(), + ]); + const subscribers = subscriberEntries + .map((entry) => entry.value) + .sort((a, b) => a.addedAtMs - b.addedAtMs); const notifiedRequestIds: Record = {}; - for (const [requestId, ts] of Object.entries(notifiedRaw)) { - const normalizedRequestId = normalizeOptionalString(requestId); - if (!normalizedRequestId) { + for (const entry of seenRequestEntries) { + const requestId = normalizeOptionalString(entry.value.requestId); + const notifiedAtMs = entry.value.notifiedAtMs; + if (!requestId || !Number.isFinite(notifiedAtMs) || notifiedAtMs <= 0) { continue; } - if (typeof ts !== "number" || !Number.isFinite(ts) || ts <= 0) { - continue; - } - notifiedRequestIds[normalizedRequestId] = Math.trunc(ts); + notifiedRequestIds[requestId] = Math.trunc(notifiedAtMs); } return { subscribers, notifiedRequestIds }; } -async function readNotifyState(filePath: string): Promise { - try { - const content = await fs.readFile(filePath, "utf8"); - return normalizeNotifyState(JSON.parse(content)); - } catch { - return { subscribers: [], notifiedRequestIds: {} }; +async function writeNotifyState(api: OpenClawPluginApi, state: NotifyStateFile): Promise { + const subscriberStore = openNotifySubscriberStore(api); + const nextSubscribers = new Map( + state.subscribers.map((subscriber) => [notifySubscriberStoreKey(subscriber), subscriber]), + ); + for (const entry of await subscriberStore.entries()) { + if (!nextSubscribers.has(entry.key)) { + await subscriberStore.delete(entry.key); + } + } + for (const [key, subscriber] of nextSubscribers) { + await subscriberStore.register(key, subscriber); } -} -async function writeNotifyState(filePath: string, state: NotifyStateFile): Promise { - const content = JSON.stringify(state, null, 2); - await replaceFileAtomic({ - filePath, - content: `${content}\n`, - tempPrefix: ".device-pair-notify", - }); -} - -function notifySubscriberKey(subscriber: { - to: string; - accountId?: string; - messageThreadId?: string | number; -}): string { - return JSON.stringify([ - subscriber.to, - subscriber.accountId ?? "", - normalizeNotifyThreadKey(subscriber.messageThreadId), - ]); -} - -function normalizeNotifyThreadKey(messageThreadId?: string | number): string { - if (typeof messageThreadId === "number" && Number.isFinite(messageThreadId)) { - return String(Math.trunc(messageThreadId)); + const seenRequestStore = openNotifySeenRequestStore(api); + const nextSeenRequests = new Map( + Object.entries(state.notifiedRequestIds).map(([requestId, notifiedAtMs]) => [ + notifyRequestStoreKey(requestId), + { requestId, notifiedAtMs }, + ]), + ); + for (const entry of await seenRequestStore.entries()) { + if (!nextSeenRequests.has(entry.key)) { + await seenRequestStore.delete(entry.key); + } } - if (typeof messageThreadId !== "string") { - return ""; - } - const normalized = normalizeOptionalString(messageThreadId); - if (!normalized) { - return ""; - } - if (!/^-?\d+$/u.test(normalized)) { - return normalized; - } - try { - return BigInt(normalized).toString(); - } catch { - return normalized; + for (const [key, value] of nextSeenRequests) { + await seenRequestStore.register(key, value, { + ttlMs: DEVICE_PAIR_NOTIFY_MAX_SEEN_AGE_MS, + }); } } @@ -317,11 +284,8 @@ async function notifySubscriber(params: { } } -async function notifyPendingPairingRequests(params: { - api: OpenClawPluginApi; - statePath: string; -}): Promise { - const state = await readNotifyState(params.statePath); +async function notifyPendingPairingRequests(params: { api: OpenClawPluginApi }): Promise { + const state = await readNotifyState(params.api); const pairing = await listDevicePairing(); const pending = pairing.pending as PendingPairingRequest[]; const now = Date.now(); @@ -329,7 +293,7 @@ async function notifyPendingPairingRequests(params: { let changed = false; for (const [requestId, ts] of Object.entries(state.notifiedRequestIds)) { - if (!pendingIds.has(requestId) || now - ts > NOTIFY_MAX_SEEN_AGE_MS) { + if (!pendingIds.has(requestId) || now - ts > DEVICE_PAIR_NOTIFY_MAX_SEEN_AGE_MS) { delete state.notifiedRequestIds[requestId]; changed = true; } @@ -376,7 +340,7 @@ async function notifyPendingPairingRequests(params: { } if (changed) { - await writeNotifyState(params.statePath, state); + await writeNotifyState(params.api, state); } } @@ -399,9 +363,7 @@ export async function armPairNotifyOnce(params: { return false; } - const stateDir = params.api.runtime.state.resolveStateDir(); - const statePath = resolveNotifyStatePath(stateDir); - const state = await readNotifyState(statePath); + const state = await readNotifyState(params.api); let changed = false; if (upsertNotifySubscriber(state.subscribers, target, "once")) { @@ -409,7 +371,7 @@ export async function armPairNotifyOnce(params: { } if (changed) { - await writeNotifyState(statePath, state); + await writeNotifyState(params.api, state); } return true; } @@ -435,15 +397,13 @@ export async function handleNotifyCommand(params: { return { text: "Could not resolve Telegram target for this chat." }; } - const stateDir = params.api.runtime.state.resolveStateDir(); - const statePath = resolveNotifyStatePath(stateDir); - const state = await readNotifyState(statePath); + const state = await readNotifyState(params.api); const targetKey = notifySubscriberKey(target); const current = state.subscribers.find((entry) => notifySubscriberKey(entry) === targetKey); if (params.action === "on" || params.action === "enable") { if (upsertNotifySubscriber(state.subscribers, target, "persistent")) { - await writeNotifyState(statePath, state); + await writeNotifyState(params.api, state); } return { text: @@ -458,7 +418,7 @@ export async function handleNotifyCommand(params: { ); if (currentIndex !== -1) { state.subscribers.splice(currentIndex, 1); - await writeNotifyState(statePath, state); + await writeNotifyState(params.api, state); } return { text: "✅ Pair request notifications disabled for this Telegram chat." }; } @@ -499,10 +459,9 @@ export function createPairingNotifierService(api: OpenClawPluginApi): OpenClawPl return { id: "device-pair-notifier", - start: async (ctx) => { - const statePath = resolveNotifyStatePath(ctx.stateDir); + start: async () => { const tick = async () => { - await notifyPendingPairingRequests({ api, statePath }); + await notifyPendingPairingRequests({ api }); }; await tick().catch((err: unknown) => {