mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-24 11:25:50 -06:00
fix(telegram): use SDK dispatch replay dedupe
This commit is contained in:
Ayaan Zaidi
@@ -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)}`));
|
||||
},
|
||||
|
||||
@@ -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<typeof import("./bot-core.js").createTelegramBotCore>;
|
||||
|
||||
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<typeof messageDispatchDedupe.setTelegramMessageDispatchDedupeStoreForTest>[0]
|
||||
>;
|
||||
await store.clear();
|
||||
messageDispatchDedupe.setTelegramMessageDispatchDedupeStoreForTest(store);
|
||||
});
|
||||
|
||||
// groupPolicy tests
|
||||
|
||||
@@ -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<typeof setTelegramMessageDispatchDedupeStoreForTest>[0]
|
||||
>;
|
||||
type SyncMessageDispatchDedupeStore = Extract<MessageDispatchDedupeStore, { entries(): unknown[] }>;
|
||||
|
||||
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<Parameters<typeof setTelegramMessageDispatchDedupeStoreForTest>[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<Parameters<typeof setTelegramMessageDispatchDedupeStoreForTest>[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",
|
||||
|
||||
@@ -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<string, number>;
|
||||
};
|
||||
|
||||
type TelegramMessageDispatchDedupeStore =
|
||||
| PluginStateKeyedStore<TelegramMessageDispatchDedupeRecord>
|
||||
| PluginStateSyncKeyedStore<TelegramMessageDispatchDedupeRecord>;
|
||||
|
||||
type PendingClaim = {
|
||||
promise: Promise<boolean>;
|
||||
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<TelegramMessageDispatchDedupeRecord>({
|
||||
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<void> {
|
||||
await new Promise<void>((resolve) => {
|
||||
setTimeout(resolve, ms);
|
||||
});
|
||||
}
|
||||
|
||||
function pruneDedupeBucketEntries(entries: Record<string, number>, 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<string, number>;
|
||||
}): 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<boolean> {
|
||||
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<string, PendingClaim>();
|
||||
const committedInMemory = new Map<string, MemoryCommittedClaim>();
|
||||
const bucketWriteQueue = new Map<string, Promise<void>>();
|
||||
|
||||
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<boolean>((resolvePromise, rejectPromise) => {
|
||||
resolve = resolvePromise;
|
||||
reject = rejectPromise;
|
||||
});
|
||||
void promise.catch(() => {});
|
||||
const pending = { promise, resolve, reject };
|
||||
inflight.set(entryKey, pending);
|
||||
return pending;
|
||||
}
|
||||
|
||||
function enqueueBucketWrite<T>(bucketKey: string, write: () => Promise<T>): Promise<T> {
|
||||
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<T>(paramsLocal: {
|
||||
store: TelegramMessageDispatchDedupeStore;
|
||||
namespace: string;
|
||||
bucketId: string;
|
||||
bucketKey: string;
|
||||
write: () => Promise<T>;
|
||||
}): Promise<T> {
|
||||
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<ClaimableDedupeClaimResult> {
|
||||
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<string, { value: TelegramMessageDispatchDedupeRecord; ttlMs: number }>();
|
||||
for (const [key, value] of Object.entries(parsed as Record<string, unknown>)) {
|
||||
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,
|
||||
}));
|
||||
}
|
||||
|
||||
@@ -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 };
|
||||
|
||||
@@ -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<string, number>;
|
||||
};
|
||||
|
||||
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<string, unknown> {
|
||||
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<string, number> = {};
|
||||
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<unknown>("telegram", {
|
||||
namespace: TELEGRAM_MESSAGE_DISPATCH_LEGACY_BUCKET_NAMESPACE,
|
||||
maxEntries: TELEGRAM_MESSAGE_DISPATCH_LEGACY_BUCKET_MAX_ENTRIES,
|
||||
env: params.env,
|
||||
});
|
||||
const latestSeenAtByKey = new Map<string, number>();
|
||||
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<PersistentDedupeEntry>("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,
|
||||
}),
|
||||
},
|
||||
];
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user