sdk(sync-agent): allow use of worker to unpack stream

packages/playground/package.json

@@ -40,6 +40,7 @@
         "viem": "^2.21.51",
         "vite-plugin-node-polyfills": "^0.22.0",
         "wagmi": "^2.13.0",
+        "workerpool": "^9.2.0",
         "zod": "^3.21.4"
     },
     "devDependencies": {

packages/playground/src/App.tsx

@@ -9,6 +9,8 @@ import { type SyncAgent } from '@river-build/sdk'
 import { router } from './routes'
 import { wagmiConfig } from './config/wagmi'
 import { loadAuth } from './utils/persist-auth'
+import { workerPool } from './utils/workers'
+import { UnpackerWorker } from './utils/unpack-worker'
 
 function App() {
     const [queryClient] = useState(() => new QueryClient())
@@ -19,6 +21,7 @@ function App() {
         if (persistedAuth) {
             connectRiver(persistedAuth.signerContext, {
                 riverConfig: persistedAuth.riverConfig,
+                unpacker: new UnpackerWorker(workerPool),
             }).then((syncAgent) => setSyncAgent(syncAgent))
         }
     }, [persistedAuth])

packages/playground/src/components/dialog/env-switcher.tsx

@@ -16,6 +16,8 @@ import { useNavigate } from 'react-router-dom'
 import { getWeb3Deployment, getWeb3Deployments } from '@river-build/web3'
 import { deleteAuth, storeAuth } from '@/utils/persist-auth'
 import { useEthersSigner } from '@/utils/viem-to-ethers'
+import { workerPool } from '@/utils/workers'
+import { UnpackerWorker } from '@/utils/unpack-worker'
 import { Button } from '../ui/button'
 import {
     Dialog,
@@ -127,6 +129,7 @@ export const RiverEnvSwitcherContent = (props: {
                                                 if (bearerToken) {
                                                     await connectUsingBearerToken(bearerToken, {
                                                         riverConfig,
+                                                        unpacker: new UnpackerWorker(workerPool),
                                                     }).then((sync) => {
                                                         if (sync?.config.context) {
                                                             storeAuth(
@@ -143,6 +146,7 @@ export const RiverEnvSwitcherContent = (props: {
                                                 }
                                                 await connect(signer, {
                                                     riverConfig,
+                                                    unpacker: new UnpackerWorker(workerPool),
                                                 }).then((sync) => {
                                                     if (sync?.config.context) {
                                                         storeAuth(sync?.config.context, riverConfig)

packages/playground/src/utils/unpack-worker.ts

@@ -0,0 +1,73 @@
+import type { Envelope, Miniblock, StreamAndCookie } from '@river-build/proto'
+import {
+    type ParsedEvent,
+    type ParsedMiniblock,
+    type ParsedStreamAndCookie,
+    type ParsedStreamResponse,
+    type UnpackEnvelopeOpts,
+    Unpacker,
+} from '@river-build/sdk'
+import workerpool from 'workerpool'
+
+const unpacker = new Unpacker()
+
+// instance that we can pass to the sync agent
+export class UnpackerWorker {
+    private pool: workerpool.Pool
+
+    constructor(pool: workerpool.Pool) {
+        this.pool = pool
+    }
+
+    unpackStream = async (
+        stream: StreamAndCookie | undefined,
+        opts: UnpackEnvelopeOpts | undefined,
+    ): Promise<ParsedStreamResponse> => {
+        const res = await this.pool.exec('unpackStream', [stream, opts])
+        return res as ParsedStreamResponse
+    }
+
+    unpackStreamEnvelopes = async (
+        stream: StreamAndCookie,
+        opts: UnpackEnvelopeOpts | undefined,
+    ) => {
+        const res = await this.pool.exec('unpackStreamEnvelopes', [stream, opts])
+        return res as ParsedEvent[]
+    }
+    unpackStreamEx = async (miniblocks: Miniblock[], opts: UnpackEnvelopeOpts | undefined) => {
+        const res = await this.pool.exec('unpackStreamEx', [miniblocks, opts])
+        return res as ParsedStreamResponse
+    }
+
+    unpackEnvelope = async (envelope: Envelope, opts: UnpackEnvelopeOpts | undefined) => {
+        const res = await this.pool.exec('unpackEnvelope', [envelope, opts])
+        return res as ParsedEvent
+    }
+
+    unpackEnvelopes = async (envelopes: Envelope[], opts: UnpackEnvelopeOpts | undefined) => {
+        const res = await this.pool.exec('unpackEnvelopes', [envelopes, opts])
+        return res as ParsedEvent[]
+    }
+
+    unpackMiniblock = async (miniblock: Miniblock, opts: UnpackEnvelopeOpts | undefined) => {
+        const res = await this.pool.exec('unpackMiniblock', [miniblock, opts])
+        return res as ParsedMiniblock
+    }
+
+    unpackStreamAndCookie = async (
+        stream: StreamAndCookie,
+        opts: UnpackEnvelopeOpts | undefined,
+    ) => {
+        const res = await this.pool.exec('unpackStreamAndCookie', [stream, opts])
+        return res as ParsedStreamAndCookie
+    }
+}
+
+workerpool.worker({
+    unpackStream: unpacker.unpackStream,
+    unpackStreamEnvelopes: unpacker.unpackStreamEnvelopes,
+    unpackEnvelope: unpacker.unpackEnvelope,
+    unpackEnvelopes: unpacker.unpackEnvelopes,
+    unpackStreamAndCookie: unpacker.unpackStreamAndCookie,
+    unpackMiniblock: unpacker.unpackMiniblock,
+})

packages/playground/src/utils/workers.ts

@@ -0,0 +1,8 @@
+import workerpool from 'workerpool'
+import workerURL from './unpack-worker?worker&url'
+
+export const workerPool = workerpool.pool(workerURL, {
+    workerOpts: {
+        type: 'module',
+    },
+})

packages/playground/vite.config.ts

@@ -6,6 +6,22 @@ import { replaceCodePlugin } from 'vite-plugin-replace'
 import tsconfigPaths from 'vite-tsconfig-paths'
 import path from 'path'
 
+const fixExport = replaceCodePlugin({
+    replacements: [
+        {
+            from: `if ((crypto && crypto.getRandomValues) || !process.browser) {
+  exports.randomFill = randomFill
+  exports.randomFillSync = randomFillSync
+} else {
+  exports.randomFill = oldBrowser
+  exports.randomFillSync = oldBrowser
+}`,
+            to: `exports.randomFill = randomFill
+exports.randomFillSync = randomFillSync`,
+        },
+    ],
+})
+
 // https://vitejs.dev/config/
 export default ({ mode }: { mode: string }) => {
     const env = loadEnv(mode, process.cwd(), '')
@@ -20,21 +36,7 @@ export default ({ mode }: { mode: string }) => {
         },
         plugins: [
             tsconfigPaths(),
-            replaceCodePlugin({
-                replacements: [
-                    {
-                        from: `if ((crypto && crypto.getRandomValues) || !process.browser) {
-  exports.randomFill = randomFill
-  exports.randomFillSync = randomFillSync
-} else {
-  exports.randomFill = oldBrowser
-  exports.randomFillSync = oldBrowser
-}`,
-                        to: `exports.randomFill = randomFill
-exports.randomFillSync = randomFillSync`,
-                    },
-                ],
-            }),
+            fixExport,
             checker({
                 typescript: true,
                 eslint: {
@@ -44,6 +46,9 @@ exports.randomFillSync = randomFillSync`,
             nodePolyfills(),
             react(),
         ],
+        worker: {
+            plugins: () => [fixExport],
+        },
         resolve: {
             alias: {
                 '@': path.resolve(__dirname, './src'),

packages/sdk/src/client.ts

@@ -82,14 +82,7 @@ import {
     contractAddressFromSpaceId,
     isUserId,
 } from './id'
-import {
-    checkEventSignature,
-    makeEvent,
-    UnpackEnvelopeOpts,
-    unpackMiniblock,
-    unpackStream,
-    unpackStreamEx,
-} from './sign'
+import { checkEventSignature, makeEvent, UnpackEnvelopeOpts } from './sign'
 import { StreamEvents } from './streamEvents'
 import { IStreamStateView, StreamStateView } from './streamStateView'
 import {
@@ -149,6 +142,7 @@ import { SyncedStreamsExtension } from './syncedStreamsExtension'
 import { SignerContext } from './signerContext'
 import { decryptAESGCM, deriveKeyAndIV, encryptAESGCM, uint8ArrayToBase64 } from './crypto_utils'
 import { makeTags } from './tags'
+import { Unpacker } from './unpacker'
 
 export type ClientEvents = StreamEvents & DecryptionEvents
 
@@ -195,6 +189,7 @@ export class Client
     private syncedStreamsExtensions?: SyncedStreamsExtension
     private persistenceStore: IPersistenceStore
     private validatedEvents: Record<string, { isValid: boolean; reason?: string }> = {}
+    private unpacker: Unpacker
 
     constructor(
         signerContext: SignerContext,
@@ -204,6 +199,7 @@ export class Client
         persistenceStoreName?: string,
         logNamespaceFilter?: string,
         highPriorityStreamIds?: string[],
+        unpacker?: Unpacker,
         unpackEnvelopeOpts?: UnpackEnvelopeOpts,
     ) {
         super()
@@ -224,6 +220,7 @@ export class Client
         this.rpcClient = rpcClient
         this.unpackEnvelopeOpts = unpackEnvelopeOpts
         this.userId = userIdFromAddress(signerContext.creatorAddress)
+        this.unpacker = unpacker ?? new Unpacker()
 
         const shortId = shortenHexString(
             this.userId.startsWith('0x') ? this.userId.slice(2) : this.userId,
@@ -457,7 +454,7 @@ export class Client
             optional: true,
         })
         if (response.stream) {
-            return unpackStream(response.stream, this.unpackEnvelopeOpts)
+            return this.unpacker.unpackStream(response.stream, this.unpackEnvelopeOpts)
         } else {
             return undefined
         }
@@ -480,7 +477,7 @@ export class Client
             streamId: streamIdAsBytes(userStreamId),
             metadata: metadata,
         })
-        return unpackStream(response.stream, this.unpackEnvelopeOpts)
+        return this.unpacker.unpackStream(response.stream, this.unpackEnvelopeOpts)
     }
 
     private async createUserMetadataStream(
@@ -501,7 +498,7 @@ export class Client
             streamId: streamIdAsBytes(userMetadataStreamId),
             metadata: metadata,
         })
-        return unpackStream(response.stream, this.unpackEnvelopeOpts)
+        return this.unpacker.unpackStream(response.stream, this.unpackEnvelopeOpts)
     }
 
     private async createUserInboxStream(
@@ -522,7 +519,7 @@ export class Client
             streamId: streamIdAsBytes(userInboxStreamId),
             metadata: metadata,
         })
-        return unpackStream(response.stream, this.unpackEnvelopeOpts)
+        return this.unpacker.unpackStream(response.stream, this.unpackEnvelopeOpts)
     }
 
     private async createUserSettingsStream(
@@ -544,7 +541,7 @@ export class Client
             streamId: userSettingsStreamId,
             metadata: metadata,
         })
-        return unpackStream(response.stream, this.unpackEnvelopeOpts)
+        return this.unpacker.unpackStream(response.stream, this.unpackEnvelopeOpts)
     }
 
     private async createStreamAndSync(
@@ -563,7 +560,10 @@ export class Client
                 // fetch the stream to get the client in the rigth state
                 response = await this.rpcClient.getStream({ streamId: request.streamId })
             }
-            const unpacked = await unpackStream(response.stream, this.unpackEnvelopeOpts)
+            const unpacked = await this.unpacker.unpackStream(
+                response.stream,
+                this.unpackEnvelopeOpts,
+            )
             await stream.initializeFromResponse(unpacked)
             if (stream.view.syncCookie) {
                 await this.streams.addStreamToSync(stream.view.syncCookie)
@@ -780,7 +780,10 @@ export class Client
             streamId: streamIdAsBytes(streamId),
         })
 
-        const unpackedResponse = await unpackStream(response.stream, this.unpackEnvelopeOpts)
+        const unpackedResponse = await this.unpacker.unpackStream(
+            response.stream,
+            this.unpackEnvelopeOpts,
+        )
         const streamView = new StreamStateView(this.userId, streamId)
         streamView.initialize(
             unpackedResponse.streamAndCookie.nextSyncCookie,
@@ -1189,7 +1192,10 @@ export class Client
             const response = await this.rpcClient.getStream({
                 streamId: streamIdAsBytes(streamId),
             })
-            const unpackedResponse = await unpackStream(response.stream, this.unpackEnvelopeOpts)
+            const unpackedResponse = await this.unpacker.unpackStream(
+                response.stream,
+                this.unpackEnvelopeOpts,
+            )
             return this.streamViewFromUnpackedResponse(streamId, unpackedResponse)
         } catch (err) {
             this.logCall('getStream', streamId, 'ERROR', err)
@@ -1268,7 +1274,10 @@ export class Client
                     )}.`,
                 )
             }
-            const unpackedResponse = await unpackStreamEx(miniblocks, this.unpackEnvelopeOpts)
+            const unpackedResponse = await this.unpacker.unpackStreamEx(
+                miniblocks,
+                this.unpackEnvelopeOpts,
+            )
             return this.streamViewFromUnpackedResponse(streamId, unpackedResponse)
         } catch (err) {
             this.logCall('getStreamEx', streamId, 'ERROR', err)
@@ -1339,7 +1348,10 @@ export class Client
                     const response = await this.rpcClient.getStream({
                         streamId: streamIdAsBytes(streamId),
                     })
-                    const unpacked = await unpackStream(response.stream, this.unpackEnvelopeOpts)
+                    const unpacked = await this.unpacker.unpackStream(
+                        response.stream,
+                        this.unpackEnvelopeOpts,
+                    )
                     this.logCall('initStream calling initializingFromResponse', streamId)
                     await stream.initializeFromResponse(unpacked)
                     if (stream.view.syncCookie) {
@@ -1917,7 +1929,10 @@ export class Client
 
         const unpackedMiniblocks: ParsedMiniblock[] = []
         for (const miniblock of response.miniblocks) {
-            const unpackedMiniblock = await unpackMiniblock(miniblock, this.unpackEnvelopeOpts)
+            const unpackedMiniblock = await this.unpacker.unpackMiniblock(
+                miniblock,
+                this.unpackEnvelopeOpts,
+            )
             unpackedMiniblocks.push(unpackedMiniblock)
         }
         await this.persistenceStore.saveMiniblocks(

packages/sdk/src/index.ts

@@ -93,4 +93,5 @@ export * from './syncedStreamsLoop'
 export * from './tags'
 export * from './types'
 export * from './unauthenticatedClient'
+export * from './unpacker'
 export * from './utils'