From bab18d567b0cd9811012a50e032f71a9c9beb440 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sun, 7 Jun 2026 02:41:45 -0700 Subject: [PATCH] refactor(plugin-sdk): persist dedupe state in sqlite --- extensions/nextcloud-talk/src/doctor.test.ts | 50 ++ extensions/nextcloud-talk/src/doctor.ts | 79 +++ extensions/nextcloud-talk/src/replay-guard.ts | 31 +- src/channels/plugins/types.adapters.ts | 1 + src/commands/doctor/shared/channel-doctor.ts | 1 + src/plugin-sdk/memory-host-events.test.ts | 56 ++- src/plugin-sdk/persistent-dedupe.ts | 454 ++++++++++++------ 7 files changed, 482 insertions(+), 190 deletions(-) diff --git a/extensions/nextcloud-talk/src/doctor.test.ts b/extensions/nextcloud-talk/src/doctor.test.ts index 42410200e0a8..ec90aea9fe45 100644 --- a/extensions/nextcloud-talk/src/doctor.test.ts +++ b/extensions/nextcloud-talk/src/doctor.test.ts @@ -1,5 +1,10 @@ // Nextcloud Talk tests cover doctor plugin behavior. +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { resetPluginStateStoreForTests } from "openclaw/plugin-sdk/plugin-state-test-runtime"; import { beforeEach, describe, expect, it, vi } from "vitest"; +import { createNextcloudTalkReplayGuard } from "./replay-guard.js"; const hoisted = vi.hoisted(() => ({ probeNextcloudTalkBotResponseFeature: vi.fn(), @@ -24,6 +29,7 @@ function getNextcloudTalkCompatibilityNormalizer(): NonNullable< describe("nextcloud-talk doctor", () => { beforeEach(() => { hoisted.probeNextcloudTalkBotResponseFeature.mockReset(); + resetPluginStateStoreForTests(); }); it("normalizes legacy private-network aliases", () => { @@ -85,4 +91,48 @@ describe("nextcloud-talk doctor", () => { '- channels.nextcloud-talk.default: Nextcloud Talk bot "OpenClaw" (1) is missing the response feature (features=9); outbound replies will fail.', ]); }); + + it("migrates legacy replay dedupe JSON into SQLite during doctor repair", async () => { + const stateDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-nextcloud-doctor-")); + const legacyDir = path.join(stateDir, "nextcloud-talk", "replay-dedupe"); + const legacyPath = path.join(legacyDir, "account-a.json"); + await fs.mkdir(legacyDir, { recursive: true }); + await fs.writeFile( + legacyPath, + JSON.stringify({ + "room-1:msg-1": Date.now(), + }), + ); + + const mutation = await nextcloudTalkDoctor.repairConfig?.({ + cfg: { + channels: { + "nextcloud-talk": { + accounts: { + "account-a": { + baseUrl: "https://cloud.example.com", + botSecret: "secret", + }, + }, + }, + }, + } as never, + doctorFixCommand: "openclaw doctor --fix", + env: { ...process.env, OPENCLAW_STATE_DIR: stateDir }, + }); + + expect(mutation?.changes.join("\n")).toContain( + 'Migrated Nextcloud Talk replay dedupe cache for account "account-a" to SQLite', + ); + await expect(fs.access(legacyPath)).rejects.toThrow(); + + const guard = createNextcloudTalkReplayGuard({ stateDir }); + await expect( + guard.shouldProcessMessage({ + accountId: "account-a", + roomToken: "room-1", + messageId: "msg-1", + }), + ).resolves.toBe(false); + }); }); diff --git a/extensions/nextcloud-talk/src/doctor.ts b/extensions/nextcloud-talk/src/doctor.ts index 0edbb6f7248b..7be90be0b127 100644 --- a/extensions/nextcloud-talk/src/doctor.ts +++ b/extensions/nextcloud-talk/src/doctor.ts @@ -1,13 +1,42 @@ // Nextcloud Talk plugin module implements doctor behavior. +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; import type { ChannelDoctorAdapter } from "openclaw/plugin-sdk/channel-contract"; +import { migratePersistentDedupeLegacyJsonFile } from "openclaw/plugin-sdk/persistent-dedupe"; +import { resolveStateDir } from "openclaw/plugin-sdk/state-paths"; import { listNextcloudTalkAccountIds, resolveNextcloudTalkAccount } from "./accounts.js"; import { probeNextcloudTalkBotResponseFeature } from "./bot-preflight.js"; import { legacyConfigRules as NEXTCLOUD_TALK_LEGACY_CONFIG_RULES, normalizeCompatibilityConfig as normalizeNextcloudTalkCompatibilityConfig, } from "./doctor-contract.js"; +import { + NEXTCLOUD_TALK_PLUGIN_ID, + NEXTCLOUD_TALK_REPLAY_DEDUPE_NAMESPACE_PREFIX, +} from "./replay-guard.js"; import type { CoreConfig } from "./types.js"; +const REPLAY_DEDUPE_TTL_MS = 24 * 60 * 60 * 1000; +const REPLAY_DEDUPE_MAX_ENTRIES = 10_000; + +function sanitizeLegacyReplaySegment(value: string): string { + const trimmed = value.trim(); + if (!trimmed) { + return "default"; + } + return trimmed.replace(/[^a-zA-Z0-9_-]/g, "_"); +} + +async function fileExists(filePath: string): Promise { + try { + await fs.access(filePath); + return true; + } catch { + return false; + } +} + async function collectNextcloudTalkBotResponseWarnings(params: { cfg: CoreConfig; }): Promise { @@ -33,9 +62,59 @@ async function collectNextcloudTalkBotResponseWarnings(params: { return warnings; } +async function repairNextcloudTalkReplayDedupeState(params: { + cfg: CoreConfig; + env?: NodeJS.ProcessEnv; +}): Promise<{ changes: string[]; warnings: string[] }> { + const changes: string[] = []; + const warnings: string[] = []; + const env = params.env ?? process.env; + const stateDir = resolveStateDir(env, os.homedir); + const replayDir = path.join(stateDir, "nextcloud-talk", "replay-dedupe"); + + for (const accountId of listNextcloudTalkAccountIds(params.cfg)) { + const legacyPath = path.join(replayDir, `${sanitizeLegacyReplaySegment(accountId)}.json`); + if (!(await fileExists(legacyPath))) { + continue; + } + try { + const result = await migratePersistentDedupeLegacyJsonFile({ + filePath: legacyPath, + namespace: accountId, + ttlMs: REPLAY_DEDUPE_TTL_MS, + memoryMaxSize: 0, + pluginId: NEXTCLOUD_TALK_PLUGIN_ID, + namespacePrefix: NEXTCLOUD_TALK_REPLAY_DEDUPE_NAMESPACE_PREFIX, + stateMaxEntries: REPLAY_DEDUPE_MAX_ENTRIES, + env, + }); + changes.push( + `Migrated Nextcloud Talk replay dedupe cache for account "${accountId}" to SQLite (${result.imported} imported, ${result.skippedExpired} expired, ${result.skippedExisting} already current).`, + ); + } catch (error) { + warnings.push( + `Skipped Nextcloud Talk replay dedupe cache for account "${accountId}": ${String(error)}`, + ); + } + } + + return { changes, warnings }; +} + export const nextcloudTalkDoctor: ChannelDoctorAdapter = { legacyConfigRules: NEXTCLOUD_TALK_LEGACY_CONFIG_RULES, normalizeCompatibilityConfig: normalizeNextcloudTalkCompatibilityConfig, collectPreviewWarnings: async ({ cfg }) => await collectNextcloudTalkBotResponseWarnings({ cfg: cfg as CoreConfig }), + repairConfig: async ({ cfg, env }) => { + const repair = await repairNextcloudTalkReplayDedupeState({ + cfg: cfg as CoreConfig, + ...(env ? { env } : {}), + }); + return { + config: cfg, + changes: repair.changes, + warnings: repair.warnings, + }; + }, }; diff --git a/extensions/nextcloud-talk/src/replay-guard.ts b/extensions/nextcloud-talk/src/replay-guard.ts index 1857c4ea2d75..288643cc13cb 100644 --- a/extensions/nextcloud-talk/src/replay-guard.ts +++ b/extensions/nextcloud-talk/src/replay-guard.ts @@ -1,18 +1,11 @@ // Nextcloud Talk plugin module implements replay guard behavior. -import path from "node:path"; import { createClaimableDedupe } from "openclaw/plugin-sdk/persistent-dedupe"; +export const NEXTCLOUD_TALK_PLUGIN_ID = "nextcloud-talk"; +export const NEXTCLOUD_TALK_REPLAY_DEDUPE_NAMESPACE_PREFIX = "replay-dedupe"; const DEFAULT_REPLAY_TTL_MS = 24 * 60 * 60 * 1000; const DEFAULT_MEMORY_MAX_SIZE = 1_000; -const DEFAULT_FILE_MAX_ENTRIES = 10_000; - -function sanitizeSegment(value: string): string { - const trimmed = value.trim(); - if (!trimmed) { - return "default"; - } - return trimmed.replace(/[^a-zA-Z0-9_-]/g, "_"); -} +const DEFAULT_STATE_MAX_ENTRIES = 10_000; function buildReplayKey(params: { roomToken: string; messageId: string }): string | null { const roomToken = params.roomToken.trim(); @@ -27,6 +20,8 @@ type NextcloudTalkReplayGuardOptions = { stateDir?: string; ttlMs?: number; memoryMaxSize?: number; + stateMaxEntries?: number; + /** @deprecated Use stateMaxEntries. */ fileMaxEntries?: number; onDiskError?: (error: unknown) => void; }; @@ -67,14 +62,14 @@ export function createNextcloudTalkReplayGuard( stateDir ? { ...baseOptions, - fileMaxEntries: options.fileMaxEntries ?? DEFAULT_FILE_MAX_ENTRIES, - resolveFilePath: (namespace) => - path.join( - stateDir, - "nextcloud-talk", - "replay-dedupe", - `${sanitizeSegment(namespace)}.json`, - ), + pluginId: NEXTCLOUD_TALK_PLUGIN_ID, + namespacePrefix: NEXTCLOUD_TALK_REPLAY_DEDUPE_NAMESPACE_PREFIX, + stateMaxEntries: + options.stateMaxEntries ?? options.fileMaxEntries ?? DEFAULT_STATE_MAX_ENTRIES, + env: { + ...process.env, + OPENCLAW_STATE_DIR: stateDir, + }, onDiskError: options.onDiskError, } : baseOptions, diff --git a/src/channels/plugins/types.adapters.ts b/src/channels/plugins/types.adapters.ts index 6ff17f0eb34d..e8f8f1cfc4f2 100644 --- a/src/channels/plugins/types.adapters.ts +++ b/src/channels/plugins/types.adapters.ts @@ -526,6 +526,7 @@ export type ChannelDoctorAdapter = { repairConfig?: (params: { cfg: OpenClawConfig; doctorFixCommand: string; + env?: NodeJS.ProcessEnv; }) => ChannelDoctorConfigMutation | Promise; runConfigSequence?: (params: { cfg: OpenClawConfig; diff --git a/src/commands/doctor/shared/channel-doctor.ts b/src/commands/doctor/shared/channel-doctor.ts index 9db9853c0159..65793da89123 100644 --- a/src/commands/doctor/shared/channel-doctor.ts +++ b/src/commands/doctor/shared/channel-doctor.ts @@ -407,6 +407,7 @@ export async function collectChannelDoctorRepairMutations(params: { const mutation = await entry.doctor.repairConfig?.({ cfg: nextCfg, doctorFixCommand: params.doctorFixCommand, + ...(params.env ? { env: params.env } : {}), }); if (!mutation || mutation.changes.length === 0) { if (mutation?.warnings?.length) { diff --git a/src/plugin-sdk/memory-host-events.test.ts b/src/plugin-sdk/memory-host-events.test.ts index 39ec6a8a28f0..7b7d747f5adc 100644 --- a/src/plugin-sdk/memory-host-events.test.ts +++ b/src/plugin-sdk/memory-host-events.test.ts @@ -3,7 +3,8 @@ */ import fs from "node:fs/promises"; import path from "node:path"; -import { describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it } from "vitest"; +import { resetPluginStateStoreForTests } from "../plugin-state/plugin-state-store.js"; import { appendMemoryHostEvent, readMemoryHostEvents, @@ -18,11 +19,17 @@ function createDedupe(root: string, overrides?: { ttlMs?: number }) { return createPersistentDedupe({ ttlMs: overrides?.ttlMs ?? 24 * 60 * 60 * 1000, memoryMaxSize: 100, - fileMaxEntries: 1000, - resolveFilePath: (namespace) => path.join(root, `${namespace}.json`), + pluginId: "test-persistent-dedupe", + namespacePrefix: "test-dedupe", + stateMaxEntries: 1000, + env: { ...process.env, OPENCLAW_STATE_DIR: root }, }); } +afterEach(() => { + resetPluginStateStoreForTests(); +}); + describe("memory host event journal helpers", () => { it("appends and reads typed workspace events", async () => { const workspaceDir = await createTempDir("memory-host-events-"); @@ -94,8 +101,10 @@ describe("createPersistentDedupe", () => { const dedupe = createPersistentDedupe({ ttlMs: Number.NaN, memoryMaxSize: Number.NaN, - fileMaxEntries: Number.NaN, - resolveFilePath: (namespace) => path.join(root, `${namespace}.json`), + pluginId: "test-persistent-dedupe", + namespacePrefix: "test-bounds", + stateMaxEntries: Number.NaN, + env: { ...process.env, OPENCLAW_STATE_DIR: root }, }); expect(await dedupe.checkAndRecord("m1", { namespace: "a", now: 100 })).toBe(true); @@ -104,33 +113,32 @@ describe("createPersistentDedupe", () => { expect(dedupe.memorySize()).toBe(0); }); - it("falls back to memory-only behavior on disk errors", async () => { + it("uses legacy JSON paths only as SQLite namespace identifiers", async () => { + const root = await createTempDir("openclaw-legacy-dedupe-"); + const legacyPath = path.join(root, "legacy.json"); const dedupe = createPersistentDedupe({ ttlMs: 10_000, memoryMaxSize: 100, fileMaxEntries: 1000, - resolveFilePath: () => path.join("/dev/null", "dedupe.json"), + resolveFilePath: () => legacyPath, + env: { ...process.env, OPENCLAW_STATE_DIR: root }, }); - expect(await dedupe.checkAndRecord("memory-only", { namespace: "x" })).toBe(true); - expect(await dedupe.checkAndRecord("memory-only", { namespace: "x" })).toBe(false); + expect(await dedupe.checkAndRecord("sqlite-only", { namespace: "x" })).toBe(true); + expect(await dedupe.checkAndRecord("sqlite-only", { namespace: "x" })).toBe(false); + await expect(fs.access(legacyPath)).rejects.toThrow(); }); - it("warms empty namespaces and skips expired disk entries", async () => { + it("warms empty namespaces and ignores retired JSON cache files", async () => { const root = await createTempDir("openclaw-dedupe-"); const emptyReader = createDedupe(root, { ttlMs: 10_000 }); expect(await emptyReader.warmup("nonexistent")).toBe(0); - const oldNow = Date.now() - 2000; - await fs.writeFile( - path.join(root, "acct.json"), - JSON.stringify({ "old-msg": oldNow, "new-msg": Date.now() }), - ); + await fs.writeFile(path.join(root, "acct.json"), JSON.stringify({ "retired-msg": Date.now() })); const reader = createDedupe(root, { ttlMs: 1000 }); - expect(await reader.warmup("acct")).toBe(1); - expect(await reader.checkAndRecord("old-msg", { namespace: "acct" })).toBe(true); - expect(await reader.checkAndRecord("new-msg", { namespace: "acct" })).toBe(false); + expect(await reader.warmup("acct")).toBe(0); + expect(await reader.checkAndRecord("retired-msg", { namespace: "acct" })).toBe(true); }); }); @@ -189,8 +197,10 @@ describe("createClaimableDedupe", () => { const writer = createClaimableDedupe({ ttlMs: 10_000, memoryMaxSize: 100, - fileMaxEntries: 1000, - resolveFilePath: (namespace) => path.join(root, `${namespace}.json`), + pluginId: "test-claimable-dedupe", + namespacePrefix: "test-claimable-dedupe", + stateMaxEntries: 1000, + env: { ...process.env, OPENCLAW_STATE_DIR: root }, }); await expect(writer.claim("m1", { namespace: "acct" })).resolves.toEqual({ kind: "claimed" }); @@ -199,8 +209,10 @@ describe("createClaimableDedupe", () => { const reader = createClaimableDedupe({ ttlMs: 10_000, memoryMaxSize: 100, - fileMaxEntries: 1000, - resolveFilePath: (namespace) => path.join(root, `${namespace}.json`), + pluginId: "test-claimable-dedupe", + namespacePrefix: "test-claimable-dedupe", + stateMaxEntries: 1000, + env: { ...process.env, OPENCLAW_STATE_DIR: root }, }); expect(await reader.hasRecent("m1", { namespace: "acct" })).toBe(true); diff --git a/src/plugin-sdk/persistent-dedupe.ts b/src/plugin-sdk/persistent-dedupe.ts index 6e90c8f9db8e..7aafa6f919dc 100644 --- a/src/plugin-sdk/persistent-dedupe.ts +++ b/src/plugin-sdk/persistent-dedupe.ts @@ -1,26 +1,81 @@ // Persistent dedupe helpers give plugins bounded replay protection across process restarts. +import { createHash } from "node:crypto"; +import fs from "node:fs/promises"; import { createDedupeCache } from "../infra/dedupe.js"; import { resolveNonNegativeIntegerOption } from "../infra/numeric-options.js"; +import { + createCorePluginStateSyncKeyedStore, + createPluginStateSyncKeyedStore, +} from "../plugin-state/plugin-state-store.js"; +import type { PluginStateSyncKeyedStore } from "../plugin-state/plugin-state-store.types.js"; import type { FileLockOptions } from "./file-lock.js"; -import { withFileLock } from "./file-lock.js"; -import { readJsonFileWithFallback, writeJsonFileAtomically } from "./json-store.js"; -type PersistentDedupeData = Record; +const LEGACY_PATH_OWNER_ID = "core:persistent-dedupe"; +const DEFAULT_NAMESPACE_PREFIX = "persistent-dedupe"; -/** Configuration for a disk-backed dedupe namespace cache. */ -export type PersistentDedupeOptions = { +type PersistentDedupeEntry = { + key: string; + seenAt: number; +}; + +type PersistentDedupeBaseOptions = { /** Milliseconds a recorded key remains recent; `0` keeps keys until cache pruning. */ ttlMs: number; - /** Maximum process-local cache entries used before consulting disk. */ + /** Maximum process-local cache entries used before consulting SQLite. */ memoryMaxSize: number; - /** Maximum persisted entries retained per namespace file. */ - fileMaxEntries: number; - /** Maps a namespace to the JSON file that stores its persisted dedupe timestamps. */ - resolveFilePath: (namespace: string) => string; - lockOptions?: Partial; onDiskError?: (error: unknown) => void; }; +/** Configuration for a SQLite plugin-state dedupe namespace cache. */ +export type PersistentDedupePluginStateOptions = PersistentDedupeBaseOptions & { + /** Plugin id that owns the persisted dedupe namespace. */ + pluginId: string; + /** Prefix for persisted plugin-state namespaces; defaults to `persistent-dedupe`. */ + namespacePrefix?: string; + /** Maximum persisted entries retained per namespace. */ + stateMaxEntries: number; + /** Test/runtime env used to resolve the shared OpenClaw state database. */ + env?: NodeJS.ProcessEnv; + resolveFilePath?: undefined; + fileMaxEntries?: undefined; + lockOptions?: undefined; +}; + +/** Legacy path-shaped configuration. Paths now name SQLite namespaces, not JSON files. */ +export type PersistentDedupeLegacyPathOptions = PersistentDedupeBaseOptions & { + pluginId?: undefined; + stateMaxEntries?: undefined; + namespacePrefix?: undefined; + /** Maximum persisted entries retained per legacy namespace. */ + fileMaxEntries: number; + /** Maps a namespace to the retired JSON path; used only to derive a stable SQLite namespace. */ + resolveFilePath: (namespace: string) => string; + /** Test/runtime env used to resolve the shared OpenClaw state database. */ + env?: NodeJS.ProcessEnv; + /** @deprecated File locks are ignored because persistence is SQLite-backed. */ + lockOptions?: Partial; +}; + +/** Configuration for a persisted dedupe namespace cache. */ +export type PersistentDedupeOptions = + | PersistentDedupePluginStateOptions + | PersistentDedupeLegacyPathOptions; + +export type PersistentDedupeLegacyJsonMigrationResult = { + imported: number; + skippedExpired: number; + skippedInvalid: number; + skippedExisting: number; + removed: boolean; +}; + +export type PersistentDedupeLegacyJsonMigrationOptions = PersistentDedupePluginStateOptions & { + filePath: string; + namespace: string; + now?: number; + removeFile?: boolean; +}; + /** Per-call options used when checking or recording a dedupe key. */ export type PersistentDedupeCheckOptions = { /** Logical bucket for the key; omitted/blank values use `global`. */ @@ -53,17 +108,15 @@ export type ClaimableDedupeClaimResult = /** Options for a claimable dedupe guard, either persistent or memory-only. */ export type ClaimableDedupeOptions = + | PersistentDedupePluginStateOptions + | PersistentDedupeLegacyPathOptions | { ttlMs: number; memoryMaxSize: number; - resolveFilePath: (namespace: string) => string; - fileMaxEntries: number; - lockOptions?: Partial; - onDiskError?: (error: unknown) => void; - } - | { - ttlMs: number; - memoryMaxSize: number; + pluginId?: undefined; + stateMaxEntries?: undefined; + namespacePrefix?: undefined; + env?: undefined; resolveFilePath?: undefined; fileMaxEntries?: undefined; lockOptions?: undefined; @@ -97,70 +150,6 @@ export type ClaimableDedupe = { memorySize: () => number; }; -const DEFAULT_LOCK_OPTIONS: FileLockOptions = { - retries: { - retries: 6, - factor: 1.35, - minTimeout: 8, - maxTimeout: 180, - randomize: true, - }, - stale: 60_000, -}; - -function mergeLockOptions(overrides?: Partial): FileLockOptions { - return { - stale: overrides?.stale ?? DEFAULT_LOCK_OPTIONS.stale, - retries: { - retries: overrides?.retries?.retries ?? DEFAULT_LOCK_OPTIONS.retries.retries, - factor: overrides?.retries?.factor ?? DEFAULT_LOCK_OPTIONS.retries.factor, - minTimeout: overrides?.retries?.minTimeout ?? DEFAULT_LOCK_OPTIONS.retries.minTimeout, - maxTimeout: overrides?.retries?.maxTimeout ?? DEFAULT_LOCK_OPTIONS.retries.maxTimeout, - randomize: overrides?.retries?.randomize ?? DEFAULT_LOCK_OPTIONS.retries.randomize, - }, - }; -} - -function sanitizeData(value: unknown): PersistentDedupeData { - if (!value || typeof value !== "object") { - return {}; - } - const out: PersistentDedupeData = {}; - for (const [key, ts] of Object.entries(value as Record)) { - if (typeof ts === "number" && Number.isFinite(ts) && ts > 0) { - out[key] = ts; - } - } - return out; -} - -function pruneData( - data: PersistentDedupeData, - now: number, - ttlMs: number, - maxEntries: number, -): void { - if (ttlMs > 0) { - for (const [key, ts] of Object.entries(data)) { - if (now - ts >= ttlMs) { - delete data[key]; - } - } - } - - const keys = Object.keys(data); - if (keys.length <= maxEntries) { - return; - } - - keys - .toSorted((a, b) => data[a] - data[b]) - .slice(0, keys.length - maxEntries) - .forEach((key) => { - delete data[key]; - }); -} - function resolveNamespace(namespace?: string): string { return namespace?.trim() || "global"; } @@ -173,43 +162,195 @@ function isRecentTimestamp(seenAt: number | undefined, ttlMs: number, now: numbe return seenAt != null && (ttlMs <= 0 || now - seenAt < ttlMs); } -/** Create a dedupe helper that combines in-memory fast checks with a lock-protected disk store. */ +function resolveEntrySeenAt(entry: PersistentDedupeEntry | undefined): number | undefined { + return typeof entry?.seenAt === "number" && Number.isFinite(entry.seenAt) + ? entry.seenAt + : undefined; +} + +function shortHash(value: string): string { + return createHash("sha256").update(value).digest("hex").slice(0, 32); +} + +function resolveEntryKey(key: string): string { + return `k.${shortHash(key)}`; +} + +function resolveRemainingTtlMs( + seenAt: number, + ttlMs: number, + now: number, +): { ttlMs: number } | undefined | null { + if (ttlMs <= 0) { + return undefined; + } + const remaining = ttlMs - (now - seenAt); + return remaining > 0 ? { ttlMs: Math.max(1, Math.floor(remaining)) } : null; +} + +function normalizeNamespacePrefix(value: string | undefined): string { + const normalized = (value ?? DEFAULT_NAMESPACE_PREFIX) + .trim() + .toLowerCase() + .replace(/[^a-z0-9._-]+/g, "-") + .replace(/^[._-]+|[._-]+$/g, "") + .slice(0, 48); + return normalized || DEFAULT_NAMESPACE_PREFIX; +} + +function resolveStateNamespace(prefix: string, namespace: string): string { + return `${prefix}.${shortHash(namespace)}`; +} + +function hasPluginStateOptions( + options: ClaimableDedupeOptions | PersistentDedupeOptions, +): options is PersistentDedupePluginStateOptions { + return typeof options.pluginId === "string"; +} + +function hasLegacyPathOptions( + options: ClaimableDedupeOptions | PersistentDedupeOptions, +): options is PersistentDedupeLegacyPathOptions { + return typeof options.resolveFilePath === "function"; +} + +function resolveStateMaxEntries(options: PersistentDedupeOptions): number { + const maxEntries = hasPluginStateOptions(options) + ? options.stateMaxEntries + : options.fileMaxEntries; + return Math.max(1, resolveNonNegativeIntegerOption(maxEntries, 1)); +} + +function resolvePersistentStoreCacheKey(pluginId: string, namespace: string): string { + return `${pluginId}\0${namespace}`; +} + +function createPersistentStoreResolver( + options: PersistentDedupeOptions, +): (namespace: string) => PluginStateSyncKeyedStore { + const maxEntries = resolveStateMaxEntries(options); + const ttlMs = resolveNonNegativeIntegerOption(options.ttlMs, 0); + const defaultTtlMs = ttlMs > 0 ? ttlMs : undefined; + const stores = new Map>(); + + if (hasPluginStateOptions(options)) { + const pluginId = options.pluginId; + const prefix = normalizeNamespacePrefix(options.namespacePrefix); + return (namespace) => { + const stateNamespace = resolveStateNamespace(prefix, namespace); + const cacheKey = resolvePersistentStoreCacheKey(pluginId, stateNamespace); + const existing = stores.get(cacheKey); + if (existing) { + return existing; + } + const store = createPluginStateSyncKeyedStore(pluginId, { + namespace: stateNamespace, + maxEntries, + ...(defaultTtlMs != null ? { defaultTtlMs } : {}), + ...(options.env ? { env: options.env } : {}), + }); + stores.set(cacheKey, store); + return store; + }; + } + + const prefix = normalizeNamespacePrefix("legacy-path"); + return (namespace) => { + const legacyPath = options.resolveFilePath(namespace); + const stateNamespace = resolveStateNamespace(prefix, legacyPath); + const cacheKey = resolvePersistentStoreCacheKey(LEGACY_PATH_OWNER_ID, stateNamespace); + const existing = stores.get(cacheKey); + if (existing) { + return existing; + } + const store = createCorePluginStateSyncKeyedStore({ + ownerId: LEGACY_PATH_OWNER_ID, + namespace: stateNamespace, + maxEntries, + ...(defaultTtlMs != null ? { defaultTtlMs } : {}), + ...(options.env ? { env: options.env } : {}), + }); + stores.set(cacheKey, store); + return store; + }; +} + +function parseLegacyDedupeData(raw: string): { + data: Record; + invalidCount: number; +} { + const parsed = JSON.parse(raw) as unknown; + if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) { + return { data: {}, invalidCount: 0 }; + } + const data: Record = {}; + let invalidCount = 0; + for (const [key, seenAt] of Object.entries(parsed)) { + if (typeof seenAt === "number" && Number.isFinite(seenAt) && seenAt > 0) { + data[key] = seenAt; + } else { + invalidCount++; + } + } + return { data, invalidCount }; +} + +/** Import one retired JSON dedupe cache file into plugin-state SQLite during doctor repair. */ +export async function migratePersistentDedupeLegacyJsonFile( + options: PersistentDedupeLegacyJsonMigrationOptions, +): Promise { + const raw = await fs.readFile(options.filePath, "utf8"); + const { data, invalidCount } = parseLegacyDedupeData(raw); + const ttlMs = resolveNonNegativeIntegerOption(options.ttlMs, 0); + const now = options.now ?? Date.now(); + const store = createPersistentStoreResolver(options)(resolveNamespace(options.namespace)); + const result: PersistentDedupeLegacyJsonMigrationResult = { + imported: 0, + skippedExpired: 0, + skippedInvalid: 0, + skippedExisting: 0, + removed: false, + }; + + for (const [key, seenAt] of Object.entries(data)) { + const ttlOption = resolveRemainingTtlMs(seenAt, ttlMs, now); + if (ttlOption === null) { + result.skippedExpired++; + continue; + } + const changed = store.update?.( + resolveEntryKey(key), + (current) => { + const currentSeenAt = resolveEntrySeenAt(current); + if (currentSeenAt != null && currentSeenAt >= seenAt) { + return undefined; + } + return { key, seenAt }; + }, + ttlOption, + ); + if (changed) { + result.imported++; + } else { + result.skippedExisting++; + } + } + + result.skippedInvalid = invalidCount; + if (options.removeFile !== false) { + await fs.rm(options.filePath, { force: true }); + result.removed = true; + } + return result; +} + +/** Create a dedupe helper that combines in-memory fast checks with SQLite-backed state. */ export function createPersistentDedupe(options: PersistentDedupeOptions): PersistentDedupe { const ttlMs = resolveNonNegativeIntegerOption(options.ttlMs, 0); const memoryMaxSize = resolveNonNegativeIntegerOption(options.memoryMaxSize, 0); - const fileMaxEntries = Math.max(1, resolveNonNegativeIntegerOption(options.fileMaxEntries, 1)); - const lockOptions = mergeLockOptions(options.lockOptions); + const getStore = createPersistentStoreResolver(options); const memory = createDedupeCache({ ttlMs, maxSize: memoryMaxSize }); const inflight = new Map>(); - // In-process write queue per file path. `withFileLock` is re-entrant - // within the same process (a second caller for the same path gets - // immediate access instead of waiting), so two concurrent - // checkAndRecordInner calls for different keys but the same file can - // race: both read the same stale data, and the last writer's - // writeJsonFileAtomically silently overwrites the first writer's - // additions. This queue serializes all read-modify-write cycles - // targeting the same file within this process, preventing the lost - // update while still allowing cross-process file-lock contention to - // be handled by the file lock itself. - const fileWriteQueues = new Map>(); - - function enqueueFileWrite(filePath: string, fn: () => Promise): Promise { - const prev = fileWriteQueues.get(filePath) ?? Promise.resolve(); - const next = prev.then(fn, fn); - fileWriteQueues.set(filePath, next); - // Cleanup: remove the queue entry once this link settles, but only if - // no newer work was chained after us. The `.catch(() => {})` prevents - // an unhandled rejection when `next` rejects — callers still observe - // the rejection through the returned `next` promise directly. - next - .finally(() => { - if (fileWriteQueues.get(filePath) === next) { - fileWriteQueues.delete(filePath); - } - }) - .catch(() => {}); - return next; - } async function checkAndRecordInner( key: string, @@ -222,24 +363,28 @@ export function createPersistentDedupe(options: PersistentDedupeOptions): Persis return false; } - const path = options.resolveFilePath(namespace); try { - const duplicate = await enqueueFileWrite(path, () => - withFileLock(path, lockOptions, async () => { - const { value } = await readJsonFileWithFallback(path, {}); - const data = sanitizeData(value); - const seenAt = data[key]; - const isRecent = seenAt != null && (ttlMs <= 0 || now - seenAt < ttlMs); - if (isRecent) { - return true; + const entryKey = resolveEntryKey(key); + const store = getStore(namespace); + let duplicateSeenAt: number | undefined; + store.update?.( + entryKey, + (entry) => { + const seenAt = resolveEntrySeenAt(entry); + if (isRecentTimestamp(seenAt, ttlMs, now)) { + duplicateSeenAt = seenAt; + return undefined; } - data[key] = now; - pruneData(data, now, ttlMs, fileMaxEntries); - await writeJsonFileAtomically(path, data); - return false; - }), + return { key, seenAt: now }; + }, + ttlMs > 0 ? { ttlMs } : undefined, ); - return !duplicate; + if (duplicateSeenAt != null) { + memory.check(scopedKey, duplicateSeenAt); + return false; + } + memory.check(scopedKey, now); + return true; } catch (error) { onDiskError?.(error); memory.check(scopedKey, now); @@ -258,11 +403,8 @@ export function createPersistentDedupe(options: PersistentDedupeOptions): Persis return true; } - const path = options.resolveFilePath(namespace); try { - const { value } = await readJsonFileWithFallback(path, {}); - const data = sanitizeData(value); - const seenAt = data[key]; + const seenAt = resolveEntrySeenAt(getStore(namespace).lookup(resolveEntryKey(key))); if (!isRecentTimestamp(seenAt, ttlMs, now)) { return false; } @@ -275,17 +417,18 @@ export function createPersistentDedupe(options: PersistentDedupeOptions): Persis } async function warmup(namespace = "global", onError?: (error: unknown) => void): Promise { - const filePath = options.resolveFilePath(namespace); const now = Date.now(); try { - const { value } = await readJsonFileWithFallback(filePath, {}); - const data = sanitizeData(value); let loaded = 0; - for (const [key, ts] of Object.entries(data)) { + for (const entry of getStore(resolveNamespace(namespace)).entries()) { + const ts = resolveEntrySeenAt(entry.value); + if (ts == null) { + continue; + } if (ttlMs > 0 && now - ts >= ttlMs) { continue; } - const scopedKey = `${namespace}:${key}`; + const scopedKey = `${resolveNamespace(namespace)}:${entry.value.key}`; memory.check(scopedKey, ts); loaded++; } @@ -354,17 +497,28 @@ export function createClaimableDedupe(options: ClaimableDedupeOptions): Claimabl const ttlMs = resolveNonNegativeIntegerOption(options.ttlMs, 0); const memoryMaxSize = resolveNonNegativeIntegerOption(options.memoryMaxSize, 0); const memory = createDedupeCache({ ttlMs, maxSize: memoryMaxSize }); - const persistent = - options.resolveFilePath != null - ? createPersistentDedupe({ - ttlMs, - memoryMaxSize, - fileMaxEntries: Math.max(1, resolveNonNegativeIntegerOption(options.fileMaxEntries, 1)), - resolveFilePath: options.resolveFilePath, - lockOptions: options.lockOptions, - onDiskError: options.onDiskError, - }) - : null; + let persistent: PersistentDedupe | null = null; + if (hasPluginStateOptions(options)) { + persistent = createPersistentDedupe({ + ttlMs, + memoryMaxSize, + pluginId: options.pluginId, + stateMaxEntries: Math.max(1, resolveNonNegativeIntegerOption(options.stateMaxEntries, 1)), + ...(options.namespacePrefix ? { namespacePrefix: options.namespacePrefix } : {}), + ...(options.env ? { env: options.env } : {}), + ...(options.onDiskError ? { onDiskError: options.onDiskError } : {}), + }); + } else if (hasLegacyPathOptions(options)) { + persistent = createPersistentDedupe({ + ttlMs, + memoryMaxSize, + fileMaxEntries: Math.max(1, resolveNonNegativeIntegerOption(options.fileMaxEntries, 1)), + resolveFilePath: options.resolveFilePath, + ...(options.env ? { env: options.env } : {}), + ...(options.lockOptions ? { lockOptions: options.lockOptions } : {}), + ...(options.onDiskError ? { onDiskError: options.onDiskError } : {}), + }); + } const inflight = new Map< string,