Storage Squid integration with Argus/Colossus

.env

@@ -49,11 +49,11 @@ GRAPHQL_SERVER_HOST=localhost
 # Websocket RPC endpoint containers will use.
 JOYSTREAM_NODE_WS=ws://joystream-node:9944/
 
-# Query node which colossus will use
-COLOSSUS_QUERY_NODE_URL=http://graphql-server:8081/graphql
+# Query node (graphql-server) which distributor will use
+COLOSSUS_QUERY_NODE_URL=http://squid-graphql-server:4352/graphql
 
-# Query node which distributor will use
-DISTRIBUTOR_QUERY_NODE_URL=http://graphql-server:8081/graphql
+# Query node (graphql-server) which distributor will use
+DISTRIBUTOR_QUERY_NODE_URL=http://squid-graphql-server:4352/graphql
 
 # Indexer gateway used by processor. If you don't use the local indexer set this to a remote gateway
 PROCESSOR_INDEXER_GATEWAY=http://hydra-indexer-gateway:4000/graphql
@@ -155,7 +155,7 @@ [email protected]
 SQD_DEBUG=api:*
 OPENAPI_PLAYGROUND=true
 
-ARCHIVE_GATEWAY_URL=${CUSTOM_ARCHIVE_GATEWAY_URL:-http://squid-archive-gateway:8000/graphql}
+ORION_ARCHIVE_GATEWAY_URL=${CUSTOM_ARCHIVE_GATEWAY_URL:-http://squid-archive-gateway:8000/graphql}
 
 # =====================================================================================
 
@@ -169,3 +169,22 @@ TELEMETRY_ENDPOINT=http://collector:4318
 # We do not provide a default value - scripts that startup a joystream-node service
 # Should be explicit about what version to use.
 # JOYSTREAM_NODE_TAG=latest
+
+# =====================================================================================
+## Storage-Squid configuration
+
+# Db config
+SQUID_DB_HOST=squid_db
+SQUID_DB_NAME=squid
+SQUID_DB_PASS=squid
+SQUID_DB_PORT=23332
+
+# Processor service prometheus port
+SQUID_PROCESSOR_PROMETHEUS_PORT=3337
+
+# Graphql server port
+SQUID_GQL_PORT=4352
+
+# Archive gateway host (Should not be set in local development)
+# For running a production storage-squid instance uncomment the following line (to use the subsquid hosted archive)
+# SQUID_ARCHIVE_GATEWAY_URL=${CUSTOM_ARCHIVE_GATEWAY_URL:-https://v2.archive.subsquid.io/network/joystream}

cli/package.json

@@ -39,7 +39,7 @@
     "cross-fetch": "^3.0.6",
     "fluent-ffmpeg": "^2.1.2",
     "form-data": "^4.0.0",
-    "graphql": "^14.7.0",
+    "graphql": "^15.3.0",
     "graphql-tag": "^2.12.6",
     "inquirer": "^7.1.0",
     "inquirer-datepicker-prompt": "^0.4.2",

distributor-node/config.yml

@@ -1,6 +1,6 @@
 id: test-node
 endpoints:
-  queryNode: http://localhost:8081/graphql
+  queryNode: http://localhost:4352/graphql
   joystreamNodeWs: ws://localhost:9944
 directories:
   assets: ./local/data

distributor-node/package.json

@@ -31,7 +31,7 @@
     "express-winston": "^4.1.0",
     "fast-safe-stringify": "^2.1.1",
     "file-type": "^16.5.1",
-    "graphql": "^14.7.0",
+    "graphql": "^15.3.0",
     "graphql-tag": "^2.12.6",
     "inquirer": "^8.1.2",
     "js-image-generator": "^1.0.3",
@@ -138,6 +138,7 @@
     "prepack": "rm -rf lib && tsc -b && oclif-dev manifest && generate:all",
     "test": "nyc --extension .ts mocha --forbid-only \"test/**/*.test.ts\"",
     "version": "generate:docs:cli && git add docs/cli/*",
+    "generate:schema:graphql": "docker run --rm joystream/storage-squid:latest npm run get-graphql-schema > src/services/networking/query-node/schema.graphql",
     "generate:types:json-schema": "yarn ts-node --transpile-only ./src/schemas/scripts/generateTypes.ts",
     "generate:types:graphql": "yarn graphql-codegen -c ./src/services/networking/query-node/codegen.yml",
     "generate:types:public-api": "yarn openapi-typescript ./src/api-spec/public.yml -o ./src/types/generated/PublicApi.ts -c ../prettierrc.js",

distributor-node/src/schemas/configSchema.ts

@@ -28,7 +28,7 @@ export const configSchema: JSONSchema4 = objectSchema({
       description: 'Specifies external endpoints that the distributor node will connect to',
       properties: {
         queryNode: {
-          description: 'Query node graphql server uri (for example: http://localhost:8081/graphql)',
+          description: 'Storage-Squid graphql server uri (for example: http://localhost:4352/graphql)',
           type: 'string',
           format: 'uri',
         },

distributor-node/src/services/networking/NetworkingService.ts

@@ -1,27 +1,28 @@
-import { ReadonlyConfig } from '../../types/config'
-import { QueryFetchPolicy, QueryNodeApi } from './query-node/api'
-import { Logger } from 'winston'
-import { LoggingService } from '../logging'
-import { StorageNodeApi } from './storage-node/api'
-import { StateCacheService } from '../cache/StateCacheService'
-import { DataObjectDetailsFragment } from './query-node/generated/queries'
 import axios from 'axios'
+import http from 'http'
+import https from 'https'
+import Mime from 'mime/lite'
+import queue from 'queue'
+import { Logger } from 'winston'
 import {
-  StorageNodeEndpointData,
   DataObjectAccessPoints,
   DataObjectData,
   DataObjectInfo,
-  StorageNodeDownloadResponse,
   DownloadData,
   StatusResponse,
+  StorageNodeDownloadResponse,
+  StorageNodeEndpointData,
 } from '../../types'
-import queue from 'queue'
-import { DistributionBucketOperatorStatus } from './query-node/generated/schema'
-import http from 'http'
-import https from 'https'
+import { ReadonlyConfig } from '../../types/config'
+import { StateCacheService } from '../cache/StateCacheService'
+import { LoggingService } from '../logging'
 import { parseAxiosError } from '../parsers/errors'
 import { PendingDownload, PendingDownloadStatusType } from './PendingDownload'
-import Mime from 'mime/lite'
+import { QueryFetchPolicy, QueryNodeApi } from './query-node/api'
+import { DataObjectDetailsFragment } from './query-node/generated/queries'
+import { DistributionBucketOperatorStatus } from './query-node/generated/schema'
+import { RuntimeApi } from './runtime/api'
+import { StorageNodeApi } from './storage-node/api'
 
 // Concurrency limits
 export const MAX_CONCURRENT_AVAILABILITY_CHECKS_PER_OBJECT = 10
@@ -30,6 +31,7 @@ export const MAX_CONCURRENT_RESPONSE_TIME_CHECKS = 10
 export class NetworkingService {
   private config: ReadonlyConfig
   private queryNodeApi: QueryNodeApi
+  private runtimeApi!: RuntimeApi
   private logging: LoggingService
   private stateCache: StateCacheService
   private logger: Logger
@@ -65,6 +67,12 @@ export class NetworkingService {
     this.downloadQueue.on('error', (err) => {
       this.logger.error('Data object download failed', { err })
     })
+
+    this.initRuntimeApi().catch((err) => this.logger.error('Runtime API initialization failed:', { err }))
+  }
+
+  private async initRuntimeApi() {
+    this.runtimeApi = await RuntimeApi.create(this.logging, this.config.endpoints.joystreamNodeWs)
   }
 
   private validateNodeEndpoint(endpoint: string): void {
@@ -97,12 +105,12 @@ export class NetworkingService {
 
   private prepareStorageNodeEndpoints(details: DataObjectDetailsFragment) {
     const endpointsData = details.storageBag.storageBuckets
-      .filter((bucket) => bucket.operatorStatus.__typename === 'StorageBucketOperatorStatusActive')
-      .map((bucket) => {
-        const rootEndpoint = bucket.operatorMetadata?.nodeEndpoint
+      .filter(({ storageBucket }) => storageBucket.operatorStatus.__typename === 'StorageBucketOperatorStatusActive')
+      .map(({ storageBucket }) => {
+        const rootEndpoint = storageBucket.operatorMetadata?.nodeEndpoint
         const apiEndpoint = rootEndpoint ? this.getApiEndpoint(rootEndpoint) : ''
         return {
-          bucketId: bucket.id,
+          bucketId: storageBucket.id,
           endpoint: apiEndpoint,
         }
       })
@@ -119,8 +127,8 @@ export class NetworkingService {
   private getDataObjectActiveDistributorsSet(objectDetails: DataObjectDetailsFragment): Set<number> {
     const activeDistributorsSet = new Set<number>()
     const { distributionBuckets } = objectDetails.storageBag
-    for (const bucket of distributionBuckets) {
-      for (const operator of bucket.operators) {
+    for (const { distributionBucket } of distributionBuckets) {
+      for (const operator of distributionBucket.operators) {
         if (operator.status === DistributionBucketOperatorStatus.Active) {
           activeDistributorsSet.add(operator.workerId)
         }
@@ -151,14 +159,16 @@ export class NetworkingService {
         isSupported = typeof this.config.workerId === 'number' ? distributors.has(this.config.workerId) : false
       } else {
         const supportedBucketIds = this.config.buckets.map((id) => id.toString())
-        isSupported = details.storageBag.distributionBuckets.some((b) => supportedBucketIds.includes(b.id))
+        isSupported = details.storageBag.distributionBuckets.some((b) =>
+          supportedBucketIds.includes(b.distributionBucket.id)
+        )
       }
       data = {
         objectId,
         accessPoints: this.parseDataObjectAccessPoints(details),
         contentHash: details.ipfsHash,
         size: parseInt(details.size),
-        fallbackMimeType: this.parseUserProvidedMimeType(details.type.subtitle?.mimeType),
+        fallbackMimeType: this.parseUserProvidedMimeType(details.type?.subtitle?.mimeType),
       }
     }
 
@@ -384,7 +394,7 @@ export class NetworkingService {
         contentHash: ipfsHash,
         objectId: id,
         size: parseInt(size),
-        fallbackMimeType: this.parseUserProvidedMimeType(type.subtitle?.mimeType),
+        fallbackMimeType: this.parseUserProvidedMimeType(type?.subtitle?.mimeType),
       })
     })
 
@@ -437,16 +447,16 @@ export class NetworkingService {
   }
 
   async getQueryNodeStatus(): Promise<StatusResponse['queryNodeStatus']> {
-    const qnState = await this.queryNodeApi.getQueryNodeState()
+    const squidStatus = await this.queryNodeApi.getQueryNodeState()
 
-    if (qnState === null) {
-      this.logger.error("Couldn't fetch the state from connected query-node")
+    if (squidStatus === null) {
+      this.logger.error("Couldn't fetch the state from connected storage-squid")
     }
 
     return {
       url: this.config.endpoints.queryNode,
-      chainHead: qnState?.chainHead || 0,
-      blocksProcessed: qnState?.lastCompleteBlock || 0,
+      chainHead: (await this.runtimeApi.derive.chain.bestNumber()).toNumber() || 0,
+      blocksProcessed: squidStatus?.height || 0,
     }
   }
 }

distributor-node/src/services/networking/query-node/api.ts

@@ -1,20 +1,9 @@
-import {
-  ApolloClient,
-  DocumentNode,
-  FetchPolicy,
-  HttpLink,
-  NormalizedCacheObject,
-  from,
-  split,
-} from '@apollo/client/core'
+import { ApolloClient, DocumentNode, FetchPolicy, HttpLink, NormalizedCacheObject, from } from '@apollo/client/core'
 import { onError } from '@apollo/client/link/error'
-import { WebSocketLink } from '@apollo/client/link/ws'
-import { getMainDefinition } from '@apollo/client/utilities'
 import { InvalidationPolicyCache } from '@nerdwallet/apollo-cache-policies'
 import fetch from 'cross-fetch'
 import { FragmentDefinitionNode } from 'graphql'
 import { Logger } from 'winston'
-import ws from 'ws'
 import { ReadonlyConfig } from '../../../types'
 import { LoggingService } from '../../logging'
 import {
@@ -35,11 +24,10 @@ import {
   GetDistributionBucketsWithBagsByWorkerIdQuery,
   GetDistributionBucketsWithBagsByWorkerIdQueryVariables,
   MinimalDataObjectFragment,
-  QueryNodeState,
-  QueryNodeStateFields,
-  QueryNodeStateFieldsFragment,
-  QueryNodeStateSubscription,
-  QueryNodeStateSubscriptionVariables,
+  SquidStatus,
+  SquidStatusFieldsFragment,
+  SquidStatusQuery,
+  SquidStatusQueryVariables,
   StorageBagWithObjectsFragment,
   StorageBucketOperatorFieldsFragment,
 } from './generated/queries'
@@ -62,8 +50,6 @@ type PaginationQueryResult<T = unknown> = {
   }
 }
 
-type CustomVariables<T> = Omit<T, keyof PaginationQueryVariables>
-
 export class QueryNodeApi {
   private config: ReadonlyConfig
   private apolloClient: ApolloClient<NormalizedCacheObject>
@@ -79,33 +65,11 @@ export class QueryNodeApi {
       exitOnError && process.exit(-1)
     })
 
-    const queryLink = from([errorLink, new HttpLink({ uri: this.config.endpoints.queryNode, fetch })])
-    const wsLink = new WebSocketLink({
-      uri: this.config.endpoints.queryNode,
-      options: {
-        reconnect: true,
-      },
-      webSocketImpl: ws,
-    })
-    const splitLink = split(
-      ({ query }) => {
-        const definition = getMainDefinition(query)
-        return definition.kind === 'OperationDefinition' && definition.operation === 'subscription'
-      },
-      wsLink,
-      queryLink
-    )
-
     this.apolloClient = new ApolloClient({
-      link: splitLink,
+      link: from([errorLink, new HttpLink({ uri: this.config.endpoints.queryNode, fetch })]),
       // Ref: https://www.apollographql.com/docs/react/api/core/ApolloClient/#assumeimmutableresults
       assumeImmutableResults: true,
       cache: new InvalidationPolicyCache({
-        typePolicies: {
-          ProcessorState: {
-            keyFields: (object) => object.__typename,
-          },
-        },
         invalidationPolicies: {
           types: {
             StorageDataObject: {
@@ -126,7 +90,7 @@ export class QueryNodeApi {
     query: DocumentNode,
     variables: VariablesT,
     resultKey: keyof QueryT,
-    fetchPolicy: QueryFetchPolicy
+    fetchPolicy: QueryFetchPolicy = 'no-cache'
   ): Promise<Required<QueryT>[keyof QueryT] | null> {
     return (
       (await this.apolloClient.query<QueryT, VariablesT>({ query, variables, fetchPolicy })).data[resultKey] || null
@@ -266,30 +230,18 @@ export class QueryNodeApi {
   }
 
   public getActiveStorageBucketOperatorsData(): Promise<StorageBucketOperatorFieldsFragment[]> {
-    return this.multipleEntitiesWithPagination<
-      StorageBucketOperatorFieldsFragment,
+    return this.multipleEntitiesQuery<
       GetActiveStorageBucketOperatorsDataQuery,
-      CustomVariables<GetActiveStorageBucketOperatorsDataQueryVariables>
-    >(GetActiveStorageBucketOperatorsData, {}, 'storageBucketsConnection')
+      GetActiveStorageBucketOperatorsDataQueryVariables
+    >(GetActiveStorageBucketOperatorsData, {}, 'storageBuckets')
   }
 
-  public async getQueryNodeState(): Promise<QueryNodeStateFieldsFragment | null> {
-    // fetch cached state
-    const cachedState = this.readFragment<QueryNodeStateFieldsFragment, QueryNodeStateSubscriptionVariables>(
-      'ProcessorState',
-      QueryNodeStateFields
-    )
-
-    // If we have the state in cache, return it
-    if (cachedState) {
-      return cachedState
-    }
-
-    // Otherwise setup the subscription (which will periodically update the cache) and return for the first result
-    return this.uniqueEntitySubscription<QueryNodeStateSubscription, QueryNodeStateSubscriptionVariables>(
-      QueryNodeState,
+  public async getQueryNodeState(): Promise<SquidStatusFieldsFragment | null> {
+    const squidStatus = await this.uniqueEntityQuery<SquidStatusQuery, SquidStatusQueryVariables>(
+      SquidStatus,
       {},
-      'stateSubscription'
+      'squidStatus'
     )
+    return squidStatus
   }
 }