From dbacc86795c86fe099ad0c10170613a13eb84050 Mon Sep 17 00:00:00 2001 From: orange-bot Date: Sun, 18 Oct 2026 01:34:36 +0800 Subject: [PATCH] message: exclude a send echo by platform message id --- packages/bridge/src/dialogs.ts | 51 ++++-- packages/bridge/src/index.ts | 6 + packages/bridge/src/send-echoes.test.ts | 50 ++++++ packages/bridge/src/send-echoes.ts | 48 ++++++ .../send-observer-reconciliation.e2e.test.ts | 160 +++++++++++++++++- packages/bridge/src/update-manager.ts | 12 +- 6 files changed, 311 insertions(+), 16 deletions(-) create mode 100644 packages/bridge/src/send-echoes.test.ts create mode 100644 packages/bridge/src/send-echoes.ts diff --git a/packages/bridge/src/dialogs.ts b/packages/bridge/src/dialogs.ts index 7fc58151..653b34c0 100644 --- a/packages/bridge/src/dialogs.ts +++ b/packages/bridge/src/dialogs.ts @@ -41,6 +41,7 @@ import { type DialogFolderStore, type StoredDialogFilter, } from './dialog-folders.js' import type { MessageProjectionPipeline } from './message-projection.js' +import type { SendEchoes } from './send-echoes.js' import type { SpeechPipeline } from './speech.js' import { convertRichMessage, markdownRichMessage, type RichMediaReference, type ResolvedRichMedia } from './rich-message.js' @@ -292,6 +293,7 @@ export class DialogRpc { private readonly _dialogFolders?: DialogFolderStore, private readonly _systemPeers?: SystemPeerService, private readonly _messageProjection?: MessageProjectionPipeline, + private readonly _sendEchoes?: SendEchoes, ) { this._actions = new PlatformMessageActions(_platform, _session) // 有 `MessageStore` 时淘汰旧条目并由数据库兜底；无存储模式不可淘汰，避免重用已分配的 ID。 @@ -2358,16 +2360,18 @@ export class DialogRpc { const conversation = this._conversation(toId) const publishedPayloads: tl.RawUpdates[] = [] const projections = [] - for (const output of forwarded!) { - const source: IMMessage = { ...output, conversationId: toId, outgoing: true } - const published = await this._publishLocalMessage(toId, source, excludeConnection) - if (published) { - publishedPayloads.push(published) - continue + await this._withSendEchoes(forwarded!, async () => { + for (const output of forwarded!) { + const source: IMMessage = { ...output, conversationId: toId, outgoing: true } + const published = await this._publishLocalMessage(toId, source, excludeConnection) + if (published) { + publishedPayloads.push(published) + continue + } + const persisted = await this._store.ingest(this._session, conversation, source) + projections.push(...persisted.projection) } - const persisted = await this._store.ingest(this._session, conversation, source) - projections.push(...persisted.projection) - } + }) if (projections.length) { let pts = await this._reservePts( projections.length, @@ -2412,7 +2416,8 @@ export class DialogRpc { conversationId: resolution.peer.id, outgoing: true, } - const published = await this._publishLocalMessage(resolution.peer.id, source, excludeConnection) + const published = await this._withSendEchoes([source], () => + this._publishLocalMessage(resolution.peer.id, source, excludeConnection)) await this._systemPeers!.receive(this._session, resolution, source, input) if (published) return this._confirmForwardRandomIds(published, randomIds) @@ -3284,9 +3289,9 @@ export class DialogRpc { outgoing: true, replyToId: sent.replyToId ?? replyToId, } - const published = await this._publishLocalMessage( + const published = await this._withSendEchoes([source], () => this._publishLocalMessage( peerId, source, excludeConnection, [req.randomId], replyTopMessageId(req.replyTo), - ) + )) if (systemPeer) await this._systemPeers!.receive(this._session, systemPeer, source, input) if (published) { const update = published.updates.find((item) => @@ -3428,6 +3433,24 @@ export class DialogRpc { } } + /** + * 平台返回发送结果后，把「这些平台消息是当前 auth key 发的」记下来，再投递本地副本。 + * 平台随后回显同一条消息时，`UpdateManager` 按平台消息 id 把发送端排除掉：客户端 + * 只从 RPC 回包里拿这一条，`updateMessageID` 能原地升级，不会多出服务端副本。 + */ + private async _withSendEchoes( + sent: readonly IMMessage[], + publish: () => Promise, + ): Promise { + const ids = sent.flatMap((message) => [message.id, ...(message.sourceIds ?? [])]) + this._sendEchoes?.bind(this._session.platformSessionId, ids, this._authKeyId) + try { + return await publish() + } finally { + this._sendEchoes?.release(this._session.platformSessionId, ids) + } + } + private async _sendToPlatform(send: () => Promise): Promise { try { return await send() @@ -3654,9 +3677,9 @@ export class DialogRpc { if (!this._store) throw new RpcError(500, 'MESSAGE_STORE_UNAVAILABLE') const replyToTopId = replyTopMessageId(replyTo) const publishAt = performance.now() - const published = await this._publishLocalMessage( + const published = await this._withSendEchoes([source], () => this._publishLocalMessage( peerId, source, excludeConnection, randomIds, replyToTopId, - ) + )) const publishMs = performance.now() - publishAt const cleanupAt = performance.now() if (systemPeer) { diff --git a/packages/bridge/src/index.ts b/packages/bridge/src/index.ts index f29f4e35..27a59a12 100644 --- a/packages/bridge/src/index.ts +++ b/packages/bridge/src/index.ts @@ -15,6 +15,7 @@ import { DialogRpc, type LegacyGetForumTopicsByIdRequest, type LegacyGetForumTopicsRequest, } from './dialogs.js' +import { SendEchoes } from './send-echoes.js' import { startupRpcHandlers } from './startup.js' import { androidRpcHandlers } from './android-rpc.js' import { MessageStore } from './message-store.js' @@ -265,6 +266,7 @@ export function apply(ctx: Context, config: BridgeConfig = {}): void { reactionRpcs.set(key, { platform, rpc }) return rpc } + const sendEchoes = new SendEchoes() const updates = new UpdateManager( ctx.database, registry, store, (authKeyId, update, excludeConnection) => @@ -280,6 +282,7 @@ export function apply(ctx: Context, config: BridgeConfig = {}): void { config.recalledMessageMode ?? 'show', (session, conversationId, definition) => reactionRpcFor(registry.require(session.platformId), session) .registerInlineCustomEmoji(conversationId, definition), + sendEchoes, ) const builtInMediaProvider = createBuiltInVoiceMediaProvider({ serverHost: config.serverHost, @@ -437,6 +440,7 @@ export function apply(ctx: Context, config: BridgeConfig = {}): void { updates.requestRecovery(rpcContext) }, messageProjection, + sendEchoes, ) new MtprotoBridgeService(ctx, requireBridgeSession, dcId) registerGroupFilesMiniApp(ctx, platforms, requireBridgeSession, config.groupFilesMiniApp) @@ -1487,6 +1491,7 @@ export function createSessionResolver( rpc: ServerRpcContext, ) => void | Promise, messageProjection?: MessageProjectionPipeline, + sendEchoes?: SendEchoes, ) { const projection = messageProjection ?? new MessageProjectionPipeline(ctx) const loading = new Map>() @@ -1570,6 +1575,7 @@ export function createSessionResolver( dialogFolders, systemPeers, projection, + sendEchoes, ) return state } diff --git a/packages/bridge/src/send-echoes.test.ts b/packages/bridge/src/send-echoes.test.ts new file mode 100644 index 00000000..68fc8cc1 --- /dev/null +++ b/packages/bridge/src/send-echoes.test.ts @@ -0,0 +1,50 @@ +import { describe, expect, it } from 'vitest' +import { SendEchoes } from './send-echoes.js' +import type { IMMessage, PlatformSession } from './platform.js' + +const session: PlatformSession = { + platformSessionId: 'send-echoes', platformId: 'qq', userId: 'self', + credentials: {}, metadata: {}, +} +const otherSession: PlatformSession = { ...session, platformSessionId: 'send-echoes-other' } + +function message(id: string, sourceIds?: string[]): IMMessage { + return { + id, sourceIds, conversationId: 'group', senderId: 'self', outgoing: true, timestamp: 1, + content: { parts: [{ type: 'text', text: id }] }, + } +} + +describe('SendEchoes', () => { + it('binds the echoed platform message to the auth key that is sending it', () => { + const echoes = new SendEchoes() + echoes.bind(session.platformSessionId, ['native-1'], 'key-a') + expect(echoes.excludeFor(session, message('native-1'))).toBe('key-a') + }) + + it('covers every part of a multi-part message', () => { + const echoes = new SendEchoes() + echoes.bind(session.platformSessionId, ['part-1', 'part-2'], 'key-a') + expect(echoes.excludeFor(session, message('part-2', ['part-1']))).toBe('key-a') + }) + + it('leaves other own messages and other sessions alone', () => { + const echoes = new SendEchoes() + echoes.bind(session.platformSessionId, ['native-1'], 'key-a') + expect(echoes.excludeFor(session, message('native-2'))).toBeUndefined() + expect(echoes.excludeFor(otherSession, message('native-1'))).toBeUndefined() + }) + + it('stops excluding once the send delivered the message', () => { + const echoes = new SendEchoes() + echoes.bind(session.platformSessionId, ['native-1'], 'key-a') + echoes.release(session.platformSessionId, ['native-1']) + expect(echoes.excludeFor(session, message('native-1'))).toBeUndefined() + }) + + it('ignores a send without an auth key', () => { + const echoes = new SendEchoes() + echoes.bind(session.platformSessionId, ['native-1'], undefined) + expect(echoes.excludeFor(session, message('native-1'))).toBeUndefined() + }) +}) diff --git a/packages/bridge/src/send-echoes.ts b/packages/bridge/src/send-echoes.ts new file mode 100644 index 00000000..ccc0d154 --- /dev/null +++ b/packages/bridge/src/send-echoes.ts @@ -0,0 +1,48 @@ +import type { IMMessage, PlatformSession } from './platform.js' + +/** + * 平台对「自己刚发出去的消息」的回显。 + * + * 发送 RPC 的回包已经用 `updateMessageID` 把发送端的乐观消息原地升级；如果平台 + * 再把同一条消息作为普通事件推回发送端，客户端就会多出一份服务端副本（桌面端 + * 合并时必然销毁一条并播放消息粉碎动画，Android 则是占位气泡消失）。 + * + * 这里按**平台消息 id** 精确记录这条回显属于哪个 auth key，而不是按发送时间段 + * 屏蔽整个 auth key：同一时间从 QQ 手机端发出的别的消息、或并发的其他发送都不受影响。 + * + * 发送响应返回前的回显由平台按 `originRequestId` 过滤（`platform-crossgram` + * 的 own-origin 规则）；这里补的是响应返回之后、本方投递完成之前的那一段。 + */ +export class SendEchoes { + private readonly _bySession = new Map>() + + /** 记录本次发送产生的平台消息（含多段消息的 `sourceIds`）属于哪个 auth key。 */ + bind(platformSessionId: string, messageIds: readonly string[], authKeyId?: string): void { + if (!authKeyId || !messageIds.length) return + let bound = this._bySession.get(platformSessionId) + if (!bound) this._bySession.set(platformSessionId, bound = new Map()) + for (const id of messageIds) bound.set(id, authKeyId) + } + + /** 本方投递完成（或发送失败）后解绑；此后的回显会被事件去重挡掉。 */ + release(platformSessionId: string, messageIds: readonly string[]): void { + const bound = this._bySession.get(platformSessionId) + if (!bound) return + for (const id of messageIds) bound.delete(id) + if (!bound.size) this._bySession.delete(platformSessionId) + } + + /** + * 这条消息的发送回显该排除哪个 auth key：命中本方发送记录时返回它的 auth key。 + * 同一会话里别人发的、或从别的客户端同步过来的消息不会命中。 + */ + excludeFor(session: PlatformSession, message: IMMessage): string | undefined { + const bound = this._bySession.get(session.platformSessionId) + if (!bound) return undefined + for (const id of [message.id, ...(message.sourceIds ?? [])]) { + const authKeyId = bound.get(id) + if (authKeyId) return authKeyId + } + return undefined + } +} diff --git a/packages/bridge/src/send-observer-reconciliation.e2e.test.ts b/packages/bridge/src/send-observer-reconciliation.e2e.test.ts index c39c5fde..457c0b0b 100644 --- a/packages/bridge/src/send-observer-reconciliation.e2e.test.ts +++ b/packages/bridge/src/send-observer-reconciliation.e2e.test.ts @@ -8,7 +8,8 @@ import { TlBinaryReader, TlBinaryWriter } from '@mtcute/tl-runtime' import Long from 'long' import type { RpcResult, ServerConnection, ServerRpcContext } from '@mtproto-relay/mtproto' import { getServerReaderMap } from '../../mtproto/src/rpc/server-reader-map.js' -import { DialogRpc, stableId } from './dialogs.js' +import { SendEchoes } from './send-echoes.js' +import { DialogRpc, peerTlSeed, stableId } from './dialogs.js' import { MessageStore } from './message-store.js' import { defineModels } from './models.js' import { PlatformRegistry } from './platform-manager.js' @@ -168,4 +169,161 @@ describe('sent-message observer reconciliation e2e', () => { expect((observerPushes[0].payload.updates[0] as tl.RawUpdateMessageID).id) .toBe((response.updates[0] as tl.RawUpdateMessageID).id) }) + + it('excludes only the sending auth key for the echoed message, not every own message', async () => { + const ctx = new Context() + const fibers = [ctx.plugin(Database), ctx.plugin(SQLiteDriver, { path: ':memory:' })] + await Promise.all(fibers) + await new Promise((resolve) => setTimeout(resolve, 25)) + defineModels(ctx) + await ctx.database.prepared() + disposals.push(async () => { + for (const fiber of fibers.reverse()) await Promise.resolve((fiber as any).dispose?.()) + }) + for (const authKeyId of [REQUESTER_AUTH_KEY, OBSERVER_AUTH_KEY]) { + await ctx.database.create('mtproto_auth_binding', { + authKeyId, platformId: session.platformId, platformSessionId: session.platformSessionId, + }) + } + + const sendEchoes = new SendEchoes() + const store = new MessageStore(ctx.database) + const platform: IMPlatform = { + capabilities: { + history: true, + send: { text: true, images: true, files: true, mixed: true, maxTextLength: 4096, maxMedia: 9 }, + conversations: { groups: true, channels: true, subchannels: false }, + }, + async subscribe() { return () => {} }, + async getDialogs() { return { dialogs: [{ conversation, unreadCount: 0 }] } }, + async getHistory() { return { messages: [] } }, + async getUser(_session, id) { return { id, firstName: id === session.userId ? 'Current' : id } }, + async sendMessage(): Promise { throw new Error('not used') }, + } + const pushes: Array<{ authKeyId: string, payload: tl.RawUpdates }> = [] + const manager = new UpdateManager( + ctx.database, new PlatformRegistry([[session.platformId, platform]]), store, + (authKeyId, update) => { + pushes.push({ + authKeyId: Buffer.from(authKeyId).toString('hex'), + payload: update as tl.RawUpdates, + }) + return 1 + }, + 1, undefined, undefined, undefined, undefined, undefined, undefined, undefined, + sendEchoes, + ) + const echo = async (id: string, text: string) => { + const message: IMMessage = { + id, conversationId: conversation.id, senderId: session.userId, outgoing: true, + timestamp: 1_800_000_000, content: { parts: [{ type: 'text', text }] }, + } + const result = await store.ingest(session, conversation, message) + await manager.publish(session, { event: { type: 'message', conversation, message }, result }, {}) + } + + // 发送 RPC 拿到平台结果后登记这条消息：它的回显只推给其他设备。 + sendEchoes.bind(session.platformSessionId, ['qq-sent-text'], REQUESTER_AUTH_KEY) + await echo('qq-sent-text', 'one bubble for the sender') + expect(pushes.map((push) => push.authKeyId)).toEqual([OBSERVER_AUTH_KEY]) + expect(pushes[0]!.payload.updates).toMatchObject([ + { _: 'updateNewChannelMessage', message: { message: 'one bubble for the sender' } }, + ]) + + // 同一时间段里从 QQ 手机端发出的另一条自己消息，不能被顺带屏蔽掉。 + await echo('qq-phone-app-text', 'sent from another client') + expect(pushes.map((push) => push.authKeyId)).toEqual([ + OBSERVER_AUTH_KEY, REQUESTER_AUTH_KEY, OBSERVER_AUTH_KEY, + ]) + + // 本方投递完成后解绑，之后到达的消息照常推给所有设备。 + sendEchoes.release(session.platformSessionId, ['qq-sent-text']) + await echo('qq-later-text', 'delivered to every device') + expect(pushes.map((push) => push.authKeyId)).toEqual([ + OBSERVER_AUTH_KEY, REQUESTER_AUTH_KEY, OBSERVER_AUTH_KEY, + REQUESTER_AUTH_KEY, OBSERVER_AUTH_KEY, + ]) + }) + + it('binds the sent platform message for the whole local publish', async () => { + const ctx = new Context() + const fibers = [ctx.plugin(Database), ctx.plugin(SQLiteDriver, { path: ':memory:' })] + await Promise.all(fibers) + await new Promise((resolve) => setTimeout(resolve, 25)) + defineModels(ctx) + await ctx.database.prepared() + disposals.push(async () => { + for (const fiber of fibers.reverse()) await Promise.resolve((fiber as any).dispose?.()) + }) + for (const authKeyId of [REQUESTER_AUTH_KEY, OBSERVER_AUTH_KEY]) { + await ctx.database.create('mtproto_auth_binding', { + authKeyId, platformId: session.platformId, platformSessionId: session.platformSessionId, + }) + } + + const sendEchoes = new SendEchoes() + const store = new MessageStore(ctx.database) + const platform: IMPlatform = { + capabilities: { + history: true, + send: { text: true, images: true, files: true, mixed: true, maxTextLength: 4096, maxMedia: 9 }, + conversations: { groups: true, channels: true, subchannels: false }, + }, + async subscribe() { return () => {} }, + async getDialogs() { return { dialogs: [{ conversation, unreadCount: 0 }] } }, + async getHistory() { return { messages: [] } }, + async getUser(_session, id) { return { id, firstName: id === session.userId ? 'Current' : id } }, + async sendMessage(_session, target, input): Promise { + const text = input.parts.find((part) => part.type === 'text') + if (!text || text.type !== 'text') throw new Error('expected sent text') + return { + id: 'qq-wired-text', conversationId: target.id, senderId: session.userId, + outgoing: true, timestamp: 1_800_000_000, + content: { parts: [{ type: 'text', text: text.text }] }, + } + }, + } + const boundDuringPublish: Array = [] + const manager = new UpdateManager( + ctx.database, new PlatformRegistry([[session.platformId, platform]]), store, + () => 1, + 1, undefined, undefined, undefined, undefined, undefined, undefined, undefined, + sendEchoes, + ) + const dialogs = new DialogRpc( + platform, session, store, + undefined, undefined, 1, undefined, undefined, undefined, + async (localSession, event, options) => { + if (event.type !== 'message') return + boundDuringPublish.push(sendEchoes.excludeFor(localSession, event.message)) + const result = await store.ingest(localSession, event.conversation, event.message) + return manager.publish(localSession, { event, result }, options) + }, + REQUESTER_AUTH_KEY, + undefined, undefined, undefined, undefined, undefined, undefined, undefined, undefined, undefined, + sendEchoes, + ) + await (dialogs as any)._hydratePeers() + const rpc = createCordisRpcTestHarness() + rpc.register('messages.sendMessage', async (context, request) => + dialogs.sendMessage(request as tl.messages.RawSendMessageRequest, context.connection)) + const targetPeer = { + _: 'inputPeerChannel' as const, + channelId: stableId(peerTlSeed(session.platformSessionId, conversation.id)), accessHash: Long.ZERO, + } + await roundTripRpc(rpc, { + _: 'messages.sendMessage', + peer: targetPeer, + message: 'bind while publishing', + randomId: Long.fromNumber(80_003), + }) + + // 投递本地副本的那一刻，这条平台消息已经归属发送它的 auth key。 + expect(boundDuringPublish).toEqual([REQUESTER_AUTH_KEY]) + // 投递完成后解绑：此后同一条消息的回显由事件去重处理。 + expect(sendEchoes.excludeFor(session, { + id: 'qq-wired-text', conversationId: conversation.id, senderId: session.userId, + outgoing: true, timestamp: 1_800_000_000, content: { parts: [{ type: 'text', text: 'x' }] }, + })).toBeUndefined() + }) }) diff --git a/packages/bridge/src/update-manager.ts b/packages/bridge/src/update-manager.ts index a9dd4040..e1bb7bae 100644 --- a/packages/bridge/src/update-manager.ts +++ b/packages/bridge/src/update-manager.ts @@ -25,6 +25,7 @@ import type { BlockedPeerStore } from './blocked-peers.js' import { customReactionDocumentId } from './reaction-rpc.js' import { updateFromJson, updateToJson } from './update-json.js' import type { MessageProjectionPipeline } from './message-projection.js' +import type { SendEchoes } from './send-echoes.js' const CHANNEL_POLL_TIMEOUT_SECONDS = 30 const PENDING_CHANNEL_POLL_TIMEOUT_SECONDS = 1 @@ -67,6 +68,7 @@ export class UpdateManager { conversationId: string, definition: IMReactionDefinition, ) => number | undefined, + private readonly _sendEchoes?: SendEchoes, ) {} /** 注册行内自定义表情的 document，使渲染消息的客户端能解析对应资源。 */ @@ -452,12 +454,20 @@ export class UpdateManager { ) return } + // 平台对自己刚发出的消息的回显：发送端已经/即将从 RPC 回包拿到带 + // `updateMessageID` 的同一份数据，再推一份普通消息过去会让客户端多出服务端副本， + // 合并时销毁其中一条（桌面端的消息粉碎动画就是这么来的）。 + // 用平台消息 id 精确匹配，只影响这一条消息，不按时间段屏蔽整个 auth key。 + const echoAuthKeyId = (committed.event.type === 'message' || committed.event.type === 'message-edit') + && !options.excludeAuthKeyId + ? this._sendEchoes?.excludeFor(session, committed.event.message) + : undefined return await this._publishMessage( session, committed as Exclude, - options, + echoAuthKeyId ? { ...options, excludeAuthKeyId: echoAuthKeyId } : options, ) } catch (error) { if (eventKey) await this._releaseAbandonedDelivery(eventKey, session)