Merge pull request #458 from xmtp/beta

v11 Release

.github/workflows/release.yml

@@ -26,9 +26,6 @@ jobs:
         uses: actions/setup-node@v3
         with:
           node-version-file: '.nvmrc'
-      - name: Use latest NPM
-        # Ensure we are using the latest version of NPM to get support for provenance
-        run: npm i -g npm
       - name: Install dependencies
         run: npm ci
       - name: Verify the integrity of provenance attestations and registry signatures for installed dependencies

bench/encode.ts

@@ -17,7 +17,7 @@ const encodeV1 = () => {
       const alice = await Client.create(newWallet(), { env: 'local' })
       const bobKeys = (await newPrivateKeyBundle()).getPublicKeyBundle()
 
-      const message = randomBytes(size)
+      const message = randomBytes(size).toString()
       const timestamp = new Date()
 
       // The returned function is the actual benchmark. Everything above is setup

package.json

@@ -28,6 +28,7 @@
     "clean:proto": "rimraf -g src/proto/*.ts",
     "package": "npm pack",
     "prepublishOnly": "npm run build",
+    "updateSnapVersion": "npm view @xmtp/snap --json | jq '{\"version\": .version, \"package\": .name}' > ./src/snapInfo.json",
     "test:setup": "./dev/up",
     "test:teardown": "./dev/down",
     "test": "npm run test:node",
@@ -83,7 +84,7 @@
   },
   "dependencies": {
     "@noble/secp256k1": "^1.5.2",
-    "@xmtp/proto": "^3.24.0",
+    "@xmtp/proto": "^3.28.0-beta.1",
     "async-mutex": "^0.4.0",
     "elliptic": "^6.5.4",
     "ethers": "^5.5.3",
@@ -92,6 +93,7 @@
   "devDependencies": {
     "@commitlint/cli": "^16.1.0",
     "@commitlint/config-conventional": "^16.0.0",
+    "@metamask/providers": "^11.1.1",
     "@types/benchmark": "^2.1.2",
     "@types/bl": "^5.0.2",
     "@types/callback-to-async-iterator": "^1.1.4",
@@ -113,12 +115,12 @@
     "eslint-plugin-prettier": "^4.0.0",
     "eslint-plugin-promise": "^6.0.1",
     "husky": "^7.0.4",
-    "jest": "^28.1.3",
+    "jest": "^29.6.0",
     "jest-environment-jsdom": "^28.1.3",
     "prettier": "^2.4.0",
     "rimraf": "^5.0.0",
     "semantic-release": "^21.0.3",
-    "ts-jest": "^28.0.0",
+    "ts-jest": "^29.1.1",
     "ts-node": "^10.9.1",
     "tsup": "^6.7.0",
     "typedoc": "^0.22.11",

src/ApiClient.ts

@@ -1,6 +1,6 @@
 import { messageApi } from '@xmtp/proto'
 import { NotifyStreamEntityArrival } from '@xmtp/proto/ts/dist/types/fetch.pb'
-import { retry, sleep, toNanoString } from './utils'
+import { b64Decode, retry, sleep, toNanoString } from './utils'
 import AuthCache from './authn/AuthCache'
 import { Authenticator } from './authn'
 import packageJson from '../package.json'
@@ -92,6 +92,13 @@ export type SubscribeCallback = NotifyStreamEntityArrival<messageApi.Envelope>
 
 export type UnsubscribeFn = () => Promise<void>
 
+export type UpdateContentTopics = (topics: string[]) => Promise<void>
+
+export type SubscriptionManager = {
+  unsubscribe: UnsubscribeFn
+  updateContentTopics?: UpdateContentTopics
+}
+
 export type OnConnectionLostCallback = () => void
 
 const isAbortError = (err?: Error): boolean => {
@@ -115,11 +122,47 @@ const isAuthError = (err?: GrpcError | Error): boolean => {
 // eslint-disable-next-line @typescript-eslint/no-explicit-any
 const isNotAuthError = (err?: Error): boolean => !isAuthError(err)
 
+export interface ApiClient {
+  query(
+    params: QueryParams,
+    options: QueryAllOptions
+  ): Promise<messageApi.Envelope[]>
+  queryIterator(
+    params: QueryParams,
+    options: QueryStreamOptions
+  ): AsyncGenerator<messageApi.Envelope>
+  queryIteratePages(
+    params: QueryParams,
+    options: QueryStreamOptions
+  ): AsyncGenerator<messageApi.Envelope[]>
+  subscribe(
+    params: SubscribeParams,
+    callback: SubscribeCallback,
+    onConnectionLost?: OnConnectionLostCallback
+  ): SubscriptionManager
+  publish(messages: PublishParams[]): ReturnType<typeof MessageApi.Publish>
+  batchQuery(queries: Query[]): Promise<messageApi.Envelope[][]>
+  setAuthenticator(
+    authenticator: Authenticator,
+    cacheExpirySeconds?: number
+  ): void
+}
+
+const normalizeEnvelope = (env: messageApi.Envelope): messageApi.Envelope => {
+  if (!env.message || !env.message.length) {
+    return env
+  }
+  if (typeof env.message === 'string') {
+    env.message = b64Decode(env.message)
+  }
+  return env
+}
+
 /**
  * ApiClient provides a wrapper for calling the GRPC Gateway generated code.
  * It adds some helpers for dealing with paginated data and automatically retries idempotent calls
  */
-export default class ApiClient {
+export default class HttpApiClient implements ApiClient {
   pathPrefix: string
   maxRetries: number
   private authCache?: AuthCache
@@ -220,7 +263,7 @@ export default class ApiClient {
     req: messageApi.SubscribeRequest,
     cb: NotifyStreamEntityArrival<messageApi.Envelope>,
     onConnectionLost?: OnConnectionLostCallback
-  ): UnsubscribeFn {
+  ): SubscriptionManager {
     const abortController = new AbortController()
 
     const doSubscribe = async () => {
@@ -262,8 +305,10 @@ export default class ApiClient {
     }
     doSubscribe()
 
-    return async () => {
-      abortController?.abort()
+    return {
+      unsubscribe: async () => {
+        abortController?.abort()
+      },
     }
   }
 
@@ -334,7 +379,7 @@ export default class ApiClient {
       })
 
       if (result.envelopes?.length) {
-        yield result.envelopes
+        yield result.envelopes.map(normalizeEnvelope)
       } else {
         return
       }
@@ -395,7 +440,7 @@ export default class ApiClient {
       }
       for (const queryResponse of batchResponse.responses) {
         if (queryResponse.envelopes) {
-          allEnvelopes.push(queryResponse.envelopes)
+          allEnvelopes.push(queryResponse.envelopes.map(normalizeEnvelope))
         } else {
           // If no envelopes provided, then add an empty list
           allEnvelopes.push([])
@@ -438,12 +483,16 @@ export default class ApiClient {
     params: SubscribeParams,
     callback: SubscribeCallback,
     onConnectionLost?: OnConnectionLostCallback
-  ): UnsubscribeFn {
+  ): SubscriptionManager {
     if (!params.contentTopics.length) {
       throw new Error('Must provide list of contentTopics to subscribe to')
     }
 
-    return this._subscribe(params, callback, onConnectionLost)
+    return this._subscribe(
+      params,
+      (env) => callback(normalizeEnvelope(env)),
+      onConnectionLost
+    )
   }
 
   private getToken(): Promise<string> {