Feat: Prefetch Function Support in Adapter (#3027)

* Feat: Prefetch Function Support in Adapter

* remove unused vars

* type error

* Fix: remove timestamp from prefetch

* Fix

* minor fix

---------

Co-authored-by: g1nt0ki <[email protected]>

Author: treeoflife2

adapters/types.ts

@@ -39,6 +39,7 @@ export type FetchOptions = {
   getStartBlock: () => Promise<number>,
   getEndBlock: () => Promise<number>,
   dateString: string,
+  preFetchedResults?: any,
 }
 
 export type FetchGetLogsOptions = {
@@ -98,6 +99,7 @@ export type AdapterBase = {
   protocolType?: ProtocolType;
   version?: number;
   deadFrom?: string;
+  prefetch?: FetchV2;
 }
 
 export type SimpleAdapter = AdapterBase & {

adapters/utils/runAdapter.ts

@@ -17,8 +17,9 @@ function getUnixTimeNow() {
 export default async function runAdapter(volumeAdapter: BaseAdapter, cleanCurrentDayTimestamp: number, chainBlocks: ChainBlocks, id?: string, version?: string, {
   adapterVersion = 1,
   isTest = false,
+  _module = {},
 }: any = {}) {
-
+  const prefetch = _module?.prefetch
   const closeToCurrentTime = Math.trunc(Date.now() / 1000) - cleanCurrentDayTimestamp < 24 * 60 * 60 // 12 hours
   const chains = Object.keys(volumeAdapter).filter(c => c !== DISABLED_ADAPTER_KEY)
   const validStart = {} as {
@@ -29,6 +30,16 @@ export default async function runAdapter(volumeAdapter: BaseAdapter, cleanCurren
   }
   await Promise.all(chains.map(setChainValidStart))
 
+  // Run prefetch if provided
+  let preFetchedResults: any = null;
+  if (typeof prefetch === 'function') {
+    const firstChain = chains.find(chain => validStart[chain]?.canRun);
+    if (firstChain) {
+      const options = await getOptionsObject(cleanCurrentDayTimestamp, firstChain, chainBlocks);
+      preFetchedResults = await prefetch(options);
+    }
+  }
+
   const response = await Promise.all(chains.filter(chain => {
     const res = validStart[chain]?.canRun
     if (isTest && !res) console.log(`Skipping ${chain} because the configured start time is ${new Date(validStart[chain]?.startTimestamp * 1e3).toUTCString()} \n\n`)
@@ -49,6 +60,10 @@ export default async function runAdapter(volumeAdapter: BaseAdapter, cleanCurren
     const fetchFunction = volumeAdapter[chain].customBackfill ?? volumeAdapter[chain].fetch
     try {
       const options = await getOptionsObject(cleanCurrentDayTimestamp, chain, chainBlocks)
+      if (preFetchedResults !== null) {
+        options.preFetchedResults = preFetchedResults;
+      }
+
       let result: any
       if (adapterVersion === 1) {
         result = await (fetchFunction as Fetch)(options.toTimestamp, chainBlocks, options);

cli/testAdapter.ts

@@ -83,6 +83,7 @@ const passedFile = path.resolve(process.cwd(), `./${adapterType}/${process.argv[
     // Get adapter
     const volumes = await runAdapter(adapter, endTimestamp, chainBlocks, undefined, undefined, {
       adapterVersion,
+      _module: module,
     })
     printVolumes(volumes, adapter)
     console.info("\n")
@@ -91,6 +92,7 @@ const passedFile = path.resolve(process.cwd(), `./${adapterType}/${process.argv[
     const allVolumes = await Promise.all(Object.entries(breakdownAdapter).map(([version, adapter]) =>
       runAdapter(adapter, endTimestamp, chainBlocks, undefined, undefined, {
         adapterVersion,
+        prefetch: adapter?.prefetch,
         isTest: true,
       }).then(res => ({ version, res }))
     ))

fees/rainbow-wallet.ts

@@ -1,25 +1,62 @@
-import { Adapter, FetchOptions, FetchResultFees } from "../adapters/types";
+import { Adapter, FetchOptions, } from "../adapters/types";
 import { CHAIN } from "../helpers/chains";
-import { getTokenDiff } from "../helpers/token";
+import { queryDuneSql } from "../helpers/dune";
 
 const rainbowRouter = '0x00000000009726632680fb29d3f7a9734e3010e2'
 
-const fetch: any = async (timestamp: number, _: any, options: FetchOptions) => {
-  const { createBalances, getLogs, } = options
-  const dailyFees = createBalances()
-
-  const ethLogs = await getLogs({ target: rainbowRouter, eventAbi: 'event EthWithdrawn(address indexed target, uint256 amount)' })
-  const tokenLogs = await getLogs({ target: rainbowRouter, eventAbi: 'event TokenWithdrawn(address indexed token, address indexed target, uint256 amount)' })
-  let extraTokens = new Set<string>()
+const RainBowRouter = {
+  [CHAIN.ETHEREUM]: rainbowRouter,
+  [CHAIN.OPTIMISM]: rainbowRouter,
+  [CHAIN.BSC]: rainbowRouter,
+  [CHAIN.UNICHAIN]: '0x2a0332E28913A06Fa924d40A3E2160f763010417',
+  [CHAIN.POLYGON]: rainbowRouter,
+  [CHAIN.BASE]: rainbowRouter,
+  [CHAIN.ARBITRUM]: rainbowRouter,
+  [CHAIN.AVAX]: rainbowRouter,
+  [CHAIN.INK]: rainbowRouter,
+  [CHAIN.BERACHAIN]: rainbowRouter,
+  [CHAIN.BLAST]: rainbowRouter,
+  [CHAIN.ZORA]: '0xA61550E9ddD2797E16489db09343162BE98d9483',
+  [CHAIN.APECHAIN]: rainbowRouter,
+  [CHAIN.GRAVITY]: rainbowRouter,
+}
 
-  ethLogs.forEach((log: any) => dailyFees.addGasToken(log.amount))
-  tokenLogs.forEach((log: any) => {
-    extraTokens.add(log.token)
-    dailyFees.add(log.token, log.amount)
-  })
+// Prefetch function that will run once before any fetch calls
+// don't do console.log(options) as there is circular dependency in ChainApi
+const prefetch = async (options: FetchOptions) => {
+  return queryDuneSql(options, `
+    SELECT 
+        CASE 
+            WHEN blockchain = 'bnb' THEN 'bsc'
+            WHEN blockchain = 'avalanche_c' THEN 'avax'
+            ELSE blockchain
+        END as chain,
+        sum(amount_usd) as volume,
+        sum(amount_usd * 0.0085) as fees
+    FROM dex.trades
+    WHERE (
+        (tx_to = 0x00000000009726632680fb29d3f7a9734e3010e2 AND blockchain NOT IN ('unichain', 'zora'))
+        OR
+        (tx_to = 0x2a0332E28913A06Fa924d40A3E2160f763010417 AND blockchain = 'unichain')
+        OR
+        (tx_to = 0xA61550E9ddD2797E16489db09343162BE98d9483 AND blockchain = 'zora')
+    )
+    AND block_time >= from_unixtime(${options.startTimestamp})
+    AND block_time <= from_unixtime(${options.endTimestamp})
+    GROUP BY 1
+  `);
+};
 
-  await getTokenDiff({ options, target: rainbowRouter, balances: dailyFees, extraTokens: [...extraTokens], })
-  dailyFees.removeNegativeBalances()
+const fetch: any = async (timestamp: number, _: any, options: FetchOptions) => {
+  const results = options.preFetchedResults || [];
+  
+  let dailyFees = 0;
+  for (const result of results) {
+    if (result.chain === options.chain) {
+      dailyFees = result.fees;
+      break;
+    }
+  }
 
   return {
     timestamp,
@@ -37,16 +74,10 @@ const methodology = {
 const chainAdapter = { fetch, start: '2023-01-01', meta: { methodology } }
 
 const adapter: Adapter = {
-  adapter: {
-    [CHAIN.ETHEREUM]: chainAdapter,
-    [CHAIN.OPTIMISM]: chainAdapter,
-    [CHAIN.ARBITRUM]: chainAdapter,
-    [CHAIN.POLYGON]: chainAdapter,
-    [CHAIN.BASE]: chainAdapter,
-    [CHAIN.BSC]: chainAdapter,
-    [CHAIN.AVAX]: chainAdapter,
-  },
-
+  adapter: Object.fromEntries(Object.entries(RainBowRouter).map(
+    ([chain]) => [chain, chainAdapter]
+  )),
+  prefetch: prefetch,
 }
 
 export default adapter;