fix(network-subgraphs): [ETH-876] limit the length of Stream entity id field

packages/network-subgraphs/schema.graphql

@@ -33,8 +33,10 @@ type StreamPermission @entity {
 }
 
 type Stream @entity {
-  "stream ID = 'creator address'/'path' where path can be any string"
+  "NOT THE SAME AS streamId if streamId is over 1000 characters"
   id: ID!
+  "stream ID = 'creator address'/'path' where path can be any string"
+  streamId: String!
   "Stream metadata JSON"
   metadata: String!
   "Permissions that each Ethereum address owns to this stream"

packages/network-subgraphs/src/sponsorshipFactory.ts

@@ -5,7 +5,7 @@ import { NewSponsorship } from '../generated/SponsorshipFactory/SponsorshipFacto
 import { Sponsorship, Stream } from '../generated/schema'
 import { Sponsorship as SponsorshipTemplate } from '../generated/templates'
 import { loadOrCreateNetwork, loadOrCreateSponsorshipDailyBucket } from './helpers'
-
+import { getStreamEntityId } from './streamRegistry'
 export function handleNewSponsorship(event: NewSponsorship): void {
     const sponsorshipContractAddress = event.params.sponsorshipContract
     const sponsorshipContractAddressString = sponsorshipContractAddress.toHexString()
@@ -40,9 +40,10 @@ export function handleNewSponsorship(event: NewSponsorship): void {
     sponsorship.save()
 
     // try to load stream entity
-    const stream = Stream.load(event.params.streamId.toString())
+    const streamEntityId = getStreamEntityId(event.params.streamId.toString())
+    const stream = Stream.load(streamEntityId)
     if (stream != null) {
-        sponsorship.stream = stream.id
+        sponsorship.stream = streamEntityId
         sponsorship.save()
     }
 

packages/network-subgraphs/src/streamRegistry.ts

@@ -11,14 +11,26 @@ import { Stream, StreamPermission } from '../generated/schema'
  * TODO: after ETH-876 is solved, streamId can't be over-long, remove the slice(0, 1000) below
  *       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()
+function getPermissionId(streamEntityId: string, userId: Bytes): string {
+    return streamEntityId + "-" + crypto.keccak256(Bytes.fromUTF8(streamEntityId).concat(userId)).toHexString()
+}
+
+/**
+ * Build the subgraph entity ID; if streamId is short enough, use it as-is (for backwards compatibility)
+ * @param streamId Stream ID in the StreamRegistry contract
+ * @returns
+ */
+export function getStreamEntityId(streamId: string): string {
+    return streamId.length <= 1000 ? streamId : streamId.slice(0, 1000) + "-" + crypto.keccak256(Bytes.fromUTF8(streamId)).toHexString()
 }
 
 export function handleStreamCreation(event: StreamCreated): void {
     log.info('handleStreamCreation: id={} metadata={} blockNumber={}',
         [event.params.id, event.params.metadata, event.block.number.toString()])
-    let stream = new Stream(event.params.id)
+    const streamId = event.params.id
+    const streamEntityId = getStreamEntityId(streamId)
+    const stream = new Stream(streamEntityId)
+    stream.streamId = streamId
     stream.metadata = event.params.metadata
     stream.createdAt = event.block.timestamp
     stream.updatedAt = event.block.timestamp
@@ -28,15 +40,19 @@ export function handleStreamCreation(event: StreamCreated): void {
 export function handleStreamDeletion(event: StreamDeleted): void {
     log.info('handleDeleteStream: id={} blockNumber={}',
         [event.params.id, event.block.number.toString()])
-    store.remove('Stream', event.params.id)
+    const streamEntityId = getStreamEntityId(event.params.id)
+    store.remove('Stream', streamEntityId)
 }
 
 export function handleStreamUpdate(event: StreamUpdated): void {
     log.info('handleUpdateStream: id={} metadata={} blockNumber={}',
         [event.params.id, event.params.metadata, event.block.number.toString()])
-    let stream = Stream.load(event.params.id)
+    const streamId = event.params.id
+    const streamEntityId = getStreamEntityId(streamId)
+    let stream = Stream.load(streamEntityId)
     if (stream === null) {
-        stream = new Stream(event.params.id)
+        stream = new Stream(streamEntityId)
+        stream.streamId = streamId
         stream.createdAt = event.block.timestamp
     }
     stream.metadata = event.params.metadata
@@ -47,14 +63,15 @@ 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()])
-    let stream = Stream.load(event.params.streamId)
+    const streamEntityId = getStreamEntityId(event.params.streamId)
+    const stream = Stream.load(streamEntityId)
     if (stream == null) { return }
 
-    let permissionId = getPermissionId(event.params.streamId, event.params.user)
-    let permission = new StreamPermission(permissionId)
+    const permissionId = getPermissionId(streamEntityId, event.params.user)
+    const permission = new StreamPermission(permissionId)
     permission.userAddress = event.params.user
     permission.userId = event.params.user
-    permission.stream = event.params.streamId
+    permission.stream = streamEntityId
     permission.canEdit = event.params.canEdit
     permission.canDelete = event.params.canDelete
     permission.publishExpiration = event.params.publishExpiration
@@ -69,11 +86,12 @@ 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()])
-    let stream = Stream.load(event.params.streamId)
+    const streamEntityId = getStreamEntityId(event.params.streamId)
+    const stream = Stream.load(streamEntityId)
     if (stream == null) { return }
 
-    let permissionId = getPermissionId(event.params.streamId, event.params.user)
-    let permission = new StreamPermission(permissionId)
+    const permissionId = getPermissionId(streamEntityId, event.params.user)
+    const permission = new StreamPermission(permissionId)
     // Backwards compatibility: pad/concatenate to 20 bytes, Ethereum addresses remain Ethereum addresses.
     // This makes it possible to use both *forUserId functions and the old functions for Ethereum addresses.
     // All new code should use userId instead of userAddress, though; userAddress is marked as deprecated

packages/network-subgraphs/src/streamStorageRegistry.ts

@@ -6,36 +6,38 @@ import {
 } from '../generated/StreamStorageRegistry/StreamStorageRegistry'
 import { Node } from '../generated/schema'
 
+import { getStreamEntityId } from './streamRegistry'
+
 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()])
+    const nodeId = event.params.nodeAddress.toHexString()
+    const streamEntityId = getStreamEntityId(event.params.streamId)
+    log.info('handleStorageNodeAddedToStream: stream={} node={} blockNumber={}', [streamEntityId, nodeId, event.block.number.toString()])
 
-    let node = Node.load(nodeId)!
+    const node = Node.load(nodeId)!
     if (!node.storedStreams) {
-        node.storedStreams = [streamId]
+        node.storedStreams = [streamEntityId]
     } else {
         let streams = node.storedStreams
         if (!streams) { streams = [] }
-        if (streams.includes(streamId)) { return }
-        streams.push(streamId)
+        if (streams.includes(streamEntityId)) { return }
+        streams.push(streamEntityId)
         node.storedStreams = streams
     }
     node.save()
 }
 
 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()])
+    const nodeId = event.params.nodeAddress.toHexString()
+    const streamEntityId = getStreamEntityId(event.params.streamId)
+    log.info('handleStorageNodeRemovedFromStream: stream={} node={} blockNumber={}', [streamEntityId, nodeId, event.block.number.toString()])
 
     let node = Node.load(nodeId)!
     if (!node) { return }
     if (!node.storedStreams) { return }
     let streams = node.storedStreams as string[]
     for (let i = 0; i < streams.length; i++) {
-        let s = streams[i] as string
-        if (s == streamId) {
+        const s = streams[i] as string
+        if (s == streamEntityId) {
             streams.splice(i, 1)
             node.storedStreams = streams
             node.save()