feat: add shouldPush to messages

bench/decode.ts

@@ -22,9 +22,10 @@ const decodeV1 = () => {
       const bob = await newPrivateKeyBundle()
 
       const message = randomBytes(size)
+      const { payload } = await alice.encodeContent(message)
       const encodedMessage = await MessageV1.encode(
         alice.keystore,
-        await alice.encodeContent(message),
+        payload,
         alice.publicKeyBundle,
         bob.getPublicKeyBundle(),
         new Date()
@@ -75,8 +76,8 @@ const decodeV2 = () => {
         new Date(),
         undefined
       )
-      const payload = await alice.encodeContent(message)
-      const encodedMessage = await convo.createMessage(payload)
+      const { payload, shouldPush } = await alice.encodeContent(message)
+      const encodedMessage = await convo.createMessage(payload, shouldPush)
       const messageBytes = encodedMessage.toBytes()
 
       const envelope = {

bench/encode.ts

@@ -22,7 +22,7 @@ const encodeV1 = () => {
 
       // The returned function is the actual benchmark. Everything above is setup
       return async () => {
-        const encodedMessage = await alice.encodeContent(message)
+        const { payload: encodedMessage } = await alice.encodeContent(message)
         await MessageV1.encode(
           alice.keystore,
           encodedMessage,
@@ -57,11 +57,11 @@ const encodeV2 = () => {
         undefined
       )
       const message = randomBytes(size)
-      const payload = await alice.encodeContent(message)
+      const { payload, shouldPush } = await alice.encodeContent(message)
 
       // The returned function is the actual benchmark. Everything above is setup
       return async () => {
-        await convo.createMessage(payload)
+        await convo.createMessage(payload, shouldPush)
       }
     })
   )

package.json

@@ -106,7 +106,7 @@
   },
   "dependencies": {
     "@noble/secp256k1": "^1.5.2",
-    "@xmtp/proto": "^3.37.0-beta.2",
+    "@xmtp/proto": "^3.39.0-beta.2",
     "@xmtp/user-preferences-bindings-wasm": "^0.3.4",
     "async-mutex": "^0.4.0",
     "elliptic": "^6.5.4",

src/Client.ts

@@ -635,7 +635,10 @@ export default class Client<ContentTypes = any> {
   async encodeContent(
     content: ContentTypes,
     options?: SendOptions
-  ): Promise<Uint8Array> {
+  ): Promise<{
+    payload: Uint8Array
+    shouldPush: boolean
+  }> {
     const contentType = options?.contentType || ContentTypeText
     const codec = this.codecFor(contentType)
     if (!codec) {
@@ -651,7 +654,10 @@ export default class Client<ContentTypes = any> {
       encoded.compression = options.compression
     }
     await compress(encoded)
-    return proto.EncodedContent.encode(encoded).finish()
+    return {
+      payload: proto.EncodedContent.encode(encoded).finish(),
+      shouldPush: codec.shouldPush(content),
+    }
   }
 
   async decodeContent(contentBytes: Uint8Array): Promise<{

src/Message.ts

@@ -192,28 +192,32 @@ export class MessageV2 extends MessageBase implements proto.MessageV2 {
   senderAddress: string | undefined
   private header: proto.MessageHeaderV2
   senderHmac: Uint8Array
+  shouldPush: boolean
 
   constructor(
     id: string,
     bytes: Uint8Array,
     obj: proto.Message,
     header: proto.MessageHeaderV2,
-    senderHmac: Uint8Array
+    senderHmac: Uint8Array,
+    shouldPush: boolean
   ) {
     super(id, bytes, obj)
     this.header = header
     this.senderHmac = senderHmac
+    this.shouldPush = shouldPush
   }
 
   static async create(
     obj: proto.Message,
     header: proto.MessageHeaderV2,
     bytes: Uint8Array,
-    senderHmac: Uint8Array
+    senderHmac: Uint8Array,
+    shouldPush: boolean
   ): Promise<MessageV2> {
     const id = bytesToHex(await sha256(bytes))
 
-    return new MessageV2(id, bytes, obj, header, senderHmac)
+    return new MessageV2(id, bytes, obj, header, senderHmac, shouldPush)
   }
 
   get sent(): Date {

src/conversations/Conversation.ts

@@ -284,7 +284,7 @@ export class ConversationV1<ContentTypes>
     } else {
       topics = [topic]
     }
-    const payload = await this.client.encodeContent(content, options)
+    const { payload } = await this.client.encodeContent(content, options)
     const msg = await this.createMessage(payload, recipient, options?.timestamp)
     const msgBytes = msg.toBytes()
 
@@ -393,7 +393,7 @@ export class ConversationV1<ContentTypes>
       topics = [topic]
     }
     const contentType = options?.contentType || ContentTypeText
-    const payload = await this.client.encodeContent(content, options)
+    const { payload } = await this.client.encodeContent(content, options)
     const msg = await this.createMessage(payload, recipient, options?.timestamp)
 
     await this.client.publishEnvelopes(
@@ -602,8 +602,15 @@ export class ConversationV2<ContentTypes>
     content: Exclude<ContentTypes, undefined>,
     options?: SendOptions
   ): Promise<DecodedMessage<ContentTypes>> {
-    const payload = await this.client.encodeContent(content, options)
-    const msg = await this.createMessage(payload, options?.timestamp)
+    const { payload, shouldPush } = await this.client.encodeContent(
+      content,
+      options
+    )
+    const msg = await this.createMessage(
+      payload,
+      shouldPush,
+      options?.timestamp
+    )
 
     const topic = options?.ephemeral ? this.ephemeralTopic : this.topic
 
@@ -637,6 +644,7 @@ export class ConversationV2<ContentTypes>
   async createMessage(
     // Payload is expected to have already gone through `client.encodeContent`
     payload: Uint8Array,
+    shouldPush: boolean,
     timestamp?: Date
   ): Promise<MessageV2> {
     const header: message.MessageHeaderV2 = {
@@ -660,13 +668,14 @@ export class ConversationV2<ContentTypes>
       signedBytes,
       headerBytes
     )
+
     const protoMsg = {
       v1: undefined,
-      v2: { headerBytes, ciphertext, senderHmac },
+      v2: { headerBytes, ciphertext, senderHmac, shouldPush },
     }
     const bytes = message.Message.encode(protoMsg).finish()
 
-    return MessageV2.create(protoMsg, header, bytes, senderHmac)
+    return MessageV2.create(protoMsg, header, bytes, senderHmac, shouldPush)
   }
 
   private async decryptBatch(
@@ -779,8 +788,15 @@ export class ConversationV2<ContentTypes>
     content: any, // eslint-disable-line @typescript-eslint/no-explicit-any
     options?: SendOptions
   ): Promise<PreparedMessage> {
-    const payload = await this.client.encodeContent(content, options)
-    const msg = await this.createMessage(payload, options?.timestamp)
+    const { payload, shouldPush } = await this.client.encodeContent(
+      content,
+      options
+    )
+    const msg = await this.createMessage(
+      payload,
+      shouldPush,
+      options?.timestamp
+    )
     const msgBytes = msg.toBytes()
 
     const topic = options?.ephemeral ? this.ephemeralTopic : this.topic
@@ -827,7 +843,13 @@ export class ConversationV2<ContentTypes>
       throw new Error('topic mismatch')
     }
 
-    return MessageV2.create(msg, header, env.message, msg.v2.senderHmac)
+    return MessageV2.create(
+      msg,
+      header,
+      env.message,
+      msg.v2.senderHmac,
+      msg.v2.shouldPush
+    )
   }
 
   async decodeMessage(

test/Client.test.ts

@@ -6,7 +6,7 @@ import {
   waitForUserContact,
   newLocalHostClientWithCustomWallet,
 } from './helpers'
-import { buildUserContactTopic } from '../src/utils'
+import { EnvelopeWithMessage, buildUserContactTopic } from '../src/utils'
 import Client, { ClientOptions } from '../src/Client'
 import {
   ApiUrls,
@@ -20,15 +20,17 @@ import {
 } from '../src'
 import NetworkKeyManager from '../src/keystore/providers/NetworkKeyManager'
 import TopicPersistence from '../src/keystore/persistence/TopicPersistence'
-import { PrivateKeyBundleV1 } from '../src/crypto'
+import { PrivateKey, PrivateKeyBundleV1 } from '../src/crypto'
 import { Wallet } from 'ethers'
 import { NetworkKeystoreProvider } from '../src/keystore/providers'
 import { PublishResponse } from '@xmtp/proto/ts/dist/types/message_api/v1/message_api.pb'
 import LocalStoragePonyfill from '../src/keystore/persistence/LocalStoragePonyfill'
+import { message } from '@xmtp/proto'
 import { createWalletClient, http } from 'viem'
 import { privateKeyToAccount } from 'viem/accounts'
 import { mainnet } from 'viem/chains'
 import { generatePrivateKey } from 'viem/accounts'
+import { ContentTypeTestKey, TestKeyCodec } from './ContentTypeTestKey'
 
 type TestCase = {
   name: string
@@ -196,11 +198,25 @@ describe('encodeContent', () => {
       173, 229,
     ])
 
-    const payload = await c.encodeContent(uncompressed, {
+    const { payload } = await c.encodeContent(uncompressed, {
       compression: Compression.COMPRESSION_DEFLATE,
     })
     assert.deepEqual(Uint8Array.from(payload), compressed)
   })
+
+  it('returns shouldPush based on content codec', async () => {
+    const alice = await newLocalHostClient()
+    alice.registerCodec(new TestKeyCodec())
+
+    const { shouldPush: result1 } = await alice.encodeContent('gm')
+    expect(result1).toBe(true)
+
+    const key = PrivateKey.generate().publicKey
+    const { shouldPush: result2 } = await alice.encodeContent(key, {
+      contentType: ContentTypeTestKey,
+    })
+    expect(result2).toBe(false)
+  })
 })
 
 describe('canMessage', () => {
@@ -296,6 +312,43 @@ describe('canMessageMultipleBatches', () => {
   })
 })
 
+describe('listEnvelopes', () => {
+  it('has envelopes with senderHmac and shouldPush', async () => {
+    const alice = await newLocalHostClient()
+    const bob = await newLocalHostClient()
+    alice.registerCodec(new TestKeyCodec())
+    const convo = await alice.conversations.newConversation(bob.address)
+    await convo.send('hi')
+    const key = PrivateKey.generate().publicKey
+    await convo.send(key, {
+      contentType: ContentTypeTestKey,
+    })
+
+    const envelopes = await alice.listEnvelopes(
+      convo.topic,
+      (env: EnvelopeWithMessage) => Promise.resolve(env)
+    )
+
+    const msg1 = message.Message.decode(envelopes[0].message)
+    if (!msg1.v2) {
+      throw new Error('unknown message version')
+    }
+    const header1 = message.MessageHeaderV2.decode(msg1.v2.headerBytes)
+    expect(header1.topic).toEqual(convo.topic)
+    expect(msg1.v2.senderHmac).toBeDefined()
+    expect(msg1.v2.shouldPush).toBe(true)
+
+    const msg2 = message.Message.decode(envelopes[1].message)
+    if (!msg2.v2) {
+      throw new Error('unknown message version')
+    }
+    const header2 = message.MessageHeaderV2.decode(msg2.v2.headerBytes)
+    expect(header2.topic).toEqual(convo.topic)
+    expect(msg2.v2.senderHmac).toBeDefined()
+    expect(msg2.v2.shouldPush).toBe(false)
+  })
+})
+
 describe('publishEnvelopes', () => {
   it('can send a valid envelope', async () => {
     const c = await newLocalHostClient()

test/Message.test.ts

@@ -154,7 +154,7 @@ describe('Message', function () {
         env: 'local',
         privateKeyOverride: alice.encode(),
       })
-      const payload = await aliceClient.encodeContent(text)
+      const { payload } = await aliceClient.encodeContent(text)
       const timestamp = new Date()
       const sender = alice.getPublicKeyBundle()
       const recipient = bob.getPublicKeyBundle()