diff --git a/docs/refactor/database-first.md b/docs/refactor/database-first.md index d84724f1374d..c6a06a8fe5c3 100644 --- a/docs/refactor/database-first.md +++ b/docs/refactor/database-first.md @@ -1129,9 +1129,10 @@ sessionId})`; create, branch, continue, list, and fork flows live in their now use shared SQLite plugin state. The old `imessage/catchup/*.json`, `imessage/reply-cache.jsonl`, and `imessage/sent-echoes.jsonl` files are doctor inputs only. -- Feishu message dedupe rows now use shared SQLite plugin state instead of - `feishu/dedup/*.json` files. Its legacy JSON import plan lives in the Feishu - plugin setup/doctor migration surface, not in core migration code. +- Feishu message dedupe rows now ride the core claimable dedupe + (`feishu.dedup.*` namespaces in shared SQLite plugin state) instead of + `feishu/dedup/*.json` files or the retired hand-rolled `dedup.*` store, with + no legacy import because replay-protection cache rebuilds after upgrade. - Microsoft Teams conversations, polls, pending upload buffers, and feedback learnings now use shared SQLite plugin state/blob tables. The pending upload path uses `plugin_blob_entries` so media buffers are stored as SQLite BLOBs diff --git a/extensions/feishu/legacy-state-migrations-api.ts b/extensions/feishu/legacy-state-migrations-api.ts deleted file mode 100644 index 2bfc9adb5925..000000000000 --- a/extensions/feishu/legacy-state-migrations-api.ts +++ /dev/null @@ -1,2 +0,0 @@ -// Feishu API module exposes the plugin public contract. -export { detectFeishuLegacyStateMigrations } from "./src/dedup-migrations.js"; diff --git a/extensions/feishu/package.json b/extensions/feishu/package.json index 8a2a0972ec65..4d02540a727e 100644 --- a/extensions/feishu/package.json +++ b/extensions/feishu/package.json @@ -29,9 +29,6 @@ "./index.ts" ], "setupEntry": "./setup-entry.ts", - "setupFeatures": { - "legacyStateMigrations": true - }, "channel": { "id": "feishu", "label": "Feishu", diff --git a/extensions/feishu/setup-entry.test.ts b/extensions/feishu/setup-entry.test.ts index 718248be9077..0d5673dd47c5 100644 --- a/extensions/feishu/setup-entry.test.ts +++ b/extensions/feishu/setup-entry.test.ts @@ -15,9 +15,9 @@ describe("feishu setup entry", () => { const { default: setupEntry } = await import("./setup-entry.js"); expect(setupEntry.kind).toBe("bundled-channel-setup-entry"); - expect(setupEntry.features).toEqual({ legacyStateMigrations: true }); + expect(setupEntry.features).toBeUndefined(); expect(typeof setupEntry.loadSetupPlugin).toBe("function"); - expect(setupEntry.loadLegacyStateMigrationDetector?.()).toBeTypeOf("function"); + expect(setupEntry.loadLegacyStateMigrationDetector).toBeUndefined(); expect(typeof setupEntry.setChannelRuntime).toBe("function"); }); diff --git a/extensions/feishu/setup-entry.ts b/extensions/feishu/setup-entry.ts index 143ff61cdc06..4b76ee798cb0 100644 --- a/extensions/feishu/setup-entry.ts +++ b/extensions/feishu/setup-entry.ts @@ -3,17 +3,10 @@ import { defineBundledChannelSetupEntry } from "openclaw/plugin-sdk/channel-entr export default defineBundledChannelSetupEntry({ importMetaUrl: import.meta.url, - features: { - legacyStateMigrations: true, - }, plugin: { specifier: "./setup-api.js", exportName: "feishuPlugin", }, - legacyStateMigrations: { - specifier: "./legacy-state-migrations-api.js", - exportName: "detectFeishuLegacyStateMigrations", - }, secrets: { specifier: "./secret-contract-api.js", exportName: "channelSecrets", diff --git a/extensions/feishu/src/bot.ts b/extensions/feishu/src/bot.ts index 43278cc3e439..8aa31f9b97d4 100644 --- a/extensions/feishu/src/bot.ts +++ b/extensions/feishu/src/bot.ts @@ -51,7 +51,7 @@ import { type FeishuPermissionError, resolveFeishuSenderName } from "./bot-sende import { getChatInfo } from "./chat.js"; import { createFeishuClient } from "./client.js"; import { resolveConfiguredFeishuGroupSessionScope } from "./conversation-id.js"; -import { finalizeFeishuMessageProcessing, tryRecordMessagePersistent } from "./dedup.js"; +import { finalizeFeishuMessageProcessing, recordProcessedFeishuMessage } from "./dedup.js"; import { resolveFeishuMessageDedupeKey } from "./dedupe-key.js"; import { maybeCreateDynamicAgent } from "./dynamic-agent.js"; import { extractMentionTargets, isMentionForwardRequest } from "./mention.js"; @@ -1534,7 +1534,7 @@ export async function handleFeishuMessage(params: { // Uses a shared "broadcast" namespace (not per-account) so the first handler // to reach this point claims the message; subsequent accounts skip. if ( - !(await tryRecordMessagePersistent(messageDedupeKey ?? ctx.messageId, "broadcast", log)) + !(await recordProcessedFeishuMessage(messageDedupeKey ?? ctx.messageId, "broadcast", log)) ) { log( `feishu[${account.accountId}]: broadcast already claimed by another account for message ${ctx.messageId}; skipping`, diff --git a/extensions/feishu/src/dedup-migrations.test.ts b/extensions/feishu/src/dedup-migrations.test.ts deleted file mode 100644 index 711abfb36586..000000000000 --- a/extensions/feishu/src/dedup-migrations.test.ts +++ /dev/null @@ -1,90 +0,0 @@ -// Feishu tests cover dedup migrations plugin behavior. -import fs from "node:fs/promises"; -import os from "node:os"; -import path from "node:path"; -import { afterEach, describe, expect, it, vi } from "vitest"; -import { detectFeishuLegacyStateMigrations } from "./dedup-migrations.js"; - -const tempDirs: string[] = []; - -async function makeStateDir(): Promise { - const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-feishu-dedup-migration-")); - tempDirs.push(dir); - return dir; -} - -afterEach(async () => { - vi.useRealTimers(); - await Promise.all(tempDirs.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true }))); -}); - -describe("Feishu dedupe migration", () => { - it("plans recent legacy dedupe rows with remaining TTL", async () => { - vi.useFakeTimers(); - vi.setSystemTime(2_000); - const stateDir = await makeStateDir(); - const sourcePath = path.join(stateDir, "feishu", "dedup", "account-a.json"); - await fs.mkdir(path.dirname(sourcePath), { recursive: true }); - await fs.writeFile( - sourcePath, - JSON.stringify({ - fresh: 1_000, - expired: 2_000 - 24 * 60 * 60 * 1000, - malformed: "nope", - }), - ); - - const plans = await Promise.resolve( - detectFeishuLegacyStateMigrations({ - cfg: {}, - env: {}, - oauthDir: path.join(stateDir, "credentials"), - stateDir, - }), - ); - - if (!plans) { - throw new Error("expected migration plans"); - } - expect(plans).toHaveLength(1); - const plan = plans[0]; - expect(plan?.kind).toBe("plugin-state-import"); - if (plan?.kind !== "plugin-state-import") { - throw new Error("expected plugin-state import plan"); - } - expect(plan.pluginId).toBe("feishu"); - expect(plan.namespace).toBe("dedup.account-a"); - const entries = await plan.readEntries(); - expect(entries).toHaveLength(1); - expect(entries[0]?.key).toMatch(/^[0-9a-f]{32}$/u); - expect(entries[0]?.value).toEqual({ - namespace: "account-a", - messageId: "fresh", - seenAt: 1_000, - }); - expect(entries[0]?.ttlMs).toBe(24 * 60 * 60 * 1000 - 1_000); - }); - - it("skips expired-only legacy dedupe files", async () => { - vi.useFakeTimers(); - vi.setSystemTime(2_000); - const stateDir = await makeStateDir(); - const sourcePath = path.join(stateDir, "feishu", "dedup", "account-a.json"); - await fs.mkdir(path.dirname(sourcePath), { recursive: true }); - await fs.writeFile( - sourcePath, - JSON.stringify({ - expired: 2_000 - 24 * 60 * 60 * 1000, - }), - ); - - expect( - detectFeishuLegacyStateMigrations({ - cfg: {}, - env: {}, - oauthDir: path.join(stateDir, "credentials"), - stateDir, - }), - ).toStrictEqual([]); - }); -}); diff --git a/extensions/feishu/src/dedup-migrations.ts b/extensions/feishu/src/dedup-migrations.ts deleted file mode 100644 index 14b674f4cb31..000000000000 --- a/extensions/feishu/src/dedup-migrations.ts +++ /dev/null @@ -1,103 +0,0 @@ -// Feishu plugin module implements dedup migrations behavior. -import { createHash } from "node:crypto"; -import fs from "node:fs"; -import path from "node:path"; -import type { BundledChannelLegacyStateMigrationDetector } from "openclaw/plugin-sdk/channel-entry-contract"; - -const DEDUP_TTL_MS = 24 * 60 * 60 * 1000; -const STORE_MAX_ENTRIES = 10_000; - -type LegacyDedupeData = Record; - -function safeNamespaceFromFileName(fileName: string): string | null { - if (!fileName.endsWith(".json")) { - return null; - } - const namespace = fileName.slice(0, -".json".length).trim(); - return namespace ? namespace : null; -} - -function readLegacyDedupeData(filePath: string): LegacyDedupeData { - try { - const parsed = JSON.parse(fs.readFileSync(filePath, "utf8")) as unknown; - if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) { - return {}; - } - const out: LegacyDedupeData = {}; - for (const [messageId, seenAt] of Object.entries(parsed as Record)) { - if (typeof seenAt === "number" && Number.isFinite(seenAt) && seenAt > 0) { - out[messageId] = seenAt; - } - } - return out; - } catch { - return {}; - } -} - -function dedupeStoreKey(namespace: string, messageId: string): string { - return createHash("sha256") - .update(`${namespace}\0${messageId}`, "utf8") - .digest("hex") - .slice(0, 32); -} - -function remainingTtlMs(seenAt: number, now: number): number { - return Math.max(1, DEDUP_TTL_MS - (now - seenAt)); -} - -function buildMigrationEntries(namespace: string, sourcePath: string, now: number) { - return Object.entries(readLegacyDedupeData(sourcePath)).flatMap(([messageId, seenAt]) => { - if (now - seenAt >= DEDUP_TTL_MS) { - return []; - } - return [ - { - key: dedupeStoreKey(namespace, messageId), - value: { namespace, messageId, seenAt }, - ttlMs: remainingTtlMs(seenAt, now), - }, - ]; - }); -} - -export const detectFeishuLegacyStateMigrations: BundledChannelLegacyStateMigrationDetector = ({ - stateDir, -}) => { - const dedupDir = path.join(stateDir, "feishu", "dedup"); - let entries: fs.Dirent[]; - try { - entries = fs.readdirSync(dedupDir, { withFileTypes: true }); - } catch { - return []; - } - const now = Date.now(); - return entries.flatMap((entry) => { - if (!entry.isFile()) { - return []; - } - const namespace = safeNamespaceFromFileName(entry.name); - if (!namespace) { - return []; - } - const sourcePath = path.join(dedupDir, entry.name); - const migrationEntries = buildMigrationEntries(namespace, sourcePath, now); - if (migrationEntries.length === 0) { - return []; - } - return [ - { - kind: "plugin-state-import" as const, - label: `Feishu ${namespace} dedupe`, - sourcePath, - targetPath: `plugin state:dedup.${namespace}`, - pluginId: "feishu", - namespace: `dedup.${namespace}`, - maxEntries: STORE_MAX_ENTRIES, - scopeKey: "", - cleanupSource: "rename" as const, - readEntries: () => buildMigrationEntries(namespace, sourcePath, now), - }, - ]; - }); -}; diff --git a/extensions/feishu/src/dedup.test.ts b/extensions/feishu/src/dedup.test.ts index 152592fac8ce..c4c050a74ca3 100644 --- a/extensions/feishu/src/dedup.test.ts +++ b/extensions/feishu/src/dedup.test.ts @@ -1,39 +1,30 @@ // Feishu tests cover dedup plugin behavior. -import fs from "node:fs/promises"; +import fs from "node:fs"; import os from "node:os"; import path from "node:path"; -import type { OpenKeyedStoreOptions } from "openclaw/plugin-sdk/plugin-state-runtime"; -import { - createPluginStateSyncKeyedStoreForTests, - resetPluginStateStoreForTests, -} from "openclaw/plugin-sdk/plugin-state-test-runtime"; +import { resetPluginStateStoreForTests } from "openclaw/plugin-sdk/plugin-state-test-runtime"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import type { PluginRuntime } from "../runtime-api.js"; import { + claimUnprocessedFeishuMessage, + finalizeFeishuMessageProcessing, hasProcessedFeishuMessage, + recordProcessedFeishuMessage, + releaseFeishuMessageProcessing, testingHooks, - tryRecordMessagePersistent, warmupDedupFromPluginState, } from "./dedup.js"; -import { setFeishuRuntime } from "./runtime.js"; let tempDir: string | undefined; let previousStateDir: string | undefined; -beforeEach(async () => { +beforeEach(() => { previousStateDir = process.env.OPENCLAW_STATE_DIR; - tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-feishu-dedup-")); + tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-feishu-dedup-")); process.env.OPENCLAW_STATE_DIR = tempDir; - setFeishuRuntime({ - state: { - openSyncKeyedStore: (options: OpenKeyedStoreOptions) => - createPluginStateSyncKeyedStoreForTests("feishu", options), - }, - } as unknown as PluginRuntime); testingHooks.resetFeishuDedupForTests(); }); -afterEach(async () => { +afterEach(() => { vi.useRealTimers(); testingHooks.resetFeishuDedupForTests(); resetPluginStateStoreForTests(); @@ -43,53 +34,132 @@ afterEach(async () => { process.env.OPENCLAW_STATE_DIR = previousStateDir; } if (tempDir) { - await fs.rm(tempDir, { recursive: true, force: true }); + fs.rmSync(tempDir, { recursive: true, force: true }); } tempDir = undefined; }); -describe("Feishu persistent dedupe", () => { - it("records message ids in plugin state", async () => { - await expect(tryRecordMessagePersistent("msg-1", "account-a")).resolves.toBe(true); - await expect(tryRecordMessagePersistent("msg-1", "account-a")).resolves.toBe(false); +// Simulates a process restart: a fresh guard has empty memory and no in-flight +// claims, so any duplicate verdict must come from the persisted SQLite rows. +function restartFeishuDedup(): void { + testingHooks.resetFeishuDedupForTests(); +} + +describe("Feishu claimable dedupe", () => { + it("drops a duplicate message within the TTL after commit", async () => { + await expect( + claimUnprocessedFeishuMessage({ messageId: "msg-1", namespace: "account-a" }), + ).resolves.toBe("claimed"); + await expect(recordProcessedFeishuMessage("msg-1", "account-a")).resolves.toBe(true); + + await expect( + claimUnprocessedFeishuMessage({ messageId: "msg-1", namespace: "account-a" }), + ).resolves.toBe("duplicate"); await expect(hasProcessedFeishuMessage("msg-1", "account-a")).resolves.toBe(true); await expect(hasProcessedFeishuMessage("msg-1", "account-b")).resolves.toBe(false); }); + it("reports an in-flight claim and lets a released claim retry", async () => { + await expect( + claimUnprocessedFeishuMessage({ messageId: "msg-2", namespace: "account-a" }), + ).resolves.toBe("claimed"); + await expect( + claimUnprocessedFeishuMessage({ messageId: "msg-2", namespace: "account-a" }), + ).resolves.toBe("inflight"); + + releaseFeishuMessageProcessing("msg-2", "account-a"); + await expect( + claimUnprocessedFeishuMessage({ messageId: "msg-2", namespace: "account-a" }), + ).resolves.toBe("claimed"); + }); + + it("does not persist released claims across a restart", async () => { + await expect( + claimUnprocessedFeishuMessage({ messageId: "msg-3", namespace: "account-a" }), + ).resolves.toBe("claimed"); + releaseFeishuMessageProcessing("msg-3", "account-a"); + + restartFeishuDedup(); + await expect( + claimUnprocessedFeishuMessage({ messageId: "msg-3", namespace: "account-a" }), + ).resolves.toBe("claimed"); + }); + + it("prevents replay after a restart once a message is committed", async () => { + await expect( + finalizeFeishuMessageProcessing({ messageId: "msg-4", namespace: "account-a" }), + ).resolves.toBe(true); + + restartFeishuDedup(); + await expect( + claimUnprocessedFeishuMessage({ messageId: "msg-4", namespace: "account-a" }), + ).resolves.toBe("duplicate"); + await expect( + finalizeFeishuMessageProcessing({ messageId: "msg-4", namespace: "account-a" }), + ).resolves.toBe(false); + }); + + it("commits a held claim without reclaiming it", async () => { + await expect( + claimUnprocessedFeishuMessage({ messageId: "msg-5", namespace: "account-a" }), + ).resolves.toBe("claimed"); + await expect( + finalizeFeishuMessageProcessing({ + messageId: "msg-5", + namespace: "account-a", + claimHeld: true, + }), + ).resolves.toBe(true); + await expect( + finalizeFeishuMessageProcessing({ + messageId: "msg-5", + namespace: "account-a", + claimHeld: true, + }), + ).resolves.toBe(false); + }); + + it("dedupes cross-account broadcast claims through the shared namespace", async () => { + // Multi-account groups deliver the same event once per bot account; the + // shared "broadcast" namespace lets the first account claim dispatch. + await expect(recordProcessedFeishuMessage("msg-6", "broadcast")).resolves.toBe(true); + await expect(recordProcessedFeishuMessage("msg-6", "broadcast")).resolves.toBe(false); + + restartFeishuDedup(); + await expect(recordProcessedFeishuMessage("msg-6", "broadcast")).resolves.toBe(false); + }); + it("warms memory from persisted plugin state", async () => { - await expect(tryRecordMessagePersistent("msg-2", "account-a")).resolves.toBe(true); - testingHooks.resetFeishuDedupMemoryForTests(); + await expect(recordProcessedFeishuMessage("msg-7", "account-a")).resolves.toBe(true); + restartFeishuDedup(); await expect(warmupDedupFromPluginState("account-a")).resolves.toBe(1); - await expect(tryRecordMessagePersistent("msg-2", "account-a")).resolves.toBe(false); + await expect(recordProcessedFeishuMessage("msg-7", "account-a")).resolves.toBe(false); }); - it("ignores expired persisted entries", async () => { + it("ignores committed messages after the TTL expires", async () => { vi.useFakeTimers(); vi.setSystemTime(1_000); - await expect(tryRecordMessagePersistent("msg-3", "account-a")).resolves.toBe(true); - testingHooks.resetFeishuDedupMemoryForTests(); + await expect(recordProcessedFeishuMessage("msg-8", "account-a")).resolves.toBe(true); + restartFeishuDedup(); vi.setSystemTime(1_000 + 24 * 60 * 60 * 1000 + 1); - await expect(hasProcessedFeishuMessage("msg-3", "account-a")).resolves.toBe(false); + await expect(hasProcessedFeishuMessage("msg-8", "account-a")).resolves.toBe(false); }); - it("ignores legacy JSON dedupe files at runtime", async () => { - vi.useFakeTimers(); - vi.setSystemTime(2_000); - const legacyPath = path.join(tempDir as string, "feishu", "dedup", "account-a.json"); - await fs.mkdir(path.dirname(legacyPath), { recursive: true }); - await fs.writeFile( - legacyPath, - JSON.stringify({ - "msg-legacy": 1_000, - "msg-expired": 2_000 - 24 * 60 * 60 * 1000 - 1, - }), - "utf8", - ); + it("keeps deduping in memory and logs when plugin-state persistence fails", async () => { + // A regular file where the state dir should be makes every SQLite open fail. + const blockedPath = path.join(tempDir as string, "not-a-dir"); + fs.writeFileSync(blockedPath, "x", "utf8"); + process.env.OPENCLAW_STATE_DIR = path.join(blockedPath, "nested"); + const log = vi.fn(); - await expect(hasProcessedFeishuMessage("msg-legacy", "account-a")).resolves.toBe(false); - await expect(tryRecordMessagePersistent("msg-legacy", "account-a")).resolves.toBe(true); - await expect(hasProcessedFeishuMessage("msg-expired", "account-a")).resolves.toBe(false); + await expect(recordProcessedFeishuMessage("msg-9", "account-a", log)).resolves.toBe(true); + await expect( + claimUnprocessedFeishuMessage({ messageId: "msg-9", namespace: "account-a", log }), + ).resolves.toBe("duplicate"); + expect(log).toHaveBeenCalledWith( + expect.stringContaining("feishu-dedup: persistent state error"), + ); }); }); diff --git a/extensions/feishu/src/dedup.ts b/extensions/feishu/src/dedup.ts index a5296e85aaaa..ebd19153cadd 100644 --- a/extensions/feishu/src/dedup.ts +++ b/extensions/feishu/src/dedup.ts @@ -1,304 +1,157 @@ -// Feishu plugin module implements dedup behavior. -import { createHash } from "node:crypto"; -import type { PluginStateSyncKeyedStore } from "openclaw/plugin-sdk/plugin-state-runtime"; -import { - releaseFeishuMessageProcessing, - tryBeginFeishuMessageProcessing, -} from "./processing-claims.js"; -import { getFeishuRuntime } from "./runtime.js"; +// Feishu inbound replay protection rides the core claimable dedupe: Feishu +// redelivers events after reconnects/restarts and multi-account groups receive +// the same event once per bot, so handlers claim a dedupe key before +// processing, commit once handling is dispatched, and release on retryable +// failure so the event can be redelivered. +import { createClaimableDedupe } from "openclaw/plugin-sdk/persistent-dedupe"; +// Persisted namespaces resolve to `feishu.dedup.` in the shared +// plugin-state SQLite store. Rows from the retired hand-rolled `dedup.*` store +// are dropped without import: replay protection is cache and rebuilds after +// upgrade, leaving only a brief unclean-shutdown redelivery gap. +const DEDUPE_NAMESPACE_PREFIX = "feishu.dedup"; // Persistent TTL: 24 hours — survives restarts & WebSocket reconnects. const DEDUP_TTL_MS = 24 * 60 * 60 * 1000; const MEMORY_MAX_SIZE = 1_000; const STORE_MAX_ENTRIES = 10_000; -type FeishuDedupStoreEntry = { - namespace: string; - messageId: string; - seenAt: number; -}; -const memory = new Map(); -const cachedDedupStores = new Map>(); +type FeishuDedupeLog = (...args: unknown[]) => void; -function normalizeMessageId(messageId: string | undefined | null): string | null { - const trimmed = messageId?.trim(); - return trimmed ? trimmed : null; -} +type FeishuMessageClaim = "claimed" | "duplicate" | "inflight"; -function normalizeNamespace(namespace?: string): string { - return namespace?.trim() || "global"; -} - -function pluginStateNamespace(namespace: string): string { - return `dedup.${namespace.replace(/[^a-zA-Z0-9_-]/g, "_")}`; -} - -function openDedupStore(namespace: string): PluginStateSyncKeyedStore { - const stateNamespace = pluginStateNamespace(namespace); - const cached = cachedDedupStores.get(stateNamespace); - if (cached) { - return cached; - } - const store = getFeishuRuntime().state.openSyncKeyedStore({ - namespace: stateNamespace, - maxEntries: STORE_MAX_ENTRIES, - defaultTtlMs: DEDUP_TTL_MS, +function createFeishuDedupeGuard() { + return createClaimableDedupe({ + pluginId: "feishu", + namespacePrefix: DEDUPE_NAMESPACE_PREFIX, + ttlMs: DEDUP_TTL_MS, + memoryMaxSize: MEMORY_MAX_SIZE, + stateMaxEntries: STORE_MAX_ENTRIES, }); - cachedDedupStores.set(stateNamespace, store); - return store; } -function dedupeStoreKey(namespace: string, messageId: string): string { - return createHash("sha256") - .update(`${namespace}\0${messageId}`, "utf8") - .digest("hex") - .slice(0, 32); +let guard = createFeishuDedupeGuard(); + +function dedupeKey(messageId: string | undefined | null): string { + return messageId?.trim() ?? ""; } -function memoryKey(namespace: string, messageId: string): string { - return `${namespace}\0${messageId}`; +function dedupeOptions(namespace: string | undefined, log: FeishuDedupeLog | undefined) { + return { + ...(namespace ? { namespace } : {}), + // Persistence is best effort: a broken state DB must never block inbound + // handling, so disk errors surface to the caller's log while the memory + // layer keeps deduping. + ...(log + ? { + onDiskError: (error: unknown) => + log(`feishu-dedup: persistent state error: ${String(error)}`), + } + : {}), + }; } -function isRecent(seenAt: number | undefined, now = Date.now()): boolean { - return typeof seenAt === "number" && Number.isFinite(seenAt) && now - seenAt < DEDUP_TTL_MS; -} - -function pruneMemory(now = Date.now()): void { - for (const [key, seenAt] of memory) { - if (!isRecent(seenAt, now)) { - memory.delete(key); - } - } - if (memory.size <= MEMORY_MAX_SIZE) { - return; - } - const toRemove = Array.from(memory.entries()) - .toSorted(([, left], [, right]) => left - right) - .slice(0, memory.size - MEMORY_MAX_SIZE); - for (const [key] of toRemove) { - memory.delete(key); - } -} - -function remember(namespace: string, messageId: string, seenAt = Date.now()): void { - memory.set(memoryKey(namespace, messageId), seenAt); - pruneMemory(seenAt); -} - -function hasMemory(namespace: string, messageId: string, now = Date.now()): boolean { - const key = memoryKey(namespace, messageId); - const seenAt = memory.get(key); - if (isRecent(seenAt, now)) { - return true; - } - memory.delete(key); - return false; -} - -export { releaseFeishuMessageProcessing, tryBeginFeishuMessageProcessing }; - +/** + * Claims a dedupe key for exclusive handling. Duplicate (already committed) + * and in-flight keys are reported; blank keys fail open as claimed so an + * unidentifiable event is never suppressed. + */ export async function claimUnprocessedFeishuMessage(params: { messageId: string | undefined | null; namespace?: string; - log?: (...args: unknown[]) => void; -}): Promise<"claimed" | "duplicate" | "inflight" | "invalid"> { - const { messageId, namespace = "global", log } = params; - const normalizedMessageId = normalizeMessageId(messageId); - if (!normalizedMessageId) { - return "invalid"; + log?: FeishuDedupeLog; +}): Promise { + const key = dedupeKey(params.messageId); + if (!key) { + return "claimed"; } - if (await hasProcessedFeishuMessage(normalizedMessageId, namespace, log)) { - return "duplicate"; - } - if (!tryBeginFeishuMessageProcessing(normalizedMessageId, namespace)) { - return "inflight"; - } - return "claimed"; + return (await guard.claim(key, dedupeOptions(params.namespace, params.log))).kind; } +/** Drops an uncommitted claim so a failed handler can retry the message. */ +export function releaseFeishuMessageProcessing( + messageId: string | undefined | null, + namespace = "global", +): void { + const key = dedupeKey(messageId); + if (key) { + guard.release(key, { namespace }); + } +} + +/** + * Claims (unless the caller already holds the claim) and commits a message. + * False means another handler owns it, it was already handled, or the key is + * blank; handlers must skip dispatch then. + */ export async function finalizeFeishuMessageProcessing(params: { messageId: string | undefined | null; namespace?: string; - log?: (...args: unknown[]) => void; + log?: FeishuDedupeLog; claimHeld?: boolean; }): Promise { - const { messageId, namespace = "global", log, claimHeld = false } = params; - const normalizedMessageId = normalizeMessageId(messageId); - if (!normalizedMessageId) { + const key = dedupeKey(params.messageId); + if (!key) { return false; } - if (!claimHeld && !tryBeginFeishuMessageProcessing(normalizedMessageId, namespace)) { + const options = dedupeOptions(params.namespace, params.log); + if (!params.claimHeld && (await guard.claim(key, options)).kind !== "claimed") { return false; } - if (!(await tryRecordMessagePersistent(normalizedMessageId, namespace, log))) { - releaseFeishuMessageProcessing(normalizedMessageId, namespace); - return false; - } - return true; + return await guard.commit(key, options); } +/** Records a handled message so restart/replay cannot dispatch it again; false when already recorded. */ export async function recordProcessedFeishuMessage( messageId: string | undefined | null, namespace = "global", - log?: (...args: unknown[]) => void, + log?: FeishuDedupeLog, ): Promise { - const normalizedMessageId = normalizeMessageId(messageId); - if (!normalizedMessageId) { + const key = dedupeKey(messageId); + if (!key) { return false; } - return await tryRecordMessagePersistent(normalizedMessageId, namespace, log); + return await guard.commit(key, dedupeOptions(namespace, log)); } +/** Forgets a recorded message so a retryable synthetic event can be handled on redelivery. */ export async function forgetProcessedFeishuMessage( messageId: string | undefined | null, namespace = "global", - log?: (...args: unknown[]) => void, + log?: FeishuDedupeLog, ): Promise { - const normalizedNamespace = normalizeNamespace(namespace); - const normalizedMessageId = normalizeMessageId(messageId); - if (!normalizedMessageId) { - return false; - } - memory.delete(memoryKey(normalizedNamespace, normalizedMessageId)); - const key = dedupeStoreKey(normalizedNamespace, normalizedMessageId); - try { - return openDedupStore(normalizedNamespace).delete(key); - } catch (error) { - log?.(`feishu-dedup: persistent delete failed: ${String(error)}`); + const key = dedupeKey(messageId); + if (!key) { return false; } + return await guard.forget(key, dedupeOptions(namespace, log)); } +/** Checks recency without claiming or recording. */ export async function hasProcessedFeishuMessage( messageId: string | undefined | null, namespace = "global", - log?: (...args: unknown[]) => void, + log?: FeishuDedupeLog, ): Promise { - const normalizedMessageId = normalizeMessageId(messageId); - if (!normalizedMessageId) { + const key = dedupeKey(messageId); + if (!key) { return false; } - return hasRecordedMessagePersistent(normalizedMessageId, namespace, log); -} - -export async function tryRecordMessagePersistent( - messageId: string, - namespace = "global", - log?: (...args: unknown[]) => void, -): Promise { - const normalizedNamespace = normalizeNamespace(namespace); - const normalizedMessageId = normalizeMessageId(messageId); - if (!normalizedMessageId) { - return true; - } - const now = Date.now(); - if (hasMemory(normalizedNamespace, normalizedMessageId, now)) { - return false; - } - const key = dedupeStoreKey(normalizedNamespace, normalizedMessageId); - try { - const store = openDedupStore(normalizedNamespace); - const existing = store.lookup(key); - const existingSeenAt = existing?.seenAt; - if (isRecent(existingSeenAt, now)) { - remember(normalizedNamespace, normalizedMessageId, existingSeenAt); - return false; - } - const recorded = store.registerIfAbsent( - key, - { - namespace: normalizedNamespace, - messageId: normalizedMessageId, - seenAt: now, - }, - { ttlMs: DEDUP_TTL_MS }, - ); - if (!recorded) { - const current = store.lookup(key); - const currentSeenAt = current?.seenAt; - if (isRecent(currentSeenAt, now)) { - remember(normalizedNamespace, normalizedMessageId, currentSeenAt); - return false; - } - store.register( - key, - { - namespace: normalizedNamespace, - messageId: normalizedMessageId, - seenAt: now, - }, - { ttlMs: DEDUP_TTL_MS }, - ); - } - remember(normalizedNamespace, normalizedMessageId, now); - return true; - } catch (error) { - log?.(`feishu-dedup: persistent state error, falling back to memory: ${String(error)}`); - remember(normalizedNamespace, normalizedMessageId, now); - return true; - } -} - -async function hasRecordedMessagePersistent( - messageId: string, - namespace = "global", - log?: (...args: unknown[]) => void, -): Promise { - const normalizedNamespace = normalizeNamespace(namespace); - const normalizedMessageId = normalizeMessageId(messageId); - if (!normalizedMessageId) { - return false; - } - const now = Date.now(); - if (hasMemory(normalizedNamespace, normalizedMessageId, now)) { - return true; - } - try { - const store = openDedupStore(normalizedNamespace); - const existing = store.lookup(dedupeStoreKey(normalizedNamespace, normalizedMessageId)); - const existingSeenAt = existing?.seenAt; - if (!isRecent(existingSeenAt, now)) { - return false; - } - remember(normalizedNamespace, normalizedMessageId, existingSeenAt); - return true; - } catch (error) { - log?.(`feishu-dedup: persistent peek failed: ${String(error)}`); - return hasMemory(normalizedNamespace, normalizedMessageId, now); - } + return await guard.hasRecent(key, dedupeOptions(namespace, log)); } +/** Loads recent persisted entries into memory at account start; returns the loaded count. */ export async function warmupDedupFromPluginState( namespace: string, - log?: (...args: unknown[]) => void, + log?: FeishuDedupeLog, ): Promise { - const normalizedNamespace = normalizeNamespace(namespace); - try { - let loaded = 0; - const now = Date.now(); - for (const entry of openDedupStore(normalizedNamespace).entries()) { - if (entry.value.namespace !== normalizedNamespace || !isRecent(entry.value.seenAt, now)) { - continue; - } - remember(normalizedNamespace, entry.value.messageId, entry.value.seenAt); - loaded++; - } - return loaded; - } catch (error) { - log?.(`feishu-dedup: warmup persistent state error: ${String(error)}`); - return 0; - } + return await guard.warmup(namespace, (error) => + log?.(`feishu-dedup: warmup persistent state error: ${String(error)}`), + ); } export const testingHooks = { + /** Drops in-flight claims and process memory; persisted rows follow the test's state dir. */ resetFeishuDedupForTests() { - memory.clear(); - for (const store of cachedDedupStores.values()) { - store.clear(); - } - cachedDedupStores.clear(); - }, - resetFeishuDedupMemoryForTests() { - memory.clear(); + guard = createFeishuDedupeGuard(); }, }; diff --git a/extensions/feishu/src/lifecycle.test-support.ts b/extensions/feishu/src/lifecycle.test-support.ts index c2ee360be920..bd5e0d229a5c 100644 --- a/extensions/feishu/src/lifecycle.test-support.ts +++ b/extensions/feishu/src/lifecycle.test-support.ts @@ -1,7 +1,6 @@ // Feishu plugin module implements lifecycle support behavior. import { vi, type Mock } from "vitest"; import { testingHooks as dedupTestingHooks } from "./dedup.js"; -import { testingHooks as processingClaimTestingHooks } from "./processing-claims.js"; type BoundConversation = { bindingId: string; @@ -92,7 +91,6 @@ export function getFeishuLifecycleTestMocks(): FeishuLifecycleTestMocks { export function resetFeishuLifecycleTestMocks(): void { dedupTestingHooks.resetFeishuDedupForTests(); - processingClaimTestingHooks.resetFeishuMessageProcessingClaimsForTests(); for (const mock of Object.values(feishuLifecycleTestMocks)) { mock.mockReset(); } diff --git a/extensions/feishu/src/monitor.message-handler.ts b/extensions/feishu/src/monitor.message-handler.ts index 5cbadd327ece..3ce385aaa7f4 100644 --- a/extensions/feishu/src/monitor.message-handler.ts +++ b/extensions/feishu/src/monitor.message-handler.ts @@ -1,13 +1,10 @@ // Feishu plugin module implements monitor.message handler behavior. import { isRecord, readStringValue as readString } from "openclaw/plugin-sdk/string-coerce-runtime"; import type { ClawdbotConfig, HistoryEntry, PluginRuntime, RuntimeEnv } from "../runtime-api.js"; +import { claimUnprocessedFeishuMessage, releaseFeishuMessageProcessing } from "./dedup.js"; import { resolveFeishuMessageDedupeKey } from "./dedupe-key.js"; import type { FeishuMessageEvent } from "./event-types.js"; import { isMentionForwardRequest } from "./mention.js"; -import { - releaseFeishuMessageProcessing, - tryBeginFeishuMessageProcessing, -} from "./processing-claims.js"; import { createSequentialQueue } from "./sequential-queue.js"; import type { FeishuChatType } from "./types.js"; @@ -347,8 +344,13 @@ export function createFeishuMessageReceiveHandler({ return; } const messageDedupeKey = resolveFeishuMessageDedupeKey(event); - if (!tryBeginFeishuMessageProcessing(messageDedupeKey, accountId)) { - log(`feishu[${accountId}]: dropping duplicate event for message ${messageId}`); + const claim = await claimUnprocessedFeishuMessage({ + messageId: messageDedupeKey, + namespace: accountId, + log, + }); + if (claim !== "claimed") { + log(`feishu[${accountId}]: dropping ${claim} event for message ${messageId}`); return; } const processMessage = async () => { diff --git a/extensions/feishu/src/monitor.reaction.test.ts b/extensions/feishu/src/monitor.reaction.test.ts index 0224e8392bfd..9eb1d0f17315 100644 --- a/extensions/feishu/src/monitor.reaction.test.ts +++ b/extensions/feishu/src/monitor.reaction.test.ts @@ -234,7 +234,7 @@ function expectParsedFirstDispatchedEvent(botOpenId = "ou_bot") { } function setDedupPassThroughMocks(): void { - vi.spyOn(dedup, "tryBeginFeishuMessageProcessing").mockReturnValue(true); + vi.spyOn(dedup, "claimUnprocessedFeishuMessage").mockResolvedValue("claimed"); vi.spyOn(dedup, "recordProcessedFeishuMessage").mockResolvedValue(true); vi.spyOn(dedup, "hasProcessedFeishuMessage").mockResolvedValue(false); } @@ -595,7 +595,7 @@ describe("Feishu inbound debounce regressions", () => { }); it("passes prefetched botName through to handleFeishuMessage", async () => { - vi.spyOn(dedup, "tryBeginFeishuMessageProcessing").mockReturnValue(true); + vi.spyOn(dedup, "claimUnprocessedFeishuMessage").mockResolvedValue("claimed"); vi.spyOn(dedup, "recordProcessedFeishuMessage").mockResolvedValue(true); vi.spyOn(dedup, "hasProcessedFeishuMessage").mockResolvedValue(false); const onMessage = await setupDebounceMonitor({ botName: "OpenClaw Bot" }); @@ -678,7 +678,7 @@ describe("Feishu inbound debounce regressions", () => { }); it("excludes previously processed retries from combined debounce text", async () => { - vi.spyOn(dedup, "tryBeginFeishuMessageProcessing").mockReturnValue(true); + vi.spyOn(dedup, "claimUnprocessedFeishuMessage").mockResolvedValue("claimed"); vi.spyOn(dedup, "recordProcessedFeishuMessage").mockResolvedValue(true); setStaleRetryMocks(); const onMessage = await setupDebounceMonitor(); @@ -704,7 +704,7 @@ describe("Feishu inbound debounce regressions", () => { }); it("uses latest fresh message id when debounce batch ends with stale retry", async () => { - vi.spyOn(dedup, "tryBeginFeishuMessageProcessing").mockReturnValue(true); + vi.spyOn(dedup, "claimUnprocessedFeishuMessage").mockResolvedValue("claimed"); const recordSpy = vi.spyOn(dedup, "recordProcessedFeishuMessage").mockResolvedValue(true); setStaleRetryMocks("om_old_latest_fresh"); const onMessage = await setupDebounceMonitor(); diff --git a/extensions/feishu/src/processing-claims.ts b/extensions/feishu/src/processing-claims.ts deleted file mode 100644 index 7eef314c604d..000000000000 --- a/extensions/feishu/src/processing-claims.ts +++ /dev/null @@ -1,66 +0,0 @@ -// Feishu plugin module implements processing claims behavior. -const EVENT_DEDUP_TTL_MS = 5 * 60 * 1000; -const EVENT_MEMORY_MAX_SIZE = 2_000; - -const processingClaims = new Map(); - -function resolveEventDedupeKey( - namespace: string, - messageId: string | undefined | null, -): string | null { - const trimmed = messageId?.trim(); - return trimmed ? `${namespace}:${trimmed}` : null; -} - -function pruneProcessingClaims(now: number): void { - const cutoff = now - EVENT_DEDUP_TTL_MS; - for (const [key, seenAt] of processingClaims) { - if (seenAt < cutoff) { - processingClaims.delete(key); - } - } - while (processingClaims.size > EVENT_MEMORY_MAX_SIZE) { - const oldestKey = processingClaims.keys().next().value; - if (!oldestKey) { - return; - } - processingClaims.delete(oldestKey); - } -} - -export function tryBeginFeishuMessageProcessing( - messageId: string | undefined | null, - namespace = "global", -): boolean { - const key = resolveEventDedupeKey(namespace, messageId); - if (!key) { - return true; - } - const now = Date.now(); - pruneProcessingClaims(now); - if (processingClaims.has(key)) { - processingClaims.delete(key); - processingClaims.set(key, now); - pruneProcessingClaims(now); - return false; - } - processingClaims.set(key, now); - pruneProcessingClaims(now); - return true; -} - -export function releaseFeishuMessageProcessing( - messageId: string | undefined | null, - namespace = "global", -): void { - const key = resolveEventDedupeKey(namespace, messageId); - if (key) { - processingClaims.delete(key); - } -} - -export const testingHooks = { - resetFeishuMessageProcessingClaimsForTests() { - processingClaims.clear(); - }, -}; diff --git a/scripts/check-no-random-messaging-tmp.mjs b/scripts/check-no-random-messaging-tmp.mjs index 040be5e2c5aa..a7a1422075db 100644 --- a/scripts/check-no-random-messaging-tmp.mjs +++ b/scripts/check-no-random-messaging-tmp.mjs @@ -2,7 +2,6 @@ // Blocks host-random tmpdir usage in messaging/channel runtime sources. import ts from "typescript"; -import { bundledPluginFile } from "./lib/bundled-plugin-paths.mjs"; import { runCallsiteGuard } from "./lib/callsite-guard.mjs"; import { collectCallExpressionLines, @@ -21,8 +20,6 @@ export const messagingTmpdirGuardSourceRoots = [ "src/media-understanding", "extensions", ]; -const allowedRelativePaths = new Set([bundledPluginFile("feishu", "src/dedup.ts")]); - function collectOsTmpdirImports(sourceFile) { const osModuleSpecifiers = new Set(["node:os", "os"]); const osNamespaceOrDefault = new Set(); @@ -85,7 +82,6 @@ export async function main() { importMetaUrl: import.meta.url, sourceRoots: messagingTmpdirGuardSourceRoots, findCallLines: findMessagingTmpdirCallLines, - skipRelativePath: (relativePath) => allowedRelativePaths.has(relativePath), header: "Found os.tmpdir()/tmpdir() usage in messaging/channel runtime sources:", footer: "Use resolvePreferredOpenClawTmpDir() or plugin-sdk temp helpers instead of host tmp defaults.", diff --git a/src/cli/program/config-guard.test.ts b/src/cli/program/config-guard.test.ts index 6fe558955828..d6f22a7b2f50 100644 --- a/src/cli/program/config-guard.test.ts +++ b/src/cli/program/config-guard.test.ts @@ -363,7 +363,6 @@ describe("ensureConfigReady", () => { it.each([ ["Discord model picker preferences", "discord/model-picker-preferences.json"], ["Discord thread bindings", "discord/thread-bindings.json"], - ["Feishu dedupe sidecar", "feishu/dedup/default.json"], ["Telegram bot info cache", "telegram/bot-info-default.json"], ["Telegram update offset", "telegram/update-offset-default.json"], ["Telegram sticker cache", "telegram/sticker-cache.json"], diff --git a/src/cli/program/config-guard.ts b/src/cli/program/config-guard.ts index 6eda8edc4045..72a62980ca11 100644 --- a/src/cli/program/config-guard.ts +++ b/src/cli/program/config-guard.ts @@ -88,9 +88,6 @@ function hasBundledChannelLegacyStateMigrationInputs(stateDir: string, oauthDir: ) { return true; } - if (dirHasFile(path.join(stateDir, "feishu", "dedup"), (name) => name.endsWith(".json"))) { - return true; - } if (hasLegacyIMessageStateFiles(stateDir)) { return true; }