From 9297c20a85e101c1951dccd432f0f1d8a18ec92c Mon Sep 17 00:00:00 2001 From: Ayaan Zaidi Date: Wed, 10 Jun 2026 12:55:45 +0530 Subject: [PATCH] fix(telegram): use SDK dispatch replay dedupe --- .../telegram/src/bot-handlers.runtime.ts | 1 - .../src/bot.create-telegram-bot.test.ts | 31 +- .../src/message-dispatch-dedupe.test.ts | 191 +----- .../telegram/src/message-dispatch-dedupe.ts | 553 +----------------- .../telegram/src/state-migrations.test.ts | 123 +++- extensions/telegram/src/state-migrations.ts | 165 +++++- 6 files changed, 331 insertions(+), 733 deletions(-) diff --git a/extensions/telegram/src/bot-handlers.runtime.ts b/extensions/telegram/src/bot-handlers.runtime.ts index f72ce1938a27..38256bc30e47 100644 --- a/extensions/telegram/src/bot-handlers.runtime.ts +++ b/extensions/telegram/src/bot-handlers.runtime.ts @@ -213,7 +213,6 @@ export const registerTelegramHandlers = ({ scope: resolveTelegramMessageCacheScope(telegramDeps.resolveStorePath(cfg.session?.store)), }); const messageDispatchReplayGuard = createTelegramMessageDispatchReplayGuard({ - storePath: telegramDeps.resolveStorePath(cfg.session?.store), onDiskError: (error) => { runtime.error?.(danger(`[telegram] message dispatch dedupe store failed: ${String(error)}`)); }, diff --git a/extensions/telegram/src/bot.create-telegram-bot.test.ts b/extensions/telegram/src/bot.create-telegram-bot.test.ts index 2bd2b2646f1a..58966ea9d097 100644 --- a/extensions/telegram/src/bot.create-telegram-bot.test.ts +++ b/extensions/telegram/src/bot.create-telegram-bot.test.ts @@ -1,4 +1,7 @@ // Telegram tests cover bot.create telegram bot plugin behavior. +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; import { escapeRegExp, formatEnvelopeTimestamp } from "openclaw/plugin-sdk/channel-test-helpers"; import type { TelegramGroupConfig } from "openclaw/plugin-sdk/config-contracts"; import type { GetReplyOptions, MsgContext } from "openclaw/plugin-sdk/reply-runtime"; @@ -14,6 +17,8 @@ const conversationRuntime = await import("openclaw/plugin-sdk/conversation-runti const configMutation = await import("openclaw/plugin-sdk/config-mutation"); const sessionStoreRuntime = await import("openclaw/plugin-sdk/session-store-runtime"); const EYES_EMOJI = "\u{1F440}"; +const tempStateDirs: string[] = []; +let previousStateDir: string | undefined; const { answerCallbackQuerySpy, botCtorSpy, @@ -60,7 +65,6 @@ const { } = await import("./bot-core.js"); const { resolveTelegramConversationRoute } = await import("./conversation-route.js"); const { clearAccountThrottlersForTest } = await import("./account-throttler.js"); -const messageDispatchDedupe = await import("./message-dispatch-dedupe.js"); const { buildTelegramGroupFrom, buildTelegramThreadParams, @@ -74,6 +78,12 @@ let createTelegramBot: ( opts: TelegramBotOptions, ) => ReturnType; +function createTelegramBotTestStateDir(): string { + const dir = mkdtempSync(path.join(tmpdir(), "openclaw-telegram-bot-")); + tempStateDirs.push(dir); + return dir; +} + const loadConfig = getLoadConfigMock(); const loadSessionStore = getLoadSessionStoreMock(); const loadWebMedia = getLoadWebMediaMock(); @@ -214,10 +224,19 @@ describe("createTelegramBot", () => { } }); afterEach(() => { - messageDispatchDedupe.setTelegramMessageDispatchDedupeStoreForTest(undefined); pluginStateTestRuntime.resetPluginStateStoreForTests(); + if (previousStateDir === undefined) { + delete process.env.OPENCLAW_STATE_DIR; + } else { + process.env.OPENCLAW_STATE_DIR = previousStateDir; + } + for (const dir of tempStateDirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }); + } }); beforeEach(async () => { + previousStateDir = process.env.OPENCLAW_STATE_DIR; + process.env.OPENCLAW_STATE_DIR = createTelegramBotTestStateDir(); resetTelegramForumFlagCacheForTest(); clearAccountThrottlersForTest(); throttlerSpy.mockReset(); @@ -230,14 +249,6 @@ describe("createTelegramBot", () => { telegramDeps: telegramBotDepsForTest, }); pluginStateTestRuntime.resetPluginStateStoreForTests({ closeDatabase: false }); - const store = pluginStateTestRuntime.createPluginStateKeyedStoreForTests("telegram", { - namespace: messageDispatchDedupe.TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE, - maxEntries: messageDispatchDedupe.TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES, - }) as NonNullable< - Parameters[0] - >; - await store.clear(); - messageDispatchDedupe.setTelegramMessageDispatchDedupeStoreForTest(store); }); // groupPolicy tests diff --git a/extensions/telegram/src/message-dispatch-dedupe.test.ts b/extensions/telegram/src/message-dispatch-dedupe.test.ts index 77b279f1629f..03f68ff8c084 100644 --- a/extensions/telegram/src/message-dispatch-dedupe.test.ts +++ b/extensions/telegram/src/message-dispatch-dedupe.test.ts @@ -3,34 +3,23 @@ import { mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; import type { Message } from "grammy/types"; -import { - createPluginStateKeyedStoreForTests, - 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 } from "vitest"; import { - TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES, - TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE, buildTelegramMessageDispatchReplayKey, claimTelegramMessageDispatchReplay, commitTelegramMessageDispatchReplay, createTelegramMessageDispatchReplayGuard, releaseTelegramMessageDispatchReplay, - setTelegramMessageDispatchDedupeStoreForTest, } from "./message-dispatch-dedupe.js"; -type MessageDispatchDedupeStore = NonNullable< - Parameters[0] ->; -type SyncMessageDispatchDedupeStore = Extract; - const tempDirs: string[] = []; +let previousStateDir: string | undefined; -function createStorePath(): string { +function createStateDir(): string { const dir = mkdtempSync(path.join(tmpdir(), "openclaw-telegram-dispatch-dedupe-")); tempDirs.push(dir); - return path.join(dir, "sessions.json"); + return dir; } function message(params?: { chatId?: number; messageId?: number }): Message { @@ -41,19 +30,19 @@ function message(params?: { chatId?: number; messageId?: number }): Message { } as Message; } -beforeEach(async () => { +beforeEach(() => { + previousStateDir = process.env.OPENCLAW_STATE_DIR; + process.env.OPENCLAW_STATE_DIR = createStateDir(); resetPluginStateStoreForTests({ closeDatabase: false }); - const store = createPluginStateKeyedStoreForTests("telegram", { - namespace: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE, - maxEntries: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES, - }) as NonNullable[0]>; - await store.clear(); - setTelegramMessageDispatchDedupeStoreForTest(store); }); afterEach(() => { - setTelegramMessageDispatchDedupeStoreForTest(undefined); resetPluginStateStoreForTests(); + if (previousStateDir === undefined) { + delete process.env.OPENCLAW_STATE_DIR; + } else { + process.env.OPENCLAW_STATE_DIR = previousStateDir; + } for (const dir of tempDirs.splice(0)) { rmSync(dir, { recursive: true, force: true }); } @@ -68,8 +57,7 @@ describe("Telegram message dispatch replay guard", () => { }); it("persists committed dispatches across guard recreation", async () => { - const storePath = createStorePath(); - const writer = createTelegramMessageDispatchReplayGuard({ storePath }); + const writer = createTelegramMessageDispatchReplayGuard(); const first = await claimTelegramMessageDispatchReplay({ guard: writer, accountId: "default", @@ -89,7 +77,7 @@ describe("Telegram message dispatch replay guard", () => { keys: [first.key], }); - const reader = createTelegramMessageDispatchReplayGuard({ storePath }); + const reader = createTelegramMessageDispatchReplayGuard(); await expect( claimTelegramMessageDispatchReplay({ guard: reader, @@ -99,9 +87,8 @@ describe("Telegram message dispatch replay guard", () => { ).resolves.toEqual({ kind: "duplicate" }); }); - it("preserves concurrent commits that share dedupe buckets", async () => { - const storePath = createStorePath(); - const writer = createTelegramMessageDispatchReplayGuard({ storePath }); + it("preserves concurrent commits", async () => { + const writer = createTelegramMessageDispatchReplayGuard(); const keys = Array.from({ length: 400 }, (_, index) => JSON.stringify(["message", "1234", index + 1]), ); @@ -112,151 +99,12 @@ describe("Telegram message dispatch replay guard", () => { keys, }); - const reader = createTelegramMessageDispatchReplayGuard({ storePath }); + const reader = createTelegramMessageDispatchReplayGuard(); await expect(reader.warmup("default")).resolves.toBe(keys.length); }); - it("falls back to same-process replay protection when plugin-state is unavailable", async () => { - setTelegramMessageDispatchDedupeStoreForTest(undefined); - const errors: unknown[] = []; - const storePath = createStorePath(); - const guard = createTelegramMessageDispatchReplayGuard({ - storePath, - onDiskError: (error) => errors.push(error), - }); - const first = await claimTelegramMessageDispatchReplay({ - guard, - accountId: "default", - msg: message(), - }); - if (first.kind !== "claimed") { - throw new Error("expected initial claim"); - } - - await expect(guard.commit(first.key, { namespace: "default" })).resolves.toBe(false); - - await expect( - claimTelegramMessageDispatchReplay({ - guard, - accountId: "default", - msg: message(), - }), - ).resolves.toEqual({ kind: "duplicate" }); - await expect(guard.hasRecent(first.key, { namespace: "default" })).resolves.toBe(true); - expect(errors).toEqual([]); - }); - - it("keeps same-process replay protection when plugin-state commit fails", async () => { - const failingStore = createPluginStateKeyedStoreForTests("telegram", { - namespace: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE, - maxEntries: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES, - }) as NonNullable[0]>; - setTelegramMessageDispatchDedupeStoreForTest({ - ...failingStore, - async register() { - throw new Error("state write failed"); - }, - }); - const storePath = createStorePath(); - const guard = createTelegramMessageDispatchReplayGuard({ storePath }); - const first = await claimTelegramMessageDispatchReplay({ - guard, - accountId: "default", - msg: message(), - }); - if (first.kind !== "claimed") { - throw new Error("expected initial claim"); - } - - await expect(guard.commit(first.key, { namespace: "default" })).resolves.toBe(false); - - await expect( - claimTelegramMessageDispatchReplay({ - guard, - accountId: "default", - msg: message(), - }), - ).resolves.toEqual({ kind: "duplicate" }); - await expect(guard.hasRecent(first.key, { namespace: "default" })).resolves.toBe(true); - await expect(guard.warmup("default")).resolves.toBe(1); - }); - - it("keeps same-process replay protection when lookup fails after a successful commit", async () => { - const backingStore = createPluginStateSyncKeyedStoreForTests("telegram", { - namespace: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE, - maxEntries: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES, - }) as SyncMessageDispatchDedupeStore; - let failLookup = false; - setTelegramMessageDispatchDedupeStoreForTest({ - ...backingStore, - lookup(key) { - if (failLookup) { - throw new Error("state read failed"); - } - return backingStore.lookup(key); - }, - }); - const storePath = createStorePath(); - const guard = createTelegramMessageDispatchReplayGuard({ storePath }); - const first = await claimTelegramMessageDispatchReplay({ - guard, - accountId: "default", - msg: message(), - }); - if (first.kind !== "claimed") { - throw new Error("expected initial claim"); - } - await expect(guard.commit(first.key, { namespace: "default" })).resolves.toBe(true); - - failLookup = true; - - await expect( - claimTelegramMessageDispatchReplay({ - guard, - accountId: "default", - msg: message(), - }), - ).resolves.toEqual({ kind: "duplicate" }); - }); - - it("keeps replay histories isolated by session store path", async () => { - const firstStorePath = createStorePath(); - const secondStorePath = createStorePath(); - const firstGuard = createTelegramMessageDispatchReplayGuard({ - storePath: firstStorePath, - }); - const first = await claimTelegramMessageDispatchReplay({ - guard: firstGuard, - accountId: "default", - msg: message(), - }); - if (first.kind !== "claimed") { - throw new Error("expected initial claim"); - } - await commitTelegramMessageDispatchReplay({ - guard: firstGuard, - accountId: "default", - keys: [first.key], - }); - - const secondGuard = createTelegramMessageDispatchReplayGuard({ - storePath: secondStorePath, - }); - await expect( - claimTelegramMessageDispatchReplay({ - guard: secondGuard, - accountId: "default", - msg: message(), - }), - ).resolves.toEqual({ - kind: "claimed", - key: first.key, - }); - }); - it("keeps accounts isolated and releases retryable pre-dispatch claims", async () => { - const storePath = createStorePath(); - const guard = createTelegramMessageDispatchReplayGuard({ storePath }); + const guard = createTelegramMessageDispatchReplayGuard(); const first = await claimTelegramMessageDispatchReplay({ guard, accountId: "default", @@ -295,8 +143,7 @@ describe("Telegram message dispatch replay guard", () => { }); it("lets an in-flight duplicate retry after the first claim is released", async () => { - const storePath = createStorePath(); - const guard = createTelegramMessageDispatchReplayGuard({ storePath }); + const guard = createTelegramMessageDispatchReplayGuard(); const first = await claimTelegramMessageDispatchReplay({ guard, accountId: "default", diff --git a/extensions/telegram/src/message-dispatch-dedupe.ts b/extensions/telegram/src/message-dispatch-dedupe.ts index dfa3879f0519..e754c5ecccd1 100644 --- a/extensions/telegram/src/message-dispatch-dedupe.ts +++ b/extensions/telegram/src/message-dispatch-dedupe.ts @@ -1,208 +1,20 @@ // Telegram plugin module implements message dispatch dedupe behavior. -import { createHash } from "node:crypto"; -import fs from "node:fs"; import path from "node:path"; import type { Message } from "grammy/types"; -import type { - ClaimableDedupe, - ClaimableDedupeClaimResult, -} from "openclaw/plugin-sdk/persistent-dedupe"; -import type { - PluginStateKeyedStore, - PluginStateSyncKeyedStore, -} from "openclaw/plugin-sdk/plugin-state-runtime"; +import { createClaimableDedupe, type ClaimableDedupe } from "openclaw/plugin-sdk/persistent-dedupe"; import { normalizeStringEntries, uniqueStrings } from "openclaw/plugin-sdk/string-coerce-runtime"; -import { getOptionalTelegramRuntime } from "./runtime.js"; -const TELEGRAM_MESSAGE_DISPATCH_TTL_MS = 7 * 24 * 60 * 60 * 1000; -export const TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE = "telegram.message-dispatch-dedupe"; -export const TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES = 4_096; -const TELEGRAM_MESSAGE_DISPATCH_DEDUPE_LOGICAL_MAX_ENTRIES = 50_000; -const TELEGRAM_MESSAGE_DISPATCH_DEDUPE_BUCKET_COUNT = 256; -const TELEGRAM_MESSAGE_DISPATCH_DEDUPE_BUCKET_MAX_KEYS = 256; -const TELEGRAM_MESSAGE_DISPATCH_DEDUPE_LOCK_TTL_MS = 30_000; -const TELEGRAM_MESSAGE_DISPATCH_DEDUPE_LOCK_RETRY_MS = 10; -const TELEGRAM_MESSAGE_DISPATCH_DEDUPE_LOCK_ATTEMPTS = 50; +export const TELEGRAM_MESSAGE_DISPATCH_DEDUPE_TTL_MS = 7 * 24 * 60 * 60 * 1000; +export const TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE_PREFIX = "telegram.message-dispatch-dedupe"; +export const TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES = 50_000; export type TelegramMessageDispatchReplayGuard = ClaimableDedupe; -type TelegramMessageDispatchDedupeRecord = { - scopeKey: string; - namespace: string; - bucketId: string; - entries: Record; -}; - -type TelegramMessageDispatchDedupeStore = - | PluginStateKeyedStore - | PluginStateSyncKeyedStore; - -type PendingClaim = { - promise: Promise; - resolve: (result: boolean) => void; - reject: (error: unknown) => void; -}; - -type MemoryCommittedClaim = { - namespace: string; - expiresAt: number; -}; - -let dispatchDedupeStoreForTest: TelegramMessageDispatchDedupeStore | undefined; export type TelegramMessageDispatchClaim = | { kind: "claimed"; key: string } | { kind: "duplicate" } | { kind: "invalid" }; -function openDispatchDedupeStore(): TelegramMessageDispatchDedupeStore | undefined { - if (dispatchDedupeStoreForTest) { - return dispatchDedupeStoreForTest; - } - return getOptionalTelegramRuntime()?.state.openKeyedStore({ - namespace: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE, - maxEntries: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES, - }); -} - -function resolveDispatchScopeKey(storePath: string): string { - return createHash("sha256").update(storePath, "utf8").digest("hex").slice(0, 24); -} - -function dedupeEntryKey(scopeKey: string, namespace: string, key: string): string { - return createHash("sha256") - .update(`${scopeKey}\0${namespace}\0${key}`, "utf8") - .digest("hex") - .slice(0, 32); -} - -function dedupeBucketId(key: string): string { - const bucketIndex = - Number.parseInt(createHash("sha256").update(key, "utf8").digest("hex").slice(0, 8), 16) % - TELEGRAM_MESSAGE_DISPATCH_DEDUPE_BUCKET_COUNT; - return bucketIndex.toString(16).padStart(2, "0"); -} - -function dedupeBucketEntryKey(scopeKey: string, namespace: string, bucketId: string): string { - return createHash("sha256") - .update(`${scopeKey}\0${namespace}\0${bucketId}`, "utf8") - .digest("hex") - .slice(0, 32); -} - -function dedupeLegacyBucketEntryKey(params: { - scopeKey: string; - namespace: string; - bucketId: string; - sourcePath: string; -}): string { - const sourceKey = createHash("sha256") - .update(params.sourcePath, "utf8") - .digest("hex") - .slice(0, 12); - return dedupeBucketEntryKey(params.scopeKey, params.namespace, `${params.bucketId}:${sourceKey}`); -} - -function dedupeBucketLockKey(bucketKey: string): string { - return `${bucketKey}:lock`; -} - -async function sleep(ms: number): Promise { - await new Promise((resolve) => { - setTimeout(resolve, ms); - }); -} - -function pruneDedupeBucketEntries(entries: Record, now: number): void { - for (const [key, timestamp] of Object.entries(entries)) { - if (typeof timestamp !== "number" || !Number.isFinite(timestamp)) { - delete entries[key]; - continue; - } - if (now - timestamp >= TELEGRAM_MESSAGE_DISPATCH_TTL_MS) { - delete entries[key]; - } - } - const keys = Object.keys(entries); - if (keys.length <= TELEGRAM_MESSAGE_DISPATCH_DEDUPE_BUCKET_MAX_KEYS) { - return; - } - for (const key of keys - .toSorted((left, right) => entries[left] - entries[right]) - .slice(0, keys.length - TELEGRAM_MESSAGE_DISPATCH_DEDUPE_BUCKET_MAX_KEYS)) { - delete entries[key]; - } -} - -function createDedupeBucketRecord(params: { - scopeKey: string; - namespace: string; - bucketId: string; - entries?: Record; -}): TelegramMessageDispatchDedupeRecord { - return { - scopeKey: params.scopeKey, - namespace: params.namespace, - bucketId: params.bucketId, - entries: { ...params.entries }, - }; -} - -function normalizeDedupeBucketRecord( - value: TelegramMessageDispatchDedupeRecord | undefined, - params: { - scopeKey: string; - namespace: string; - bucketId: string; - now: number; - }, -): TelegramMessageDispatchDedupeRecord { - const entries = - value?.scopeKey === params.scopeKey && - value.namespace === params.namespace && - value.bucketId === params.bucketId && - value.entries && - typeof value.entries === "object" - ? { ...value.entries } - : {}; - pruneDedupeBucketEntries(entries, params.now); - return createDedupeBucketRecord({ - scopeKey: params.scopeKey, - namespace: params.namespace, - bucketId: params.bucketId, - entries, - }); -} - -async function lookupDedupeBucketContains(params: { - store: TelegramMessageDispatchDedupeStore; - scopeKey: string; - namespace: string; - bucketId: string; - bucketKey: string; - key: string; - now: number; -}): Promise { - const bucket = normalizeDedupeBucketRecord(await params.store.lookup(params.bucketKey), params); - if (bucket.entries[params.key] !== undefined) { - return true; - } - for (const entry of await params.store.entries()) { - if ( - entry.key === params.bucketKey || - entry.value.scopeKey !== params.scopeKey || - entry.value.namespace !== params.namespace || - entry.value.bucketId !== params.bucketId - ) { - continue; - } - const legacyBucket = normalizeDedupeBucketRecord(entry.value, params); - if (legacyBucket.entries[params.key] !== undefined) { - return true; - } - } - return false; -} - function sanitizeFileSegment(value: string): string { const trimmed = value.trim(); if (!trimmed) { @@ -232,289 +44,19 @@ export function buildTelegramMessageDispatchReplayKey(msg: Message): string | nu return JSON.stringify(["message", String(chatId), messageId]); } -export function createTelegramMessageDispatchReplayGuard(params: { - storePath: string; - onDiskError?: (error: unknown) => void; -}): TelegramMessageDispatchReplayGuard { - const scopeKey = resolveDispatchScopeKey(params.storePath); - const onStateError = params.onDiskError; - let store: TelegramMessageDispatchDedupeStore | undefined; - const inflight = new Map(); - const committedInMemory = new Map(); - const bucketWriteQueue = new Map>(); - - function getStore(): TelegramMessageDispatchDedupeStore | undefined { - if (store) { - return store; - } - try { - store = openDispatchDedupeStore(); - return store; - } catch (error) { - onStateError?.(error); - return undefined; - } - } - - function pruneCommittedInMemory(now = Date.now()) { - for (const [entryKey, entry] of committedInMemory) { - if ( - entry.expiresAt <= now || - committedInMemory.size > TELEGRAM_MESSAGE_DISPATCH_DEDUPE_LOGICAL_MAX_ENTRIES - ) { - committedInMemory.delete(entryKey); - } - } - } - - function rememberCommittedInMemory(entryKey: string, namespace: string, now: number) { - committedInMemory.set(entryKey, { - namespace, - expiresAt: now + TELEGRAM_MESSAGE_DISPATCH_TTL_MS, - }); - pruneCommittedInMemory(now); - } - - function hasCommittedInMemory(entryKey: string, now = Date.now()): boolean { - const entry = committedInMemory.get(entryKey); - if (!entry) { - return false; - } - if (entry.expiresAt <= now) { - committedInMemory.delete(entryKey); - return false; - } - return true; - } - - function rememberPendingClaim(entryKey: string): PendingClaim { - let resolve!: (result: boolean) => void; - let reject!: (error: unknown) => void; - const promise = new Promise((resolvePromise, rejectPromise) => { - resolve = resolvePromise; - reject = rejectPromise; - }); - void promise.catch(() => {}); - const pending = { promise, resolve, reject }; - inflight.set(entryKey, pending); - return pending; - } - - function enqueueBucketWrite(bucketKey: string, write: () => Promise): Promise { - const previous = bucketWriteQueue.get(bucketKey) ?? Promise.resolve(); - const next = previous.catch(() => undefined).then(write); - const queued = next.then( - () => undefined, - () => undefined, - ); - bucketWriteQueue.set(bucketKey, queued); - void queued.finally(() => { - if (bucketWriteQueue.get(bucketKey) === queued) { - bucketWriteQueue.delete(bucketKey); - } - }); - return next; - } - - async function withBucketLock(paramsLocal: { - store: TelegramMessageDispatchDedupeStore; - namespace: string; - bucketId: string; - bucketKey: string; - write: () => Promise; - }): Promise { - const lockKey = dedupeBucketLockKey(paramsLocal.bucketKey); - const lockValue = createDedupeBucketRecord({ - scopeKey, - namespace: `${paramsLocal.namespace}:lock`, - bucketId: paramsLocal.bucketId, - }); - let locked = false; - for (let attempt = 0; attempt < TELEGRAM_MESSAGE_DISPATCH_DEDUPE_LOCK_ATTEMPTS; attempt += 1) { - if ( - await paramsLocal.store.registerIfAbsent(lockKey, lockValue, { - ttlMs: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_LOCK_TTL_MS, - }) - ) { - locked = true; - break; - } - await sleep(TELEGRAM_MESSAGE_DISPATCH_DEDUPE_LOCK_RETRY_MS); - } - if (!locked) { - throw new Error( - `timed out acquiring Telegram dispatch dedupe bucket lock: ${paramsLocal.bucketId}`, - ); - } - try { - return await paramsLocal.write(); - } finally { - await paramsLocal.store.delete(lockKey); - } - } - - return { - async claim(key, options): Promise { - const namespace = options?.namespace?.trim() || "global"; - const entryKey = dedupeEntryKey(scopeKey, namespace, key); - const bucketId = dedupeBucketId(key); - const bucketKey = dedupeBucketEntryKey(scopeKey, namespace, bucketId); - if (hasCommittedInMemory(entryKey)) { - return { kind: "duplicate" }; - } - const existing = inflight.get(entryKey); - if (existing) { - return { kind: "inflight", pending: existing.promise }; - } - const pending = rememberPendingClaim(entryKey); - const storeEntry = getStore(); - if (!storeEntry) { - return { kind: "claimed" }; - } - try { - if ( - await lookupDedupeBucketContains({ - store: storeEntry, - scopeKey, - namespace, - bucketId, - bucketKey, - key, - now: Date.now(), - }) - ) { - pending.resolve(false); - inflight.delete(entryKey); - return { kind: "duplicate" }; - } - return { kind: "claimed" }; - } catch (error) { - onStateError?.(error); - return { kind: "claimed" }; - } - }, - async commit(key, options) { - const namespace = options?.namespace?.trim() || "global"; - const now = options?.now ?? Date.now(); - const entryKey = dedupeEntryKey(scopeKey, namespace, key); - const bucketId = dedupeBucketId(key); - const bucketKey = dedupeBucketEntryKey(scopeKey, namespace, bucketId); - const storeResult = getStore(); - if (!storeResult) { - rememberCommittedInMemory(entryKey, namespace, now); - inflight.get(entryKey)?.resolve(true); - inflight.delete(entryKey); - return false; - } - try { - await enqueueBucketWrite(bucketKey, async () => { - await withBucketLock({ - store: storeResult, - namespace, - bucketId, - bucketKey, - write: async () => { - const bucket = normalizeDedupeBucketRecord(await storeResult.lookup(bucketKey), { - scopeKey, - namespace, - bucketId, - now, - }); - bucket.entries[key] = now; - pruneDedupeBucketEntries(bucket.entries, now); - await storeResult.register(bucketKey, bucket, { - ttlMs: TELEGRAM_MESSAGE_DISPATCH_TTL_MS, - }); - }, - }); - }); - rememberCommittedInMemory(entryKey, namespace, now); - inflight.get(entryKey)?.resolve(true); - return true; - } catch (error) { - rememberCommittedInMemory(entryKey, namespace, now); - inflight.get(entryKey)?.resolve(true); - onStateError?.(error); - return false; - } finally { - inflight.delete(entryKey); - } - }, - release(key, options) { - const namespace = options?.namespace?.trim() || "global"; - const entryKey = dedupeEntryKey(scopeKey, namespace, key); - const pending = inflight.get(entryKey); - if (pending) { - pending.reject(options?.error ?? new Error(`claim released before commit: ${namespace}`)); - inflight.delete(entryKey); - } - }, - async hasRecent(key, options) { - const namespace = options?.namespace?.trim() || "global"; - const entryKey = dedupeEntryKey(scopeKey, namespace, key); - const bucketId = dedupeBucketId(key); - const bucketKey = dedupeBucketEntryKey(scopeKey, namespace, bucketId); - if (hasCommittedInMemory(entryKey)) { - return true; - } - const storeValue = getStore(); - if (!storeValue) { - return false; - } - try { - return await lookupDedupeBucketContains({ - store: storeValue, - scopeKey, - namespace, - bucketId, - bucketKey, - key, - now: Date.now(), - }); - } catch (error) { - onStateError?.(error); - return false; - } - }, - async warmup(namespace = "global") { - pruneCommittedInMemory(); - const memoryCount = [...committedInMemory.values()].filter( - (entry) => entry.namespace === namespace, - ).length; - const storeLocal = getStore(); - if (!storeLocal) { - return memoryCount; - } - try { - const now = Date.now(); - const persistedCount = (await storeLocal.entries()) - .filter( - (entry) => entry.value.scopeKey === scopeKey && entry.value.namespace === namespace, - ) - .reduce((count, entry) => { - const bucket = normalizeDedupeBucketRecord(entry.value, { - scopeKey, - namespace, - bucketId: entry.value.bucketId, - now, - }); - return count + Object.keys(bucket.entries).length; - }, 0); - return persistedCount + memoryCount; - } catch (error) { - onStateError?.(error); - return memoryCount; - } - }, - clearMemory() { - inflight.clear(); - committedInMemory.clear(); - }, - memorySize() { - pruneCommittedInMemory(); - return inflight.size + committedInMemory.size; - }, - }; +export function createTelegramMessageDispatchReplayGuard( + params: { + onDiskError?: (error: unknown) => void; + } = {}, +): TelegramMessageDispatchReplayGuard { + return createClaimableDedupe({ + ttlMs: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_TTL_MS, + memoryMaxSize: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES, + pluginId: "telegram", + namespacePrefix: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE_PREFIX, + stateMaxEntries: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES, + ...(params.onDiskError ? { onDiskError: params.onDiskError } : {}), + }); } export async function claimTelegramMessageDispatchReplay(params: { @@ -572,64 +114,3 @@ export function releaseTelegramMessageDispatchReplay(params: { params.guard.release(key, { namespace: params.accountId, error: params.error }); } } - -export function setTelegramMessageDispatchDedupeStoreForTest( - store: TelegramMessageDispatchDedupeStore | undefined, -): void { - dispatchDedupeStoreForTest = store; -} - -export function listTelegramLegacyMessageDispatchDedupeEntries(params: { - storePath: string; - namespace: string; - persistedPath?: string; -}): Array<{ key: string; value: TelegramMessageDispatchDedupeRecord; ttlMs?: number }> { - const filePath = params.persistedPath ?? resolveTelegramMessageDispatchLegacyPath(params); - if (!fs.existsSync(filePath)) { - return []; - } - const now = Date.now(); - let parsed: unknown; - try { - parsed = JSON.parse(fs.readFileSync(filePath, "utf8")); - } catch { - return []; - } - if (!parsed || typeof parsed !== "object") { - return []; - } - const scopeKey = resolveDispatchScopeKey(params.storePath); - const buckets = new Map(); - for (const [key, value] of Object.entries(parsed as Record)) { - if (typeof value !== "number" || !Number.isFinite(value)) { - continue; - } - const ttlMs = TELEGRAM_MESSAGE_DISPATCH_TTL_MS - Math.max(0, now - value); - if (ttlMs <= 0) { - continue; - } - const bucketId = dedupeBucketId(key); - const bucketKey = dedupeLegacyBucketEntryKey({ - scopeKey, - namespace: params.namespace, - bucketId, - sourcePath: filePath, - }); - const bucket = buckets.get(bucketKey) ?? { - value: createDedupeBucketRecord({ - scopeKey, - namespace: params.namespace, - bucketId, - }), - ttlMs: 0, - }; - bucket.value.entries[key] = value; - bucket.ttlMs = Math.max(bucket.ttlMs, ttlMs); - buckets.set(bucketKey, bucket); - } - return [...buckets.entries()].map(([key, bucket]) => ({ - key, - value: bucket.value, - ttlMs: bucket.ttlMs, - })); -} diff --git a/extensions/telegram/src/state-migrations.test.ts b/extensions/telegram/src/state-migrations.test.ts index 4e0d94153800..dc5901ef347e 100644 --- a/extensions/telegram/src/state-migrations.test.ts +++ b/extensions/telegram/src/state-migrations.test.ts @@ -4,11 +4,20 @@ import os from "node:os"; import path from "node:path"; import type { Message } from "grammy/types"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { resolvePersistentDedupePluginStateNamespace } from "openclaw/plugin-sdk/persistent-dedupe"; +import { + createPluginStateSyncKeyedStoreForTests, + resetPluginStateStoreForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; import { resolveStorePath } from "openclaw/plugin-sdk/session-store-runtime"; -import { describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it } from "vitest"; import { resolveTelegramBotInfoCachePath } from "./bot-info-cache.js"; import { resolveTelegramMessageCachePath } from "./message-cache.js"; -import { resolveTelegramMessageDispatchLegacyPath } from "./message-dispatch-dedupe.js"; +import { + resolveTelegramMessageDispatchLegacyPath, + TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES, + TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE_PREFIX, +} from "./message-dispatch-dedupe.js"; import { detectTelegramLegacyStateMigrations } from "./state-migrations.js"; import { resolveTopicNameCacheNamespace, @@ -38,6 +47,10 @@ function persistedCacheEntry(messageId: number, text: string): PersistedCacheEnt }; } +afterEach(() => { + resetPluginStateStoreForTests(); +}); + describe("telegram state migrations", () => { it("detects legacy bot-info cache import", async () => { const dir = await mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-state-migration-")); @@ -354,6 +367,10 @@ describe("telegram state migrations", () => { }, } as OpenClawConfig; const plans = await detectTelegramLegacyStateMigrations({ cfg, env }); + const dispatchNamespace = resolvePersistentDedupePluginStateNamespace({ + namespace: "ops", + namespacePrefix: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE_PREFIX, + }); const byLabel = new Map(plans.map((plan) => [plan.label, plan])); expect(byLabel.get("Telegram update offset")).toMatchObject({ @@ -379,8 +396,21 @@ describe("telegram state migrations", () => { expect(byLabel.get("Telegram message dispatch dedupe")).toMatchObject({ kind: "plugin-state-import", sourcePath: dispatchPath, - namespace: "telegram.message-dispatch-dedupe", + namespace: dispatchNamespace, }); + const dispatchPlan = byLabel.get("Telegram message dispatch dedupe"); + if (!dispatchPlan || dispatchPlan.kind !== "plugin-state-import") { + throw new Error("expected Telegram message dispatch dedupe import plan"); + } + await expect(dispatchPlan.readEntries()).resolves.toMatchObject([ + { + key: expect.stringMatching(/^k\.[a-f0-9]{32}$/), + value: { + key: JSON.stringify(["message", "7", 42]), + seenAt: now, + }, + }, + ]); for (const label of [ "Telegram update offset", @@ -400,6 +430,93 @@ describe("telegram state migrations", () => { } }); + it("migrates shipped Telegram message dispatch plugin-state buckets", async () => { + const dir = await mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-state-migration-")); + const env = { ...process.env, OPENCLAW_STATE_DIR: dir }; + const now = Date.now(); + const replayKey = JSON.stringify(["message", "7", 42]); + const dispatchNamespace = resolvePersistentDedupePluginStateNamespace({ + namespace: "ops", + namespacePrefix: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE_PREFIX, + }); + try { + const legacyStore = createPluginStateSyncKeyedStoreForTests("telegram", { + namespace: "telegram.message-dispatch-dedupe", + maxEntries: 4_096, + env, + }); + legacyStore.register("legacy-bucket", { + scopeKey: "old-session-store", + namespace: "ops", + bucketId: "00", + entries: { + [replayKey]: now, + }, + }); + + const cfg = { + channels: { + telegram: { + accounts: { + ops: { + botToken: "123456:secret", + }, + }, + }, + }, + } as OpenClawConfig; + const plans = await detectTelegramLegacyStateMigrations({ cfg, env }); + const plan = plans.find( + (candidate) => + candidate.kind === "plugin-state-import" && + candidate.label === "Telegram message dispatch dedupe" && + candidate.sourcePath === "plugin state:telegram.message-dispatch-dedupe:ops", + ); + + expect(plan).toMatchObject({ + kind: "plugin-state-import", + namespace: dispatchNamespace, + }); + if (!plan || plan.kind !== "plugin-state-import") { + throw new Error("expected Telegram message dispatch plugin-state import plan"); + } + const entries = await plan.readEntries(); + expect(entries).toMatchObject([ + { + key: expect.stringMatching(/^k\.[a-f0-9]{32}$/), + value: { + key: replayKey, + seenAt: now, + }, + }, + ]); + + const targetStore = createPluginStateSyncKeyedStoreForTests("telegram", { + namespace: dispatchNamespace, + maxEntries: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES, + env, + }); + for (const entry of entries) { + targetStore.register( + entry.key, + entry.value, + entry.ttlMs ? { ttlMs: entry.ttlMs } : undefined, + ); + } + + const plansAfterImport = await detectTelegramLegacyStateMigrations({ cfg, env }); + expect( + plansAfterImport.some( + (candidate) => + candidate.kind === "plugin-state-import" && + candidate.sourcePath === "plugin state:telegram.message-dispatch-dedupe:ops", + ), + ).toBe(false); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + it("detects Telegram account sidecars even after the account was removed from config", async () => { const dir = await mkdtemp(path.join(os.tmpdir(), "openclaw-telegram-state-migration-")); const env = { ...process.env, OPENCLAW_STATE_DIR: dir }; diff --git a/extensions/telegram/src/state-migrations.ts b/extensions/telegram/src/state-migrations.ts index 7c641f1e60b9..446cffba08dd 100644 --- a/extensions/telegram/src/state-migrations.ts +++ b/extensions/telegram/src/state-migrations.ts @@ -4,6 +4,15 @@ import path from "node:path"; import type { ChannelLegacyStateMigrationPlan } from "openclaw/plugin-sdk/channel-contract"; import { resolveChannelAllowFromPath } from "openclaw/plugin-sdk/channel-pairing"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { + type PersistentDedupeEntry, + type PersistentDedupeLegacyJsonImportEntry, + createPersistentDedupeImportEntry, + listPersistentDedupeLegacyJsonFileEntries, + resolvePersistentDedupePluginStateNamespace, + shouldReplacePersistentDedupeEntry, +} from "openclaw/plugin-sdk/persistent-dedupe"; +import { createPluginStateSyncKeyedStore } from "openclaw/plugin-sdk/runtime-doctor"; import { statRegularFileSync } from "openclaw/plugin-sdk/security-runtime"; import { resolveStorePath } from "openclaw/plugin-sdk/session-store-runtime"; import { uniqueStrings } from "openclaw/plugin-sdk/string-coerce-runtime"; @@ -22,10 +31,10 @@ import { TELEGRAM_MESSAGE_CACHE_PERSISTENT_NAMESPACE, } from "./message-cache.js"; import { - listTelegramLegacyMessageDispatchDedupeEntries, resolveTelegramMessageDispatchLegacyPath, TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES, - TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE, + TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE_PREFIX, + TELEGRAM_MESSAGE_DISPATCH_DEDUPE_TTL_MS, } from "./message-dispatch-dedupe.js"; import { listTelegramLegacySentMessageCacheEntries, @@ -59,6 +68,14 @@ import { TELEGRAM_UPDATE_OFFSET_NAMESPACE, } from "./update-offset-store.js"; +const TELEGRAM_MESSAGE_DISPATCH_LEGACY_BUCKET_NAMESPACE = "telegram.message-dispatch-dedupe"; +const TELEGRAM_MESSAGE_DISPATCH_LEGACY_BUCKET_MAX_ENTRIES = 4_096; + +type TelegramLegacyMessageDispatchDedupeRecord = { + namespace: string; + entries: Record; +}; + function fileExists(pathValue: string): boolean { try { return !statRegularFileSync(pathValue).missing; @@ -83,6 +100,91 @@ function resolveMigrationStateDir(params: { env: NodeJS.ProcessEnv; stateDir?: s ); } +function isObjectRecord(value: unknown): value is Record { + return Boolean(value) && typeof value === "object" && !Array.isArray(value); +} + +function readLegacyMessageDispatchDedupeRecord( + value: unknown, +): TelegramLegacyMessageDispatchDedupeRecord | undefined { + if (!isObjectRecord(value) || typeof value.namespace !== "string") { + return undefined; + } + if (!isObjectRecord(value.entries)) { + return undefined; + } + const entries: Record = {}; + for (const [key, seenAt] of Object.entries(value.entries)) { + if (typeof seenAt === "number" && Number.isFinite(seenAt) && seenAt > 0) { + entries[key] = seenAt; + } + } + return { namespace: value.namespace, entries }; +} + +function remainingMessageDispatchDedupeTtlMs(seenAt: number, now: number): number | undefined { + const ttlMs = TELEGRAM_MESSAGE_DISPATCH_DEDUPE_TTL_MS - Math.max(0, now - seenAt); + return ttlMs > 0 ? ttlMs : undefined; +} + +function listTelegramLegacyMessageDispatchPluginStateEntries(params: { + accountId: string; + env: NodeJS.ProcessEnv; + now?: number; +}): PersistentDedupeLegacyJsonImportEntry[] { + const store = createPluginStateSyncKeyedStore("telegram", { + namespace: TELEGRAM_MESSAGE_DISPATCH_LEGACY_BUCKET_NAMESPACE, + maxEntries: TELEGRAM_MESSAGE_DISPATCH_LEGACY_BUCKET_MAX_ENTRIES, + env: params.env, + }); + const latestSeenAtByKey = new Map(); + for (const entry of store.entries()) { + const record = readLegacyMessageDispatchDedupeRecord(entry.value); + if (!record || record.namespace !== params.accountId) { + continue; + } + for (const [key, seenAt] of Object.entries(record.entries)) { + latestSeenAtByKey.set(key, Math.max(latestSeenAtByKey.get(key) ?? 0, seenAt)); + } + } + const now = params.now ?? Date.now(); + return [...latestSeenAtByKey.entries()].flatMap(([key, seenAt]) => { + const ttlMs = remainingMessageDispatchDedupeTtlMs(seenAt, now); + return ttlMs == null + ? [] + : [ + createPersistentDedupeImportEntry({ + key, + seenAt, + ttlMs, + }), + ]; + }); +} + +function hasCurrentMessageDispatchDedupeTargets(params: { + namespace: string; + entries: PersistentDedupeLegacyJsonImportEntry[]; + env: NodeJS.ProcessEnv; +}): boolean { + const store = createPluginStateSyncKeyedStore("telegram", { + namespace: params.namespace, + maxEntries: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES, + env: params.env, + }); + const existingByKey = new Map(store.entries().map((entry) => [entry.key, entry.value])); + return params.entries.every((entry) => { + const existingValue = existingByKey.get(entry.key); + return ( + existingValue != null && + !shouldReplacePersistentDedupeEntry({ + existingValue, + incomingValue: entry.value, + }) + ); + }); +} + function listTelegramLegacySidecarAccountIds(params: { cfg: OpenClawConfig; stateDir: string; @@ -322,15 +424,19 @@ function detectTelegramMessageDispatchLegacyStateMigration(params: { }): ChannelLegacyStateMigrationPlan[] { const storePath = resolveStorePath(params.cfg.session?.store, { env: params.env }); const legacyStorePath = resolveLegacySessionStorePath(params); + const env = params.stateDir ? { ...params.env, OPENCLAW_STATE_DIR: params.stateDir } : params.env; return listTelegramAccountIds(params.cfg).flatMap((accountId) => { + const namespace = resolvePersistentDedupePluginStateNamespace({ + namespace: accountId, + namespacePrefix: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE_PREFIX, + }); const sources = uniqueStrings([storePath, legacyStorePath]).map((sourceStorePath) => ({ - targetStorePath: storePath, sourcePath: resolveTelegramMessageDispatchLegacyPath({ storePath: sourceStorePath, namespace: accountId, }), })); - return sources.flatMap((source) => { + const jsonPlans = sources.flatMap((source) => { const sourcePath = source.sourcePath; if (!fileExists(sourcePath)) { return []; @@ -339,21 +445,58 @@ function detectTelegramMessageDispatchLegacyStateMigration(params: { kind: "plugin-state-import", label: "Telegram message dispatch dedupe", sourcePath, - targetPath: `plugin state:${TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE}`, + targetPath: `plugin state:${namespace}`, pluginId: "telegram", - namespace: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE, + namespace, maxEntries: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES, scopeKey: "", cleanupSource: "rename", - preview: `- Telegram message dispatch dedupe: ${sourcePath} → plugin state (${TELEGRAM_MESSAGE_DISPATCH_DEDUPE_NAMESPACE})`, + preview: `- Telegram message dispatch dedupe: ${sourcePath} → plugin state (${namespace})`, + shouldReplaceExistingEntry: ({ existingValue, incomingValue }) => + shouldReplacePersistentDedupeEntry({ existingValue, incomingValue }), readEntries: () => - listTelegramLegacyMessageDispatchDedupeEntries({ - storePath: source.targetStorePath, - namespace: accountId, - persistedPath: source.sourcePath, + listPersistentDedupeLegacyJsonFileEntries({ + filePath: source.sourcePath, + ttlMs: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_TTL_MS, }), }; }); + let pluginStateEntries: PersistentDedupeLegacyJsonImportEntry[]; + try { + pluginStateEntries = listTelegramLegacyMessageDispatchPluginStateEntries({ + accountId, + env, + }); + } catch { + pluginStateEntries = []; + } + if ( + pluginStateEntries.length === 0 || + hasCurrentMessageDispatchDedupeTargets({ namespace, entries: pluginStateEntries, env }) + ) { + return jsonPlans; + } + return [ + ...jsonPlans, + { + kind: "plugin-state-import", + label: "Telegram message dispatch dedupe", + sourcePath: `plugin state:${TELEGRAM_MESSAGE_DISPATCH_LEGACY_BUCKET_NAMESPACE}:${accountId}`, + targetPath: `plugin state:${namespace}`, + pluginId: "telegram", + namespace, + maxEntries: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_MAX_ENTRIES, + scopeKey: "", + preview: `- Telegram message dispatch dedupe: plugin state (${TELEGRAM_MESSAGE_DISPATCH_LEGACY_BUCKET_NAMESPACE}) → plugin state (${namespace})`, + shouldReplaceExistingEntry: ({ existingValue, incomingValue }) => + shouldReplacePersistentDedupeEntry({ existingValue, incomingValue }), + readEntries: () => + listTelegramLegacyMessageDispatchPluginStateEntries({ + accountId, + env, + }), + }, + ]; }); }