feature(cardano-services): support Blockfrost NetworkInfoProvider in PgBossWorker

nix/cardano-services/deployments/pg-boss-worker-deployment.nix

@@ -48,7 +48,7 @@
                 METADATA_FETCH_MODE = values.pg-boss-worker.metadata-fetch-mode;
 
                 STAKE_POOL_PROVIDER_URL = "http://${config.name}-backend.${config.namespace}.svc.cluster.local";
-                NETWORK_INFO_PROVIDER_URL = "http://${config.name}-backend.${config.namespace}.svc.cluster.local";
+                NETWORK_INFO_PROVIDER = "blockfrost";
 
                 POSTGRES_POOL_MAX_STAKE_POOL = "5";
                 POSTGRES_HOST_STAKE_POOL = values.postgresName;

packages/cardano-services/src/PgBoss/stakePoolRewardsHandler.ts

@@ -15,7 +15,9 @@ import { ServiceNames } from '../Program/programs/types';
 import { accountActiveStake, poolDelegators, poolRewards } from './stakePoolRewardsQueries';
 import { computeROS } from '../StakePool/TypeormStakePoolProvider/util';
 import { missingProviderUrlOption } from '../Program/options/common';
-import { networkInfoHttpProvider } from '@cardano-sdk/cardano-services-client';
+import { BlockfrostNetworkInfoProvider, networkInfoHttpProvider } from '@cardano-sdk/cardano-services-client';
+import { getBlockfrostClient } from '../util';
+import { ProviderImplementation } from '../Program';
 
 /** The version of the algorithm to compute rewards. */
 export const REWARDS_COMPUTE_VERSION = 1;
@@ -211,16 +213,22 @@ const getLastSlot = async (provider: NetworkInfoProvider, epochNo: Cardano.Epoch
 
 /** Creates a `stakePoolRewardsHandler`. */
 export const stakePoolRewardsHandlerFactory: WorkerHandlerFactory = (options) => {
-  const { dataSource, db, lastRosEpochs, logger, networkInfoProviderUrl } = options;
+  const { dataSource, db, lastRosEpochs, logger, networkInfoProvider, networkInfoProviderUrl } = options;
 
   // Introduced following code repetition as the correct form is source of a circular-deps:check failure.
   // Solving it would require an invasive refactoring action, probably better to defer it.
   // if (!lastRosEpochs) throw new MissingProgramOption(STAKE_POOL_REWARDS, Descriptions.LastRosEpochs);
   if (!lastRosEpochs)
     throw new MissingProgramOption(STAKE_POOL_REWARDS, 'Number of epochs over which lastRos is computed');
-  if (!networkInfoProviderUrl) throw missingProviderUrlOption(STAKE_POOL_REWARDS, ServiceNames.NetworkInfo);
 
-  const provider = networkInfoHttpProvider({ baseUrl: networkInfoProviderUrl, logger });
+  let provider: NetworkInfoProvider;
+  if (networkInfoProvider === ProviderImplementation.BLOCKFROST) {
+    provider = new BlockfrostNetworkInfoProvider(getBlockfrostClient(), logger);
+  } else if (networkInfoProviderUrl) {
+    provider = networkInfoHttpProvider({ baseUrl: networkInfoProviderUrl, logger });
+  } else {
+    throw missingProviderUrlOption(STAKE_POOL_REWARDS, ServiceNames.NetworkInfo);
+  }
 
   return async (data: StakePoolRewardsJob) => {
     const { epochNo } = data;

packages/cardano-services/src/Program/options/providers.ts

@@ -19,7 +19,7 @@ export type ProviderImplementations = {
   stakePoolProvider?: ProviderImplementation;
 };
 export const ProviderImplementationDescription = 'Select one of the available provider implementations';
-const argParser = (impl: string) => ProviderImplementation[impl.toUpperCase() as keyof typeof ProviderImplementation];
+export const argParser = (impl: string) => ProviderImplementation[impl.toUpperCase() as keyof typeof ProviderImplementation];
 export const providerSelectionOptions = [
   newOption(
     '--asset-provider <assetProvider>',

packages/cardano-services/src/Program/programs/pgBossWorker.ts

@@ -13,6 +13,7 @@ export const PARALLEL_JOBS_DEFAULT = 10;
 export const PG_BOSS_WORKER_API_URL_DEFAULT = new URL('http://localhost:3003');
 
 export enum PgBossWorkerOptionDescriptions {
+  NetworkInfoProvider = 'Select one of the available provider implementations',
   ParallelJobs = 'Parallel jobs to run',
   Queues = 'Comma separated queue names',
   Schedules = 'File path for schedules configurations'
@@ -21,6 +22,7 @@ export enum PgBossWorkerOptionDescriptions {
 export interface LoadPgBossWorkerDependencies {
   dnsResolver?: (serviceName: string) => Promise<SrvRecord>;
   logger?: Logger;
+  networkInfoProvider?: string;
 }
 
 const pgBossWorker = 'pg-boss-worker';

packages/cardano-services/src/Program/services/pgboss.ts

@@ -12,7 +12,7 @@ import {
   createPgBoss,
   isRecoverableTypeormError
 } from '@cardano-sdk/projection-typeorm';
-import { CommonProgramOptions, PosgresProgramOptions } from '../options';
+import { CommonProgramOptions, PosgresProgramOptions, ProviderImplementation } from '../options';
 import { DataSource } from 'typeorm';
 import { HealthCheckResponse } from '@cardano-sdk/core';
 import { HttpService } from '../../Http/HttpService';
@@ -88,6 +88,7 @@ export type PgBossWorkerArgs = CommonProgramOptions &
   StakePoolMetadataProgramOptions &
   PosgresProgramOptions<'DbSync'> &
   PosgresProgramOptions<'StakePool'> & {
+    networkInfoProvider?: ProviderImplementation;
     parallelJobs: number;
     queues: PgBossQueue[];
     schedules: Array<ScheduleConfig>;

packages/cardano-services/src/cli.ts

@@ -50,7 +50,10 @@ import {
   withHandlePolicyIdsOptions,
   withOgmiosOptions,
   withPostgresOptions,
-  withStakePoolMetadataOptions
+  withStakePoolMetadataOptions,
+  ProviderImplementationDescription,
+  ProviderImplementation,
+  argParser as providerImplArgParser,
 } from './Program';
 import { Command } from 'commander';
 import { DB_CACHE_TTL_DEFAULT } from './InMemoryCache';
@@ -491,7 +494,14 @@ addOptions(
         return readScheduleConfig(schedules);
       },
       []
-    )
+    ),
+    newOption(
+        '--network-info-provider <networkInfoProvider>',
+        ProviderImplementationDescription,
+        'NETWORK_INFO_PROVIDER',
+        providerImplArgParser,
+        ProviderImplementation.DBSYNC
+    ).choices([ProviderImplementation.BLOCKFROST, ProviderImplementation.DBSYNC]),
   ]
 ).action(async (args: PgBossWorkerArgs) =>
   runServer('pg-boss worker', { args }, () =>