@@ -5,7 +5,6 @@ import type { ChannelLegacyStateMigrationPlan } from "openclaw/plugin-sdk/channe
55import { resolveChannelAllowFromPath } from "openclaw/plugin-sdk/channel-pairing";
66import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
77import {
8-type PersistentDedupeEntry,
98type PersistentDedupeLegacyJsonImportEntry,
109createPersistentDedupeImportEntry,
1110listPersistentDedupeLegacyJsonFileEntries,
@@ -130,28 +129,45 @@ function remainingMessageDispatchDedupeTtlMs(seenAt: number, now: number): numbe
130129return ttlMs > 0 ? ttlMs : undefined;
131130}
132131133-function listTelegramLegacyMessageDispatchPluginStateEntries(params: {
134-accountId: string;
135-env: NodeJS.ProcessEnv;
136-now?: number;
137-}): PersistentDedupeLegacyJsonImportEntry[] {
138-const store = createPluginStateSyncKeyedStore<unknown>("telegram", {
132+function openTelegramLegacyMessageDispatchBucketStore(env: NodeJS.ProcessEnv) {
133+return createPluginStateSyncKeyedStore<unknown>("telegram", {
139134namespace: TELEGRAM_MESSAGE_DISPATCH_LEGACY_BUCKET_NAMESPACE,
140135maxEntries: TELEGRAM_MESSAGE_DISPATCH_LEGACY_BUCKET_MAX_ENTRIES,
141-env: params.env,
136+ env,
142137});
138+}
139+140+function readTelegramLegacyMessageDispatchBuckets(params: {
141+accountId: string;
142+env: NodeJS.ProcessEnv;
143+now?: number;
144+}): { importEntries: PersistentDedupeLegacyJsonImportEntry[]; recordKeys: string[] } {
145+const store = openTelegramLegacyMessageDispatchBucketStore(params.env);
143146const latestSeenAtByKey = new Map<string, number>();
147+const recordKeys: string[] = [];
144148for (const entry of store.entries()) {
145149const record = readLegacyMessageDispatchDedupeRecord(entry.value);
146-if (!record || record.namespace !== params.accountId) {
150+if (!record) {
151+continue;
152+}
153+// Lock rows persist as `<accountId>:lock` buckets without dedupe entries;
154+// track them as removable so cleanup empties the retired namespace.
155+const ownsRecord =
156+record.namespace === params.accountId ||
157+record.namespace.startsWith(`${params.accountId}:`);
158+if (!ownsRecord) {
159+continue;
160+}
161+recordKeys.push(entry.key);
162+if (record.namespace !== params.accountId) {
147163continue;
148164}
149165for (const [key, seenAt] of Object.entries(record.entries)) {
150166latestSeenAtByKey.set(key, Math.max(latestSeenAtByKey.get(key) ?? 0, seenAt));
151167}
152168}
153169const now = params.now ?? Date.now();
154-return [...latestSeenAtByKey.entries()].flatMap(([key, seenAt]) => {
170+const importEntries = [...latestSeenAtByKey.entries()].flatMap(([key, seenAt]) => {
155171const ttlMs = remainingMessageDispatchDedupeTtlMs(seenAt, now);
156172return ttlMs == null
157173 ? []
@@ -166,33 +182,17 @@ function listTelegramLegacyMessageDispatchPluginStateEntries(params: {
166182}),
167183];
168184});
185+return { importEntries, recordKeys };
169186}
170187171-function hasCurrentMessageDispatchDedupeTargets(params: {
172-namespace: string;
173-entries: PersistentDedupeLegacyJsonImportEntry[];
188+function removeTelegramLegacyMessageDispatchBuckets(params: {
189+accountId: string;
174190env: NodeJS.ProcessEnv;
175-}): boolean {
176-const store = createPluginStateSyncKeyedStore<PersistentDedupeEntry>(
177-TELEGRAM_MESSAGE_DISPATCH_DEDUPE_STATE_PLUGIN_ID,
178-{
179-namespace: params.namespace,
180-maxEntries: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_STATE_MAX_ENTRIES,
181-defaultTtlMs: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_TTL_MS,
182-env: params.env,
183-},
184-);
185-const existingByKey = new Map(store.entries().map((entry) => [entry.key, entry.value]));
186-return params.entries.every((entry) => {
187-const existingValue = existingByKey.get(entry.key);
188-return (
189-existingValue != null &&
190-!shouldReplacePersistentDedupeEntry({
191- existingValue,
192-incomingValue: entry.value,
193-})
194-);
195-});
191+}): void {
192+const store = openTelegramLegacyMessageDispatchBucketStore(params.env);
193+for (const key of readTelegramLegacyMessageDispatchBuckets(params).recordKeys) {
194+store.delete(key);
195+}
196196}
197197198198function mapTelegramMessageDispatchDedupeImportEntries(params: {
@@ -491,19 +491,15 @@ function detectTelegramMessageDispatchLegacyStateMigration(params: {
491491}),
492492};
493493});
494-let pluginStateEntries: PersistentDedupeLegacyJsonImportEntry[];
494+let legacyRecordKeys: string[];
495495try {
496-pluginStateEntries = listTelegramLegacyMessageDispatchPluginStateEntries({
497- accountId,
498- env,
499-});
496+legacyRecordKeys = readTelegramLegacyMessageDispatchBuckets({ accountId, env }).recordKeys;
500497} catch {
501-pluginStateEntries = [];
498+legacyRecordKeys = [];
502499}
503-if (
504-pluginStateEntries.length === 0 ||
505-hasCurrentMessageDispatchDedupeTargets({ namespace, entries: pluginStateEntries, env })
506-) {
500+// Emit the plan while any retired bucket rows remain (even TTL-expired ones)
501+// so doctor --fix imports live entries and then deletes the legacy source.
502+if (legacyRecordKeys.length === 0) {
507503return jsonPlans;
508504}
509505const pluginStatePlan: ChannelLegacyStateMigrationPlan = {
@@ -516,14 +512,12 @@ function detectTelegramMessageDispatchLegacyStateMigration(params: {
516512maxEntries: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_STATE_MAX_ENTRIES,
517513defaultTtlMs: TELEGRAM_MESSAGE_DISPATCH_DEDUPE_TTL_MS,
518514scopeKey: "",
515+cleanupWhenEmpty: true,
519516preview: `- Telegram message dispatch dedupe: plugin state (${TELEGRAM_MESSAGE_DISPATCH_LEGACY_BUCKET_NAMESPACE}) → plugin state (${namespace})`,
520517shouldReplaceExistingEntry: ({ existingValue, incomingValue }) =>
521518shouldReplacePersistentDedupeEntry({ existingValue, incomingValue }),
522-readEntries: () =>
523-listTelegramLegacyMessageDispatchPluginStateEntries({
524- accountId,
525- env,
526-}),
519+readEntries: () => readTelegramLegacyMessageDispatchBuckets({ accountId, env }).importEntries,
520+removeSource: () => removeTelegramLegacyMessageDispatchBuckets({ accountId, env }),
527521};
528522return jsonPlans.concat(pluginStatePlan);
529523});
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。