diff --git a/CHANGELOG.md b/CHANGELOG.md index 67718f9..e0d74ab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,13 @@ This project follows the [Keep a Changelog](https://keepachangelog.com/) format. +## [Unreleased] + +### Added + +- **Quote reconstruction for newer WeChat clients:** Losslessly parse message IDs and resolve ID-only `svr_id` text and partial quotes through a per-account, per-conversation SQLite side store. Images, video, voice, and attachments are written directly into plugin-owned OpenClaw managed storage, using one retained file with time, count, byte-budget, and single-file eviction limits. Restored attachments include tool-access hints so agents can locate and read the file when automatic extraction fails. +- **Compatible degradation:** Disable quote caching when `node:sqlite` is unavailable or `quoteCache.enabled=false`, with no in-memory fallback. Cache failures never interrupt normal message delivery. + ## [2.4.7] - 2026-08-31 ### Fixed diff --git a/CHANGELOG.zh_CN.md b/CHANGELOG.zh_CN.md index c052898..6a65b6d 100644 --- a/CHANGELOG.zh_CN.md +++ b/CHANGELOG.zh_CN.md @@ -4,6 +4,13 @@ 本项目遵循 [Keep a Changelog](https://keepachangelog.com/) 格式。 +## [未发布] + +### 新增 + +- **新版微信引用消息还原:** 对消息 ID 做无损解析,并用按账号、会话隔离的 SQLite 旁路存储还原只携带 `svr_id` 的文本引用和局部引用。图片、视频、语音及附件首次直接写入插件专属的 OpenClaw 受管目录,以单份文件支持按时间、条数、空间和单文件大小淘汰;还原引用附件时会提供工具访问提示,便于 agent 在自动抽取失败后定位并读取文件。 +- **兼容降级:** `node:sqlite` 不可用或 `quoteCache.enabled=false` 时关闭引用缓存,不启用内存替代方案;缓存异常不会影响正常消息收发。 + ## [2.4.7] - 2026-08-31 ### 修复 diff --git a/README.md b/README.md index 651eccd..c9f8a7f 100644 --- a/README.md +++ b/README.md @@ -113,6 +113,78 @@ authentication or routing. All registered agents on this plugin instance currently share the same `botAgent` declaration; per-agent overrides may be added in a future version if needed. +## Local quote cache + +Newer WeChat clients may send only a server message ID for a quoted message. The +plugin therefore stores text and media metadata in SQLite and writes images, +video, voice, and attachments directly into plugin-owned OpenClaw managed +storage, so current delivery and later quotes reuse one file. When a quoted +attachment is restored, the agent also receives its original filename, managed +source path, and workspace `media/inbound/` hint, allowing file/PDF tools to read +it if automatic extraction fails. Records are isolated by account and +conversation. By default, text is retained for 30 days with a limit of 10,000 +messages per account; media is retained for 7 days with a 256 MiB per-account +budget and a 25 MiB single-file limit. Cleanup runs at startup, hourly, every 100 +writes, whenever the media budget is exceeded, and when an account is deleted. + +The quote cache is **enabled by default**. No configuration is required after +upgrading: this preserves the quoted-message experience that existed before +newer WeChat clients stopped including the quoted body. Configure it under +`channels.openclaw-weixin.quoteCache` only when you need different limits. + +| Option | Default | Description | +| --- | ---: | --- | +| `enabled` | `true` | Enables local quote reconstruction. | +| `retentionDays` | `30` | Number of days to retain text and message metadata. | +| `maxMessagesPerAccount` | `10000` | Maximum message records retained per account. | +| `mediaRetentionDays` | `7` | Number of days to retain quoted media files. | +| `maxMediaBytesPerAccount` | `268435456` | Maximum retained media per account, in bytes (256 MiB). | +| `maxSingleMediaBytes` | `26214400` | Maximum size of one retained media file, in bytes (25 MiB). Larger files are still delivered normally but are not retained for later quote reconstruction. | + +Example with custom limits while keeping the feature enabled: + +```json +{ + "channels": { + "openclaw-weixin": { + "quoteCache": { + "enabled": true, + "retentionDays": 14, + "maxMessagesPerAccount": 5000, + "mediaRetentionDays": 3, + "maxMediaBytesPerAccount": 134217728, + "maxSingleMediaBytes": 26214400 + } + } + } +} +``` + +To disable local quote storage explicitly: + +```json +{ + "channels": { + "openclaw-weixin": { + "quoteCache": { + "enabled": false + } + } + } +} +``` + +Restart the gateway after changing the configuration: + +```bash +openclaw gateway restart +``` + +If `node:sqlite` is unavailable or the database cannot be opened, the plugin +logs a warning and disables this feature automatically. It does not fall back to +an in-memory cache, and quote-cache failures do not interrupt normal message +delivery. + ## Backend API Protocol This plugin communicates with the backend gateway via HTTP JSON API. Developers integrating with their own backend need to implement the following interfaces. diff --git a/README.zh_CN.md b/README.zh_CN.md index c0184f9..1b9a166 100644 --- a/README.zh_CN.md +++ b/README.zh_CN.md @@ -109,6 +109,71 @@ openclaw config set session.dmScope per-account-channel-peer 已注册的 agent 共享同一个 `botAgent` 声明;如有需要按 agent 单独标识的场景, 可在后续版本扩展配置。 +## 引用消息本地缓存 + +新版微信的引用消息可能只携带服务端消息 ID。插件默认使用 SQLite 保存文本和媒体元数据, +并将图片、视频、语音及附件首次直接写入插件专属的 OpenClaw 受管目录,让当前消息和后续引用 +复用同一份文件。命中引用附件时,还会向 agent 提供原文件名、受管源路径及工作区 +`media/inbound/` 提示;自动抽取失败时,agent 可调用文件/PDF 工具读取。缓存按账号和会话隔离; +文本默认保留 30 天且每账号最多 10,000 条,媒体默认保留 7 天、每账号最多 256 MiB、单文件 +最多 25 MiB。淘汰在启动时、每小时、每写入 100 条及媒体超出空间上限时触发;删除账号时 +会同步删除其引用缓存。 + +引用缓存**默认开启**,升级后无需增加任何配置。这样可以在新版微信客户端不再携带引用正文后, +继续保持此前的引用消息体验。只有需要调整缓存限制时,才需要配置 +`channels.openclaw-weixin.quoteCache`。 + +| 配置项 | 默认值 | 说明 | +| --- | ---: | --- | +| `enabled` | `true` | 是否启用本地引用消息还原。 | +| `retentionDays` | `30` | 文本及消息元数据的保留天数。 | +| `maxMessagesPerAccount` | `10000` | 每个账号最多保留的消息记录数。 | +| `mediaRetentionDays` | `7` | 引用媒体文件的保留天数。 | +| `maxMediaBytesPerAccount` | `268435456` | 每个账号最多保留的媒体总大小,单位为字节(256 MiB)。 | +| `maxSingleMediaBytes` | `26214400` | 单个可保留媒体文件的最大大小,单位为字节(25 MiB)。超出后仍正常投递当前消息,但不会为后续引用保留文件。 | + +保持功能开启并自定义限制的示例: + +```json +{ + "channels": { + "openclaw-weixin": { + "quoteCache": { + "enabled": true, + "retentionDays": 14, + "maxMessagesPerAccount": 5000, + "mediaRetentionDays": 3, + "maxMediaBytesPerAccount": 134217728, + "maxSingleMediaBytes": 26214400 + } + } + } +} +``` + +如需明确关闭本地引用存储,只需配置: + +```json +{ + "channels": { + "openclaw-weixin": { + "quoteCache": { + "enabled": false + } + } + } +} +``` + +修改配置后重启 gateway: + +```bash +openclaw gateway restart +``` + +如当前 Node.js 不提供 `node:sqlite`,或数据库无法打开,插件会记录警告并自动关闭此功能。 +插件不会使用内存缓存降级,引用缓存异常也不会中断正常消息收发。 + ## 后端 API 协议 本插件通过 HTTP JSON API 与后端网关通信。二次开发者若需对接自有后端,需实现以下接口。 diff --git a/src/api/api.test.ts b/src/api/api.test.ts index a12408d..de1fe01 100644 --- a/src/api/api.test.ts +++ b/src/api/api.test.ts @@ -37,6 +37,7 @@ import { sanitizeBotAgent, readPackageJsonFromDir, classifyFetchError, + parseWeixinApiJson, } from "./api.js"; function mockResponse(body: object | string, status = 200, ok = true): Response { @@ -187,7 +188,14 @@ describe("sendMessage", () => { } as Response); await expect( sendMessage({ baseUrl: "https://api.example.com/", body: { msg: { to_user_id: "u" } } }), - ).resolves.toBeUndefined(); + ).resolves.toEqual({}); + }); + + it("returns a lossless server message ID", async () => { + mockFetch.mockResolvedValueOnce(mockResponse('{"ret":0,"message_id":18446744073709551615}')); + await expect( + sendMessage({ baseUrl: "https://api.example.com/", body: { msg: { to_user_id: "u" } } }), + ).resolves.toMatchObject({ message_id: "18446744073709551615" }); }); it("throws on non-ok response", async () => { @@ -196,6 +204,54 @@ describe("sendMessage", () => { sendMessage({ baseUrl: "https://api.example.com/", body: { msg: {} } }), ).rejects.toThrow("sendMessage 403"); }); + + it("throws on a successful HTTP response with a non-zero API ret", async () => { + mockFetch.mockResolvedValueOnce(mockResponse({ ret: 1 })); + await expect( + sendMessage({ baseUrl: "https://api.example.com/", body: { msg: {} } }), + ).rejects.toThrow("sendMessage ret=1 errmsg=(none)"); + }); +}); + +describe("parseWeixinApiJson", () => { + it("preserves all known uint64 message ID fields as strings", () => { + const parsed = parseWeixinApiJson<{ + message_id: string; + nested: { msg_id: string; ref: { svr_id: string } }; + }>( + '{"message_id":18446744073709551615,"nested":{"msg_id" : 9007199254740993,"ref":{"svr_id": 123}}}', + ); + expect(parsed).toEqual({ + message_id: "18446744073709551615", + nested: { msg_id: "9007199254740993", ref: { svr_id: "123" } }, + }); + }); + + it("does not rewrite matching text inside JSON strings", () => { + const raw = '{"text":"\\\"message_id\\\":18446744073709551615","message_id":"7"}'; + expect(parseWeixinApiJson<{ text: string; message_id: string }>(raw)).toEqual({ + text: '"message_id":18446744073709551615', + message_id: "7", + }); + }); + + it("leaves non-ID keys and non-numeric ID values unchanged", () => { + expect( + parseWeixinApiJson( + '{ "other": 9007199254740993, "message_id": null, "msg_id": "already-string" }', + ), + ).toEqual({ + other: 9007199254740992, + message_id: null, + msg_id: "already-string", + }); + }); + + it("handles negative numeric IDs without losing precision", () => { + expect(parseWeixinApiJson<{ svr_id: string }>('{"svr_id" : -9007199254740993}')).toEqual({ + svr_id: "-9007199254740993", + }); + }); }); describe("getConfig", () => { diff --git a/src/api/api.ts b/src/api/api.ts index d59de2b..5febfea 100644 --- a/src/api/api.ts +++ b/src/api/api.ts @@ -467,7 +467,7 @@ export async function getUpdates( label: "getUpdates", abortSignal: params.abortSignal, }); - const resp: GetUpdatesResp = JSON.parse(rawText); + const resp = parseWeixinApiJson(rawText); return resp; } catch (err) { // Long-poll timeout or external abort are both normal control-flow exits. @@ -515,10 +515,71 @@ export async function getUploadUrl( return resp; } -/** Send a single message downstream. */ +const LOSSLESS_ID_FIELDS = new Set(["message_id", "msg_id", "svr_id"]); + +/** + * Quote uint64 message identifiers before JSON.parse sees them. This scanner + * only rewrites actual object properties, never matching text inside JSON strings. + */ +export function parseWeixinApiJson(rawText: string): T { + let output = ""; + let index = 0; + while (index < rawText.length) { + if (rawText[index] !== '"') { + output += rawText[index++]; + continue; + } + + const stringStart = index; + index++; + let escaped = false; + while (index < rawText.length) { + const char = rawText[index++]; + if (escaped) { + escaped = false; + } else if (char === "\\") { + escaped = true; + } else if (char === '"') { + break; + } + } + const stringToken = rawText.slice(stringStart, index); + output += stringToken; + + let cursor = index; + while (/\s/.test(rawText[cursor] ?? "")) cursor++; + if (rawText[cursor] !== ":") continue; + + let key: unknown; + try { + key = JSON.parse(stringToken); + } catch { + continue; + } + if (typeof key !== "string" || !LOSSLESS_ID_FIELDS.has(key)) continue; + + output += rawText.slice(index, cursor + 1); + cursor++; + while (/\s/.test(rawText[cursor] ?? "")) { + output += rawText[cursor++]; + } + const numberStart = cursor; + if (rawText[cursor] === "-") cursor++; + while (/\d/.test(rawText[cursor] ?? "")) cursor++; + if (cursor > numberStart && !(cursor === numberStart + 1 && rawText[numberStart] === "-")) { + output += `"${rawText.slice(numberStart, cursor)}"`; + index = cursor; + } else { + index = numberStart; + } + } + return JSON.parse(output) as T; +} + +/** Send a single message downstream and return the server-assigned message ID. */ export async function sendMessage( params: WeixinApiOptions & { body: SendMessageReq }, -): Promise { +): Promise { const rawText = await apiPostFetch({ baseUrl: params.baseUrl, endpoint: "ilink/bot/sendmessage", @@ -527,10 +588,11 @@ export async function sendMessage( timeoutMs: params.timeoutMs ?? DEFAULT_API_TIMEOUT_MS, label: "sendMessage", }); - const resp: SendMessageResp = JSON.parse(rawText); + const resp = parseWeixinApiJson(rawText); if (resp.ret && resp.ret !== 0) { throw new Error(`sendMessage ret=${resp.ret} errmsg=${resp.errmsg ?? "(none)"}`); } + return resp; } /** Fetch bot config (includes typing_ticket) for a given user. */ diff --git a/src/api/types.ts b/src/api/types.ts index f63656d..276e845 100644 --- a/src/api/types.ts +++ b/src/api/types.ts @@ -147,6 +147,18 @@ export interface VideoItem { export interface RefMessage { message_item?: MessageItem; title?: string; // 摘要 + /** Server message ID used when newer clients omit the quoted content. */ + svr_id?: string; + /** Optional metadata describing a selected substring of the quoted message. */ + partial_text?: PartialText; +} + +export interface PartialText { + start: string; + end: string; + startindex: number; + endindex: number; + quotemd5: string; } export interface ToolCallStartItem { @@ -179,7 +191,8 @@ export interface MessageItem { /** Unified message (proto: WeixinMessage). Replaces the old split Message + MessageContent + FullMessage. */ export interface WeixinMessage { seq?: number; - message_id?: number; + /** uint64 on the wire; parsed losslessly as a string. */ + message_id?: string; from_user_id?: string; to_user_id?: string; client_id?: string; @@ -224,6 +237,8 @@ export interface SendMessageReq { } export interface SendMessageResp { + /** uint64 on the wire; parsed losslessly as a string. */ + message_id?: string; ret?: number; errmsg?: string; } diff --git a/src/auth/accounts.ts b/src/auth/accounts.ts index 43d613e..dbfa878 100644 --- a/src/auth/accounts.ts +++ b/src/auth/accounts.ts @@ -5,6 +5,7 @@ import { normalizeAccountId } from "openclaw/plugin-sdk/account-id"; import type { OpenClawConfig } from "openclaw/plugin-sdk/core"; import { resolveStateDir } from "../storage/state-dir.js"; +import { deleteQuoteCacheForAccount } from "../messaging/quote-store.js"; import { resolveFrameworkAllowFromPath } from "./pairing.js"; import { logger } from "../util/logger.js"; @@ -224,6 +225,7 @@ export function saveWeixinAccount( * - credentials/openclaw-weixin-{accountId}-allowFrom.json (authorized users) */ export function clearWeixinAccount(accountId: string): void { + deleteQuoteCacheForAccount(accountId); const dir = resolveAccountsDir(); const accountFiles = [ `${accountId}.json`, diff --git a/src/channel.ts b/src/channel.ts index 87917c4..1899793 100644 --- a/src/channel.ts +++ b/src/channel.ts @@ -23,6 +23,7 @@ import { restoreContextTokens, clearContextTokensForAccount, } from "./messaging/inbound.js"; +import { deactivateQuoteStoreAccount, initializeQuoteStore } from "./messaging/quote-store.js"; import { logger } from "./util/logger.js"; import { DEFAULT_ILINK_BOT_TYPE, @@ -158,6 +159,7 @@ async function sendWeixinOutbound(params: { baseUrl: account.baseUrl, token: account.token, contextToken: params.contextToken, + accountId: account.accountId, }, }); emitWeixinMessageSent({ @@ -200,6 +202,27 @@ export const weixinPlugin: ChannelPlugin = { default: true, description: "Send structured tool-call progress messages.", }, + quoteCache: { + type: "object", + additionalProperties: false, + description: "Persist message content locally so ID-only Weixin quotes can be resolved.", + properties: { + enabled: { type: "boolean", default: true }, + retentionDays: { type: "number", default: 30, minimum: 0.01 }, + maxMessagesPerAccount: { type: "integer", default: 10000, minimum: 1 }, + mediaRetentionDays: { type: "number", default: 7, minimum: 0.01 }, + maxMediaBytesPerAccount: { + type: "integer", + default: 268435456, + minimum: 1, + }, + maxSingleMediaBytes: { + type: "integer", + default: 26214400, + minimum: 1, + }, + }, + }, }, }, }, @@ -298,7 +321,12 @@ export const weixinPlugin: ChannelPlugin = { filePath, to: ctx.to, text, - opts: { baseUrl: account.baseUrl, token: account.token, contextToken }, + opts: { + baseUrl: account.baseUrl, + token: account.token, + contextToken, + accountId: account.accountId, + }, cdnBaseUrl: account.cdnBaseUrl, }); emitWeixinMessageSent({ @@ -329,6 +357,7 @@ export const weixinPlugin: ChannelPlugin = { baseUrl: account.baseUrl, token: account.token, contextToken, + accountId: account.accountId, }, }); emitWeixinMessageSent({ @@ -483,6 +512,8 @@ export const weixinPlugin: ChannelPlugin = { throw new Error("weixin not configured: missing token"); } + await initializeQuoteStore(ctx.cfg, account.accountId); + ctx.log?.info?.(`[${account.accountId}] starting weixin provider (${DEFAULT_BASE_URL})`); try { @@ -527,6 +558,7 @@ export const weixinPlugin: ChannelPlugin = { stopAccount: async (ctx) => { const account = ctx.account; const aLog = logger.withAccount(account.accountId); + deactivateQuoteStoreAccount(account.accountId); if (!account.configured || !account.token?.trim()) { aLog.debug(`gateway.stopAccount: skip notifyStop (not configured or no token)`); return; diff --git a/src/config/config-schema.test.ts b/src/config/config-schema.test.ts index f65397a..56bcade 100644 --- a/src/config/config-schema.test.ts +++ b/src/config/config-schema.test.ts @@ -7,6 +7,14 @@ describe("WeixinConfigSchema", () => { expect(result.baseUrl).toBe("https://ilinkai.weixin.qq.com"); expect(result.cdnBaseUrl).toBe("https://novac2c.cdn.weixin.qq.com/c2c"); expect(result.replyProgressMessages).toBe(true); + expect(result.quoteCache).toEqual({ + enabled: true, + retentionDays: 30, + maxMessagesPerAccount: 10_000, + mediaRetentionDays: 7, + maxMediaBytesPerAccount: 256 * 1024 * 1024, + maxSingleMediaBytes: 25 * 1024 * 1024, + }); }); it("accepts custom baseUrl and cdnBaseUrl", () => { @@ -34,6 +42,12 @@ describe("WeixinConfigSchema", () => { expect(result.replyProgressMessages).toBe(false); }); + it("accepts disabling quote caching for hosts without SQLite", () => { + const result = WeixinConfigSchema.parse({ quoteCache: { enabled: false } }); + expect(result.quoteCache.enabled).toBe(false); + expect(result.quoteCache.retentionDays).toBe(30); + }); + it("accepts accounts map", () => { const result = WeixinConfigSchema.parse({ accounts: { diff --git a/src/config/config-schema.ts b/src/config/config-schema.ts index 4d152cc..da256ec 100644 --- a/src/config/config-schema.ts +++ b/src/config/config-schema.ts @@ -14,10 +14,35 @@ const weixinAccountSchema = z.object({ routeTag: z.number().optional(), }); +const quoteCacheSchema = z.object({ + enabled: z.boolean().default(true), + retentionDays: z.number().positive().default(30), + maxMessagesPerAccount: z.number().int().positive().default(10_000), + mediaRetentionDays: z.number().positive().default(7), + maxMediaBytesPerAccount: z + .number() + .int() + .positive() + .default(256 * 1024 * 1024), + maxSingleMediaBytes: z + .number() + .int() + .positive() + .default(25 * 1024 * 1024), +}); + /** Top-level weixin config schema (token is stored in credentials file, not config). */ export const WeixinConfigSchema = weixinAccountSchema.extend({ accounts: z.record(z.string(), weixinAccountSchema).optional(), replyProgressMessages: z.boolean().default(true), + quoteCache: quoteCacheSchema.default({ + enabled: true, + retentionDays: 30, + maxMessagesPerAccount: 10_000, + mediaRetentionDays: 7, + maxMediaBytesPerAccount: 256 * 1024 * 1024, + maxSingleMediaBytes: 25 * 1024 * 1024, + }), /** ISO 8601; bumped on each successful login to refresh gateway config from disk. */ channelConfigUpdatedAt: z.string().optional(), }); diff --git a/src/media/media-download.ts b/src/media/media-download.ts index 5a2fda7..3f66d02 100644 --- a/src/media/media-download.ts +++ b/src/media/media-download.ts @@ -26,12 +26,15 @@ export async function downloadMediaFromItem( deps: { cdnBaseUrl: string; saveMedia: SaveMediaFn; + /** OpenClaw media-store subdirectory; defaults to the transient inbound bucket. */ + mediaSubdir?: string; log: (msg: string) => void; errLog: (msg: string) => void; label: string; }, ): Promise { const { cdnBaseUrl, saveMedia, log, errLog, label } = deps; + const mediaSubdir = deps.mediaSubdir ?? "inbound"; const result: WeixinInboundMediaOpts = {}; if (item.type === MessageItemType.IMAGE) { @@ -58,7 +61,7 @@ export async function downloadMediaFromItem( `${label} image-plain`, img.media.full_url, ); - const saved = await saveMedia(buf, undefined, "inbound", WEIXIN_MEDIA_MAX_BYTES); + const saved = await saveMedia(buf, undefined, mediaSubdir, WEIXIN_MEDIA_MAX_BYTES); result.decryptedPicPath = saved.path; logger.debug(`${label} image saved: ${saved.path}`); } catch (err) { @@ -80,12 +83,12 @@ export async function downloadMediaFromItem( logger.debug(`${label} voice: decrypted ${silkBuf.length} bytes, attempting silk transcode`); const wavBuf = await silkToWav(silkBuf); if (wavBuf) { - const saved = await saveMedia(wavBuf, "audio/wav", "inbound", WEIXIN_MEDIA_MAX_BYTES); + const saved = await saveMedia(wavBuf, "audio/wav", mediaSubdir, WEIXIN_MEDIA_MAX_BYTES); result.decryptedVoicePath = saved.path; result.voiceMediaType = "audio/wav"; logger.debug(`${label} voice: saved WAV to ${saved.path}`); } else { - const saved = await saveMedia(silkBuf, "audio/silk", "inbound", WEIXIN_MEDIA_MAX_BYTES); + const saved = await saveMedia(silkBuf, "audio/silk", mediaSubdir, WEIXIN_MEDIA_MAX_BYTES); result.decryptedVoicePath = saved.path; result.voiceMediaType = "audio/silk"; logger.debug(`${label} voice: silk transcode unavailable, saved raw SILK to ${saved.path}`); @@ -113,7 +116,7 @@ export async function downloadMediaFromItem( const saved = await saveMedia( buf, mime, - "inbound", + mediaSubdir, WEIXIN_MEDIA_MAX_BYTES, fileItem.file_name ?? undefined, ); @@ -139,7 +142,7 @@ export async function downloadMediaFromItem( `${label} video`, videoItem.media.full_url, ); - const saved = await saveMedia(buf, "video/mp4", "inbound", WEIXIN_MEDIA_MAX_BYTES); + const saved = await saveMedia(buf, "video/mp4", mediaSubdir, WEIXIN_MEDIA_MAX_BYTES); result.decryptedVideoPath = saved.path; logger.debug(`${label} video: saved to ${saved.path}`); } catch (err) { diff --git a/src/messaging/error-notice.ts b/src/messaging/error-notice.ts index a2d299d..0532832 100644 --- a/src/messaging/error-notice.ts +++ b/src/messaging/error-notice.ts @@ -13,6 +13,7 @@ export async function sendWeixinErrorNotice(params: { baseUrl: string; token?: string; runId?: string; + accountId?: string; errLog: (m: string) => void; }): Promise { if (!params.contextToken) { @@ -28,6 +29,7 @@ export async function sendWeixinErrorNotice(params: { baseUrl: params.baseUrl, token: params.token, contextToken: params.contextToken, + accountId: params.accountId, ...(params.runId ? { runId: params.runId } : {}), }, }); diff --git a/src/messaging/inbound.test.ts b/src/messaging/inbound.test.ts index 927fd9b..91918b5 100644 --- a/src/messaging/inbound.test.ts +++ b/src/messaging/inbound.test.ts @@ -1,8 +1,14 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + import { describe, it, expect, vi, beforeEach } from "vitest"; import { + getContextTokenFromMsgContext, + getWeixinMessageId, isMediaItem, + resolveStoredQuoteContext, weixinMessageToMsgContext, - getContextTokenFromMsgContext, } from "./inbound.js"; import type { WeixinMsgContext } from "./inbound.js"; import { MessageItemType } from "../api/types.js"; @@ -51,6 +57,17 @@ describe("isMediaItem", () => { }); }); +describe("getMediaLabel", () => { + it("labels every supported media kind", async () => { + const { getMediaLabel } = await import("./inbound.js"); + expect(getMediaLabel(MessageItemType.IMAGE)).toBe("[图片]"); + expect(getMediaLabel(MessageItemType.VIDEO)).toBe("[视频]"); + expect(getMediaLabel(MessageItemType.FILE)).toBe("[文件]"); + expect(getMediaLabel(MessageItemType.VOICE)).toBe("[语音]"); + expect(getMediaLabel(MessageItemType.TEXT)).toBe(""); + }); +}); + describe("weixinMessageToMsgContext", () => { beforeEach(() => { vi.spyOn(Date, "now").mockReturnValue(1700000000000); @@ -144,6 +161,17 @@ describe("weixinMessageToMsgContext", () => { expect(ctx.MediaType).toBe("audio/wav"); }); + it("uses transcribed voice text as the body", () => { + const ctx = weixinMessageToMsgContext( + { + from_user_id: "u", + item_list: [{ type: MessageItemType.VOICE, voice_item: { text: "voice transcript" } }], + }, + "acc", + ); + expect(ctx.Body).toBe("voice transcript"); + }); + it("prioritizes pic > video > file > voice", () => { const ctx = weixinMessageToMsgContext(baseMsg, "acc", { decryptedPicPath: "/tmp/pic.png", @@ -155,6 +183,17 @@ describe("weixinMessageToMsgContext", () => { expect(ctx.MediaType).toBe("image/*"); }); + it("keeps downloaded inline-quote media out of the current message attachment", () => { + const ctx = weixinMessageToMsgContext(baseMsg, "acc", { + decryptedPicPath: "/tmp/quoted.png", + referencedMedia: true, + }); + expect(ctx.MediaPath).toBeUndefined(); + expect(ctx.MediaType).toBeUndefined(); + expect(ctx.MediaPaths).toEqual(["/tmp/quoted.png"]); + expect(ctx.MediaTypes).toEqual(["image/*"]); + }); + it("builds quoted context from ref_msg title", () => { const msg: WeixinMessage = { from_user_id: "u", @@ -167,7 +206,9 @@ describe("weixinMessageToMsgContext", () => { ], }; const ctx = weixinMessageToMsgContext(msg, "acc"); - expect(ctx.Body).toBe("[引用: original title]\nreply"); + expect(ctx.Body).toBe("reply"); + expect(ctx.ReplyToBody).toBe("original title"); + expect(ctx.ReplyToIsQuote).toBe(true); }); it("skips quoted context when ref_msg is a media item", () => { @@ -205,7 +246,9 @@ describe("weixinMessageToMsgContext", () => { ], }; const ctx = weixinMessageToMsgContext(msg, "acc"); - expect(ctx.Body).toBe("[引用: Author | original text]\nmy reply"); + expect(ctx.Body).toBe("my reply"); + expect(ctx.ReplyToBody).toBe("Author | original text"); + expect(ctx.ReplyToIsQuote).toBe(true); }); it("builds quoted context with only message_item (no title)", () => { @@ -225,7 +268,8 @@ describe("weixinMessageToMsgContext", () => { ], }; const ctx = weixinMessageToMsgContext(msg, "acc"); - expect(ctx.Body).toBe("[引用: quoted]\nreply"); + expect(ctx.Body).toBe("reply"); + expect(ctx.ReplyToBody).toBe("quoted"); }); it("returns text when ref_msg has no extractable content", () => { @@ -243,13 +287,262 @@ describe("weixinMessageToMsgContext", () => { expect(ctx.Body).toBe("reply"); }); - it("returns empty body when item_list has only non-text items", () => { + it("uses a stable label when item_list has only a media item", () => { const msg: WeixinMessage = { from_user_id: "u", item_list: [{ type: MessageItemType.IMAGE }], }; const ctx = weixinMessageToMsgContext(msg, "acc"); - expect(ctx.Body).toBe(""); + expect(ctx.Body).toBe("[图片]"); + }); + + it("keeps the provider uint64 ID separately from the generated MessageSid", () => { + const ctx = weixinMessageToMsgContext( + { ...baseMsg, message_id: "18446744073709551615" }, + "acc", + ); + expect(ctx.MessageSidFull).toBe("18446744073709551615"); + expect(ctx.MessageSid).not.toBe(ctx.MessageSidFull); + }); +}); + +describe("stored quote resolution", () => { + const quotedMessage = ( + partial_text?: NonNullable["partial_text"]>, + ): WeixinMessage => ({ + from_user_id: "user1", + item_list: [ + { + type: MessageItemType.TEXT, + text_item: { text: "reply" }, + ref_msg: { svr_id: "9007199254740993123", ...(partial_text ? { partial_text } : {}) }, + }, + ], + }); + + it("looks up an ID-only quote in the account and conversation scope", () => { + const msg = quotedMessage(); + const ctx = weixinMessageToMsgContext(msg, "acc"); + const find = vi.fn(() => ({ + accountId: "acc", + conversationId: "user1", + messageId: "9007199254740993123", + direction: "inbound" as const, + body: "original text", + createdAt: Date.now(), + })); + resolveStoredQuoteContext(ctx, msg, "acc", { find }); + expect(find).toHaveBeenCalledWith("acc", "user1", "9007199254740993123"); + expect(ctx).toMatchObject({ + Body: "reply", + ReplyToId: "9007199254740993123", + ReplyToBody: "original text", + ReplyToIsQuote: true, + }); + }); + + it("marks a cache miss without modifying the current message body", () => { + const msg = quotedMessage(); + const ctx = weixinMessageToMsgContext(msg, "acc"); + resolveStoredQuoteContext(ctx, msg, "acc", { find: () => null }); + expect(ctx.Body).toBe("reply"); + expect(ctx.ReplyToBody).toBe("[引用消息内容未缓存]"); + }); + + it("resolves the legacy nested message ID", () => { + const msg: WeixinMessage = { + from_user_id: "user1", + item_list: [ + { + type: MessageItemType.TEXT, + text_item: { text: "reply" }, + ref_msg: { message_item: { type: MessageItemType.TEXT, msg_id: "legacy-id" } }, + }, + ], + }; + const ctx = weixinMessageToMsgContext(msg, "acc"); + resolveStoredQuoteContext(ctx, msg, "acc", { + find: () => ({ + accountId: "acc", + conversationId: "user1", + messageId: "legacy-id", + direction: "inbound", + body: "legacy body", + createdAt: Date.now(), + }), + }); + expect(ctx.ReplyToId).toBe("legacy-id"); + expect(ctx.ReplyToBody).toBe("legacy body"); + }); + + it("keeps complete inline quote content instead of replacing it from storage", () => { + const msg: WeixinMessage = { + from_user_id: "user1", + item_list: [ + { + type: MessageItemType.TEXT, + text_item: { text: "reply" }, + ref_msg: { + svr_id: "id", + message_item: { type: MessageItemType.TEXT, text_item: { text: "inline body" } }, + }, + }, + ], + }; + const ctx = weixinMessageToMsgContext(msg, "acc"); + const find = vi.fn(); + resolveStoredQuoteContext(ctx, msg, "acc", { find }); + expect(find).not.toHaveBeenCalled(); + expect(ctx.ReplyToBody).toBe("inline body"); + }); + + it("sets ReplyToQuoteText for a partial quote", () => { + const selected = "second abc"; + const msg = quotedMessage({ + start: "s", + end: "c", + startindex: 1, + endindex: 2, + quotemd5: "", + }); + const ctx = weixinMessageToMsgContext(msg, "acc"); + resolveStoredQuoteContext(ctx, msg, "acc", { + find: () => ({ + accountId: "acc", + conversationId: "user1", + messageId: "9007199254740993123", + direction: "inbound", + body: "start abc, second abc", + createdAt: Date.now(), + }), + }); + expect(ctx.ReplyToQuoteText).toBe(selected); + }); + + it.each([ + ["image/png", "picture.png", "[引用的图片已过期: picture.png]"], + ["video/mp4", undefined, "[引用的视频已过期]"], + ["audio/mpeg", undefined, "[引用的语音已过期]"], + ["application/pdf", "doc.pdf", "[引用的附件已过期: doc.pdf]"], + ])("describes expired managed media (%s)", (mediaMime, mediaName, expected) => { + const msg = quotedMessage(); + const ctx = weixinMessageToMsgContext(msg, "acc"); + resolveStoredQuoteContext(ctx, msg, "acc", { + find: () => ({ + accountId: "acc", + conversationId: "user1", + messageId: "9007199254740993123", + direction: "inbound", + body: "[媒体]", + mediaMime, + ...(mediaName ? { mediaName } : {}), + createdAt: Date.now(), + }), + }); + expect(ctx.ReplyToBody).toBe(expected); + }); + + it("adds an existing quoted media file alongside the current attachment", () => { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "weixin-inbound-quote-")); + const quotedPath = path.join(tempDir, "quoted.png"); + fs.writeFileSync(quotedPath, "image"); + try { + const msg = quotedMessage(); + const ctx = weixinMessageToMsgContext(msg, "acc", { decryptedFilePath: "/tmp/current.pdf" }); + ctx.ChannelPromptContext = ["existing channel context"]; + resolveStoredQuoteContext(ctx, msg, "acc", { + find: () => ({ + accountId: "acc", + conversationId: "user1", + messageId: "9007199254740993123", + direction: "inbound", + body: "[图片]", + mediaPath: quotedPath, + mediaMime: "image/png", + mediaName: "quoted-original.png", + createdAt: Date.now(), + }), + }); + expect(ctx.MediaPaths).toEqual(["/tmp/current.pdf", quotedPath]); + expect(ctx.MediaTypes).toEqual(["application/octet-stream", "image/png"]); + expect(ctx.media).toEqual([ + { path: "/tmp/current.pdf", contentType: "application/octet-stream" }, + { + path: quotedPath, + contentType: "image/png", + fileName: "quoted-original.png", + messageId: "9007199254740993123", + }, + ]); + expect(ctx.ChannelPromptContext).toEqual([ + "existing channel context", + [ + "Quoted attachment tool access:", + JSON.stringify({ + message_id: "9007199254740993123", + original_filename: "quoted-original.png", + managed_source_path: quotedPath, + workspace_directory: "media/inbound/", + }), + "The attachment is staged into the agent workspace under media/inbound/. " + + "If automatic extraction fails and the user asks about its contents, use the available " + + "file/PDF tools to locate it by original_filename and read it.", + ].join("\n"), + ]); + } finally { + fs.rmSync(tempDir, { recursive: true, force: true }); + } + }); + + it("marks a referenced media path that disappeared from disk as expired", () => { + const msg = quotedMessage(); + const ctx = weixinMessageToMsgContext(msg, "acc"); + resolveStoredQuoteContext(ctx, msg, "acc", { + find: () => ({ + accountId: "acc", + conversationId: "user1", + messageId: "9007199254740993123", + direction: "inbound", + body: "[视频]", + mediaPath: "/definitely/missing/video.mp4", + mediaMime: "video/mp4", + createdAt: Date.now(), + }), + }); + expect(ctx.ReplyToBody).toBe("[引用的视频已过期]"); + }); + + it("adds a tool-access hint and falls back to the managed basename", () => { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "weixin-inbound-quote-hint-")); + const quotedPath = path.join(tempDir, "managed-document.pdf"); + fs.writeFileSync(quotedPath, "pdf"); + try { + const msg = quotedMessage(); + const ctx = weixinMessageToMsgContext(msg, "acc"); + resolveStoredQuoteContext(ctx, msg, "acc", { + find: () => ({ + accountId: "acc", + conversationId: "user1", + messageId: "9007199254740993123", + direction: "inbound", + body: "[文件]", + mediaPath: quotedPath, + mediaMime: "application/pdf", + createdAt: Date.now(), + }), + }); + expect(ctx.ChannelPromptContext?.[0]).toContain('"original_filename":"managed-document.pdf"'); + expect(ctx.ChannelPromptContext?.[0]).toContain('"workspace_directory":"media/inbound/"'); + } finally { + fs.rmSync(tempDir, { recursive: true, force: true }); + } + }); +}); + +describe("getWeixinMessageId", () => { + it("prefers the top-level ID and falls back to an item ID", () => { + expect(getWeixinMessageId({ message_id: "10", item_list: [{ msg_id: "11" }] })).toBe("10"); + expect(getWeixinMessageId({ item_list: [{ msg_id: "11" }] })).toBe("11"); }); }); diff --git a/src/messaging/inbound.ts b/src/messaging/inbound.ts index cc47d4c..fea1b40 100644 --- a/src/messaging/inbound.ts +++ b/src/messaging/inbound.ts @@ -6,6 +6,9 @@ import { generateId } from "../util/random.js"; import type { WeixinMessage, MessageItem } from "../api/types.js"; import { MessageItemType } from "../api/types.js"; import { resolveStateDir } from "../storage/state-dir.js"; +import { resolvePartialQuote } from "./partial-quote.js"; +import { getQuoteStore } from "./quote-store.js"; +import type { QuoteStore } from "./quote-store.js"; // --------------------------------------------------------------------------- // Context token store (in-process cache + disk persistence) @@ -141,6 +144,8 @@ export type WeixinMsgContext = { OriginatingChannel: "openclaw-weixin"; OriginatingTo: string; MessageSid: string; + /** Provider message ID retained losslessly for transcript metadata. */ + MessageSidFull?: string; Timestamp?: number; Provider: "openclaw-weixin"; ChatType: "direct"; @@ -150,6 +155,19 @@ export type WeixinMsgContext = { MediaUrl?: string; MediaPath?: string; MediaType?: string; + MediaPaths?: string[]; + MediaTypes?: string[]; + media?: Array<{ + path?: string; + contentType?: string; + fileName?: string; + messageId?: string; + }>; + ChannelPromptContext?: string[]; + ReplyToId?: string; + ReplyToBody?: string; + ReplyToQuoteText?: string; + ReplyToIsQuote?: boolean; /** Raw message body for framework command authorization. */ CommandBody?: string; /** Whether the sender is authorized to execute slash commands. */ @@ -166,29 +184,42 @@ export function isMediaItem(item: MessageItem): boolean { ); } +export function getMediaLabel(type: number | undefined): string { + switch (type) { + case MessageItemType.IMAGE: + return "[图片]"; + case MessageItemType.VIDEO: + return "[视频]"; + case MessageItemType.FILE: + return "[文件]"; + case MessageItemType.VOICE: + return "[语音]"; + default: + return ""; + } +} + +export function getWeixinMessageId(msg: WeixinMessage): string | undefined { + const topLevel = msg.message_id?.trim(); + if (topLevel) return topLevel; + for (const item of msg.item_list ?? []) { + const itemId = item.msg_id?.trim(); + if (itemId) return itemId; + } + return undefined; +} + function bodyFromItemList(itemList?: MessageItem[]): string { if (!itemList?.length) return ""; for (const item of itemList) { if (item.type === MessageItemType.TEXT && item.text_item?.text != null) { - const text = String(item.text_item.text); - const ref = item.ref_msg; - if (!ref) return text; - // Quoted media is passed as MediaPath; only include the current text as body. - if (ref.message_item && isMediaItem(ref.message_item)) return text; - // Build quoted context from both title and message_item content. - const parts: string[] = []; - if (ref.title) parts.push(ref.title); - if (ref.message_item) { - const refBody = bodyFromItemList([ref.message_item]); - if (refBody) parts.push(refBody); - } - if (!parts.length) return text; - return `[引用: ${parts.join(" | ")}]\n${text}`; + return String(item.text_item.text); } // 语音转文字:如果语音消息有 text 字段,直接使用文字内容 if (item.type === MessageItemType.VOICE && item.voice_item?.text) { return item.voice_item.text; } + if (isMediaItem(item)) return getMediaLabel(item.type); } return ""; } @@ -206,6 +237,8 @@ export type WeixinInboundMediaOpts = { fileMediaType?: string; /** Local path to decrypted video file. */ decryptedVideoPath?: string; + /** The downloaded media belongs to an inline quoted message, not the current message. */ + referencedMedia?: boolean; }; /** @@ -235,6 +268,8 @@ export function weixinMessageToMsgContext( if (msg.context_token) { ctx.context_token = msg.context_token; } + const providerMessageId = getWeixinMessageId(msg); + if (providerMessageId) ctx.MessageSidFull = providerMessageId; if (opts?.decryptedPicPath) { ctx.MediaPath = opts.decryptedPicPath; @@ -250,9 +285,123 @@ export function weixinMessageToMsgContext( ctx.MediaType = opts.voiceMediaType ?? "audio/wav"; } + if (opts?.referencedMedia && ctx.MediaPath) { + ctx.MediaPaths = [ctx.MediaPath]; + ctx.MediaTypes = [ctx.MediaType ?? "application/octet-stream"]; + delete ctx.MediaPath; + delete ctx.MediaType; + } + + applyInlineQuoteContext(ctx, msg); + return ctx; } +function findReferenceItem(msg: WeixinMessage): MessageItem | undefined { + return msg.item_list?.find((item) => item.ref_msg); +} + +function inlineQuoteBody(item: MessageItem): string | undefined { + const ref = item.ref_msg; + if (!ref) return undefined; + const parts: string[] = []; + if (ref.title?.trim()) parts.push(ref.title.trim()); + if (ref.message_item) { + const body = bodyFromItemList([ref.message_item]); + if (body) parts.push(body); + } + return parts.length ? parts.join(" | ") : undefined; +} + +function applyInlineQuoteContext(ctx: WeixinMsgContext, msg: WeixinMessage): void { + const item = findReferenceItem(msg); + if (!item?.ref_msg) return; + const replyToId = item.ref_msg.svr_id?.trim() || item.ref_msg.message_item?.msg_id?.trim(); + const body = inlineQuoteBody(item); + ctx.ReplyToIsQuote = true; + if (replyToId) ctx.ReplyToId = replyToId; + if (body) ctx.ReplyToBody = body; +} + +function expiredMediaLabel(mime?: string, name?: string): string { + const kind = mime?.startsWith("image/") + ? "图片" + : mime?.startsWith("video/") + ? "视频" + : mime?.startsWith("audio/") + ? "语音" + : "附件"; + return name ? `[引用的${kind}已过期: ${name}]` : `[引用的${kind}已过期]`; +} + +/** Resolve an ID-only quote after sender authorization has succeeded. */ +export function resolveStoredQuoteContext( + ctx: WeixinMsgContext, + msg: WeixinMessage, + accountId: string, + store: Pick | null = getQuoteStore(), +): void { + const item = findReferenceItem(msg); + const ref = item?.ref_msg; + const referenceId = ref?.svr_id?.trim() || ref?.message_item?.msg_id?.trim(); + if (!item || !ref || !referenceId) return; + + ctx.ReplyToId = referenceId; + ctx.ReplyToIsQuote = true; + if (ctx.ReplyToBody && (ref.title?.trim() || ref.message_item)) return; + const record = store?.find(accountId, ctx.From, referenceId); + if (!record) { + if (!ctx.ReplyToBody) ctx.ReplyToBody = "[引用消息内容未缓存]"; + return; + } + + ctx.ReplyToBody = record.body || expiredMediaLabel(record.mediaMime, record.mediaName); + if (ref.partial_text && record.body) { + const partial = resolvePartialQuote(record.body, ref.partial_text); + if (!partial.fallback && partial.resolved) ctx.ReplyToQuoteText = partial.resolved; + } + + if (!record.mediaPath) { + if (record.mediaMime || record.mediaName) { + ctx.ReplyToBody = expiredMediaLabel(record.mediaMime, record.mediaName); + } + return; + } + if (!fs.existsSync(record.mediaPath)) { + ctx.ReplyToBody = expiredMediaLabel(record.mediaMime, record.mediaName); + return; + } + const currentPaths = ctx.MediaPaths ?? (ctx.MediaPath ? [ctx.MediaPath] : []); + const currentTypes = + ctx.MediaTypes ?? (ctx.MediaPath ? [ctx.MediaType ?? "application/octet-stream"] : []); + ctx.MediaPaths = [...currentPaths, record.mediaPath]; + ctx.MediaTypes = [...currentTypes, record.mediaMime ?? "application/octet-stream"]; + ctx.media = ctx.MediaPaths.map((mediaPath, index) => ({ + path: mediaPath, + contentType: ctx.MediaTypes?.[index] ?? "application/octet-stream", + ...(index === ctx.MediaPaths!.length - 1 && record.mediaName + ? { fileName: record.mediaName } + : {}), + ...(index === ctx.MediaPaths!.length - 1 ? { messageId: referenceId } : {}), + })); + const quotedAttachment = { + message_id: referenceId, + original_filename: record.mediaName ?? path.basename(record.mediaPath), + managed_source_path: record.mediaPath, + workspace_directory: "media/inbound/", + }; + ctx.ChannelPromptContext = [ + ...(ctx.ChannelPromptContext ?? []), + [ + "Quoted attachment tool access:", + JSON.stringify(quotedAttachment), + "The attachment is staged into the agent workspace under media/inbound/. " + + "If automatic extraction fails and the user asks about its contents, use the available " + + "file/PDF tools to locate it by original_filename and read it.", + ].join("\n"), + ]; +} + /** Extract the context_token from an inbound WeixinMsgContext. */ export function getContextTokenFromMsgContext(ctx: WeixinMsgContext): string | undefined { return ctx.context_token; diff --git a/src/messaging/partial-quote.test.ts b/src/messaging/partial-quote.test.ts new file mode 100644 index 0000000..2fbcd9f --- /dev/null +++ b/src/messaging/partial-quote.test.ts @@ -0,0 +1,58 @@ +import crypto from "node:crypto"; + +import { describe, expect, it } from "vitest"; + +import { resolvePartialQuote } from "./partial-quote.js"; + +function md5(value: string): string { + return crypto.createHash("md5").update(value, "utf8").digest("hex"); +} + +describe("resolvePartialQuote", () => { + it("resolves the protocol example using global occurrence indexes", () => { + expect( + resolvePartialQuote("abcedfabcgh", { + start: "a", + end: "c", + startindex: 1, + endindex: 1, + quotemd5: md5("abc"), + }), + ).toEqual({ resolved: "abc", fallback: false }); + }); + + it("uses the hash to support end indexes relative to the selected start", () => { + const full = "x-end start first-end second-end"; + const selected = "start first-end second-end"; + expect( + resolvePartialQuote(full, { + start: "start", + end: "end", + startindex: 0, + endindex: 1, + quotemd5: md5(selected), + }), + ).toEqual({ resolved: selected, fallback: false }); + }); + + it("falls back when indexes are invalid or the hash does not match", () => { + expect( + resolvePartialQuote("hello", { + start: "h", + end: "o", + startindex: -1, + endindex: 0, + quotemd5: "", + }), + ).toEqual({ resolved: null, fallback: true }); + expect( + resolvePartialQuote("hello", { + start: "h", + end: "o", + startindex: 0, + endindex: 0, + quotemd5: "not-the-md5", + }), + ).toEqual({ resolved: null, fallback: true }); + }); +}); diff --git a/src/messaging/partial-quote.ts b/src/messaging/partial-quote.ts new file mode 100644 index 0000000..f00a546 --- /dev/null +++ b/src/messaging/partial-quote.ts @@ -0,0 +1,62 @@ +import crypto from "node:crypto"; + +import type { PartialText } from "../api/types.js"; + +export type PartialQuoteResult = { + resolved: string | null; + fallback: boolean; +}; + +function nthIndexOf(text: string, value: string, occurrence: number, fromIndex = 0): number { + if (!value || !Number.isInteger(occurrence) || occurrence < 0) return -1; + let position = fromIndex; + for (let current = 0; current <= occurrence; current++) { + position = text.indexOf(value, position); + if (position < 0) return -1; + if (current < occurrence) position += value.length; + } + return position; +} + +function hashQuote(text: string): string { + return crypto.createHash("md5").update(text, "utf8").digest("hex"); +} + +function candidate( + fullText: string, + partial: PartialText, + endSearchMode: "global" | "relative", +): string | null { + const start = nthIndexOf(fullText, partial.start, partial.startindex); + if (start < 0) return null; + const end = + endSearchMode === "global" + ? nthIndexOf(fullText, partial.end, partial.endindex) + : nthIndexOf(fullText, partial.end, partial.endindex, start + partial.start.length); + if (end < start) return null; + return fullText.slice(start, end + partial.end.length); +} + +/** + * Resolve both observed interpretations of endindex. When quotemd5 is present + * it disambiguates the protocol variants; without a hash, global indexes match + * the examples supplied with the newer Weixin payload definition. + */ +export function resolvePartialQuote(fullText: string, partial: PartialText): PartialQuoteResult { + if (!fullText || !partial.start || !partial.end) { + return { resolved: null, fallback: true }; + } + const candidates = [ + candidate(fullText, partial, "global"), + candidate(fullText, partial, "relative"), + ].filter((value, index, all): value is string => Boolean(value) && all.indexOf(value) === index); + + if (!partial.quotemd5) { + return candidates[0] + ? { resolved: candidates[0], fallback: false } + : { resolved: null, fallback: true }; + } + const expected = partial.quotemd5.toLowerCase(); + const resolved = candidates.find((value) => hashQuote(value) === expected) ?? null; + return { resolved, fallback: resolved === null }; +} diff --git a/src/messaging/process-message.ts b/src/messaging/process-message.ts index 633a6e6..b4c4322 100644 --- a/src/messaging/process-message.ts +++ b/src/messaging/process-message.ts @@ -27,13 +27,16 @@ import { setContextToken, weixinMessageToMsgContext, getContextTokenFromMsgContext, + getWeixinMessageId, isMediaItem, + resolveStoredQuoteContext, } from "./inbound.js"; import type { WeixinInboundMediaOpts } from "./inbound.js"; import { sendWeixinMediaFile } from "./send-media.js"; import { StreamingMarkdownFilter } from "./markdown-filter.js"; import { sendMessageWeixin } from "./send.js"; import { WeixinReplyProgressSender } from "./reply-progress-sender.js"; +import { getActiveQuoteMediaSubdir, getQuoteStore } from "./quote-store.js"; import { handleSlashCommand } from "./slash-commands.js"; const MEDIA_OUTBOUND_TEMP_DIR = path.join( @@ -156,11 +159,13 @@ export async function processOneMessage( const downloaded = await downloadMediaFromItem(mediaItem, { cdnBaseUrl: deps.cdnBaseUrl, saveMedia: deps.channelRuntime.media.saveMediaBuffer, + mediaSubdir: getActiveQuoteMediaSubdir(deps.accountId), log: deps.log, errLog: deps.errLog, label, }); Object.assign(mediaOpts, downloaded); + if (refMediaItem) mediaOpts.referencedMedia = true; } const mediaDownloadMs = Date.now() - mediaDownloadStart; @@ -223,6 +228,30 @@ export async function processOneMessage( ); } + resolveStoredQuoteContext(ctx, full, deps.accountId); + const inboundMessageId = getWeixinMessageId(full); + if (inboundMessageId) { + try { + await getQuoteStore()?.put({ + accountId: deps.accountId, + conversationId: senderId, + messageId: inboundMessageId, + direction: "inbound", + body: ctx.Body, + ...(ctx.MediaPath ? { sourceMediaPath: ctx.MediaPath } : {}), + ...(ctx.MediaType ? { mediaMime: ctx.MediaType } : {}), + ...(mainMediaItem?.file_item?.file_name + ? { mediaName: mainMediaItem.file_item.file_name } + : {}), + createdAt: full.create_time_ms ?? Date.now(), + }); + } catch (err) { + logger.warn( + `quote cache: failed to save inbound message id=${inboundMessageId}: ${String(err)}`, + ); + } + } + const route = deps.channelRuntime.routing.resolveAgentRoute({ cfg: deps.config, channel: "openclaw-weixin", @@ -290,6 +319,7 @@ export async function processOneMessage( baseUrl: deps.baseUrl, token: deps.token, contextToken, + accountId: deps.accountId, }, }) : undefined; @@ -397,6 +427,7 @@ export async function processOneMessage( token: deps.token, contextToken, runId, + accountId: deps.accountId, }, }); emitWeixinMessageSent({ @@ -413,7 +444,13 @@ export async function processOneMessage( filePath, to: ctx.To, text, - opts: { baseUrl: deps.baseUrl, token: deps.token, contextToken, runId }, + opts: { + baseUrl: deps.baseUrl, + token: deps.token, + contextToken, + runId, + accountId: deps.accountId, + }, cdnBaseUrl: deps.cdnBaseUrl, }); emitWeixinMessageSent({ @@ -434,6 +471,7 @@ export async function processOneMessage( token: deps.token, contextToken, runId, + accountId: deps.accountId, }, }); emitWeixinMessageSent({ @@ -482,6 +520,7 @@ export async function processOneMessage( baseUrl: deps.baseUrl, token: deps.token, runId, + accountId: deps.accountId, errLog: deps.errLog, }); }, @@ -553,7 +592,13 @@ export async function processOneMessage( await sendMessageWeixin({ to: ctx.To, text: timingText, - opts: { baseUrl: deps.baseUrl, token: deps.token, contextToken, runId }, + opts: { + baseUrl: deps.baseUrl, + token: deps.token, + contextToken, + runId, + accountId: deps.accountId, + }, }); logger.info(`debug-timing: sent OK`); } catch (debugErr) { diff --git a/src/messaging/quote-store.test.ts b/src/messaging/quote-store.test.ts new file mode 100644 index 0000000..432aef9 --- /dev/null +++ b/src/messaging/quote-store.test.ts @@ -0,0 +1,616 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import type { QuoteCachePolicy } from "./quote-store.js"; +import { + QuoteStore, + closeQuoteStore, + deactivateQuoteStoreAccount, + deleteQuoteCacheForAccount, + getActiveQuoteMediaSubdir, + getQuoteStore, + initializeQuoteStore, + resolveQuoteMediaSubdir, + resolveQuoteCachePolicy, +} from "./quote-store.js"; + +vi.mock("../util/logger.js", () => ({ + logger: { + info: vi.fn(), + debug: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }, +})); + +let rootDir: string; +let mediaRoot: string; +let stores: QuoteStore[]; + +function policy(overrides: Partial = {}): QuoteCachePolicy { + return { + enabled: true, + retentionMs: 30 * 24 * 60 * 60 * 1000, + maxMessagesPerAccount: 10_000, + mediaRetentionMs: 7 * 24 * 60 * 60 * 1000, + maxMediaBytesPerAccount: 256 * 1024 * 1024, + maxSingleMediaBytes: 25 * 1024 * 1024, + ...overrides, + }; +} + +async function open(overrides: Partial = {}): Promise { + const store = await QuoteStore.open({ rootDir, mediaRoot, policy: policy(overrides) }); + expect(store).not.toBeNull(); + stores.push(store!); + return store!; +} + +function managedMediaPath(accountId: string, fileName: string): string { + const accountDir = path.join(mediaRoot, path.basename(resolveQuoteMediaSubdir(accountId))); + fs.mkdirSync(accountDir, { recursive: true }); + return path.join(accountDir, fileName); +} + +beforeEach(() => { + rootDir = fs.mkdtempSync(path.join(os.tmpdir(), "weixin-quote-store-")); + mediaRoot = path.join(rootDir, "managed-media"); + stores = []; +}); + +afterEach(() => { + closeQuoteStore(); + delete process.env.OPENCLAW_STATE_DIR; + for (const store of stores) store.close(); + fs.rmSync(rootDir, { recursive: true, force: true }); +}); + +describe("QuoteStore", () => { + it("persists lossless IDs and scopes records by account and conversation", async () => { + const store = await open(); + const createdAt = Date.now(); + await store.put({ + accountId: "account-a", + conversationId: "user-a", + messageId: "18446744073709551615", + direction: "inbound", + body: "hello", + createdAt, + }); + + expect(store.find("account-a", "user-a", "18446744073709551615")?.body).toBe("hello"); + expect(store.find("account-b", "user-a", "18446744073709551615")).toBeNull(); + expect(store.find("account-a", "user-b", "18446744073709551615")).toBeNull(); + + store.close(); + const reopened = await open(); + expect(reopened.find("account-a", "user-a", "18446744073709551615")).toMatchObject({ + messageId: "18446744073709551615", + body: "hello", + createdAt, + }); + }); + + it("evicts expired and over-count messages during GC", async () => { + const now = Date.now(); + const store = await open({ retentionMs: 100, maxMessagesPerAccount: 2 }); + for (let index = 0; index < 3; index++) { + await store.put({ + accountId: "account", + conversationId: "user", + messageId: String(index), + direction: "inbound", + body: `message-${index}`, + createdAt: now + index, + }); + } + store.runGc(now + 3); + expect(store.find("account", "user", "0")).toBeNull(); + expect(store.find("account", "user", "1")?.body).toBe("message-1"); + expect(store.find("account", "user", "2")?.body).toBe("message-2"); + + store.runGc(now + 1000); + expect(store.find("account", "user", "1")).toBeNull(); + expect(store.find("account", "user", "2")).toBeNull(); + }); + + it("registers one managed media copy and deletes it with the account", async () => { + const store = await open(); + const source = managedMediaPath("account", "source.png"); + fs.writeFileSync(source, "same-content"); + for (const messageId of ["1", "2"]) { + await store.put({ + accountId: "account", + conversationId: "user", + messageId, + direction: "inbound", + body: "[图片]", + sourceMediaPath: source, + mediaMime: "image/png", + createdAt: Date.now(), + }); + } + const first = store.find("account", "user", "1"); + const second = store.find("account", "user", "2"); + expect(first?.mediaPath).toBe(second?.mediaPath); + expect(first?.mediaPath).toBe(source); + expect(fs.readFileSync(first!.mediaPath!, "utf8")).toBe("same-content"); + + const managedPath = first!.mediaPath!; + store.deleteAccount("account"); + expect(store.find("account", "user", "1")).toBeNull(); + expect(fs.existsSync(managedPath)).toBe(false); + }); + + it("does not let sanitized account names share a media directory", async () => { + const store = await open(); + for (const accountId of ["a/b", "a_b"]) { + const source = managedMediaPath(accountId, "source.png"); + fs.writeFileSync(source, "same-content"); + await store.put({ + accountId, + conversationId: "user", + messageId: "1", + direction: "inbound", + body: "[图片]", + sourceMediaPath: source, + createdAt: Date.now(), + }); + } + const firstPath = store.find("a/b", "user", "1")!.mediaPath!; + const secondPath = store.find("a_b", "user", "1")!.mediaPath!; + expect(path.dirname(firstPath)).not.toBe(path.dirname(secondPath)); + store.deleteAccount("a/b"); + expect(fs.existsSync(firstPath)).toBe(false); + expect(fs.existsSync(secondPath)).toBe(true); + }); + + it("keeps message metadata but skips oversized media", async () => { + const store = await open({ maxSingleMediaBytes: 4 }); + const source = managedMediaPath("account", "large.bin"); + fs.writeFileSync(source, "12345"); + await store.put({ + accountId: "account", + conversationId: "user", + messageId: "large", + direction: "inbound", + body: "[文件]", + sourceMediaPath: source, + mediaMime: "application/octet-stream", + mediaName: "large.bin", + createdAt: Date.now(), + }); + expect(store.find("account", "user", "large")).toMatchObject({ + body: "[文件]", + mediaMime: "application/octet-stream", + mediaName: "large.bin", + }); + expect(store.find("account", "user", "large")?.mediaPath).toBeUndefined(); + }); + + it("enforces the per-account media byte budget oldest-first", async () => { + const store = await open({ maxMediaBytesPerAccount: 4 }); + const firstSource = managedMediaPath("account", "first.bin"); + const secondSource = managedMediaPath("account", "second.bin"); + fs.writeFileSync(firstSource, "1111"); + fs.writeFileSync(secondSource, "2222"); + const now = Date.now(); + await store.put({ + accountId: "account", + conversationId: "user", + messageId: "1", + direction: "inbound", + body: "[文件]", + sourceMediaPath: firstSource, + createdAt: now, + }); + await store.put({ + accountId: "account", + conversationId: "user", + messageId: "2", + direction: "inbound", + body: "[文件]", + sourceMediaPath: secondSource, + createdAt: now + 1, + }); + expect(store.find("account", "user", "1")?.mediaPath).toBeUndefined(); + expect(store.find("account", "user", "2")?.mediaPath).toBeDefined(); + }); + + it("expires managed media without discarding the message body", async () => { + const store = await open({ mediaRetentionMs: 10 }); + const source = managedMediaPath("account", "old.mp3"); + fs.writeFileSync(source, "voice"); + const now = Date.now(); + await store.put({ + accountId: "account", + conversationId: "user", + messageId: "old-media", + direction: "inbound", + body: "[语音]", + sourceMediaPath: source, + mediaMime: "audio/mpeg", + createdAt: now, + }); + const managedPath = store.find("account", "user", "old-media")!.mediaPath!; + store.runGc(now + 20); + expect(store.find("account", "user", "old-media")).toMatchObject({ + body: "[语音]", + mediaMime: "audio/mpeg", + }); + expect(store.find("account", "user", "old-media")?.mediaPath).toBeUndefined(); + expect(fs.existsSync(managedPath)).toBe(false); + }); + + it("removes an expired message but preserves media shared by a newer record", async () => { + const store = await open({ retentionMs: 100 }); + const source = managedMediaPath("account", "shared.png"); + fs.writeFileSync(source, "shared"); + const now = Date.now(); + await store.put({ + accountId: "account", + conversationId: "user", + messageId: "old", + direction: "inbound", + body: "old", + sourceMediaPath: source, + createdAt: now - 1000, + }); + await store.put({ + accountId: "account", + conversationId: "user", + messageId: "new", + direction: "inbound", + body: "new", + sourceMediaPath: source, + createdAt: now, + }); + const managedPath = store.find("account", "user", "new")!.mediaPath!; + expect(store.find("account", "user", "old")).toBeNull(); + expect(fs.existsSync(managedPath)).toBe(true); + }); + + it("prunes orphan files and tolerates a missing media source", async () => { + const store = await open(); + await store.put({ + accountId: "account", + conversationId: "user", + messageId: "missing", + direction: "inbound", + body: "[文件]", + sourceMediaPath: path.join(rootDir, "does-not-exist.bin"), + createdAt: Date.now(), + }); + expect(store.find("account", "user", "missing")?.mediaPath).toBeUndefined(); + + const orphanDir = path.join(mediaRoot, "nested"); + fs.mkdirSync(orphanDir, { recursive: true }); + const orphan = path.join(orphanDir, "orphan.bin"); + fs.writeFileSync(orphan, "orphan"); + store.runGc(); + expect(fs.existsSync(orphan)).toBe(false); + }); + + it("applies policy updates asynchronously", async () => { + const store = await open(); + const now = Date.now(); + for (const messageId of ["1", "2"]) { + await store.put({ + accountId: "account", + conversationId: "user", + messageId, + direction: "inbound", + body: messageId, + createdAt: now, + }); + } + store.updatePolicy(policy({ maxMessagesPerAccount: 1 })); + await new Promise((resolve) => queueMicrotask(resolve)); + const remaining = ["1", "2"].filter((id) => store.find("account", "user", id)); + expect(remaining).toHaveLength(1); + }); + + it("does not open when explicitly disabled", async () => { + await expect( + QuoteStore.open({ rootDir, policy: policy({ enabled: false }) }), + ).resolves.toBeNull(); + }); + + it("returns null when the database directory cannot be created", async () => { + const fileInsteadOfDirectory = path.join(rootDir, "plain-file"); + fs.writeFileSync(fileInsteadOfDirectory, "not a directory"); + await expect( + QuoteStore.open({ + rootDir: path.join(fileInsteadOfDirectory, "child"), + policy: policy(), + }), + ).resolves.toBeNull(); + }); + + it("ignores invalid writes, isolates unsafe account names, and substitutes invalid timestamps", async () => { + const store = await open(); + const source = managedMediaPath(" ", "media"); + fs.writeFileSync(source, "content"); + await store.put({ + accountId: " ", + conversationId: "user", + messageId: "media", + direction: "inbound", + body: "[文件]", + sourceMediaPath: source, + createdAt: Number.NaN, + }); + const record = store.find(" ", "user", "media"); + expect(record?.createdAt).toBeGreaterThan(0); + expect(record?.mediaPath).toContain(`${path.sep}default-`); + expect(path.extname(record!.mediaPath!)).toBe(""); + + await store.put({ + accountId: "account", + conversationId: "user", + messageId: "", + direction: "inbound", + body: "ignored", + createdAt: Date.now(), + }); + await store.put({ + accountId: "account", + conversationId: "user", + messageId: "empty", + direction: "inbound", + body: "", + createdAt: Date.now(), + }); + expect(store.find("account", "user", "")).toBeNull(); + expect(store.find("account", "user", "empty")).toBeNull(); + + store.close(); + await store.put({ + accountId: "account", + conversationId: "user", + messageId: "closed", + direction: "inbound", + body: "ignored", + createdAt: Date.now(), + }); + expect(store.find("account", "user", "closed")).toBeNull(); + store.deleteAccount("account"); + store.runGc(); + }); + + it("refuses to claim media outside its account-owned directory", async () => { + const source = path.join(rootDir, "outside.bin"); + fs.writeFileSync(source, "content"); + const store = await open(); + await store.put({ + accountId: "account", + conversationId: "user", + messageId: "outside", + direction: "inbound", + body: "[文件]", + sourceMediaPath: source, + createdAt: Date.now(), + }); + expect(store.find("account", "user", "outside")?.mediaPath).toBeUndefined(); + expect(fs.existsSync(source)).toBe(true); + }); + + it("moves legacy media into the OpenClaw-managed cache and rewrites its database path", async () => { + const legacyRoot = path.join(rootDir, "ref-media"); + const legacyStore = await QuoteStore.open({ + rootDir, + mediaRoot: legacyRoot, + policy: policy(), + }); + expect(legacyStore).not.toBeNull(); + stores.push(legacyStore!); + const legacyPath = path.join( + legacyRoot, + path.basename(resolveQuoteMediaSubdir("account")), + "legacy.pdf", + ); + fs.mkdirSync(path.dirname(legacyPath), { recursive: true }); + fs.writeFileSync(legacyPath, "pdf"); + await legacyStore!.put({ + accountId: "account", + conversationId: "user", + messageId: "legacy", + direction: "inbound", + body: "[文件]", + sourceMediaPath: legacyPath, + mediaName: "report.pdf", + createdAt: Date.now(), + }); + await legacyStore!.put({ + accountId: "account", + conversationId: "user", + messageId: "legacy-shared", + direction: "inbound", + body: "[文件]", + sourceMediaPath: legacyPath, + mediaName: "report.pdf", + createdAt: Date.now(), + }); + + const existingLegacyPath = path.join(path.dirname(legacyPath), "existing.pdf"); + fs.writeFileSync(existingLegacyPath, "existing"); + await legacyStore!.put({ + accountId: "account", + conversationId: "user", + messageId: "existing", + direction: "inbound", + body: "[文件]", + sourceMediaPath: existingLegacyPath, + createdAt: Date.now(), + }); + const existingManagedPath = managedMediaPath("account", "existing.pdf"); + fs.writeFileSync(existingManagedPath, "existing"); + + const missingLegacyPath = path.join(path.dirname(legacyPath), "missing.pdf"); + fs.writeFileSync(missingLegacyPath, "missing"); + await legacyStore!.put({ + accountId: "account", + conversationId: "user", + messageId: "missing-legacy", + direction: "inbound", + body: "[文件]", + sourceMediaPath: missingLegacyPath, + createdAt: Date.now(), + }); + fs.unlinkSync(missingLegacyPath); + legacyStore!.close(); + + const migratedStore = await open(); + const migrated = migratedStore.find("account", "user", "legacy"); + expect(migrated?.mediaPath).toBe(managedMediaPath("account", "legacy.pdf")); + expect(migrated?.mediaName).toBe("report.pdf"); + expect(fs.existsSync(migrated!.mediaPath!)).toBe(true); + expect(fs.existsSync(legacyPath)).toBe(false); + expect(migratedStore.find("account", "user", "legacy-shared")?.mediaPath).toBe( + migrated?.mediaPath, + ); + expect(migratedStore.find("account", "user", "existing")?.mediaPath).toBe(existingManagedPath); + expect(fs.existsSync(existingLegacyPath)).toBe(false); + }); +}); + +describe("global quote store lifecycle", () => { + it("shares one store across active accounts and closes after the last account stops", async () => { + process.env.OPENCLAW_STATE_DIR = rootDir; + expect(getActiveQuoteMediaSubdir("account-a")).toBeUndefined(); + const first = await initializeQuoteStore({}, "account-a"); + const second = await initializeQuoteStore({}, "account-b"); + expect(first).toBe(second); + expect(getQuoteStore()).toBe(first); + expect(getActiveQuoteMediaSubdir("account-a")).toBe(resolveQuoteMediaSubdir("account-a")); + + deactivateQuoteStoreAccount("account-a"); + expect(getQuoteStore()).toBe(first); + deactivateQuoteStoreAccount("account-b"); + expect(getQuoteStore()).toBeNull(); + }); + + it("honors explicit disablement", async () => { + process.env.OPENCLAW_STATE_DIR = rootDir; + await expect( + initializeQuoteStore( + { + channels: { "openclaw-weixin": { quoteCache: { enabled: false } } }, + }, + "account", + ), + ).resolves.toBeNull(); + expect(getQuoteStore()).toBeNull(); + }); + + it("closes an already-active store when the feature is disabled", async () => { + process.env.OPENCLAW_STATE_DIR = rootDir; + expect(await initializeQuoteStore({}, "account")).not.toBeNull(); + await expect( + initializeQuoteStore( + { + channels: { "openclaw-weixin": { quoteCache: { enabled: false } } }, + }, + "account", + ), + ).resolves.toBeNull(); + expect(getQuoteStore()).toBeNull(); + }); + + it("deletes an account through the active singleton", async () => { + process.env.OPENCLAW_STATE_DIR = rootDir; + const store = await initializeQuoteStore({}, "account"); + await store!.put({ + accountId: "account", + conversationId: "user", + messageId: "1", + direction: "inbound", + body: "hello", + createdAt: Date.now(), + }); + deleteQuoteCacheForAccount("account"); + expect(store!.find("account", "user", "1")).toBeNull(); + }); + + it("deletes an account from a closed on-disk database", async () => { + process.env.OPENCLAW_STATE_DIR = rootDir; + const managedRoot = path.join(rootDir, "openclaw-weixin"); + const store = await QuoteStore.open({ rootDir: managedRoot, policy: policy() }); + expect(store).not.toBeNull(); + await store!.put({ + accountId: "account", + conversationId: "user", + messageId: "1", + direction: "inbound", + body: "hello", + createdAt: Date.now(), + }); + store!.close(); + + deleteQuoteCacheForAccount("account"); + const reopened = await QuoteStore.open({ rootDir: managedRoot, policy: policy() }); + expect(reopened?.find("account", "user", "1")).toBeNull(); + reopened?.close(); + }); +}); + +describe("resolveQuoteCachePolicy", () => { + it("uses safe defaults and honors explicit disablement", () => { + expect(resolveQuoteCachePolicy({})).toMatchObject({ + enabled: true, + maxMessagesPerAccount: 10_000, + maxMediaBytesPerAccount: 256 * 1024 * 1024, + }); + expect( + resolveQuoteCachePolicy({ + channels: { "openclaw-weixin": { quoteCache: { enabled: false } } }, + }), + ).toMatchObject({ enabled: false }); + }); + + it("accepts positive limits and replaces invalid values with defaults", () => { + expect( + resolveQuoteCachePolicy({ + channels: { + "openclaw-weixin": { + quoteCache: { + retentionDays: 2, + maxMessagesPerAccount: 3.9, + mediaRetentionDays: 4, + maxMediaBytesPerAccount: 5, + maxSingleMediaBytes: 6, + }, + }, + }, + }), + ).toMatchObject({ + retentionMs: 2 * 24 * 60 * 60 * 1000, + maxMessagesPerAccount: 3, + mediaRetentionMs: 4 * 24 * 60 * 60 * 1000, + maxMediaBytesPerAccount: 5, + maxSingleMediaBytes: 6, + }); + expect( + resolveQuoteCachePolicy({ + channels: { + "openclaw-weixin": { + quoteCache: { + retentionDays: Number.NaN, + maxMessagesPerAccount: 0, + mediaRetentionDays: -1, + maxMediaBytesPerAccount: Number.POSITIVE_INFINITY, + maxSingleMediaBytes: 0, + }, + }, + }, + }), + ).toMatchObject({ + maxMessagesPerAccount: 10_000, + maxMediaBytesPerAccount: 256 * 1024 * 1024, + maxSingleMediaBytes: 25 * 1024 * 1024, + }); + }); +}); diff --git a/src/messaging/quote-store.ts b/src/messaging/quote-store.ts new file mode 100644 index 0000000..d0e3f74 --- /dev/null +++ b/src/messaging/quote-store.ts @@ -0,0 +1,656 @@ +import crypto from "node:crypto"; +import fs from "node:fs"; +import { createRequire } from "node:module"; +import path from "node:path"; + +import type { OpenClawConfig } from "openclaw/plugin-sdk/core"; + +import { resolveStateDir } from "../storage/state-dir.js"; +import { logger } from "../util/logger.js"; + +const DEFAULT_RETENTION_DAYS = 30; +const DEFAULT_MAX_MESSAGES_PER_ACCOUNT = 10_000; +const DEFAULT_MEDIA_RETENTION_DAYS = 7; +const DEFAULT_MAX_MEDIA_BYTES_PER_ACCOUNT = 256 * 1024 * 1024; +const DEFAULT_MAX_SINGLE_MEDIA_BYTES = 25 * 1024 * 1024; +const GC_WRITE_INTERVAL = 100; +const GC_TIMER_MS = 60 * 60 * 1000; +const QUOTE_MEDIA_SUBDIR = "inbound/openclaw-weixin-quotes"; + +type SqliteStatement = { + run: (...params: unknown[]) => unknown; + get: (...params: unknown[]) => unknown; + all: (...params: unknown[]) => unknown[]; +}; + +type SqliteDatabase = { + exec: (sql: string) => void; + prepare: (sql: string) => SqliteStatement; + close: () => void; +}; + +type DatabaseSyncConstructor = new (filePath: string) => SqliteDatabase; + +export type QuoteCachePolicy = { + enabled: boolean; + retentionMs: number; + maxMessagesPerAccount: number; + mediaRetentionMs: number; + maxMediaBytesPerAccount: number; + maxSingleMediaBytes: number; +}; + +export type QuoteMessageRecord = { + accountId: string; + conversationId: string; + messageId: string; + direction: "inbound" | "outbound"; + body: string; + mediaPath?: string; + mediaMime?: string; + mediaName?: string; + mediaSize?: number; + createdAt: number; +}; + +export type QuoteMessageInput = Omit & { + sourceMediaPath?: string; +}; + +type QuoteCacheConfig = { + enabled?: boolean; + retentionDays?: number; + maxMessagesPerAccount?: number; + mediaRetentionDays?: number; + maxMediaBytesPerAccount?: number; + maxSingleMediaBytes?: number; +}; + +type WeixinChannelConfig = { + quoteCache?: QuoteCacheConfig; +}; + +type StoredRow = { + account_id: string; + conversation_id: string; + message_id: string; + direction: "inbound" | "outbound"; + body: string; + media_path: string | null; + media_mime: string | null; + media_name: string | null; + media_size: number | null; + created_at: number; +}; + +type MediaRow = { + media_path: string; + media_size: number; + newest_at: number; +}; + +function positiveNumber(value: unknown, fallback: number): number { + return typeof value === "number" && Number.isFinite(value) && value > 0 ? value : fallback; +} + +function positiveInteger(value: unknown, fallback: number): number { + return Math.max(1, Math.floor(positiveNumber(value, fallback))); +} + +export function resolveQuoteCachePolicy(cfg: OpenClawConfig): QuoteCachePolicy { + const section = cfg.channels?.["openclaw-weixin"] as WeixinChannelConfig | undefined; + const quote = section?.quoteCache; + return { + enabled: quote?.enabled !== false, + retentionMs: positiveNumber(quote?.retentionDays, DEFAULT_RETENTION_DAYS) * 24 * 60 * 60 * 1000, + maxMessagesPerAccount: positiveInteger( + quote?.maxMessagesPerAccount, + DEFAULT_MAX_MESSAGES_PER_ACCOUNT, + ), + mediaRetentionMs: + positiveNumber(quote?.mediaRetentionDays, DEFAULT_MEDIA_RETENTION_DAYS) * 24 * 60 * 60 * 1000, + maxMediaBytesPerAccount: positiveInteger( + quote?.maxMediaBytesPerAccount, + DEFAULT_MAX_MEDIA_BYTES_PER_ACCOUNT, + ), + maxSingleMediaBytes: positiveInteger( + quote?.maxSingleMediaBytes, + DEFAULT_MAX_SINGLE_MEDIA_BYTES, + ), + }; +} + +function safePathSegment(raw: string): string { + const safe = raw + .trim() + .toLowerCase() + .replace(/[^a-z0-9._-]+/g, "_") + .replace(/\.\.+/g, "_"); + return safe && safe !== "." && safe !== ".." ? safe : "default"; +} + +function accountMediaDirName(accountId: string): string { + const readable = safePathSegment(accountId).slice(0, 48); + const digest = crypto.createHash("sha256").update(accountId, "utf8").digest("hex").slice(0, 16); + return `${readable}-${digest}`; +} + +/** OpenClaw media-store subdirectory owned and garbage-collected by this plugin. */ +export function resolveQuoteMediaSubdir(accountId: string): string { + return path.posix.join(QUOTE_MEDIA_SUBDIR, accountMediaDirName(accountId)); +} + +function resolveQuoteMediaRoot(): string { + return path.join(resolveStateDir(), "media", ...QUOTE_MEDIA_SUBDIR.split("/")); +} + +function asStoredRow(value: unknown): StoredRow | null { + if (!value || typeof value !== "object") return null; + return value as StoredRow; +} + +export class QuoteStore { + private readonly db: SqliteDatabase; + private readonly rootDir: string; + private readonly mediaRoot: string; + private policy: QuoteCachePolicy; + private writesSinceGc = 0; + private gcRunning = false; + private gcRequested = false; + private gcScheduled = false; + private closed = false; + private readonly gcTimer: ReturnType; + + private constructor( + db: SqliteDatabase, + rootDir: string, + mediaRoot: string, + policy: QuoteCachePolicy, + ) { + this.db = db; + this.rootDir = rootDir; + this.mediaRoot = mediaRoot; + this.policy = policy; + this.initializeSchema(); + this.migrateLegacyMedia(); + this.gcTimer = setInterval(() => this.requestGc(), GC_TIMER_MS); + this.gcTimer.unref(); + this.runGc(); + } + + static async open(params: { + policy: QuoteCachePolicy; + rootDir?: string; + mediaRoot?: string; + }): Promise { + if (!params.policy.enabled) return null; + try { + const sqlite = (await import("node:sqlite")) as unknown as { + DatabaseSync: DatabaseSyncConstructor; + }; + const rootDir = params.rootDir ?? path.join(resolveStateDir(), "openclaw-weixin"); + const mediaRoot = params.mediaRoot ?? resolveQuoteMediaRoot(); + fs.mkdirSync(rootDir, { recursive: true, mode: 0o700 }); + fs.mkdirSync(mediaRoot, { recursive: true, mode: 0o700 }); + try { + fs.chmodSync(rootDir, 0o700); + } catch { + // best-effort on filesystems without POSIX permissions + } + const dbPath = path.join(rootDir, "ref-messages.sqlite"); + const db = new sqlite.DatabaseSync(dbPath); + try { + fs.chmodSync(dbPath, 0o600); + } catch { + // best-effort + } + return new QuoteStore(db, rootDir, mediaRoot, params.policy); + } catch (err) { + logger.warn( + `quote cache disabled: node:sqlite is unavailable or the database could not be opened: ${String(err)}`, + ); + return null; + } + } + + updatePolicy(policy: QuoteCachePolicy): void { + this.policy = policy; + this.requestGc(); + } + + private initializeSchema(): void { + this.db.exec(` + PRAGMA journal_mode = WAL; + PRAGMA synchronous = NORMAL; + PRAGMA busy_timeout = 5000; + CREATE TABLE IF NOT EXISTS quote_messages ( + account_id TEXT NOT NULL, + conversation_id TEXT NOT NULL, + message_id TEXT NOT NULL, + direction TEXT NOT NULL, + body TEXT NOT NULL DEFAULT '', + media_path TEXT, + media_mime TEXT, + media_name TEXT, + media_size INTEGER, + created_at INTEGER NOT NULL, + PRIMARY KEY (account_id, conversation_id, message_id) + ); + CREATE INDEX IF NOT EXISTS quote_messages_account_created + ON quote_messages (account_id, created_at); + CREATE INDEX IF NOT EXISTS quote_messages_media_path + ON quote_messages (media_path); + `); + } + + /** Move media written by pre-managed-root builds without duplicating file contents. */ + private migrateLegacyMedia(): void { + const legacyRoots = [ + path.join(this.rootDir, "ref-media"), + // A short-lived local build accidentally omitted the media/ segment. + path.join(path.dirname(this.rootDir), ...QUOTE_MEDIA_SUBDIR.split("/")), + ]; + for (const legacyRoot of new Set(legacyRoots)) { + this.migrateLegacyMediaRoot(legacyRoot); + } + } + + private migrateLegacyMediaRoot(legacyRoot: string): void { + if (path.resolve(legacyRoot) === path.resolve(this.mediaRoot) || !fs.existsSync(legacyRoot)) { + return; + } + const rows = this.db + .prepare(` + SELECT account_id, media_path + FROM quote_messages + WHERE media_path IS NOT NULL + `) + .all() as Array<{ account_id: string; media_path: string }>; + const canonicalLegacyRoot = fs.realpathSync(legacyRoot); + const migrated = new Map(); + for (const row of rows) { + if (migrated.has(row.media_path)) continue; + if (!fs.existsSync(row.media_path)) continue; + const canonicalLegacyPath = fs.realpathSync(row.media_path); + const relative = path.relative(canonicalLegacyRoot, canonicalLegacyPath); + if (!relative || relative.startsWith("..") || path.isAbsolute(relative)) continue; + const destinationDir = path.join(this.mediaRoot, accountMediaDirName(row.account_id)); + const destination = path.join(destinationDir, path.basename(row.media_path)); + try { + fs.mkdirSync(destinationDir, { recursive: true, mode: 0o700 }); + if (!fs.existsSync(destination)) { + try { + fs.renameSync(row.media_path, destination); + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== "EXDEV") throw err; + fs.copyFileSync(row.media_path, destination, fs.constants.COPYFILE_EXCL); + fs.unlinkSync(row.media_path); + } + } else { + fs.unlinkSync(row.media_path); + } + migrated.set(row.media_path, destination); + } catch (err) { + logger.warn(`quote cache: failed to migrate media path=${row.media_path}: ${String(err)}`); + } + } + for (const [legacyPath, managedPath] of migrated) { + this.db + .prepare("UPDATE quote_messages SET media_path = ? WHERE media_path = ?") + .run(managedPath, legacyPath); + } + } + + find(accountId: string, conversationId: string, messageId: string): QuoteMessageRecord | null { + if (this.closed || !messageId) return null; + const row = asStoredRow( + this.db + .prepare(` + SELECT account_id, conversation_id, message_id, direction, body, + media_path, media_mime, media_name, media_size, created_at + FROM quote_messages + WHERE account_id = ? AND conversation_id = ? AND message_id = ? + `) + .get(accountId, conversationId, messageId), + ); + if (!row) return null; + if (Date.now() - row.created_at > this.policy.retentionMs) { + this.deleteMessage(accountId, conversationId, messageId); + return null; + } + return { + accountId: row.account_id, + conversationId: row.conversation_id, + messageId: row.message_id, + direction: row.direction, + body: row.body, + ...(row.media_path ? { mediaPath: row.media_path } : {}), + ...(row.media_mime ? { mediaMime: row.media_mime } : {}), + ...(row.media_name ? { mediaName: row.media_name } : {}), + ...(typeof row.media_size === "number" ? { mediaSize: row.media_size } : {}), + createdAt: row.created_at, + }; + } + + async put(input: QuoteMessageInput): Promise { + if (this.closed || !input.messageId || (!input.body && !input.sourceMediaPath)) return; + const media = input.sourceMediaPath + ? await this.registerManagedMedia(input.accountId, input.sourceMediaPath, input.mediaMime) + : null; + this.db + .prepare(` + INSERT INTO quote_messages ( + account_id, conversation_id, message_id, direction, body, + media_path, media_mime, media_name, media_size, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(account_id, conversation_id, message_id) DO UPDATE SET + direction = excluded.direction, + body = excluded.body, + media_path = COALESCE(excluded.media_path, quote_messages.media_path), + media_mime = COALESCE(excluded.media_mime, quote_messages.media_mime), + media_name = COALESCE(excluded.media_name, quote_messages.media_name), + media_size = COALESCE(excluded.media_size, quote_messages.media_size), + created_at = excluded.created_at + `) + .run( + input.accountId, + input.conversationId, + input.messageId, + input.direction, + input.body, + media?.path ?? null, + input.mediaMime ?? media?.mime ?? null, + input.mediaName ?? media?.name ?? null, + media?.size ?? null, + Number.isFinite(input.createdAt) && input.createdAt > 0 ? input.createdAt : Date.now(), + ); + + this.writesSinceGc += 1; + if (media) { + this.enforceMediaBudget(input.accountId); + } + if (this.writesSinceGc >= GC_WRITE_INTERVAL) { + this.writesSinceGc = 0; + this.requestGc(); + } + } + + deleteAccount(accountId: string): void { + if (this.closed) return; + this.db.prepare("DELETE FROM quote_messages WHERE account_id = ?").run(accountId); + const accountMediaDir = path.join(this.mediaRoot, accountMediaDirName(accountId)); + try { + fs.rmSync(accountMediaDir, { recursive: true, force: true }); + } catch (err) { + logger.warn(`quote cache: failed to remove media for account=${accountId}: ${String(err)}`); + } + } + + private deleteMessage(accountId: string, conversationId: string, messageId: string): void { + const existing = this.findMediaPath(accountId, conversationId, messageId); + this.db + .prepare( + "DELETE FROM quote_messages WHERE account_id = ? AND conversation_id = ? AND message_id = ?", + ) + .run(accountId, conversationId, messageId); + if (existing) this.removeMediaIfOrphaned(existing); + } + + private findMediaPath( + accountId: string, + conversationId: string, + messageId: string, + ): string | null { + const row = this.db + .prepare( + "SELECT media_path FROM quote_messages WHERE account_id = ? AND conversation_id = ? AND message_id = ?", + ) + .get(accountId, conversationId, messageId) as { media_path?: unknown } | undefined; + return typeof row?.media_path === "string" ? row.media_path : null; + } + + private async registerManagedMedia( + accountId: string, + sourcePath: string, + mime?: string, + ): Promise<{ path: string; mime?: string; name: string; size: number } | null> { + try { + const accountDir = path.join(this.mediaRoot, accountMediaDirName(accountId)); + const canonicalRoot = await fs.promises + .realpath(accountDir) + .catch(() => path.resolve(accountDir)); + const canonicalSource = await fs.promises.realpath(sourcePath); + const relative = path.relative(canonicalRoot, canonicalSource); + if (!relative || relative.startsWith("..") || path.isAbsolute(relative)) { + logger.warn(`quote cache: refusing unmanaged media path=${sourcePath}`); + return null; + } + const stat = await fs.promises.stat(canonicalSource); + if (!stat.isFile() || stat.size > this.policy.maxSingleMediaBytes) return null; + return { + path: path.resolve(sourcePath), + ...(mime ? { mime } : {}), + name: path.basename(canonicalSource), + size: stat.size, + }; + } catch (err) { + logger.warn(`quote cache: failed to register media path=${sourcePath}: ${String(err)}`); + return null; + } + } + + private requestGc(): void { + if (this.closed) return; + if (this.gcRunning) { + this.gcRequested = true; + return; + } + if (this.gcScheduled) return; + this.gcScheduled = true; + queueMicrotask(() => { + this.gcScheduled = false; + this.runGc(); + }); + } + + runGc(now = Date.now()): void { + if (this.closed || this.gcRunning) { + this.gcRequested = true; + return; + } + this.gcRunning = true; + try { + const messageCutoff = now - this.policy.retentionMs; + this.db.prepare("DELETE FROM quote_messages WHERE created_at < ?").run(messageCutoff); + + const accounts = this.db + .prepare("SELECT DISTINCT account_id FROM quote_messages") + .all() as Array<{ account_id: string }>; + for (const { account_id: accountId } of accounts) { + this.db + .prepare(` + DELETE FROM quote_messages + WHERE rowid IN ( + SELECT rowid FROM quote_messages + WHERE account_id = ? + ORDER BY created_at DESC, rowid DESC + LIMIT -1 OFFSET ? + ) + `) + .run(accountId, this.policy.maxMessagesPerAccount); + this.expireOldMedia(accountId, now - this.policy.mediaRetentionMs); + this.enforceMediaBudget(accountId); + } + this.pruneOrphanMediaFiles(); + } catch (err) { + logger.warn(`quote cache GC failed: ${String(err)}`); + } finally { + this.gcRunning = false; + if (this.gcRequested) { + this.gcRequested = false; + this.requestGc(); + } + } + } + + private listMedia(accountId: string): MediaRow[] { + return this.db + .prepare(` + SELECT media_path, MAX(COALESCE(media_size, 0)) AS media_size, + MAX(created_at) AS newest_at + FROM quote_messages + WHERE account_id = ? AND media_path IS NOT NULL + GROUP BY media_path + ORDER BY newest_at ASC + `) + .all(accountId) as MediaRow[]; + } + + private expireOldMedia(accountId: string, cutoff: number): void { + for (const media of this.listMedia(accountId)) { + if (media.newest_at >= cutoff) continue; + this.detachMedia(media.media_path); + } + } + + private enforceMediaBudget(accountId: string): void { + const rows = this.listMedia(accountId); + let total = rows.reduce((sum, row) => sum + Math.max(0, row.media_size), 0); + for (const row of rows) { + if (total <= this.policy.maxMediaBytesPerAccount) break; + this.detachMedia(row.media_path); + total -= Math.max(0, row.media_size); + } + } + + private detachMedia(mediaPath: string): void { + this.db + .prepare("UPDATE quote_messages SET media_path = NULL WHERE media_path = ?") + .run(mediaPath); + this.removeMediaIfOrphaned(mediaPath); + } + + private removeMediaIfOrphaned(mediaPath: string): void { + const row = this.db + .prepare("SELECT 1 AS found FROM quote_messages WHERE media_path = ? LIMIT 1") + .get(mediaPath) as { found?: number } | undefined; + if (row?.found) return; + try { + fs.unlinkSync(mediaPath); + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== "ENOENT") { + logger.warn(`quote cache: failed to remove media path=${mediaPath}: ${String(err)}`); + } + } + } + + private pruneOrphanMediaFiles(): void { + if (!fs.existsSync(this.mediaRoot)) return; + const known = new Set( + ( + this.db + .prepare("SELECT DISTINCT media_path FROM quote_messages WHERE media_path IS NOT NULL") + .all() as Array<{ media_path: string }> + ).map((row) => row.media_path), + ); + const visit = (dir: string) => { + for (const entry of fs.readdirSync(dir, { withFileTypes: true })) { + const fullPath = path.join(dir, entry.name); + if (entry.isDirectory()) visit(fullPath); + else if (entry.isFile() && !known.has(fullPath)) { + try { + fs.unlinkSync(fullPath); + } catch { + // retry during the next GC pass + } + } + } + }; + visit(this.mediaRoot); + } + + close(): void { + if (this.closed) return; + this.closed = true; + clearInterval(this.gcTimer); + this.db.close(); + } +} + +let activeStore: QuoteStore | null = null; +let initialization: Promise | null = null; +const activeAccounts = new Set(); + +export async function initializeQuoteStore( + cfg: OpenClawConfig, + accountId?: string, +): Promise { + if (accountId) activeAccounts.add(accountId); + const policy = resolveQuoteCachePolicy(cfg); + if (!policy.enabled) { + activeStore?.close(); + activeStore = null; + initialization = null; + return null; + } + if (activeStore) { + activeStore.updatePolicy(policy); + return activeStore; + } + if (!initialization) { + initialization = QuoteStore.open({ policy }).then((store) => { + activeStore = store; + return store; + }); + } + return initialization; +} + +export function getQuoteStore(): QuoteStore | null { + return activeStore; +} + +export function getActiveQuoteMediaSubdir(accountId: string): string | undefined { + return activeStore ? resolveQuoteMediaSubdir(accountId) : undefined; +} + +export function closeQuoteStore(): void { + activeStore?.close(); + activeStore = null; + initialization = null; + activeAccounts.clear(); +} + +export function deleteQuoteCacheForAccount(accountId: string): void { + if (activeStore) { + activeStore.deleteAccount(accountId); + return; + } + + const rootDir = path.join(resolveStateDir(), "openclaw-weixin"); + const mediaRoot = resolveQuoteMediaRoot(); + const dbPath = path.join(rootDir, "ref-messages.sqlite"); + if (!fs.existsSync(dbPath)) return; + try { + const require = createRequire(import.meta.url); + const sqlite = require("node:sqlite") as { DatabaseSync: DatabaseSyncConstructor }; + const db = new sqlite.DatabaseSync(dbPath); + db.prepare("DELETE FROM quote_messages WHERE account_id = ?").run(accountId); + db.close(); + fs.rmSync(path.join(mediaRoot, accountMediaDirName(accountId)), { + recursive: true, + force: true, + }); + fs.rmSync(path.join(rootDir, "ref-media", accountMediaDirName(accountId)), { + recursive: true, + force: true, + }); + } catch (err) { + logger.warn(`quote cache: failed to delete account=${accountId}: ${String(err)}`); + } +} + +export function deactivateQuoteStoreAccount(accountId: string): void { + activeAccounts.delete(accountId); + if (activeAccounts.size === 0) closeQuoteStore(); +} diff --git a/src/messaging/reply-progress-sender.ts b/src/messaging/reply-progress-sender.ts index a2069a1..b60f607 100644 --- a/src/messaging/reply-progress-sender.ts +++ b/src/messaging/reply-progress-sender.ts @@ -12,6 +12,7 @@ export type WeixinReplyProgressSenderDeps = { opts: WeixinApiOptions & { contextToken?: string; runId?: string; + accountId?: string; }; }; @@ -44,7 +45,7 @@ export class WeixinReplyProgressSender { this.runId = deps.runId; this.to = deps.to; this.accountId = deps.accountId; - this.opts = { ...deps.opts, runId: deps.runId }; + this.opts = { ...deps.opts, runId: deps.runId, accountId: deps.accountId }; } get replyOptions() { diff --git a/src/messaging/send-media.test.ts b/src/messaging/send-media.test.ts index 910f9cb..a5cfce3 100644 --- a/src/messaging/send-media.test.ts +++ b/src/messaging/send-media.test.ts @@ -87,6 +87,8 @@ describe("sendWeixinMediaFile", () => { fileName: "doc.pdf", uploaded: fakeUploaded, opts: baseParams.opts, + filePath: "/tmp/doc.pdf", + mediaMime: "application/pdf", }); }); diff --git a/src/messaging/send-media.ts b/src/messaging/send-media.ts index c838251..e6ddbb7 100644 --- a/src/messaging/send-media.ts +++ b/src/messaging/send-media.ts @@ -22,7 +22,7 @@ export async function sendWeixinMediaFile(params: { filePath: string; to: string; text: string; - opts: WeixinApiOptions & { contextToken?: string; runId?: string }; + opts: WeixinApiOptions & { contextToken?: string; runId?: string; accountId?: string }; cdnBaseUrl: string; }): Promise<{ messageId: string }> { const { filePath, to, text, opts, cdnBaseUrl } = params; @@ -40,7 +40,7 @@ export async function sendWeixinMediaFile(params: { logger.info( `[weixin] sendWeixinMediaFile: video upload done filekey=${uploaded.filekey} size=${uploaded.fileSize}`, ); - return sendVideoMessageWeixin({ to, text, uploaded, opts }); + return sendVideoMessageWeixin({ to, text, uploaded, opts, filePath, mediaMime: mime }); } if (mime.startsWith("image/")) { @@ -54,7 +54,7 @@ export async function sendWeixinMediaFile(params: { logger.info( `[weixin] sendWeixinMediaFile: image upload done filekey=${uploaded.filekey} size=${uploaded.fileSize}`, ); - return sendImageMessageWeixin({ to, text, uploaded, opts }); + return sendImageMessageWeixin({ to, text, uploaded, opts, filePath, mediaMime: mime }); } // File attachment: pdf, doc, zip, etc. @@ -72,5 +72,5 @@ export async function sendWeixinMediaFile(params: { logger.info( `[weixin] sendWeixinMediaFile: file upload done filekey=${uploaded.filekey} size=${uploaded.fileSize}`, ); - return sendFileMessageWeixin({ to, text, fileName, uploaded, opts }); + return sendFileMessageWeixin({ to, text, fileName, uploaded, opts, filePath, mediaMime: mime }); } diff --git a/src/messaging/send.test.ts b/src/messaging/send.test.ts index 7cbad70..6490b03 100644 --- a/src/messaging/send.test.ts +++ b/src/messaging/send.test.ts @@ -9,7 +9,8 @@ vi.mock("../util/logger.js", () => ({ }, })); -const { mockSendMessageApi } = vi.hoisted(() => ({ +const { mockQuotePut, mockSendMessageApi } = vi.hoisted(() => ({ + mockQuotePut: vi.fn(), mockSendMessageApi: vi.fn(), })); @@ -17,6 +18,10 @@ vi.mock("../api/api.js", () => ({ sendMessage: mockSendMessageApi, })); +vi.mock("./quote-store.js", () => ({ + getQuoteStore: () => ({ put: mockQuotePut }), +})); + vi.mock("node:crypto", () => ({ default: { randomBytes: vi.fn(() => Buffer.from("deadbeef", "hex")), @@ -108,6 +113,38 @@ describe("sendMessageWeixin", () => { }), ).rejects.toThrow("api fail"); }); + + it("stores outbound text under the lossless server ID", async () => { + mockSendMessageApi.mockResolvedValueOnce({ message_id: "18446744073709551615" }); + const result = await sendMessageWeixin({ + to: "user1", + text: "hello", + opts: { baseUrl: "https://api.com", accountId: "account1" }, + }); + expect(result.serverMessageId).toBe("18446744073709551615"); + expect(result.messageId).not.toBe(result.serverMessageId); + expect(mockQuotePut).toHaveBeenCalledWith( + expect.objectContaining({ + accountId: "account1", + conversationId: "user1", + messageId: "18446744073709551615", + body: "hello", + direction: "outbound", + }), + ); + }); + + it("does not turn a successful send into a failure when caching fails", async () => { + mockSendMessageApi.mockResolvedValueOnce({ message_id: "99" }); + mockQuotePut.mockRejectedValueOnce(new Error("disk full")); + await expect( + sendMessageWeixin({ + to: "user1", + text: "hello", + opts: { baseUrl: "https://api.com", accountId: "account1" }, + }), + ).resolves.toMatchObject({ serverMessageId: "99" }); + }); }); describe("sendMessageItemWeixin", () => { @@ -138,6 +175,44 @@ describe("sendMessageItemWeixin", () => { }, ]); }); + + it("caches a structured text item and preserves an explicit client ID", async () => { + mockSendMessageApi.mockResolvedValueOnce({ message_id: "123" }); + const result = await sendMessageItemWeixin({ + to: "user1", + item: { type: MessageItemType.TEXT, text_item: { text: "structured text" } }, + opts: { baseUrl: "https://api.com", accountId: "account1" }, + clientId: "explicit-client-id", + }); + expect(result).toEqual({ messageId: "explicit-client-id", serverMessageId: "123" }); + expect(mockQuotePut).toHaveBeenCalledWith( + expect.objectContaining({ + messageId: "123", + body: "structured text", + }), + ); + }); + + it("returns the client ID when a structured item has no server ID", async () => { + mockSendMessageApi.mockResolvedValueOnce({}); + const result = await sendMessageItemWeixin({ + to: "user1", + item: { type: MessageItemType.TEXT, text_item: {} }, + opts: { baseUrl: "https://api.com" }, + }); + expect(result.serverMessageId).toBeUndefined(); + }); + + it("rethrows structured-item API errors", async () => { + mockSendMessageApi.mockRejectedValueOnce(new Error("structured fail")); + await expect( + sendMessageItemWeixin({ + to: "user1", + item: { type: MessageItemType.TEXT, text_item: { text: "hello" } }, + opts: { baseUrl: "https://api.com" }, + }), + ).rejects.toThrow("structured fail"); + }); }); function makeUploadedFileInfo(overrides?: Partial): UploadedFileInfo { @@ -224,6 +299,27 @@ describe("sendImageMessageWeixin", () => { }), ).rejects.toThrow("cdn fail"); }); + + it("stores managed metadata for an outbound image", async () => { + mockSendMessageApi.mockResolvedValueOnce({ message_id: "image-server-id" }); + await sendImageMessageWeixin({ + to: "user1", + text: "", + uploaded: makeUploadedFileInfo(), + opts: { baseUrl: "https://api.com", accountId: "account1" }, + filePath: "/tmp/photo.png", + mediaMime: "image/png", + }); + expect(mockQuotePut).toHaveBeenCalledWith( + expect.objectContaining({ + messageId: "image-server-id", + body: "[图片]", + sourceMediaPath: "/tmp/photo.png", + mediaMime: "image/png", + mediaName: "photo.png", + }), + ); + }); }); describe("sendVideoMessageWeixin", () => { diff --git a/src/messaging/send.ts b/src/messaging/send.ts index 8d1a50f..0aaea1b 100644 --- a/src/messaging/send.ts +++ b/src/messaging/send.ts @@ -7,14 +7,48 @@ import { generateId } from "../util/random.js"; import type { MessageItem, SendMessageReq } from "../api/types.js"; import { MessageItemType, MessageState, MessageType } from "../api/types.js"; import type { UploadedFileInfo } from "../cdn/upload.js"; +import { getMediaLabel } from "./inbound.js"; +import { getQuoteStore } from "./quote-store.js"; export { StreamingMarkdownFilter } from "./markdown-filter.js"; type WeixinMessageSendOptions = WeixinApiOptions & { contextToken?: string; runId?: string; + accountId?: string; }; +type WeixinSendResult = { messageId: string; serverMessageId?: string }; + +async function cacheOutboundMessage(params: { + opts: WeixinMessageSendOptions; + to: string; + serverMessageId?: string; + body: string; + sourceMediaPath?: string; + mediaMime?: string; + mediaName?: string; +}): Promise { + if (!params.opts.accountId || !params.serverMessageId) return; + try { + await getQuoteStore()?.put({ + accountId: params.opts.accountId, + conversationId: params.to, + messageId: params.serverMessageId, + direction: "outbound", + body: params.body, + ...(params.sourceMediaPath ? { sourceMediaPath: params.sourceMediaPath } : {}), + ...(params.mediaMime ? { mediaMime: params.mediaMime } : {}), + ...(params.mediaName ? { mediaName: params.mediaName } : {}), + createdAt: Date.now(), + }); + } catch (err) { + logger.warn( + `quote cache: failed to save outbound message id=${params.serverMessageId}: ${String(err)}`, + ); + } +} + function generateClientId(): string { return generateId("openclaw-weixin"); } @@ -70,7 +104,7 @@ export async function sendMessageWeixin(params: { to: string; text: string; opts: WeixinMessageSendOptions; -}): Promise<{ messageId: string }> { +}): Promise { const { to, text, opts } = params; if (!opts.contextToken) { logger.warn(`sendMessageWeixin: contextToken missing for to=${to}, sending without context`); @@ -84,17 +118,22 @@ export async function sendMessageWeixin(params: { clientId, }); try { - await sendMessageApi({ + const response = await sendMessageApi({ baseUrl: opts.baseUrl, token: opts.token, timeoutMs: opts.timeoutMs, body: req, }); + const serverMessageId = response?.message_id; + await cacheOutboundMessage({ opts, to, serverMessageId, body: text }); + return { + messageId: clientId, + ...(serverMessageId ? { serverMessageId } : {}), + }; } catch (err) { logger.error(`sendMessageWeixin: failed to=${to} clientId=${clientId} err=${String(err)}`); throw err; } - return { messageId: clientId }; } /** Send a single structured MessageItem downstream. */ @@ -104,7 +143,7 @@ export async function sendMessageItemWeixin(params: { opts: WeixinMessageSendOptions; clientId?: string; label?: string; -}): Promise<{ messageId: string }> { +}): Promise { const { to, item, opts } = params; if (!opts.contextToken) { logger.warn( @@ -125,19 +164,25 @@ export async function sendMessageItemWeixin(params: { }, }; try { - await sendMessageApi({ + const response = await sendMessageApi({ baseUrl: opts.baseUrl, token: opts.token, timeoutMs: opts.timeoutMs, body: req, }); + const serverMessageId = response?.message_id; + const itemText = item.type === MessageItemType.TEXT ? (item.text_item?.text ?? "") : ""; + if (itemText) await cacheOutboundMessage({ opts, to, serverMessageId, body: itemText }); + return { + messageId: clientId, + ...(serverMessageId ? { serverMessageId } : {}), + }; } catch (err) { logger.error( `${params.label ?? "sendMessageItemWeixin"}: failed to=${to} clientId=${clientId} err=${String(err)}`, ); throw err; } - return { messageId: clientId }; } /** @@ -150,7 +195,10 @@ async function sendMediaItems(params: { mediaItem: MessageItem; opts: WeixinMessageSendOptions; label: string; -}): Promise<{ messageId: string }> { + sourceMediaPath?: string; + mediaMime?: string; + mediaName?: string; +}): Promise { const { to, text, mediaItem, opts, label } = params; const runId = opts.runId; @@ -161,6 +209,7 @@ async function sendMediaItems(params: { items.push(mediaItem); let lastClientId = ""; + let lastServerMessageId: string | undefined; for (const item of items) { lastClientId = generateClientId(); const req: SendMessageReq = { @@ -176,12 +225,31 @@ async function sendMediaItems(params: { }, }; try { - await sendMessageApi({ + const response = await sendMessageApi({ baseUrl: opts.baseUrl, token: opts.token, timeoutMs: opts.timeoutMs, body: req, }); + lastServerMessageId = response?.message_id; + if (item.type === MessageItemType.TEXT) { + await cacheOutboundMessage({ + opts, + to, + serverMessageId: lastServerMessageId, + body: item.text_item?.text ?? text, + }); + } else { + await cacheOutboundMessage({ + opts, + to, + serverMessageId: lastServerMessageId, + body: getMediaLabel(item.type), + sourceMediaPath: params.sourceMediaPath, + mediaMime: params.mediaMime, + mediaName: params.mediaName, + }); + } } catch (err) { logger.error(`${label}: failed to=${to} clientId=${lastClientId} err=${String(err)}`); throw err; @@ -189,7 +257,10 @@ async function sendMediaItems(params: { } logger.info(`${label}: success to=${to} clientId=${lastClientId}`); - return { messageId: lastClientId }; + return { + messageId: lastClientId, + ...(lastServerMessageId ? { serverMessageId: lastServerMessageId } : {}), + }; } /** @@ -206,7 +277,9 @@ export async function sendImageMessageWeixin(params: { text: string; uploaded: UploadedFileInfo; opts: WeixinMessageSendOptions; -}): Promise<{ messageId: string }> { + filePath?: string; + mediaMime?: string; +}): Promise { const { to, text, uploaded, opts } = params; if (!opts.contextToken) { logger.warn( @@ -229,7 +302,16 @@ export async function sendImageMessageWeixin(params: { }, }; - return sendMediaItems({ to, text, mediaItem: imageItem, opts, label: "sendImageMessageWeixin" }); + return sendMediaItems({ + to, + text, + mediaItem: imageItem, + opts, + label: "sendImageMessageWeixin", + sourceMediaPath: params.filePath, + mediaMime: params.mediaMime, + mediaName: params.filePath?.split(/[\\/]/).pop(), + }); } /** @@ -242,7 +324,9 @@ export async function sendVideoMessageWeixin(params: { text: string; uploaded: UploadedFileInfo; opts: WeixinMessageSendOptions; -}): Promise<{ messageId: string }> { + filePath?: string; + mediaMime?: string; +}): Promise { const { to, text, uploaded, opts } = params; if (!opts.contextToken) { logger.warn( @@ -262,7 +346,16 @@ export async function sendVideoMessageWeixin(params: { }, }; - return sendMediaItems({ to, text, mediaItem: videoItem, opts, label: "sendVideoMessageWeixin" }); + return sendMediaItems({ + to, + text, + mediaItem: videoItem, + opts, + label: "sendVideoMessageWeixin", + sourceMediaPath: params.filePath, + mediaMime: params.mediaMime, + mediaName: params.filePath?.split(/[\\/]/).pop(), + }); } /** @@ -276,7 +369,9 @@ export async function sendFileMessageWeixin(params: { fileName: string; uploaded: UploadedFileInfo; opts: WeixinMessageSendOptions; -}): Promise<{ messageId: string }> { + filePath?: string; + mediaMime?: string; +}): Promise { const { to, text, fileName, uploaded, opts } = params; if (!opts.contextToken) { logger.warn( @@ -296,5 +391,14 @@ export async function sendFileMessageWeixin(params: { }, }; - return sendMediaItems({ to, text, mediaItem: fileItem, opts, label: "sendFileMessageWeixin" }); + return sendMediaItems({ + to, + text, + mediaItem: fileItem, + opts, + label: "sendFileMessageWeixin", + sourceMediaPath: params.filePath, + mediaMime: params.mediaMime, + mediaName: fileName, + }); } diff --git a/src/messaging/slash-commands.ts b/src/messaging/slash-commands.ts index e20f56d..5cde974 100644 --- a/src/messaging/slash-commands.ts +++ b/src/messaging/slash-commands.ts @@ -28,10 +28,11 @@ export interface SlashCommandContext { /** 发送回复消息 */ async function sendReply(ctx: SlashCommandContext, text: string): Promise { - const opts: WeixinApiOptions & { contextToken?: string } = { + const opts: WeixinApiOptions & { contextToken?: string; accountId?: string } = { baseUrl: ctx.baseUrl, token: ctx.token, contextToken: ctx.contextToken, + accountId: ctx.accountId, }; await sendMessageWeixin({ to: ctx.to, text, opts }); }