feat: XIP-5 support different payload content types (#68)

BREAKING CHANGE: The protocol format of messages is changing, old clients won't be able to decode new messages correctly and new clients won't be able to decode old messages correctly. The API changes are backward compatible.

README.md

@@ -149,7 +149,7 @@ const newConversation = await xmtp.conversations.newConversation(
 
 #### Sending messages
 
-To be able to send a message, the recipient must have already started their Client at least once and consequently advertised their key bundle on the network. Messages are addressed using wallet addresses. The message payload is a string but neither the SDK nor the network put any constraints on its contents or interpretation.
+To be able to send a message, the recipient must have already started their Client at least once and consequently advertised their key bundle on the network. Messages are addressed using wallet addresses. The message payload can be a plain string, but other types of content can be supported through the use of SendOptions (see [Different types of content](#different-types-of-content) for more details)
 
 ```ts
 const conversation = await xmtp.conversations.newConversation(
@@ -197,6 +197,35 @@ for await (const message of conversation.streamMessages()) {
 }
 ```
 
+#### Different types of content
+
+All the send functions support `SendOptions` as an optional parameter. Option `contentType` allows specifying different types of content than the default simple string, which is identified with content type identifier `ContentTypeText`. Support for other types of content can be added by registering additional `ContentCodecs` with the `Client`. Every codec is associated with a content type identifier, `ContentTypeId`, which is used to signal to the Client which codec should be used to process the content that is being sent or received. See XIP-5 for more details on Codecs and content types, new Codecs and content types are defined through XRCs.
+
+If there is a concern that the recipient may not be able to handle particular content type, the sender can use `contentFallback` option to provide a string that describes the content being sent. If the recipient fails to decode the original content, the fallback will replace it and can be used to inform the recipient what the original content was.
+
+```ts
+// Assuming we've loaded a fictional NumberCodec that can be used to encode numbers,
+// and is identified with ContentTypeNumber, we can use it as follows.
+
+xmtp.registerCodec:(new NumberCodec())
+conversation.send(3.14, {
+  contentType: ContentTypeNumber,
+  contentFallback: 'sending you a pie'
+})
+```
+
+#### Compression
+
+Message content can be optionally compressed using the `compression` option. The value of the option is the name of the compression algorithm to use. Currently supported are `gzip` and `deflate`. Compression is applied to the bytes produced by the content codec.
+
+Content will be decompressed transparently on the receiving end. Note that `Client` enforces maximum content size. The default limit can be overridden through the `ClientOptions`. Consequently a message that would expand beyond that limit on the receiving end will fail to decode.
+
+```ts
+conversation.send('#'.repeat(1000), {
+  compression: 'deflate',
+})
+```
+
 #### Under the hood
 
 Using `xmtp.conversations` hides the details of this, but for the curious this is how sending a message on XMTP works. The first message and first response between two parties is sent to three separate [Waku](https://rfc.vac.dev/spec/10/) content topics:

package.json

@@ -61,6 +61,7 @@
   },
   "dependencies": {
     "@noble/secp256k1": "^1.5.2",
+    "@stardazed/streams-polyfill": "^2.4.0",
     "cross-fetch": "^3.1.5",
     "ethers": "^5.5.3",
     "js-waku": "^0.18",

src/Client.ts

@@ -14,6 +14,21 @@ import Stream, { messageStream } from './Stream'
 import { Signer } from 'ethers'
 import { EncryptedStore, LocalStorageStore, PrivateTopicStore } from './store'
 import { Conversations } from './conversations'
+import {
+  ContentTypeId,
+  EncodedContent,
+  ContentCodec,
+  ContentTypeText,
+  TextCodec,
+  decompress,
+  compress,
+  ContentTypeFallback,
+} from './MessageContent'
+import { Compression } from './proto/messaging'
+import * as proto from './proto/messaging'
+
+/* eslint-disable @typescript-eslint/explicit-module-boundary-types */
+/* eslint-disable @typescript-eslint/no-explicit-any */
 
 const NODES_LIST_URL = 'https://nodes.xmtp.com/'
 
@@ -23,6 +38,9 @@ type NodesList = {
   testnet: Nodes
 }
 
+// Default maximum allowed content size
+const MaxContentSize = 100 * 1024 * 1024 // 100M
+
 // Parameters for the listMessages functions
 export type ListMessagesOptions = {
   pageSize?: number
@@ -35,6 +53,14 @@ export enum KeyStoreType {
   localStorage,
 }
 
+// Parameters for the send functions
+export { Compression }
+export type SendOptions = {
+  contentType: ContentTypeId
+  contentFallback?: string
+  compression?: Compression
+}
+
 /**
  * Network startup options
  */
@@ -50,6 +76,15 @@ type NetworkOptions = {
   waitForPeersTimeoutMs: number
 }
 
+type ContentOptions = {
+  // Allow configuring codecs for additional content types
+  codecs: ContentCodec<any>[]
+
+  // Set the maximum content size in bytes that is allowed by the Client.
+  // Currently only checked when decompressing compressed content.
+  maxContentSize: number
+}
+
 type KeyStoreOptions = {
   /** Specify the keyStore which should be used for loading or saving privateKeyBundles */
   keyStoreType: KeyStoreType
@@ -59,20 +94,24 @@ type KeyStoreOptions = {
  * Aggregate type for client options. Optional properties are used when the default value is calculated on invocation, and are computed
  * as needed by each function. All other defaults are specified in defaultOptions.
  */
-export type ClientOptions = NetworkOptions & KeyStoreOptions
+export type ClientOptions = NetworkOptions & KeyStoreOptions & ContentOptions
 
 /**
  * Provide a default client configuration. These settings can be used on their own, or as a starting point for custom configurations
  *
  * @param opts additional options to override the default settings
  */
 export function defaultOptions(opts?: Partial<ClientOptions>): ClientOptions {
-  const _defaultOptions = {
+  const _defaultOptions: ClientOptions = {
     keyStoreType: KeyStoreType.networkTopicStoreV1,
     env: 'testnet',
     waitForPeersTimeoutMs: 10000,
+    codecs: [new TextCodec()],
+    maxContentSize: MaxContentSize,
+  }
+  if (opts?.codecs) {
+    opts.codecs = _defaultOptions.codecs.concat(opts.codecs)
   }
-
   return { ..._defaultOptions, ...opts } as ClientOptions
 }
 
@@ -87,6 +126,8 @@ export default class Client {
   private contacts: Set<string> // address which we have connected to
   private knownPublicKeyBundles: Map<string, PublicKeyBundle> // addresses and key bundles that we have witnessed
   private _conversations: Conversations
+  private _codecs: Map<string, ContentCodec<any>>
+  private _maxContentSize: number
 
   constructor(waku: Waku, keys: PrivateKeyBundle) {
     this.waku = waku
@@ -95,6 +136,8 @@ export default class Client {
     this.keys = keys
     this.address = keys.identityKey.publicKey.walletSignatureAddress()
     this._conversations = new Conversations(this)
+    this._codecs = new Map()
+    this._maxContentSize = MaxContentSize
   }
 
   /**
@@ -119,6 +162,10 @@ export default class Client {
     const keyStore = createKeyStoreFromConfig(options, wallet, waku)
     const keys = await loadOrCreateKeys(wallet, keyStore)
     const client = new Client(waku, keys)
+    options.codecs.forEach((codec) => {
+      client.registerCodec(codec)
+    })
+    client._maxContentSize = options.maxContentSize
     await client.publishUserContact()
     return client
   }
@@ -192,7 +239,11 @@ export default class Client {
   /**
    * Send a message to the wallet identified by @peerAddress
    */
-  async sendMessage(peerAddress: string, msgString: string): Promise<void> {
+  async sendMessage(
+    peerAddress: string,
+    content: any,
+    options?: SendOptions
+  ): Promise<void> {
     let topics: string[]
     const recipient = await this.getUserContact(peerAddress)
 
@@ -213,7 +264,7 @@ export default class Client {
       topics = [buildDirectMessageTopic(this.address, peerAddress)]
     }
     const timestamp = new Date()
-    const msg = await Message.encode(this.keys, recipient, msgString, timestamp)
+    const msg = await this.encodeMessage(recipient, timestamp, content, options)
     await Promise.all(
       topics.map(async (topic) => {
         const wakuMsg = await WakuMessage.fromBytes(msg.toBytes(), topic, {
@@ -231,6 +282,75 @@ export default class Client {
     }
   }
 
+  registerCodec(codec: ContentCodec<any>): void {
+    const id = codec.contentType
+    const key = `${id.authorityId}/${id.typeId}`
+    this._codecs.set(key, codec)
+  }
+
+  codecFor(contentType: ContentTypeId): ContentCodec<any> | undefined {
+    const key = `${contentType.authorityId}/${contentType.typeId}`
+    const codec = this._codecs.get(key)
+    if (!codec) {
+      return undefined
+    }
+    if (contentType.versionMajor > codec.contentType.versionMajor) {
+      return undefined
+    }
+    return codec
+  }
+
+  async encodeMessage(
+    recipient: PublicKeyBundle,
+    timestamp: Date,
+    content: any,
+    options?: SendOptions
+  ): Promise<Message> {
+    const contentType = options?.contentType || ContentTypeText
+    const codec = this.codecFor(contentType)
+    if (!codec) {
+      throw new Error('unknown content type ' + contentType)
+    }
+    const encoded = codec.encode(content, this)
+    if (options?.contentFallback) {
+      encoded.fallback = options.contentFallback
+    }
+    if (options?.compression) {
+      encoded.compression = options.compression
+    }
+    await compress(encoded)
+    const payload = proto.EncodedContent.encode(encoded).finish()
+    return Message.encode(this.keys, recipient, payload, timestamp)
+  }
+
+  async decodeMessage(payload: Uint8Array): Promise<Message> {
+    const message = await Message.decode(this.keys, payload)
+    if (message.error) {
+      return message
+    }
+    if (!message.decrypted) {
+      throw new Error('decrypted bytes missing')
+    }
+    const encoded = proto.EncodedContent.decode(message.decrypted)
+    await decompress(encoded, this._maxContentSize)
+    if (!encoded.type) {
+      throw new Error('missing content type')
+    }
+    const contentType = new ContentTypeId(encoded.type)
+    const codec = this.codecFor(contentType)
+    if (codec) {
+      message.content = codec.decode(encoded as EncodedContent, this)
+      message.contentType = contentType
+    } else {
+      message.error = new Error('unknown content type ' + contentType)
+      if (encoded.fallback) {
+        message.content = encoded.fallback
+        message.contentType = ContentTypeFallback
+      }
+    }
+    return message
+  }
+
   streamIntroductionMessages(): Stream<Message> {
     return this.streamMessages(buildUserIntroTopic(this.address))
   }
@@ -293,7 +413,7 @@ export default class Client {
       wakuMsgs
         .filter((wakuMsg) => wakuMsg?.payload)
         .map(async (wakuMsg) =>
-          Message.decode(this.keys, wakuMsg.payload as Uint8Array)
+          this.decodeMessage(wakuMsg.payload as Uint8Array)
         )
     )
   }

src/Message.ts

@@ -10,6 +10,7 @@ import {
 import { NoMatchingPreKeyError } from './crypto/errors'
 import { bytesToHex } from './crypto/utils'
 import { sha256 } from './crypto/encryption'
+import { ContentTypeId } from './MessageContent'
 
 const extractV1Message = (msg: proto.Message): proto.V1Message => {
   if (!msg.v1) {
@@ -22,10 +23,14 @@ const extractV1Message = (msg: proto.Message): proto.V1Message => {
 // Message header carries the sender and recipient keys used to protect message.
 // Message timestamp is set by the sender.
 export default class Message implements proto.V1Message {
-  header: proto.MessageHeader | undefined // eslint-disable-line camelcase
+  header: proto.MessageHeader // eslint-disable-line camelcase
   headerBytes: Uint8Array // encoded header bytes
-  ciphertext: Ciphertext | undefined
-  decrypted: string | undefined
+  ciphertext: Ciphertext
+  decrypted?: Uint8Array
+  // content allows attaching decoded content to the Message
+  // the message receiving APIs need to return a Message to provide access to the header fields like sender/recipient
+  contentType?: ContentTypeId
+  content?: any // eslint-disable-line @typescript-eslint/no-explicit-any
   error?: Error
   /**
    * Identifier that is deterministically derived from the bytes of the message
@@ -46,9 +51,10 @@ export default class Message implements proto.V1Message {
     this.bytes = bytes
     this.headerBytes = msg.headerBytes
     this.header = header
-    if (msg.ciphertext) {
-      this.ciphertext = new Ciphertext(msg.ciphertext)
+    if (!msg.ciphertext) {
+      throw new Error('missing message ciphertext')
     }
+    this.ciphertext = new Ciphertext(msg.ciphertext)
   }
 
   toBytes(): Uint8Array {
@@ -71,10 +77,6 @@ export default class Message implements proto.V1Message {
     return Message.create(msg, header, bytes)
   }
 
-  get text(): string | undefined {
-    return this.decrypted
-  }
-
   get sent(): Date | undefined {
     return this.header ? new Date(this.header?.timestamp) : undefined
   }
@@ -103,11 +105,9 @@ export default class Message implements proto.V1Message {
   static async encode(
     sender: PrivateKeyBundle,
     recipient: PublicKeyBundle,
-    message: string,
+    message: Uint8Array,
     timestamp: Date
   ): Promise<Message> {
-    const msgBytes = new TextEncoder().encode(message)
-
     const secret = await sender.sharedSecret(
       recipient,
       sender.getCurrentPreKey().publicKey,
@@ -120,7 +120,7 @@ export default class Message implements proto.V1Message {
       timestamp: timestamp.getTime(),
     }
     const headerBytes = proto.MessageHeader.encode(header).finish()
-    const ciphertext = await encrypt(msgBytes, secret, headerBytes)
+    const ciphertext = await encrypt(message, secret, headerBytes)
     const protoMsg = { v1: { headerBytes: headerBytes, ciphertext } }
     const bytes = proto.Message.encode(protoMsg).finish()
     const msg = await Message.create(protoMsg, header, bytes)
@@ -188,8 +188,7 @@ export default class Message implements proto.V1Message {
       msg.error = e
       return msg
     }
-    bytes = await decrypt(ciphertext, secret, v1Message.headerBytes)
-    msg.decrypted = new TextDecoder().decode(bytes)
+    msg.decrypted = await decrypt(ciphertext, secret, v1Message.headerBytes)
     return msg
   }
 }