refactor(sdk)!: Stateless Stream (#2883)

**This is a breaking changes as this changes the API** 

`Stream` instances no longer have metadata field. Instead the metadata
is queried from `StreamRegistry`.
- `StreamRegistry` already caches the metadata. Now we explicity
populate the cache e.g. in `searchStreams()` (which queries `id` +
`metadata`) from The Graph).

## API

These `Stream` methods are now async:
- `getMetadata()`
- `getPartitionCount()`
- `getDescription()`
- `getStorageDayCount()`

Also added `StreamrClient#getStreamMetadata()` method which is used by
`Stream#getMetadata()`.

## Testing

Added new end-to-end tests to assert cache population logic. The new
`ProxyHttpServer` helper class which is needed by those tests.

Author: teogeb

CHANGELOG.md

@@ -16,10 +16,11 @@ Changes before Tatum release are not documented in this file.
   - it is supported for `PUBLISH` and `SUBSCRIBE` permissions
   - new `StreamrClient#getUserId()` method
 - Method `StreamrClient#getDiagnosticInfo()` provides diagnostic info about network (https://github.com/streamr-dev/network/pull/2740, https://github.com/streamr-dev/network/pull/2741)
-- Add accessors for stream metadata fields: (https://github.com/streamr-dev/network/pull/2825, https://github.com/streamr-dev/network/pull/2845)
+- Add accessors for stream metadata fields: (https://github.com/streamr-dev/network/pull/2825, https://github.com/streamr-dev/network/pull/2845, https://github.com/streamr-dev/network/pull/2883)
   - `Stream#getPartitionCount()`
   - `Stream#getDescription()` and `Stream#setDescription()`
   - `Stream#getStorageDayCount()` and `Stream#setStorageDayCount()`
+- Add method `StreamrClient#getStreamMetadata()` (https://github.com/streamr-dev/network/pull/2883)
 - Add validation for public permissions (https://github.com/streamr-dev/network/pull/2819)
 - Add `opts` parameter to `StreamrClient#addStreamToStorageNode` (https://github.com/streamr-dev/network/pull/2858)
   - controls how long to wait for storage node to pick up on assignment
@@ -40,6 +41,7 @@ Changes before Tatum release are not documented in this file.
 - **BREAKING CHANGE:** Replace methods `StreamrClient#updateStream()` and `Stream#update()`: (https://github.com/streamr-dev/network/pull/2826, https://github.com/streamr-dev/network/pull/2855, https://github.com/streamr-dev/network/pull/2859, https://github.com/streamr-dev/network/pull/2862)
   - use `StreamrClient#setStreamMetadata()` and `Stream#setMetadata()` instead
   - both methods overwrite metadata instead of merging it
+- **BREAKING CHANGE:** Methods `Stream#getMetadata()` and `Stream#getStreamParts()` are async (https://github.com/streamr-dev/network/pull/2883)
 - Caching changes:
   - storage node addresses (https://github.com/streamr-dev/network/pull/2877, https://github.com/streamr-dev/network/pull/2878)
   - stream metadata and permissions (https://github.com/streamr-dev/network/pull/2889)

docs/docs/usage/streams/partitioning.md

@@ -38,7 +38,7 @@ const stream = await streamr.createStream({
 });
 console.log(
   `Stream created: ${stream.id}. It has ${
-    stream.getPartitionCount()
+    await stream.getPartitionCount()
   } partitions.`
 );
 ```

packages/cli-tools/bin/streamr-storage-node-list-streams.ts

@@ -8,12 +8,12 @@ import { createClientCommand } from '../src/command'
 createClientCommand((async (client: StreamrClient, storageNodeAddress: string) => {
     const { streams } = await client.getStoredStreams(storageNodeAddress)
     if (streams.length > 0) {
-        console.info(EasyTable.print(streams.map((stream) => {
+        console.info(EasyTable.print(await Promise.all(streams.map(async (stream) => {
             return {
                 id: stream.id,
-                partitions: stream.getPartitionCount()
+                partitions: await stream.getPartitionCount()
             }
-        })))
+        }))))
     }
 }))
     .arguments('<storageNodeAddress>')

packages/cli-tools/bin/streamr-stream-create.ts

@@ -19,7 +19,7 @@ createClientCommand(async (client: StreamrClient, streamIdOrPath: string, option
         partitions: options.partitions
     }
     const stream = await client.createStream(body)
-    console.info(JSON.stringify({ id: stream.id, ...stream.getMetadata() }, null, 2))
+    console.info(JSON.stringify({ id: stream.id, ...await stream.getMetadata() }, null, 2))
 })
     .arguments('<streamId>')
     .description('create a new stream')

packages/cli-tools/bin/streamr-stream-show.ts

@@ -19,7 +19,7 @@ const withRenamedField = (obj: any, from: string, to: string) => {
 
 createClientCommand(async (client: StreamrClient, streamId: string, options: Options) => {
     const stream = await client.getStream(streamId)
-    const obj: any = { id: stream.id, ...stream.getMetadata() }
+    const obj: any = { id: stream.id, ...await stream.getMetadata() }
     if (options.includePermissions) {
         const assigments = await stream.getPermissions()
         obj.permissions = assigments.map((assignment) => {

packages/cli-tools/test/stream-create.test.ts

@@ -19,7 +19,7 @@ describe('create stream', () => {
         })
         const client = createTestClient()
         const stream = await client.getStream(streamId)
-        expect(stream.getPartitionCount()).toBe(1)
+        expect(await stream.getPartitionCount()).toBe(1)
         await client.destroy()
     }, 20 * 1000)
 

packages/node/src/plugins/storage/DeleteExpiredCmd.ts

@@ -116,7 +116,7 @@ export class DeleteExpiredCmd {
                     return {
                         streamId: stream.streamId,
                         partition: stream.partition,
-                        storageDays: streamFromChain.getStorageDayCount() ?? 365
+                        storageDays: (await streamFromChain.getStorageDayCount()) ?? 365
                     }
                 } catch (err) { logger.error('Failed to fetch stream info', { err }) }
             })

packages/node/src/plugins/storage/StorageConfig.ts

@@ -51,14 +51,14 @@ export class StorageConfig {
         this.clusterSize = clusterSize
         this.myIndexInCluster = myIndexInCluster
         this.listener = listener
-        this.storagePoller = new StoragePoller(clusterId, pollInterval, streamrClient, (streams, block) => {
-            const streamParts = streams.flatMap((stream: Stream) => ([
-                ...this.createMyStreamParts(stream)
-            ]))
+        this.storagePoller = new StoragePoller(clusterId, pollInterval, streamrClient, async (streams, block) => {
+            const streamParts = (await Promise.all(streams.map(async (stream: Stream) => {
+                return [...await this.createMyStreamParts(stream)]
+            }))).flat()
             this.handleDiff(this.synchronizer.ingestSnapshot(new Set<StreamPartID>(streamParts), block))
         })
-        this.storageEventListener = new StorageEventListener(clusterId, streamrClient, (stream, type, block) => {
-            const streamParts = this.createMyStreamParts(stream)
+        this.storageEventListener = new StorageEventListener(clusterId, streamrClient, async (stream, type, block) => {
+            const streamParts = await this.createMyStreamParts(stream)
             this.handleDiff(this.synchronizer.ingestPatch(streamParts, type, block))
         })
         this.abortController = new AbortController()
@@ -82,8 +82,8 @@ export class StorageConfig {
         return this.synchronizer.getState()
     }
 
-    private createMyStreamParts(stream: Stream): Set<StreamPartID> {
-        return new Set<StreamPartID>(stream.getStreamParts().filter((streamPart) => {
+    private async createMyStreamParts(stream: Stream): Promise<Set<StreamPartID>> {
+        return new Set<StreamPartID>((await stream.getStreamParts()).filter((streamPart) => {
             const hashedIndex = keyToArrayIndex(this.clusterSize, streamPart)
             return hashedIndex === this.myIndexInCluster
         }))

packages/node/src/plugins/storage/StorageEventListener.ts

@@ -10,14 +10,14 @@ const logger = new Logger(module)
 export class StorageEventListener {
     private readonly clusterId: EthereumAddress
     private readonly streamrClient: StreamrClient
-    private readonly onEvent: (stream: Stream, type: 'added' | 'removed', block: number) => void
+    private readonly onEvent: (stream: Stream, type: 'added' | 'removed', block: number) => Promise<void>
     private readonly onAddToStorageNode: (event: StorageNodeAssignmentEvent) => void
     private readonly onRemoveFromStorageNode: (event: StorageNodeAssignmentEvent) => void
 
     constructor(
         clusterId: EthereumAddress,
         streamrClient: StreamrClient,
-        onEvent: (stream: Stream, type: 'added' | 'removed', block: number) => void
+        onEvent: (stream: Stream, type: 'added' | 'removed', block: number) => Promise<void>
     ) {
         this.clusterId = clusterId
         this.streamrClient = streamrClient
@@ -26,14 +26,14 @@ export class StorageEventListener {
         this.onRemoveFromStorageNode = (event: StorageNodeAssignmentEvent) => this.handleEvent(event, 'removed')
     }
 
-    private async handleEvent(event: StorageNodeAssignmentEvent, type: 'added' | 'removed') {
+    private async handleEvent(event: StorageNodeAssignmentEvent, type: 'added' | 'removed'): Promise<void> {
         if (event.nodeAddress !== this.clusterId) {
             return
         }
         logger.info('Received StorageNodeAssignmentEvent', { type, event })
         try {
             const stream = await this.streamrClient.getStream(event.streamId)
-            this.onEvent(stream, type, event.blockNumber)
+            await this.onEvent(stream, type, event.blockNumber)
         } catch (err) {
             logger.warn('Encountered error handling StorageNodeAssignmentEvent', { err, event, type })
         }

packages/node/src/plugins/storage/StoragePoller.ts

@@ -10,13 +10,13 @@ export class StoragePoller {
     private readonly clusterId: string
     private readonly pollInterval: number
     private readonly streamrClient: StreamrClient
-    private readonly onNewSnapshot: (streams: Stream[], block: number) => void
+    private readonly onNewSnapshot: (streams: Stream[], block: number) => Promise<void>
 
     constructor(
         clusterId: string,
         pollInterval: number,
         streamrClient: StreamrClient,
-        onNewSnapshot: (streams: Stream[], block: number) => void
+        onNewSnapshot: (streams: Stream[], block: number) => Promise<void>
     ) {
         this.clusterId = clusterId
         this.pollInterval = pollInterval
@@ -39,7 +39,7 @@ export class StoragePoller {
             foundStreams: streams.length,
             blockNumber
         })
-        this.onNewSnapshot(streams, blockNumber)
+        await this.onNewSnapshot(streams, blockNumber)
     }
 
     private async tryPoll(): Promise<void> {