@@ -3,7 +3,7 @@ import type { Message } from "@grammyjs/types";
33import { formatLocationText } from "openclaw/plugin-sdk/channel-inbound";
44import type { MsgContext } from "openclaw/plugin-sdk/reply-runtime";
55import { logVerbose } from "openclaw/plugin-sdk/runtime-env";
6-import { replaceFileAtomicSync } from "openclaw/plugin-sdk/security-runtime";
6+import { appendRegularFileSync, replaceFileAtomicSync } from "openclaw/plugin-sdk/security-runtime";
77import { resolveTelegramPrimaryMedia } from "./bot/body-helpers.js";
88import {
99buildSenderName,
@@ -36,9 +36,11 @@ type MessageWithExternalReply = Message & { external_reply?: Message };
36363737type TelegramMessageCacheBucket = {
3838messages: Map<string, TelegramCachedMessageNode>;
39+persistedEntryCount: number;
3940};
40414142const DEFAULT_MAX_MESSAGES = 5000;
43+const COMPACT_THRESHOLD_RATIO = 2;
4244const persistedMessageCacheBuckets = new Map<string, TelegramMessageCacheBucket>();
43454446function telegramMessageCacheKey(params: {
@@ -136,55 +138,100 @@ function parsePersistedNode(value: unknown): TelegramCachedMessageNode | null {
136138return normalizeMessageNode(value.sourceMessage, Number.isFinite(threadId) ? { threadId } : {});
137139}
138140141+function parsePersistedEntry(value: unknown): {
142+key: string;
143+node: TelegramCachedMessageNode;
144+} | null {
145+if (!isRecord(value) || !isString(value.key)) {
146+return null;
147+}
148+const node = parsePersistedNode(value.node);
149+return node ? { key: value.key, node } : null;
150+}
151+152+function trimMessages(messages: Map<string, TelegramCachedMessageNode>, maxMessages: number): void {
153+while (messages.size > maxMessages) {
154+const oldest = messages.keys().next().value;
155+if (oldest === undefined) {
156+break;
157+}
158+messages.delete(oldest);
159+}
160+}
161+139162function readPersistedMessages(filePath: string, maxMessages: number) {
140163const messages = new Map<string, TelegramCachedMessageNode>();
164+let persistedEntryCount = 0;
141165if (!fs.existsSync(filePath)) {
142-return messages;
166+return { messages, persistedEntryCount };
143167}
144168try {
145-const parsed = JSON.parse(fs.readFileSync(filePath, "utf-8"));
146-if (!Array.isArray(parsed)) {
147-return messages;
148-}
149-for (const entry of parsed.slice(-maxMessages)) {
150-if (!isRecord(entry) || !isString(entry.key)) {
169+for (const line of fs.readFileSync(filePath, "utf-8").split("\n")) {
170+if (!line.trim()) {
151171continue;
152172}
153-const node = parsePersistedNode(entry.node);
154-if (node) {
155-messages.set(entry.key, node);
173+const entry = parsePersistedEntry(JSON.parse(line));
174+if (!entry) {
175+continue;
156176}
177+persistedEntryCount++;
178+messages.delete(entry.key);
179+messages.set(entry.key, entry.node);
180+trimMessages(messages, maxMessages);
157181}
158182} catch (error) {
159183logVerbose(`telegram: failed to read message cache: ${String(error)}`);
160184}
161-return messages;
185+return { messages, persistedEntryCount };
186+}
187+188+function serializePersistedEntry(key: string, node: TelegramCachedMessageNode): string {
189+return `${JSON.stringify({
190+ key,
191+ node: {
192+ sourceMessage: node.sourceMessage,
193+ ...(node.threadId ? { threadId: node.threadId } : {}),
194+ },
195+ })}\n`;
162196}
163197164-function persistMessages(params: {
198+function replacePersistedMessages(params: {
165199messages: Map<string, TelegramCachedMessageNode>;
166200persistedPath?: string;
167-}) {
201+}): number {
168202const { persistedPath, messages } = params;
169203if (!persistedPath) {
170-return;
204+return messages.size;
171205}
172206if (messages.size === 0) {
173207fs.rmSync(persistedPath, { force: true });
174-return;
208+return 0;
175209}
176-const serialized = Array.from(messages, ([key, node]) => ({
177- key,
178-node: {
179-sourceMessage: node.sourceMessage,
180- ...(node.threadId ? { threadId: node.threadId } : {}),
181-},
182-}));
210+const serialized = Array.from(messages, ([key, node]) => serializePersistedEntry(key, node)).join(
211+"",
212+);
183213replaceFileAtomicSync({
184214filePath: persistedPath,
185-content: JSON.stringify(serialized),
215+content: serialized,
186216tempPrefix: ".telegram-message-cache",
187217});
218+return messages.size;
219+}
220+221+function appendPersistedMessage(params: {
222+key: string;
223+node: TelegramCachedMessageNode;
224+persistedPath?: string;
225+}): number {
226+const { persistedPath } = params;
227+if (!persistedPath) {
228+return 0;
229+}
230+appendRegularFileSync({
231+filePath: persistedPath,
232+content: serializePersistedEntry(params.key, params.node),
233+});
234+return 1;
188235}
189236190237function resolveMessageCacheBucket(params: {
@@ -193,17 +240,20 @@ function resolveMessageCacheBucket(params: {
193240}): TelegramMessageCacheBucket {
194241const { persistedPath, maxMessages } = params;
195242if (!persistedPath) {
196-return { messages: new Map<string, TelegramCachedMessageNode>() };
243+return { messages: new Map<string, TelegramCachedMessageNode>(), persistedEntryCount: 0 };
197244}
198245const existing = persistedMessageCacheBuckets.get(persistedPath);
199246if (existing) {
200247if (!fs.existsSync(persistedPath)) {
201248existing.messages.clear();
249+existing.persistedEntryCount = 0;
202250}
203251return existing;
204252}
253+const persisted = readPersistedMessages(persistedPath, maxMessages);
205254const bucket = {
206-messages: readPersistedMessages(persistedPath, maxMessages),
255+messages: persisted.messages,
256+persistedEntryCount: persisted.persistedEntryCount,
207257};
208258persistedMessageCacheBuckets.set(persistedPath, bucket);
209259return bucket;
@@ -214,10 +264,11 @@ export function createTelegramMessageCache(params?: {
214264persistedPath?: string;
215265}): TelegramMessageCache {
216266const maxMessages = params?.maxMessages ?? DEFAULT_MAX_MESSAGES;
217-const { messages } = resolveMessageCacheBucket({
267+const bucket = resolveMessageCacheBucket({
218268persistedPath: params?.persistedPath,
219269 maxMessages,
220270});
271+const { messages } = bucket;
221272222273const get: TelegramMessageCache["get"] = ({ accountId, chatId, messageId }) => {
223274if (!messageId) {
@@ -242,15 +293,19 @@ export function createTelegramMessageCache(params?: {
242293const key = telegramMessageCacheKey({ accountId, chatId, messageId: entry.messageId });
243294messages.delete(key);
244295messages.set(key, entry);
245-while (messages.size > maxMessages) {
246-const oldest = messages.keys().next().value;
247-if (oldest === undefined) {
248-break;
249-}
250-messages.delete(oldest);
251-}
296+trimMessages(messages, maxMessages);
252297try {
253-persistMessages({ messages, persistedPath: params?.persistedPath });
298+bucket.persistedEntryCount += appendPersistedMessage({
299+ key,
300+node: entry,
301+persistedPath: params?.persistedPath,
302+});
303+if (bucket.persistedEntryCount > maxMessages * COMPACT_THRESHOLD_RATIO) {
304+bucket.persistedEntryCount = replacePersistedMessages({
305+ messages,
306+persistedPath: params?.persistedPath,
307+});
308+}
254309} catch (error) {
255310logVerbose(`telegram: failed to persist message cache: ${String(error)}`);
256311}
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。