fix(network-subgraphs): [ETH-876] Ignore overlong stream ids (#993)

Added stream ID length check to all events handlers which process stream
related events. If the stream ID is more than 1000 characters, the event
is just ignored. That causes the stream to be fully ignored in The Graph
(it won't appear e.g. in the `streams` query).

## TODO (in a separate PR)

We need to document to our public document that overlong streams are not
supported (i.e. the behavior is undefined in our system for those
streams).

## Manual test

```bash
PRIVATE_KEY="1111111111111111111111111111111111111111111111111111111111111111"
NOW=$(date +%Y%m%d%H%M%S)
LONG_SNIPPET=$(printf 'a%.0s' {1..2000})
echo 'Create streams'
npx tsx bin/streamr.ts internal token-mint self 1000000 1000000 --env dev2 --private-key $PRIVATE_KEY
npx tsx bin/streamr.ts stream create /short1-$NOW --env dev2 --private-key $PRIVATE_KEY
npx tsx bin/streamr.ts stream create /long-$NOW-$LONG_SNIPPET --env dev2 --private-key $PRIVATE_KEY
npx tsx bin/streamr.ts stream create /short2-$NOW --env dev2 --private-key $PRIVATE_KEY
echo 'Wait for The Graph to index'
sleep 10s
echo 'Query streams:'
npx tsx bin/streamr.ts stream search $NOW --env dev2
```

## Related changed

- Simplified `getPermissionId(`) as the slicing and concatenation which
is no longer needed

## Future improvements

- We could avoid code repetition by using some kind of wrapper function
which does the length check

Author: teogeb

packages/network-subgraphs/src/helpers.ts

@@ -17,6 +17,7 @@ import { Operator as OperatorContract } from '../generated/templates/Operator/Op
 
 const BUCKET_SECONDS = BigInt.fromI32(60 * 60 * 24) // 1 day
 const NETWORK_ENTITY_ID = "network-entity-id"
+export const MAX_STREAM_ID_LENGTH = 1000
 
 /**
  * Helper function to load a project or create a project with default values. It will probably silence some errors.

packages/network-subgraphs/src/projectRegistry.ts

@@ -10,7 +10,7 @@ import {
     StreamRemoved,
     PaymentDetailsByChainUpdated,
 } from '../generated/ProjectRegistryV1/ProjectRegistryV1'
-import { getIsDataUnionValue, loadOrCreateProject } from './helpers'
+import { getIsDataUnionValue, loadOrCreateProject, MAX_STREAM_ID_LENGTH } from './helpers'
 
 export function handleProjectCreation(event: ProjectCreated): void {
     const id = event.params.id.toHexString()
@@ -151,7 +151,11 @@ export function handleStreamAddition(event: StreamAdded): void {
     const streamId = event.params.streamId
     log.info('handleStreamAddition: projectId={} streamId={} blockNumber={}',
         [projectId, streamId, event.block.number.toString()])
-
+    if (streamId.length > MAX_STREAM_ID_LENGTH) {
+        log.warning("Overlong stream id not supported: {}", [streamId]) 
+        return
+    }
+        
     let project = loadOrCreateProject(event.params.projectId)
 
     const streams = project.streams
@@ -165,6 +169,10 @@ export function handleStreamRemoval(event: StreamRemoved): void {
     const streamId = event.params.streamId
     log.info('handleStreamRemoval: projectId={} streamId={} blockNumber={}',
         [projectId, streamId, event.block.number.toString()])
+    if (streamId.length > MAX_STREAM_ID_LENGTH) {
+        log.warning("Overlong stream id not supported: {}", [streamId]) 
+        return
+    }
 
     let project = loadOrCreateProject(event.params.projectId)
 

packages/network-subgraphs/src/sponsorshipFactory.ts

@@ -4,7 +4,7 @@ import { Sponsorship as SponsorshipContract } from '../generated/templates/Spons
 import { NewSponsorship } from '../generated/SponsorshipFactory/SponsorshipFactory'
 import { Sponsorship, Stream } from '../generated/schema'
 import { Sponsorship as SponsorshipTemplate } from '../generated/templates'
-import { loadOrCreateNetwork, loadOrCreateSponsorshipDailyBucket } from './helpers'
+import { loadOrCreateNetwork, loadOrCreateSponsorshipDailyBucket, MAX_STREAM_ID_LENGTH } from './helpers'
 
 export function handleNewSponsorship(event: NewSponsorship): void {
     const sponsorshipContractAddress = event.params.sponsorshipContract
@@ -14,6 +14,10 @@ export function handleNewSponsorship(event: NewSponsorship): void {
         [event.block.number.toString(), sponsorshipContractAddressString,
             event.params.policies.map<string>((x) => x.toHexString()).join(", "), event.params.policyParams.toString(), creator]
     )
+    if (event.params.streamId.length > MAX_STREAM_ID_LENGTH) {
+        log.warning("Overlong stream id not supported: {}", [event.params.streamId]) 
+        return
+    }
 
     const sponsorship = new Sponsorship(sponsorshipContractAddressString)
     sponsorship.totalStakedWei = BigInt.zero()

packages/network-subgraphs/src/streamRegistry.ts

@@ -3,6 +3,7 @@ import { ByteArray, Bytes, log, store, crypto } from '@graphprotocol/graph-ts'
 import { StreamCreated, StreamDeleted, StreamUpdated, PermissionUpdated, PermissionUpdatedForUserId }
     from '../generated/StreamRegistry/StreamRegistry'
 import { Stream, StreamPermission } from '../generated/schema'
+import { MAX_STREAM_ID_LENGTH } from './helpers'
 
 /**
  * Hash the streamId and the userId, in order to get constant-length permission IDs (ETH-867)
@@ -12,12 +13,16 @@ import { Stream, StreamPermission } from '../generated/schema'
  *       because it could cause some streams with same 1k-prefix to mix up when sorting
  **/
 function getPermissionId(streamId: string, userId: Bytes): string {
-    return streamId.slice(0, 1000) + "-" + crypto.keccak256(Bytes.fromUTF8(streamId).concat(userId)).toHexString()
+    return streamId + "-" + crypto.keccak256(userId).toHexString()
 }
 
 export function handleStreamCreation(event: StreamCreated): void {
     log.info('handleStreamCreation: id={} metadata={} blockNumber={}',
         [event.params.id, event.params.metadata, event.block.number.toString()])
+    if (event.params.id.length > MAX_STREAM_ID_LENGTH) {
+        log.warning("Overlong stream id not supported: {}", [event.params.id]) 
+        return
+    }
     let stream = new Stream(event.params.id)
     stream.metadata = event.params.metadata
     stream.createdAt = event.block.timestamp
@@ -47,6 +52,10 @@ export function handleStreamUpdate(event: StreamUpdated): void {
 export function handlePermissionUpdate(event: PermissionUpdated): void {
     log.info('handlePermissionUpdate: user={} streamId={} blockNumber={}',
         [event.params.user.toHexString(), event.params.streamId, event.block.number.toString()])
+    if (event.params.streamId.length > MAX_STREAM_ID_LENGTH) {
+        log.warning("Overlong stream id not supported: {}", [event.params.streamId]) 
+        return
+    }
     let stream = Stream.load(event.params.streamId)
     if (stream == null) { return }
 
@@ -69,6 +78,10 @@ export function handlePermissionUpdate(event: PermissionUpdated): void {
 export function handlePermissionUpdateForUserId(event: PermissionUpdatedForUserId): void {
     log.info('handlePermissionUpdateForUserId: user={} streamId={} blockNumber={}',
         [event.params.user.toHexString(), event.params.streamId, event.block.number.toString()])
+    if (event.params.streamId.length > MAX_STREAM_ID_LENGTH) {
+        log.warning("Overlong stream id not supported: {}", [event.params.streamId]) 
+        return
+    }
     let stream = Stream.load(event.params.streamId)
     if (stream == null) { return }
 

packages/network-subgraphs/src/streamStorageRegistry.ts

@@ -5,11 +5,16 @@ import {
     Removed
 } from '../generated/StreamStorageRegistry/StreamStorageRegistry'
 import { Node } from '../generated/schema'
+import { MAX_STREAM_ID_LENGTH } from './helpers'
 
 export function handleStorageNodeAddedToStream(event: Added): void {
     let nodeId = event.params.nodeAddress.toHexString()
     let streamId = event.params.streamId.toString()
     log.info('handleStorageNodeAddedToStream: stream={} node={} blockNumber={}', [streamId, nodeId, event.block.number.toString()])
+    if (event.params.streamId.length > MAX_STREAM_ID_LENGTH) {
+        log.warning("Overlong stream id not supported: {}", [event.params.streamId]) 
+        return
+    }
 
     let node = Node.load(nodeId)!
     if (!node.storedStreams) {
@@ -28,7 +33,11 @@ export function handleStorageNodeRemovedFromStream(event: Removed): void {
     let nodeId = event.params.nodeAddress.toHexString()
     let streamId = event.params.streamId.toString()
     log.info('handleStorageNodeRemovedFromStream: stream={} node={} blockNumber={}', [streamId, nodeId, event.block.number.toString()])
-
+    if (event.params.streamId.length > MAX_STREAM_ID_LENGTH) {
+        log.warning("Overlong stream id not supported: {}", [event.params.streamId]) 
+        return
+    }
+    
     let node = Node.load(nodeId)!
     if (!node) { return }
     if (!node.storedStreams) { return }