platform-crossgram: replace unstorable text in the named fields at the entry point

packages/platform-crossgram/src/index.ts

@@ -29,6 +29,9 @@ import { defineQQMediaPreviewModel, mediaPreviewKey, QQMediaPreviewer } from './
 import { extractRemoteVideoFrame, type RemoteVideoFrameExtractor } from './video-frame.js'
 import { migrateLegacyQQMessageMedia } from './raw-media-migration.js'
 import { migrateLegacyQQGroupAliasUsers } from './user-name-migration.js'
+import {
+  sanitizeWireConversation, sanitizeWireEvent, sanitizeWireMessage, sanitizeWireRequest, sanitizeWireUser,
+} from './sanitize.js'
 import type {
   QQMediaLocator, QQStickerReference, WireCallSignalEvent, WireCard, WireConversation, WireEvent, WireGroupFilePage, WireMedia, WireMessage,
   WireMultiForwardLocator, WireNativeAvsdkEvent, WireReactionState, WireRequest, WireTextPart,
@@ -309,8 +312,9 @@ export class QQNTPlatform implements IMPlatform<QQMediaLocator> {
       || status.protocolVersion < MIN_PROTOCOL_VERSION) {
       throw new Error(`QQNT bridge protocol ${status.protocolVersion} is unsupported; protocol ${MIN_PROTOCOL_VERSION} or newer is required`)
     }
-    const user = await this.client.getUser(userId)
-    if (!user) throw new Error(`QQNT current user is unavailable: ${userId}`)
+    const profile = await this.client.getUser(userId)
+    if (!profile) throw new Error(`QQNT current user is unavailable: ${userId}`)
+    const user = sanitizeWireUser(profile)
     const qq = typeof status.selfUin === 'string' && /^\d+$/.test(status.selfUin)
       ? status.selfUin
       : typeof user.numericId === 'string' && /^\d+$/.test(user.numericId)
@@ -418,7 +422,11 @@ export class QQNTPlatform implements IMPlatform<QQMediaLocator> {
         platformSessionId, attempt, this.client.webSocketEndpoint, lastEventId ?? '<none>',
       )
       try {
-        await this.client.subscribe(async (event, eventId) => {
+        await this.client.subscribe(async (rawEvent, eventId) => {
+          // Cleaned before any step of the stream reads the event: the
+          // checkpoint persists the event only after the handler commits, so
+          // one string PostgreSQL cannot store would stall every later event.
+          const event = sanitizeWireEvent(rawEvent)
           consecutiveStreamFailures = 0
           try {
             if (event.type === 'native-avsdk') {
@@ -837,14 +845,17 @@ export class QQNTPlatform implements IMPlatform<QQMediaLocator> {
   async getContacts(_session: PlatformSession, query: IMPageQuery = {}): Promise<IMUserPage<QQMediaLocator>> {
     const page = await this.client.getContacts({ cursor: query.cursor, limit: query.limit })
     return {
-      users: page.users.map((user) => ({
-        id: user.id,
-        firstName: user.name,
-        username: user.numericId,
-        about: typeof user.signature === 'string' ? user.signature : undefined,
-        avatar: user.avatar ? mapMedia(user.avatar) : undefined,
-        metadata: qqUserMetadata(user),
-      })),
+      users: page.users.map((wireUser) => {
+        const user = sanitizeWireUser(wireUser)
+        return {
+          id: user.id,
+          firstName: user.name,
+          username: user.numericId,
+          about: typeof user.signature === 'string' ? user.signature : undefined,
+          avatar: user.avatar ? mapMedia(user.avatar) : undefined,
+          metadata: qqUserMetadata(user),
+        }
+      }),
       nextCursor: page.nextCursor,
     }
   }
@@ -997,8 +1008,9 @@ export class QQNTPlatform implements IMPlatform<QQMediaLocator> {
   }
 
   async getUser(_session: PlatformSession, userId: string): Promise<IMUser<QQMediaLocator> | null> {
-    const user = await this.client.getUser(userId)
-    if (!user) return null
+    const wireUser = await this.client.getUser(userId)
+    if (!wireUser) return null
+    const user = sanitizeWireUser(wireUser)
     return {
       id: user.id,
       firstName: user.name,
@@ -1089,18 +1101,21 @@ export class QQNTPlatform implements IMPlatform<QQMediaLocator> {
     if (isFilteredConversationId(conversation.id)) return { members: [], total: 0 }
     const page = await this.client.getMembers(conversation.id, { cursor: query.cursor, limit: query.limit })
     return {
-      members: page.members.map((member): IMConversationMember<QQMediaLocator> => ({
-        user: {
-          id: member.user.id,
-          firstName: member.user.name,
-          username: member.user.numericId,
-          avatar: member.user.avatar ? mapMedia(member.user.avatar) : undefined,
-          metadata: { ...qqUserMetadata(member.user), qqName: member.user.name },
-        },
-        role: member.role,
-        permissions: permissions(member.role),
-        title: member.user.alias?.trim() || undefined,
-      })),
+      members: page.members.map((member): IMConversationMember<QQMediaLocator> => {
+        const user = sanitizeWireUser(member.user)
+        return {
+          user: {
+            id: user.id,
+            firstName: user.name,
+            username: user.numericId,
+            avatar: user.avatar ? mapMedia(user.avatar) : undefined,
+            metadata: { ...qqUserMetadata(user), qqName: user.name },
+          },
+          role: member.role,
+          permissions: permissions(member.role),
+          title: user.alias?.trim() || undefined,
+        }
+      }),
       total: page.total,
       nextCursor: page.nextCursor,
     }
@@ -2238,6 +2253,7 @@ export class QQNTPlatform implements IMPlatform<QQMediaLocator> {
   }
 
   private mapConversation(input: WireConversation, session?: PlatformSession): IMConversation<QQMediaLocator> {
+    input = sanitizeWireConversation(input)
     const id = session && this.savedMessagesConversationIds.get(session.platformSessionId) === input.id
       ? session.userId
       : input.id
@@ -2337,7 +2353,7 @@ export class QQNTPlatform implements IMPlatform<QQMediaLocator> {
   }
 
   private mapMessage(input: WireMessage, conversationId?: string): IMMessage<QQMediaLocator> {
-    input = this.splitOutgoingMessages.get(input.id) ?? input
+    input = sanitizeWireMessage(this.splitOutgoingMessages.get(input.id) ?? input)
     const message = mapMessage(
       conversationId ? { ...input, conversationId } : input,
       this.reactionCatalog, this.stickerProviderId,
@@ -2597,6 +2613,7 @@ function mapConversation(input: WireConversation): IMConversation<QQMediaLocator
 }
 
 function mapRequest(input: WireRequest): IMRequest<QQMediaLocator> {
+  input = sanitizeWireRequest(input)
   return {
     id: input.id,
     kind: input.kind,

packages/platform-crossgram/src/message-order.e2e.test.ts

@@ -2601,6 +2601,144 @@ describe('QQNT join notice E2E', () => {
   })
 })
 
+describe('QQNT unstorable text E2E', () => {
+  it('persists a NUL nickname, remark, body and request note, and keeps the checkpoint moving', 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)
+    defineQQNTEventCheckpointModel(ctx)
+    await ctx.database.prepared()
+    disposals.push(async () => {
+      for (const fiber of fibers.reverse()) await Promise.resolve((fiber as any).dispose?.())
+    })
+
+    const checkpointSession = { ...session, platformSessionId: 'qqnt-unstorable-text-e2e' }
+    const conversation = {
+      id: 'unstorable-group', kind: 'group' as const, title: 'Gr\u0000oup',
+      peerUid: 'unstorable-group', peerUin: '43', chatType: 2 as const,
+    }
+    const webSocketServer = new WebSocketServer({ noServer: true })
+    const server = createServer()
+    server.on('upgrade', (request, socket, head) => {
+      webSocketServer.handleUpgrade(request, socket, head, (webSocket) => {
+        webSocketServer.emit('connection', webSocket, request)
+      })
+    })
+    webSocketServer.on('connection', (webSocket) => {
+      // QQ hands over a NUL in a group title, a nickname, a remark and a body,
+      // and a lone surrogate in the remark: PostgreSQL stores none of them, so
+      // before the fix the whole event failed and this checkpoint never moved.
+      webSocket.send(JSON.stringify({
+        id: '7001',
+        event: {
+          type: 'message', conversation,
+          message: {
+            id: 'unstorable-message', conversationId: conversation.id, senderId: 'peer',
+            timestamp: 1_787_215_100, outgoing: false,
+            sender: { id: 'peer', name: 'Bad\u0000Name', alias: 're\uD800mark' },
+            parts: [{
+              type: 'text', text: '@alice\u0000 hello',
+              entities: [{ type: 'mention', offset: 0, length: 6, userId: 'alice' }],
+            }],
+          },
+        },
+      }))
+      webSocket.send(JSON.stringify({
+        id: '7002',
+        event: {
+          type: 'request',
+          request: {
+            id: 'unstorable-request', kind: 'group-join', status: 'pending',
+            requester: { id: 'requester', name: 'Frie\u0000nd' },
+            group: { id: conversation.id, name: 'Gr\u0000oup' },
+            message: 'let me in\u0000',
+            timestamp: 1_787_215_101,
+          },
+        },
+      }))
+    })
+    server.listen(0, '127.0.0.1')
+    await once(server, 'listening')
+    const address = server.address()
+    if (!address || typeof address === 'string') throw new Error('missing unstorable text test address')
+    disposals.push(async () => {
+      for (const client of webSocketServer.clients) client.terminate()
+      webSocketServer.close()
+      if (!server.listening) return
+      const closed = new Promise<void>((resolve, reject) => {
+        server.close((error) => error ? reject(error) : resolve())
+      })
+      server.closeAllConnections()
+      await closed
+    })
+
+    const platform = new QQNTPlatform({
+      endpoint: 'http://127.0.0.1:1/v1',
+      webSocketEndpoint: `ws://127.0.0.1:${address.port}/events`,
+    }, 'qqnt:stickers', undefined, undefined, ctx.database)
+    platform.client.getReactionCatalog = vi.fn(async () => ({
+      available: [], reactions: [], maxSelected: 20,
+    }))
+    platform.client.getDialogs = vi.fn(async () => ({ conversations: [] }))
+    const store = new MessageStore(ctx.database)
+    const handled: string[] = []
+    const done = Promise.withResolvers<void>()
+    const unsubscribe = await platform.subscribe(checkpointSession, async (event) => {
+      if (event.type === 'message') await store.ingest(checkpointSession, event.conversation, event.message)
+      else if (event.type === 'request') await store.ingestRequest(checkpointSession, event.request)
+      else return
+      handled.push(event.type)
+      if (handled.length === 2) done.resolve()
+    })
+    try {
+      await Promise.race([
+        done.promise,
+        new Promise<never>((_, reject) => setTimeout(() => reject(new Error('unstorable text E2E timed out')), 5_000)),
+      ])
+      expect(handled).toEqual(['message', 'request'])
+      // The reported symptom: the checkpoint stops before the event that cannot
+      // be persisted, and every later event waits behind it.
+      await vi.waitFor(async () => expect(await ctx.database.get(
+        'mtproto_qqnt_event_checkpoint', { platformSessionId: checkpointSession.platformSessionId },
+      )).toMatchObject([{ lastEventId: '7002' }]))
+      await expect(ctx.database.get('mtproto_im_conversation', {
+        platformSessionId: checkpointSession.platformSessionId, platformConversationId: conversation.id,
+      })).resolves.toMatchObject([{ title: 'Gr.oup' }])
+      await expect(ctx.database.get('mtproto_im_user', {
+        platformId: 'qqnt', platformUserId: 'peer',
+      })).resolves.toMatchObject([{ firstName: 'Bad.Name' }])
+      const [message] = await ctx.database.get('mtproto_im_message', {
+        platformSessionId: checkpointSession.platformSessionId,
+        primaryPlatformMessageId: 'unstorable-message',
+      })
+      expect(message.text).toBe('@alice. hello')
+      // The mention still covers the same six code units it covered in QQ's text.
+      expect(message.content).toMatchObject({
+        parts: [{
+          type: 'text', text: '@alice. hello',
+          entities: [{ type: 'mention', offset: 0, length: 6, userId: 'alice' }],
+        }],
+      })
+      expect(message.metadata).toMatchObject({ __mtprotoRelaySenderTitle: 're.mark' })
+      const [requestRow] = await ctx.database.get('mtproto_im_request', {
+        platformSessionId: checkpointSession.platformSessionId, platformRequestId: 'unstorable-request',
+      })
+      expect(requestRow.request).toMatchObject({
+        message: 'let me in.',
+        requester: { id: 'requester', firstName: 'Frie.nd' },
+        group: { id: conversation.id, title: 'Gr.oup' },
+      })
+    } finally {
+      await unsubscribe()
+    }
+  })
+})
+
 async function collect(source: AsyncIterable<Uint8Array>): Promise<Buffer> {
   const chunks: Buffer[] = []
   for await (const chunk of source) chunks.push(Buffer.from(chunk))

packages/platform-crossgram/src/platform.test.ts

@@ -72,6 +72,57 @@ describe('QQNTPlatform mapping', () => {
     ])
   })
 
+  it('keeps a mention range and the sender text when QQ hands over a NUL', async () => {
+    const platform = new QQNTPlatform()
+    platform.client.getHistory = vi.fn(async () => ({
+      messages: [{
+        id: 'nul-message', conversationId: 'group', senderId: 'peer', timestamp: 1, outgoing: false,
+        sender: { id: 'peer', name: 'Bad\u0000Name', alias: 're\u0000mark' },
+        parts: [{
+          type: 'text' as const, text: '@alice\u0000 hello',
+          entities: [{ type: 'mention' as const, offset: 0, length: 6, userId: 'alice' }],
+        }],
+      }],
+    }))
+
+    const [message] = (await platform.getHistory(session, { id: 'group' })).messages
+    expect(message.sender).toMatchObject({ firstName: 'Bad.Name' })
+    expect((message.sender?.metadata as { qqName?: string }).qqName).toBe('Bad.Name')
+    expect(message.senderTitle).toBe('re.mark')
+    const part = message.content.parts[0]
+    expect(part).toMatchObject({ type: 'text', text: '@alice. hello' })
+    expect(entitySlices(part as never)).toEqual(['mention:@alice'])
+  })
+
+  it('keeps a group title and a request note free of the units PostgreSQL rejects', async () => {
+    const platform = new QQNTPlatform()
+    platform.client.getConversation = vi.fn(async () => ({
+      id: 'group', kind: 'group' as const, title: 'Gr\u0000oup\uD800',
+      peerUid: 'group', peerUin: '42', chatType: 2 as const,
+    }))
+    platform.client.getRequests = vi.fn(async () => ({ requests: [{
+      id: 'request', kind: 'friend' as const, status: 'pending' as const,
+      requester: { id: 'peer', name: 'Frie\u0000nd' },
+      group: { id: 'group', name: 'Gr\u0000oup' },
+      message: 'let me in\u0000',
+      timestamp: 1,
+    }] }))
+    platform.client.getUser = vi.fn(async () => ({
+      id: 'peer', name: 'Na\u0000me', signature: 'sig\uD800',
+    }))
+
+    await expect(platform.getConversation(session, 'group')).resolves.toMatchObject({ title: 'Gr.oup.' })
+    const [request] = (await platform.getRequests(session)).requests
+    expect(request).toMatchObject({
+      requester: { firstName: 'Frie.nd' },
+      group: { title: 'Gr.oup' },
+      message: 'let me in.',
+    })
+    await expect(platform.getUser(session, 'peer')).resolves.toMatchObject({
+      firstName: 'Na.me', about: 'sig.',
+    })
+  })
+
   it('maps QQNT ranged-read past EOF responses to an empty Telegram chunk', async () => {
     const platform = new QQNTPlatform()
     platform.client.downloadFile = vi.fn(async function* () {