feat(indexer): add fast near in indexers (#512)

apps/indexer-balance/README.md

@@ -6,7 +6,6 @@ Balance indexer collects the info about native NEAR token balance changes 📊 (
 
 ```
 DATABASE_URL=
-RPC_URL=
 NETWORK=mainnet
 AWS_ACCESS_KEY_ID=
 AWS_SECRET_ACCESS_KEY=
@@ -16,5 +15,6 @@ DATABASE_CA=
 DATABASE_CERT=
 DATABASE_KEY=
 BALANCE_START_BLOCK=
+BALANCE_DATA_SOURCE=   # NEAR_LAKE | FAST_NEAR
 SENTRY_DSN=
 ```

apps/indexer-balance/package.json

@@ -29,6 +29,7 @@
     "nb-lake": "*",
     "nb-logger": "*",
     "nb-near": "*",
+    "nb-neardata": "*",
     "nb-tsconfig": "*",
     "nb-types": "*",
     "nb-utils": "*",

apps/indexer-balance/src/config.ts

@@ -3,9 +3,14 @@ import { cleanEnv, num, str, url } from 'envalid';
 import { types } from 'nb-lake';
 import { Network } from 'nb-types';
 
+import { DataSource } from '#types/enum';
 import { Config } from '#types/types';
 
 const env = cleanEnv(process.env, {
+  BALANCE_DATA_SOURCE: str({
+    choices: [DataSource.FAST_NEAR, DataSource.NEAR_LAKE],
+    default: DataSource.NEAR_LAKE,
+  }),
   BALANCE_START_BLOCK: num({ default: 0 }),
   DATABASE_CA: str({ default: '' }),
   DATABASE_CERT: str({ default: '' }),
@@ -14,11 +19,11 @@ const env = cleanEnv(process.env, {
   NETWORK: str({
     choices: [Network.MAINNET, Network.TESTNET],
   }),
-  RPC_URL: str(),
   S3_ENDPOINT: url({ default: '' }),
   SENTRY_DSN: str({ default: '' }),
 });
 
+const genesisHeight = env.NETWORK === Network.MAINNET ? 9_820_210 : 42_376_888;
 let s3Endpoint: null | types.EndpointConfig = null;
 const s3BucketName =
   env.NETWORK === Network.MAINNET
@@ -37,15 +42,16 @@ if (env.S3_ENDPOINT) {
 
 const config: Config = {
   cacheExpiry: 60 * 60, // cache expiry time in seconds (1 hour)
+  dataSource: env.BALANCE_DATA_SOURCE,
   dbCa: env.DATABASE_CA,
   dbCert: env.DATABASE_CERT,
   dbKey: env.DATABASE_KEY,
   dbUrl: env.DATABASE_URL,
-  delta: 100, // start from blocks earlier on sync interuption
+  delta: 1_000, // start from blocks earlier on sync interuption
+  genesisHeight,
   insertLimit: 1_000, // records to insert into the db at a time
   network: env.NETWORK,
-  preloadSize: 10, // blocks to preload in nearlake
-  rpcUrl: env.RPC_URL,
+  preloadSize: 100, // blocks to preload in nearlake
   s3BucketName,
   s3Endpoint,
   s3RegionName: 'eu-central-1',

apps/indexer-balance/src/index.ts

@@ -7,7 +7,10 @@ import { syncData } from '#services/stream';
 
 (async () => {
   try {
-    logger.info({ network: config.network }, 'initializing balance indexer...');
+    logger.info(
+      { data_source: config.dataSource, network: config.network },
+      'initializing balance indexer...',
+    );
     logger.info('syncing balance data...');
     await syncData();
   } catch (error) {

apps/indexer-balance/src/services/stream.ts

@@ -1,10 +1,12 @@
 import { stream, types } from 'nb-lake';
 import { logger } from 'nb-logger';
+import { streamBlock } from 'nb-neardata';
 
 import config from '#config';
 import knex from '#libs/knex';
 import sentry from '#libs/sentry';
 import { storeBalance } from '#services/balance';
+import { DataSource } from '#types/enum';
 
 const fetchBlocks = async (block: number, limit: number) => {
   try {
@@ -40,16 +42,45 @@ export const syncData = async () => {
   const settings = await knex('settings').where({ key: balanceKey }).first();
   const latestBlock = settings?.value?.sync;
 
-  if (!lakeConfig.startBlockHeight && latestBlock) {
-    const next = +latestBlock - config.delta;
+  if (config.dataSource === DataSource.FAST_NEAR) {
+    let startBlockHeight = config.startBlockHeight;
 
-    logger.info(`last synced block: ${latestBlock}`);
-    logger.info(`syncing from block: ${next}`);
-    lakeConfig.startBlockHeight = next;
-  }
+    if (!startBlockHeight && latestBlock) {
+      const next = +latestBlock - config.delta / 2;
+      startBlockHeight = next;
+      logger.info(`last synced block: ${latestBlock}`);
+      logger.info(`syncing from block: ${next}`);
+    }
+
+    const stream = streamBlock({
+      limit: config.preloadSize / 2,
+      network: config.network,
+      start: startBlockHeight || config.genesisHeight,
+    });
+
+    for await (const message of stream) {
+      await onMessage(message);
+    }
 
-  for await (const message of stream(lakeConfig)) {
-    await onMessage(message);
+    stream.on('end', () => {
+      logger.error('stream ended');
+      process.exit();
+    });
+    stream.on('error', (error: Error) => {
+      logger.error(error);
+      process.exit();
+    });
+  } else {
+    if (!lakeConfig.startBlockHeight && latestBlock) {
+      const next = +latestBlock - config.delta;
+      lakeConfig.startBlockHeight = next;
+      logger.info(`last synced block: ${latestBlock}`);
+      logger.info(`syncing from block: ${next}`);
+    }
+
+    for await (const message of stream(lakeConfig)) {
+      await onMessage(message);
+    }
   }
 };
 

apps/indexer-balance/src/types/enum.ts

@@ -0,0 +1,4 @@
+export enum DataSource {
+  FAST_NEAR = 'FAST_NEAR',
+  NEAR_LAKE = 'NEAR_LAKE',
+}

apps/indexer-balance/src/types/types.ts

@@ -3,15 +3,16 @@ import { Network, StateChangeCauseView, StateChangeValueView } from 'nb-types';
 
 export type Config = {
   cacheExpiry: number;
+  dataSource: string;
   dbCa: string;
   dbCert: string;
   dbKey: string;
   dbUrl: string;
   delta: number;
+  genesisHeight: number;
   insertLimit: number;
   network: Network;
   preloadSize: number;
-  rpcUrl: string;
   s3BucketName: string;
   s3Endpoint: null | types.EndpointConfig;
   s3RegionName: string;

apps/indexer-base/README.md

@@ -19,5 +19,6 @@ DATABASE_CERT=
 DATABASE_KEY=
 REDIS_URL=
 BASE_START_BLOCK=
+BASE_DATA_SOURCE=   # NEAR_LAKE | FAST_NEAR
 SENTRY_DSN=
 ```

apps/indexer-base/package.json

@@ -33,6 +33,7 @@
     "nb-json": "*",
     "nb-knex": "*",
     "nb-logger": "*",
+    "nb-neardata": "*",
     "nb-redis": "*",
     "nb-tsconfig": "*",
     "nb-types": "*",

apps/indexer-base/src/config.ts

@@ -3,9 +3,14 @@ import { types } from 'near-lake-framework';
 
 import { Network } from 'nb-types';
 
+import { DataSource } from '#types/enum';
 import { Config } from '#types/types';
 
 const env = cleanEnv(process.env, {
+  BASE_DATA_SOURCE: str({
+    choices: [DataSource.FAST_NEAR, DataSource.NEAR_LAKE],
+    default: DataSource.NEAR_LAKE,
+  }),
   BASE_START_BLOCK: num({ default: 0 }),
   DATABASE_CA: str({ default: '' }),
   DATABASE_CERT: str({ default: '' }),
@@ -46,6 +51,7 @@ if (env.S3_ENDPOINT) {
 
 const config: Config = {
   cacheExpiry: 5 * 60, // cache expiry time in seconds (5 min)
+  dataSource: env.BASE_DATA_SOURCE,
   dbCa: env.DATABASE_CA,
   dbCert: env.DATABASE_CERT,
   dbKey: env.DATABASE_KEY,

apps/indexer-base/src/index.ts

@@ -8,7 +8,10 @@ import { syncData } from '#services/stream';
 
 (async () => {
   try {
-    logger.info({ network: config.network }, 'initializing base indexer...');
+    logger.info(
+      { data_source: config.dataSource, network: config.network },
+      'initializing base indexer...',
+    );
     logger.info('syncing genesis data...');
     await syncGenesis();
     logger.info('syncing blockchain data...');

apps/indexer-base/src/services/stream.ts

@@ -1,6 +1,7 @@
 import { stream, types } from 'near-lake-framework';
 
 import { logger } from 'nb-logger';
+import { streamBlock } from 'nb-neardata';
 
 import config from '#config';
 import knex from '#libs/knex';
@@ -13,6 +14,7 @@ import { storeChunks } from '#services/chunk';
 import { storeExecutionOutcomes } from '#services/executionOutcome';
 import { storeReceipts } from '#services/receipt';
 import { storeTransactions } from '#services/transaction';
+import { DataSource } from '#types/enum';
 
 const lakeConfig: types.LakeConfig = {
   blocksPreloadPoolSize: config.preloadSize,
@@ -29,16 +31,45 @@ if (config.s3Endpoint) {
 export const syncData = async () => {
   const block = await knex('blocks').orderBy('block_height', 'desc').first();
 
-  if (!lakeConfig.startBlockHeight && block) {
-    const next = +block.block_height - config.delta;
+  if (config.dataSource === DataSource.FAST_NEAR) {
+    let startBlockHeight = config.startBlockHeight;
 
-    logger.info(`last synced block: ${block.block_height}`);
-    logger.info(`syncing from block: ${next}`);
-    lakeConfig.startBlockHeight = next;
-  }
+    if (!startBlockHeight && block) {
+      const next = +block.block_height - config.delta / 2;
+      startBlockHeight = next;
+      logger.info(`last synced block: ${block.block_height}`);
+      logger.info(`syncing from block: ${next}`);
+    }
+
+    const stream = streamBlock({
+      limit: config.preloadSize / 2,
+      network: config.network,
+      start: startBlockHeight || config.genesisHeight,
+    });
+
+    for await (const message of stream) {
+      await onMessage(message);
+    }
+
+    stream.on('end', () => {
+      logger.error('stream ended');
+      process.exit();
+    });
+    stream.on('error', (error: Error) => {
+      logger.error(error);
+      process.exit();
+    });
+  } else {
+    if (!lakeConfig.startBlockHeight && block) {
+      const next = +block.block_height - config.delta;
+      lakeConfig.startBlockHeight = next;
+      logger.info(`last synced block: ${block.block_height}`);
+      logger.info(`syncing from block: ${next}`);
+    }
 
-  for await (const message of stream(lakeConfig)) {
-    await onMessage(message);
+    for await (const message of stream(lakeConfig)) {
+      await onMessage(message);
+    }
   }
 };
 

apps/indexer-base/src/types/enum.ts

@@ -0,0 +1,4 @@
+export enum DataSource {
+  FAST_NEAR = 'FAST_NEAR',
+  NEAR_LAKE = 'NEAR_LAKE',
+}

apps/indexer-base/src/types/types.ts

@@ -4,6 +4,7 @@ import { AccessKeyPermissionKind, ActionKind, Network } from 'nb-types';
 
 export interface Config {
   cacheExpiry: number;
+  dataSource: string;
   dbCa: string;
   dbCert: string;
   dbKey: string;

apps/indexer-dex/README.md

@@ -10,11 +10,11 @@ NETWORK=mainnet
 S3_ENDPOINT=
 AWS_ACCESS_KEY_ID=
 AWS_SECRET_ACCESS_KEY=
-DATA_SOURCE=   # NEAR_LAKE | FAST_NEAR
 
 # Optional
 DATABASE_CA=
 DATABASE_CERT=
 DATABASE_KEY=
+DEX_DATA_SOURCE=   # NEAR_LAKE | FAST_NEAR
 SENTRY_DSN=
 ```

apps/indexer-dex/src/config.ts

@@ -7,21 +7,22 @@ import { DataSource } from '#types/enum';
 import { Config } from '#types/types';
 
 const env = cleanEnv(process.env, {
-  DATA_SOURCE: str({
-    choices: [DataSource.FAST_NEAR, DataSource.NEAR_LAKE],
-    default: DataSource.NEAR_LAKE,
-  }),
   DATABASE_CA: str({ default: '' }),
   DATABASE_CERT: str({ default: '' }),
   DATABASE_KEY: str({ default: '' }),
   DATABASE_URL: str(),
+  DEX_DATA_SOURCE: str({
+    choices: [DataSource.FAST_NEAR, DataSource.NEAR_LAKE],
+    default: DataSource.NEAR_LAKE,
+  }),
   NETWORK: str({
     choices: [Network.MAINNET, Network.TESTNET],
   }),
   S3_ENDPOINT: url({ default: '' }),
   SENTRY_DSN: str({ default: '' }),
 });
 
+const genesisHeight = env.NETWORK === Network.MAINNET ? 9_820_210 : 42_376_888;
 let s3Endpoint: null | types.EndpointConfig = null;
 const s3BucketName =
   env.NETWORK === Network.MAINNET
@@ -39,12 +40,13 @@ if (env.S3_ENDPOINT) {
 }
 
 const config: Config = {
-  dataSource: env.DATA_SOURCE,
+  dataSource: env.DEX_DATA_SOURCE,
   dbCa: env.DATABASE_CA,
   dbCert: env.DATABASE_CERT,
   dbKey: env.DATABASE_KEY,
   dbUrl: env.DATABASE_URL,
   delta: 500,
+  genesisHeight,
   NEAR_TOKEN: 'wrap.near',
   network: env.NETWORK,
   preloadSize: 50,

apps/indexer-dex/src/index.ts

@@ -9,7 +9,10 @@ import { syncData } from '#services/stream';
 (async () => {
   try {
     if (config.network === Network.MAINNET) {
-      logger.info({ network: config.network }, 'initializing dex indexer...');
+      logger.info(
+        { data_source: config.dataSource, network: config.network },
+        'initializing dex indexer...',
+      );
       logger.info('syncing dex data...');
       await syncData();
     }

apps/indexer-dex/src/libs/big.ts

@@ -0,0 +1,5 @@
+import Big from 'big.js';
+
+Big.DP = 24;
+
+export default Big;