add arm swap processing

update how we record the erc20 holder count for otokens

schema.graphql

@@ -628,6 +628,18 @@ type ArmWithdrawalRequest @entity {
   claimed: Boolean!
 }
 
+type ArmSwap @entity {
+  id: ID!
+  chainId: Int! @index
+  txHash: String! @index
+  timestamp: DateTime! @index
+  blockNumber: Int! @index
+  address: String! @index
+  from: String!
+  assets0: BigInt!
+  assets1: BigInt!
+}
+
 type TraderateChanged @entity {
   id: ID!
   chainId: Int! @index

squid.yaml

@@ -6,6 +6,7 @@ build:
 deploy:
   env:
     TS_NODE_BASEURL: './lib'
+    # DEBUG_PERF: 'true'
   addons:
     postgres:
     rpc:

src/model/generated/armSwap.model.ts

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

src/model/generated/index.ts

@@ -49,6 +49,7 @@ export * from "./arm.model"
 export * from "./armState.model"
 export * from "./armDailyStat.model"
 export * from "./armWithdrawalRequest.model"
+export * from "./armSwap.model"
 export * from "./traderateChanged.model"
 export * from "./erc20.model"
 export * from "./erc20Holder.model"

src/processor.ts

@@ -131,6 +131,9 @@ export const run = ({ chainId = 1, stateSchema, processors, postProcessors, vali
   if (process.env.PROCESSOR) {
     processors = processors.filter((p) => p.name?.includes(process.env.PROCESSOR!))
   }
+  if (process.env.PROCESSOR) {
+    postProcessors = postProcessors?.filter((p) => p.name?.includes(process.env.PROCESSOR!))
+  }
 
   console.log('Processors:\n  - ', processors.map((p) => p.name).join('\n  - '))
 
@@ -154,15 +157,18 @@ export const run = ({ chainId = 1, stateSchema, processors, postProcessors, vali
   processors.forEach((p) => p.setup?.(evmBatchProcessor, config.chain))
   postProcessors?.forEach((p) => p.setup?.(evmBatchProcessor, config.chain))
   const frequencyTracker = blockFrequencyTracker({ from })
+  let contextTime = Date.now()
   evmBatchProcessor.run(
     new TypeormDatabase({
       stateSchema,
       supportHotBlocks: true,
       isolationLevel: 'READ COMMITTED',
     }),
     async (_ctx) => {
-      const contextTime = Date.now()
       const ctx = _ctx as Context
+      if (!ctx.isHead && Date.now() - contextTime > 5000) {
+        ctx.log.info(`===== !! Slow Context !! ===== (${Date.now() - contextTime}ms)`)
+      }
       try {
         ctx.chain = config.chain
         ctx.__state = new Map<string, unknown>()
@@ -242,6 +248,7 @@ export const run = ({ chainId = 1, stateSchema, processors, postProcessors, vali
             `===== End of Context ===== (${Date.now() - contextTime}ms, ${ctx.blocks.at(-1)?.header.height})`,
           )
         }
+        contextTime = Date.now()
       }
     },
   )

src/templates/origin-arm/origin-arm.graphql

@@ -77,6 +77,18 @@ type ArmWithdrawalRequest @entity {
   claimed: Boolean!
 }
 
+type ArmSwap @entity {
+  id: ID!
+  chainId: Int! @index
+  txHash: String! @index
+  timestamp: DateTime! @index
+  blockNumber: Int! @index
+  address: String! @index
+  from: String!
+  assets0: BigInt!
+  assets1: BigInt!
+}
+
 type TraderateChanged @entity {
   id: ID!
   chainId: Int! @index

src/templates/origin-arm/origin-arm.ts

@@ -6,7 +6,7 @@ import { formatUnits } from 'viem'
 import * as erc20Abi from '@abi/erc20'
 import * as originLidoArmAbi from '@abi/origin-lido-arm'
 import * as originLidoArmCapManagerAbi from '@abi/origin-lido-arm-cap-manager'
-import { Arm, ArmDailyStat, ArmState, ArmWithdrawalRequest, TraderateChanged } from '@model'
+import { Arm, ArmDailyStat, ArmState, ArmSwap, ArmWithdrawalRequest, TraderateChanged } from '@model'
 import { Block, Context, Processor } from '@processor'
 import { ensureExchangeRate } from '@shared/post-processors/exchange-rates'
 import { EvmBatchProcessor } from '@subsquid/evm-processor'
@@ -15,6 +15,7 @@ import { createEventProcessor } from '@templates/events/createEventProcessor'
 import { blockFrequencyTracker } from '@utils/blockFrequencyUpdater'
 import { calculateAPY } from '@utils/calculateAPY'
 import { logFilter } from '@utils/logFilter'
+import { traceFilter } from '@utils/traceFilter'
 
 export const createOriginARMProcessors = ({
   name,
@@ -52,6 +53,20 @@ export const createOriginARMProcessors = ({
     topic0: [originLidoArmAbi.events.FeeCollected.topic],
     range: { from },
   })
+  const swapFilter = traceFilter({
+    type: ['call'],
+    callTo: [armAddress],
+    callSighash: [
+      originLidoArmAbi.functions['swapExactTokensForTokens(uint256,uint256,address[],address,uint256)'].sighash,
+      originLidoArmAbi.functions['swapExactTokensForTokens(address,address,uint256,uint256,address)'].sighash,
+      originLidoArmAbi.functions['swapTokensForExactTokens(uint256,uint256,address[],address,uint256)'].sighash,
+      originLidoArmAbi.functions['swapTokensForExactTokens(address,address,uint256,uint256,address)'].sighash,
+    ],
+    range: { from },
+    transaction: true,
+    transactionLogs: true,
+  })
+
   const tradeRateProcessor = createEventProcessor({
     event: originLidoArmAbi.events.TraderateChanged,
     address: armAddress,
@@ -113,6 +128,7 @@ export const createOriginARMProcessors = ({
         p.addLog(depositFilter.value)
         p.addLog(withdrawalFilter.value)
         p.addLog(feeCollectedFilter.value)
+        p.addTrace(swapFilter.value)
         tradeRateProcessor.setup(p)
       },
       initialize,
@@ -124,6 +140,7 @@ export const createOriginARMProcessors = ({
         const states: ArmState[] = []
         const dailyStatsMap = new Map<string, ArmDailyStat>()
         const redemptionMap = new Map<string, ArmWithdrawalRequest>()
+        const swaps: ArmSwap[] = []
         const getStateId = (block: Block) => `${ctx.chain.id}:${block.header.height}:${armAddress}`
         const getPreviousState = async () => {
           return (
@@ -188,7 +205,7 @@ export const createOriginARMProcessors = ({
         const calculateTotalYield = (state: ArmState) =>
           state.totalAssets - state.totalDeposits + state.totalWithdrawals
 
-        for (const block of ctx.blocks) {
+        for (const block of ctx.blocksWithContent) {
           for (const log of block.logs) {
             // ArmWithdrawalRequest
             if (redeemRequestedFilter.matches(log)) {
@@ -236,6 +253,61 @@ export const createOriginARMProcessors = ({
               state.totalFees += event.fee
             }
           }
+
+          const swapHandledTransactions = new Set<string>()
+          for (const trace of block.traces) {
+            if (trace.type === 'call' && swapFilter.matches(trace)) {
+              const transactionHash = trace.transaction?.hash ?? ''
+              if (!swapHandledTransactions.has(transactionHash)) {
+                swapHandledTransactions.add(transactionHash)
+                if (!trace.transaction) throw new Error('Transaction not found')
+                const transfers = trace.transaction.logs.filter(
+                  (log) => log.topics[0] === erc20Abi.events.Transfer.topic,
+                )
+                const transfers0 = transfers
+                  .filter((log) => log.address === armEntity.token0)
+                  .map((log) => erc20Abi.events.Transfer.decode(log))
+                const transfers1 = transfers
+                  .filter((log) => log.address === armEntity.token1)
+                  .map((log) => erc20Abi.events.Transfer.decode(log))
+
+                const assets0 = transfers0.reduce(
+                  (acc, data) =>
+                    data.from.toLowerCase() === armAddress
+                      ? acc - data.value
+                      : data.to.toLowerCase() === armAddress
+                      ? acc + data.value
+                      : acc,
+                  0n,
+                )
+                const assets1 = transfers1.reduce(
+                  (acc, data) =>
+                    data.from.toLowerCase() === armAddress
+                      ? acc - data.value
+                      : data.to.toLowerCase() === armAddress
+                      ? acc + data.value
+                      : acc,
+                  0n,
+                )
+
+                swaps.push(
+                  new ArmSwap({
+                    id: `${ctx.chain.id}::${transactionHash}:${trace.traceAddress.join(':')}`,
+                    chainId: ctx.chain.id,
+                    txHash: transactionHash,
+                    timestamp: new Date(block.header.timestamp),
+                    blockNumber: block.header.height,
+                    address: armAddress,
+                    from: trace.action.from,
+                    assets0,
+                    assets1,
+                  }),
+                )
+              }
+            }
+          }
+        }
+        for (const block of ctx.blocks) {
           if (tracker(ctx, block)) {
             // ArmState
             const [state, yesterdayState, rateUSD] = await Promise.all([
@@ -288,6 +360,7 @@ export const createOriginARMProcessors = ({
         await ctx.store.insert(states)
         await ctx.store.upsert([...dailyStatsMap.values()])
         await ctx.store.upsert([...redemptionMap.values()])
+        await ctx.store.insert(swaps)
         await tradeRateProcessor.process(ctx)
       },
     },

src/templates/otoken/otoken.ts

@@ -760,7 +760,7 @@ export const createOTokenProcessor = (params: {
           totalSupply: latest?.totalSupply ?? 0n,
           rebasingSupply: latest?.rebasingSupply ?? 0n,
           nonRebasingSupply: latest?.nonRebasingSupply ?? 0n,
-          holderCount: owners!.size ?? 0n,
+          holderCount: [...(owners?.values() ?? [])].reduce((acc, owner) => acc + (owner.balance > 0 ? 1 : 0), 0),
         })
         result.otokens.push(otokenObject)
       }

src/utils/traceFilter.ts

@@ -12,37 +12,28 @@ const lower = (hex: string) => hex.toLowerCase()
 export const traceFilter = (
   filter: Pick<
     Parameters<EvmBatchProcessor['addTrace']>[0] & { type: ['call'] },
-    'type' | 'callTo' | 'callSighash' | 'transaction' | 'range'
+    'type' | 'callTo' | 'callSighash' | 'transaction' | 'transactionLogs' | 'range'
   >,
 ) => {
   filter = {
     type: filter.type,
     callTo: filter.callTo?.map(lower),
     callSighash: filter.callSighash?.map(lower),
     transaction: filter.transaction,
+    transactionLogs: filter.transactionLogs,
     range: filter.range,
   }
   return {
     value: filter,
     matches(trace: Trace) {
       if (filter.type && !filter.type.includes(trace.type)) return false
-      if (
-        filter.callTo &&
-        trace.type === 'call' &&
-        !filter.callTo.includes(trace.action.to)
-      )
-        return false
-      if (
-        filter.callSighash &&
-        trace.type === 'call' &&
-        !filter.callSighash.includes(trace.action.sighash)
-      )
+      if (filter.callTo && trace.type === 'call' && !filter.callTo.includes(trace.action.to)) return false
+      if (filter.callSighash && trace.type === 'call' && !filter.callSighash.includes(trace.action.sighash))
         return false
 
       if (
         filter.range &&
-        (trace.block.height < filter.range.from ||
-          (filter.range.to && trace.block.height > filter.range.to))
+        (trace.block.height < filter.range.from || (filter.range.to && trace.block.height > filter.range.to))
       ) {
         return false
       }