message: exclude a send echo by platform message id

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<any> = { ...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<any> = { ...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<T>(
+    sent: readonly IMMessage<any>[],
+    publish: () => Promise<T>,
+  ): Promise<T> {
+    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<IMMessage>): Promise<IMMessage> {
     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) {

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<void>,
   messageProjection?: MessageProjectionPipeline,
+  sendEchoes?: SendEchoes,
 ) {
   const projection = messageProjection ?? new MessageProjectionPipeline(ctx)
   const loading = new Map<string, Promise<BridgeSessionState>>()
@@ -1570,6 +1575,7 @@ export function createSessionResolver(
         dialogFolders,
         systemPeers,
         projection,
+        sendEchoes,
       )
       return state
     }

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()
+  })
+})

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<string, Map<string, string>>()
+
+  /** 记录本次发送产生的平台消息（含多段消息的 `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
+  }
+}