Create createEventProcessor to offer a simpler way to do basic event processing.

- Add processing of `AccountingConsensusRewards` and `ExecutionRewardsCollected` events

Author: apexearth

abi/fee-accumulator.json

@@ -0,0 +1,31 @@
+[
+  {
+    "inputs": [{ "internalType": "address", "name": "_strategy", "type": "address" }],
+    "stateMutability": "nonpayable",
+    "type": "constructor"
+  },
+  {
+    "anonymous": false,
+    "inputs": [
+      { "indexed": true, "internalType": "address", "name": "strategy", "type": "address" },
+      { "indexed": false, "internalType": "uint256", "name": "amount", "type": "uint256" }
+    ],
+    "name": "ExecutionRewardsCollected",
+    "type": "event"
+  },
+  {
+    "inputs": [],
+    "name": "STRATEGY",
+    "outputs": [{ "internalType": "address", "name": "", "type": "address" }],
+    "stateMutability": "view",
+    "type": "function"
+  },
+  {
+    "inputs": [],
+    "name": "collect",
+    "outputs": [{ "internalType": "uint256", "name": "eth", "type": "uint256" }],
+    "stateMutability": "nonpayable",
+    "type": "function"
+  },
+  { "stateMutability": "payable", "type": "receive" }
+]

db/migrations/1729880820715-Data.js renamed to db/migrations/1729897704620-Data.js

@@ -1384,7 +1384,7 @@ type LegacyStaker @entity {
   rewardAmount: BigInt!
 }
 
-type BeaconDepositEvent @entity  {
+type BeaconDepositEvent @entity {
   id: ID! # `chainId:logId`
   chainId: Int! @index
   address: String! @index
@@ -1400,13 +1400,33 @@ type BeaconDepositEvent @entity  {
   index: String!
 }
 
-type BeaconDepositPubkey @entity  {
+type BeaconDepositPubkey @entity {
   id: ID! # `pubkey`
   createDate: DateTime!
   lastUpdated: DateTime!
   count: Int!
   deposits: [BeaconDepositEvent!] @derivedFrom(field: "pubkey")
-}# OETH Vault: 0x39254033945AA2E4809Cc2977E7087BEE48bd7Ab
+}
+
+type AccountingConsensusRewards @entity {
+  id: ID! # `chainId:logId`
+  chainId: Int! @index
+  timestamp: DateTime! @index
+  blockNumber: Int! @index
+  address: String! @index
+  rewards: BigInt!
+}
+
+type ExecutionRewardsCollected @entity {
+  id: ID! # `chainId:logId`
+  chainId: Int! @index
+  timestamp: DateTime! @index
+  blockNumber: Int! @index
+  address: String! @index
+  strategy: String! @index
+  amount: BigInt!
+}
+# OETH Vault: 0x39254033945AA2E4809Cc2977E7087BEE48bd7Ab
 """
 The Vault entity tracks the OETH vault balance over time.
 """

schema.graphql

@@ -1,4 +1,4 @@
-type BeaconDepositEvent @entity  {
+type BeaconDepositEvent @entity {
   id: ID! # `chainId:logId`
   chainId: Int! @index
   address: String! @index
@@ -14,10 +14,29 @@ type BeaconDepositEvent @entity  {
   index: String!
 }
 
-type BeaconDepositPubkey @entity  {
+type BeaconDepositPubkey @entity {
   id: ID! # `pubkey`
   createDate: DateTime!
   lastUpdated: DateTime!
   count: Int!
   deposits: [BeaconDepositEvent!] @derivedFrom(field: "pubkey")
-}
+}
+
+type AccountingConsensusRewards @entity {
+  id: ID! # `chainId:logId`
+  chainId: Int! @index
+  timestamp: DateTime! @index
+  blockNumber: Int! @index
+  address: String! @index
+  rewards: BigInt!
+}
+
+type ExecutionRewardsCollected @entity {
+  id: ID! # `chainId:logId`
+  chainId: Int! @index
+  timestamp: DateTime! @index
+  blockNumber: Int! @index
+  address: String! @index
+  strategy: String! @index
+  amount: BigInt!
+}

schema/native-staking.graphql

@@ -0,0 +1,30 @@
+import * as p from '@subsquid/evm-codec'
+import { event, fun, viewFun, indexed, ContractBase } from '@subsquid/evm-abi'
+import type { EventParams as EParams, FunctionArguments, FunctionReturn } from '@subsquid/evm-abi'
+
+export const events = {
+    ExecutionRewardsCollected: event("0xc2acb502a0dc166a61cd83b914b480d76050e91a6797d7a833be84c4eace1dfe", "ExecutionRewardsCollected(address,uint256)", {"strategy": indexed(p.address), "amount": p.uint256}),
+}
+
+export const functions = {
+    STRATEGY: viewFun("0x185025ef", "STRATEGY()", {}, p.address),
+    collect: fun("0xe5225381", "collect()", {}, p.uint256),
+}
+
+export class Contract extends ContractBase {
+
+    STRATEGY() {
+        return this.eth_call(functions.STRATEGY, {})
+    }
+}
+
+/// Event types
+export type ExecutionRewardsCollectedEventArgs = EParams<typeof events.ExecutionRewardsCollected>
+
+/// Function types
+export type STRATEGYParams = FunctionArguments<typeof functions.STRATEGY>
+export type STRATEGYReturn = FunctionReturn<typeof functions.STRATEGY>
+
+export type CollectParams = FunctionArguments<typeof functions.collect>
+export type CollectReturn = FunctionReturn<typeof functions.collect>
+

src/abi/fee-accumulator.ts

@@ -0,0 +1,30 @@
+import {Entity as Entity_, Column as Column_, PrimaryColumn as PrimaryColumn_, IntColumn as IntColumn_, Index as Index_, DateTimeColumn as DateTimeColumn_, StringColumn as StringColumn_, BigIntColumn as BigIntColumn_} from "@subsquid/typeorm-store"
+
+@Entity_()
+export class AccountingConsensusRewards {
+    constructor(props?: Partial<AccountingConsensusRewards>) {
+        Object.assign(this, props)
+    }
+
+    @PrimaryColumn_()
+    id!: string
+
+    @Index_()
+    @IntColumn_({nullable: false})
+    chainId!: number
+
+    @Index_()
+    @DateTimeColumn_({nullable: false})
+    timestamp!: Date
+
+    @Index_()
+    @IntColumn_({nullable: false})
+    blockNumber!: number
+
+    @Index_()
+    @StringColumn_({nullable: false})
+    address!: string
+
+    @BigIntColumn_({nullable: false})
+    rewards!: bigint
+}

src/model/generated/accountingConsensusRewards.model.ts

@@ -0,0 +1,34 @@
+import {Entity as Entity_, Column as Column_, PrimaryColumn as PrimaryColumn_, IntColumn as IntColumn_, Index as Index_, DateTimeColumn as DateTimeColumn_, StringColumn as StringColumn_, BigIntColumn as BigIntColumn_} from "@subsquid/typeorm-store"
+
+@Entity_()
+export class ExecutionRewardsCollected {
+    constructor(props?: Partial<ExecutionRewardsCollected>) {
+        Object.assign(this, props)
+    }
+
+    @PrimaryColumn_()
+    id!: string
+
+    @Index_()
+    @IntColumn_({nullable: false})
+    chainId!: number
+
+    @Index_()
+    @DateTimeColumn_({nullable: false})
+    timestamp!: Date
+
+    @Index_()
+    @IntColumn_({nullable: false})
+    blockNumber!: number
+
+    @Index_()
+    @StringColumn_({nullable: false})
+    address!: string
+
+    @Index_()
+    @StringColumn_({nullable: false})
+    strategy!: string
+
+    @BigIntColumn_({nullable: false})
+    amount!: bigint
+}

src/model/generated/executionRewardsCollected.model.ts

@@ -108,6 +108,8 @@ export * from "./_governanceVoteType"
 export * from "./legacyStaker.model"
 export * from "./beaconDepositEvent.model"
 export * from "./beaconDepositPubkey.model"
+export * from "./accountingConsensusRewards.model"
+export * from "./executionRewardsCollected.model"
 export * from "./oethVault.model"
 export * from "./oethCurveLp.model"
 export * from "./oethFraxStaking.model"

src/model/generated/index.ts

@@ -1,6 +1,10 @@
+import * as feeAccumulatorAbi from '@abi/fee-accumulator'
+import * as nativeStakingAbi from '@abi/strategy-native-staking'
+import { AccountingConsensusRewards, ExecutionRewardsCollected } from '@model'
 import { Context } from '@processor'
 import { mainnetCurrencies } from '@shared/post-processors/exchange-rates/mainnetCurrencies'
 import { EvmBatchProcessor } from '@subsquid/evm-processor'
+import { createEventProcessor } from '@templates/events/createEventProcessor'
 import { IStrategyData, createStrategyProcessor, createStrategySetup } from '@templates/strategy'
 import { createStrategyRewardProcessor, createStrategyRewardSetup } from '@templates/strategy-rewards'
 import {
@@ -163,11 +167,50 @@ export const oethStrategies: readonly IStrategyData[] = [
 
 const strategies = oethStrategies
 
+const eventProcessors = [
+  ...OETH_NATIVE_STRATEGY_ADDRESSES.map((address) =>
+    createEventProcessor({
+      address,
+      eventName: 'AccountingConsensusRewards',
+      event: nativeStakingAbi.events.AccountingConsensusRewards,
+      from: 20046251,
+      Entity: AccountingConsensusRewards,
+      mapEntity: (ctx, block, log, decoded) =>
+        new AccountingConsensusRewards({
+          id: `${ctx.chain.id}:${log.id}`,
+          chainId: ctx.chain.id,
+          timestamp: new Date(block.header.timestamp),
+          blockNumber: block.header.height,
+          address,
+          rewards: decoded.amount,
+        }),
+    }),
+  ),
+  createEventProcessor({
+    address: addresses.oeth.nativeStakingFeeAccumulator,
+    eventName: 'ExecutionRewardsCollected',
+    event: feeAccumulatorAbi.events.ExecutionRewardsCollected,
+    from: 20046238,
+    Entity: ExecutionRewardsCollected,
+    mapEntity: (ctx, block, log, decoded) =>
+      new ExecutionRewardsCollected({
+        id: `${ctx.chain.id}:${log.id}`,
+        chainId: ctx.chain.id,
+        timestamp: new Date(block.header.timestamp),
+        blockNumber: block.header.height,
+        address: addresses.oeth.nativeStakingFeeAccumulator,
+        strategy: decoded.strategy,
+        amount: decoded.amount,
+      }),
+  }),
+]
+
 export const from = Math.min(...strategies.map((s) => s.from))
 
 export const setup = (processor: EvmBatchProcessor) => {
   strategies.forEach((s) => createStrategySetup(s)(processor))
   strategies.filter((s) => s.kind !== 'Vault').forEach((s) => createStrategyRewardSetup(s)(processor))
+  eventProcessors.forEach((p) => p.setup(processor))
 }
 
 const processors = [
@@ -176,5 +219,5 @@ const processors = [
 ]
 
 export const process = async (ctx: Context) => {
-  await Promise.all(processors.map((p) => p(ctx)))
+  await Promise.all([...processors.map((p) => p(ctx)), ...eventProcessors.map((p) => p.process(ctx))])
 }

src/oeth/processors/strategies.ts

@@ -0,0 +1,47 @@
+import { Block, Context, Log } from '@processor'
+import { event } from '@subsquid/evm-abi'
+import { IndexedCodecs } from '@subsquid/evm-abi/lib/abi-components/event'
+import { DecodedStruct, Struct } from '@subsquid/evm-codec'
+import { EvmBatchProcessor } from '@subsquid/evm-processor'
+import { Entity } from '@subsquid/typeorm-store/lib/store'
+import { logFilter } from '@utils/logFilter'
+import { EntityClassT } from '@utils/type'
+
+export const createEventProcessor = <T extends Struct, EventEntity extends Entity>(params: {
+  eventName: string
+  event: ReturnType<typeof event<T>>
+  address: string
+  from: number
+  Entity: EntityClassT<EventEntity>
+  mapEntity: (ctx: Context, block: Block, log: Log, decoded: DecodedStruct<IndexedCodecs<T>>) => EventEntity
+  extraFilterArgs?: {
+    topic1?: string[]
+    topic2?: string[]
+    topic3?: string[]
+  }
+}) => {
+  const filter = logFilter({
+    address: [params.address],
+    range: { from: params.from },
+    topic0: [params.event.topic],
+    ...params.extraFilterArgs,
+  })
+  const setup = (p: EvmBatchProcessor) => {
+    p.addLog(filter.value)
+  }
+  const process = async (ctx: Context) => {
+    const entities: EventEntity[] = []
+    for (const block of ctx.blocks) {
+      for (const log of block.logs) {
+        if (filter.matches(log)) {
+          const decoded = params.event.decode(log)
+          const entity = new params.Entity(params.mapEntity(ctx, block, log, decoded))
+          entities.push(entity)
+        }
+      }
+    }
+    await ctx.store.insert(entities)
+  }
+
+  return { from: params.from, setup, process }
+}