Suite of fixes & improvements in stats calculation and Anchor support

package.json

@@ -15,7 +15,8 @@
     "start-alone": "./run.sh",
     "test": "node --experimental-vm-modules node_modules/jest/bin/jest.js",
     "test:cov": "node --experimental-vm-modules node_modules/jest/bin/jest.js --coverage",
-    "test:rebuild": "npm run clean:all && npm i && npm run test",
+    "test:rebuild": "npm run build && npm run test",
+    "test:reinstall": "npm run clean:all && npm i && npm run test",
     "lint": "eslint --cache --ignore-path .gitignore './packages/**/src/**/*.{js,ts,tsx}'",
     "lint:fix": "npm run lint -- --fix --quiet",
     "build": "npm run clean:dist && npm run build:ts",

packages/core/package.json

@@ -39,7 +39,7 @@
   "dependencies": {
     "@graphql-tools/schema": "^9.0.4",
     "@solana/spl-token-registry": "^0.2.7",
-    "@solana/web3.js": "^1.65.0",
+    "@solana/web3.js": "^1.66.2",
     "bn.js": "^5.2.0",
     "cacheable-lookup": "^6.0.4",
     "compression": "^1.7.4",

packages/core/src/storage/__tests__/storage.spec.ts

@@ -35,7 +35,7 @@ function getMockedEntity(i: number): MyEntity {
   }
 }
 
-xdescribe('storage v2 integration tests', () => {
+describe('storage v2 integration tests', () => {
   const entities = getMockedEntities()
 
   it('should work as expected', async () => {
@@ -66,7 +66,7 @@ xdescribe('storage v2 integration tests', () => {
       ],
     })
 
-    // await storage.save(entities)
+    await storage.save(entities)
 
     for await (const entry of await storage.getAll()) {
       console.log('by id', entry)

packages/core/src/utils/time.ts

@@ -37,11 +37,11 @@ export function splitDurationIntoIntervals(
       ? DateTime.fromISO(end, zone)
       : DateTime.fromMillis(end, zone)
 
-  if (startDate >= endDate) {
+  if (startDate > endDate) {
     return []
   }
 
-  const length = Math.ceil(
+  const length = startDate.equals(endDate) ? 1 : Math.ceil(
     endDate
       .diff(startDate, intervalUnit as DurationUnit)
       .get(intervalUnit as DurationUnit) / intervalSize,
@@ -66,10 +66,14 @@ export function splitDurationIntoIntervals(
   // Override leftmost bound with "startDate" and rightmost bound with "endDate"
   // [(startDate, A), (A, B), (B, C), (C, endDate)]
   if (preserveExactBounds) {
-    intervals[0] = intervals[0].set({ start: startDate })
-    intervals[intervals.length - 1] = intervals[intervals.length - 1].set({
-      end: endDate,
-    })
+    if (startDate) {
+      intervals[0] = intervals[0].set({ start: startDate })
+    }
+    if (endDate) {
+      intervals[intervals.length - 1] = intervals[intervals.length - 1].set({
+        end: endDate,
+      })
+    }
   }
 
   // console.log(

packages/framework/package.json

@@ -35,7 +35,7 @@
   "dependencies": {
     "@aleph-indexer/core": "^1.0.21",
     "@aleph-indexer/layout": "^1.0.14",
-    "@solana/web3.js": "^1.65.0",
+    "@solana/web3.js": "^1.66.2",
     "bn.js": "^5.2.0",
     "bs58": "^5.0.0",
     "buffer-layout": "^1.2.2",

packages/framework/src/services/fetcher/main.ts

@@ -87,7 +87,7 @@ export class FetcherMsMain implements FetcherMsI, PrivateFetcherMsI {
     this.pendingTransactions = new PendingWorkPool({
       id: 'pending-transactions',
       interval: 0,
-      chunkSize: 1000,
+      chunkSize: 500,
       concurrency: 1,
       dal: this.pendingTransactionDAL,
       handleWork: this._handlePendingTransactions.bind(this),
@@ -107,7 +107,7 @@ export class FetcherMsMain implements FetcherMsI, PrivateFetcherMsI {
     this.pendingTransactionsFetch = new PendingWorkPool({
       id: 'pending-transactions-fetch',
       interval: 0,
-      chunkSize: 200,
+      chunkSize: 100,
       concurrency: 5,
       dal: this.pendingTransactionFetchDAL,
       handleWork: this._handlePendingTransactionsFetch.bind(this),

packages/framework/src/services/parser/src/anchorInstructionParser.ts

@@ -0,0 +1,100 @@
+import bs58 from 'bs58'
+import {
+  AlephParsedInnerInstruction,
+  AlephParsedInstruction,
+  AlephParsedParsedInstruction,
+  RawInstruction,
+} from '@aleph-indexer/core'
+import { DefinedParser } from './parser.js'
+
+/**
+ * Parses a raw instruction, if a parser for given solana program is available.
+ * This parser is automatically used for indexers generated from IDL with anchor-ts-generator.
+ */
+export class AnchorInstructionParser<
+  EventTypeEnum extends string,
+> extends DefinedParser<
+  RawInstruction,
+  RawInstruction | AlephParsedInstruction
+> {
+  constructor(
+    public programId: string,
+    public programName: string,
+    protected getInstructionType: (data: Buffer) => EventTypeEnum | undefined,
+    protected accountLayouts: Partial<Record<EventTypeEnum, any>>,
+    protected dataLayouts: Partial<Record<EventTypeEnum, any>>,
+  ) {
+    super()
+  }
+
+  parse(
+    rawIx: RawInstruction | AlephParsedInstruction,
+  ): RawInstruction | AlephParsedInstruction {
+    if (!this.isCompatibleInstruction(rawIx)) return rawIx
+
+    const decoded = this.getInstructionData(rawIx)
+    if (!decoded) return rawIx
+
+    const type = this.getInstructionType(decoded)
+    if (!type) return rawIx
+
+    const parsedIx: AlephParsedParsedInstruction = rawIx as any
+    parsedIx.program = this.programName
+
+    const { instructionDiscriminator, ...data } = this.parseInstructionData(type, decoded)[0]
+    const accounts = this.parseInstructionAccounts(type, parsedIx)
+
+    parsedIx.parsed = {
+      type,
+      info: {
+        ...(rawIx as any).parsed?.info,
+        data,
+        accounts,
+      },
+    }
+
+    return parsedIx
+  }
+
+  protected isCompatibleInstruction(
+    ix: RawInstruction | AlephParsedInstruction | AlephParsedInnerInstruction,
+  ): boolean {
+    return ix.programId === this.programId
+  }
+
+  protected getInstructionData(
+    rawIx: RawInstruction | AlephParsedInstruction,
+  ): Buffer | undefined {
+    if (!('data' in rawIx)) return
+    return Buffer.from(bs58.decode(rawIx.data))
+  }
+
+  protected parseInstructionData(type: EventTypeEnum, data: Buffer): any {
+    try {
+      const template = this.dataLayouts[type]
+      if (!template) return {}
+
+      return this.dataLayouts[type].deserialize(data)
+    } catch (e) {
+      console.error(e)
+    }
+  }
+
+  protected parseInstructionAccounts(
+    type: EventTypeEnum,
+    rawIx: RawInstruction | AlephParsedInstruction,
+  ): any {
+    const info: any = {}
+
+    const template = this.dataLayouts[type]
+    if (!template) return {}
+
+    if ('accounts' in rawIx) {
+      for (const [index, name] of this.accountLayouts[type].entries()) {
+        info[name] = rawIx.accounts[index]
+      }
+    }
+
+    return info
+  }
+}

packages/framework/src/services/parser/src/instructionParserLibrary.ts

@@ -1,8 +1,9 @@
 import { ParsedInstructionV1, RawInstruction } from '@aleph-indexer/core'
 import { LayoutFactory } from './layout/layoutFactory.js'
 import { DefinedParser } from './parser.js'
-import { InstructionParser } from './instructionParser.js'
+import { SplInstructionParser } from './splInstructionParser.js'
 import { LayoutImplementation } from './layout/types.js'
+import { AnchorInstructionParser } from "./anchorInstructionParser.js";
 
 /**
  * Finds all available instruction parsers and aggregates them for use.
@@ -62,13 +63,24 @@ export class InstructionParserLibrary extends DefinedParser<
 
     if (!implementation) return
 
-    parser = new InstructionParser(
-      implementation.programID,
-      implementation.name,
-      implementation.getInstructionType,
-      implementation.accountLayoutMap,
-      implementation.dataLayoutMap,
-    )
+    // @note: deserialize() is used in Beet, so we will use AnchorInstructionParser here
+    if (Object.values(implementation.dataLayoutMap)[0].deserialize) {
+      parser = new AnchorInstructionParser(
+        implementation.programID,
+        implementation.name,
+        implementation.getInstructionType,
+        implementation.accountLayoutMap,
+        implementation.dataLayoutMap
+      )
+    } else {
+      parser = new SplInstructionParser(
+        implementation.programID,
+        implementation.name,
+        implementation.getInstructionType,
+        implementation.accountLayoutMap,
+        implementation.dataLayoutMap,
+      )
+    }
 
     this.instructionParsers[programId] = parser
 

packages/framework/src/services/parser/src/instructionParser.ts → packages/framework/src/services/parser/src/splInstructionParser.ts

@@ -7,8 +7,12 @@ import {
 } from '@aleph-indexer/core'
 import { DefinedParser } from './parser.js'
 
-// @note: Based on solana-program-library (would need different implementation for Anchor-based programs).
-export class InstructionParser<
+/**
+ * Parses a raw instruction, if a parser for given solana program is available.
+ * Based on solana-program-library, use {@link AnchorInstructionParser} for Anchor instructions, when using Beet as a
+ * layout descriptor.
+ */
+export class SplInstructionParser<
   EventTypeEnum extends string,
 > extends DefinedParser<
   RawInstruction,

packages/framework/src/utils/domain/indexer/main.ts

@@ -161,11 +161,11 @@ export class IndexerMainDomain {
    * @param type The type of time-series to get.
    * @param filters The transformations and clipping to apply to the time-series.
    */
-  async getAccountTimeSeriesStats(
+  async getAccountTimeSeriesStats<V>(
     accounts: string[] = [],
     type: string,
     filters: AccountStatsFilters,
-  ): Promise<AccountTimeSeriesStats[]> {
+  ): Promise<AccountTimeSeriesStats<V>[]> {
     this.checkStats()
 
     accounts =
@@ -177,7 +177,7 @@ export class IndexerMainDomain {
           account,
           method: 'getTimeSeriesStats',
           args: [type, filters],
-        })) as AccountTimeSeriesStats
+        })) as AccountTimeSeriesStats<V>
 
         return stats
       }),
@@ -188,19 +188,18 @@ export class IndexerMainDomain {
    * Returns the global stats for the given accounts.
    * @param accounts The accounts to get the summary from.
    */
-  async getAccountStats(accounts: string[] = []): Promise<AccountStats[]> {
+  async getAccountStats<V>(accounts: string[] = []): Promise<AccountStats<V>[]> {
     this.checkStats()
 
     accounts =
       accounts.length !== 0 ? accounts : Array.from(this.accounts.values())
-
     return Promise.all(
       accounts.map(async (account) => {
         const stats = (await this.context.apiClient.invokeDomainMethod({
           account,
           method: 'getStats',
           args: [],
-        })) as AccountStats
+        })) as AccountStats<V>
 
         return stats
       }),

packages/framework/src/utils/moleculer/config.ts

@@ -31,7 +31,7 @@ export const defaultBrokerConfig: BrokerOptions = {
   requestTimeout: 0,
   serializer: 'JSON',
   internalServices: false,
-  logLevel: 'debug',
+  logLevel: 'warn',
   cacher: 'Memory',
   validator: false,
   heartbeatInterval: HEART_BEATS,

packages/framework/src/utils/stats/accountTimeSeries.ts

@@ -24,12 +24,12 @@ const { JobRunner } = Utils
 /**
  * Defines the account stats handler class.
  */
-export class AccountTimeSeriesStatsManager {
+export class AccountTimeSeriesStatsManager<V> {
   protected compactionJob!: Utils.JobRunner
-  protected stats!: AccountStats
+  protected stats!: AccountStats<V>
 
   constructor(
-    public config: AccountTimeSeriesStatsConfig,
+    public config: AccountTimeSeriesStatsConfig<V>,
     protected indexerApi: IndexerMsI,
     protected stateDAL: StatsStateStorage,
     protected timeSeriesDAL: StatsTimeSeriesStorage,
@@ -71,15 +71,15 @@ export class AccountTimeSeriesStatsManager {
     }
   }
 
-  async getStats(): Promise<AccountStats> {
+  async getStats(): Promise<AccountStats<V>> {
     if (!this.stats) {
       await this.aggregateAccountStats(Date.now())
     }
-
     return this.stats
   }
 
   async process(now: number): Promise<void> {
+    console.log(`📊 processing time series stats for ${this.config.account}`)
     await this.aggregateTimeSeries(now)
     await this.aggregateAccountStats(now)
   }
@@ -126,7 +126,8 @@ export class AccountTimeSeriesStatsManager {
     const { timeSeriesDAL } = this
 
     if (aggregate) {
-      const stats = await aggregate({ now, account, timeSeriesDAL })
+      console.log(`📊 aggregating account stats for ${account}`)
+      const stats: V = await aggregate({ now, account, timeSeriesDAL })
       this.stats = { account, stats }
       return
     }