Merge pull request #954 from Merit-Systems/feat/base-bitquery-transfers

Switch Base transfer sync to Bitquery

Author: shafu0x

sync/transfers/trigger/chains/evm/base/bitquery/config.ts

@@ -0,0 +1,23 @@
+import { FACILITATORS_BY_CHAIN } from '@/trigger/lib/facilitators';
+import { ONE_MINUTE_IN_SECONDS } from '@/trigger/lib/constants';
+import type { SyncConfig } from '@/trigger/types';
+import { Network, PaginationStrategy, QueryProvider } from '@/trigger/types';
+import { buildQuery, transformResponse } from './query';
+
+export const baseBitqueryConfig: SyncConfig = {
+  cron: '*/5 * * * *',
+  maxDurationInSeconds: ONE_MINUTE_IN_SECONDS * 15,
+  chain: 'base',
+  provider: QueryProvider.BITQUERY,
+  apiUrl: 'https://streaming.bitquery.io/graphql',
+  paginationStrategy: PaginationStrategy.OFFSET,
+  limit: 25_000,
+  facilitators: FACILITATORS_BY_CHAIN(Network.BASE),
+  buildQuery,
+  transformResponse,
+  enabled: true,
+  machine: 'medium-1x',
+  splitSyncByFacilitator: true,
+  resumeFromProviders: [QueryProvider.CDP, QueryProvider.BITQUERY],
+  resumeOverlapMs: ONE_MINUTE_IN_SECONDS * 1000,
+};

sync/transfers/trigger/chains/evm/base/bitquery/query.ts

@@ -0,0 +1,120 @@
+import type {
+  EvmBitQueryTransferRow,
+  Facilitator,
+  FacilitatorConfig,
+  SyncConfig,
+  TransferEventData,
+} from '@/trigger/types';
+
+export function buildQuery(
+  config: SyncConfig,
+  facilitatorConfig: FacilitatorConfig,
+  since: Date,
+  now: Date,
+  offset = 0
+): string {
+  return `
+    {
+      EVM(dataset: combined, network: base) {
+        Transfers(
+          limit: { count: ${config.limit}, offset: ${offset} }
+          where: {
+            Transfer: {
+              Success: true
+              Currency: {
+                SmartContract: {
+                  is: "${facilitatorConfig.token.address}"
+                }
+              }
+            }
+            Transaction: {
+              From: {
+                is: "${facilitatorConfig.address.toLowerCase()}"
+              }
+            }
+            Block: {
+              Time: {
+                since: "${since.toISOString()}"
+                till: "${now.toISOString()}"
+              }
+            }
+          }
+          orderBy: {
+            ascending: [
+              Block_Number,
+              Transaction_Index,
+              Call_Index,
+              Log_Index,
+              Transfer_Index,
+              Transfer_Type
+            ]
+          }
+        ) {
+          Block {
+            Time
+            Number
+          }
+          Transaction {
+            Hash
+            From
+            Index
+          }
+          Transfer {
+            Amount
+            Sender
+            Receiver
+            Index
+            Type
+            Success
+            Currency {
+              SmartContract
+              Decimals
+              Symbol
+              Name
+            }
+          }
+          Log {
+            Index
+          }
+          Call {
+            Index
+          }
+        }
+      }
+    }
+  `;
+}
+
+export function transformResponse(
+  data: unknown,
+  config: SyncConfig,
+  facilitator: Facilitator,
+  facilitatorConfig: FacilitatorConfig
+): TransferEventData[] {
+  const transfers = (data as { EVM: { Transfers: EvmBitQueryTransferRow[] } })
+    .EVM.Transfers;
+  const multiplier = 10 ** facilitatorConfig.token.decimals;
+
+  return transfers.map(transfer => {
+    if (transfer.Log?.Index === undefined || transfer.Log.Index === null) {
+      throw new Error(
+        `Bitquery transfer is missing Log.Index for ${transfer.Transaction.Hash}`
+      );
+    }
+
+    return {
+      address: transfer.Transfer.Currency.SmartContract.toLowerCase(),
+      transaction_from: transfer.Transaction.From.toLowerCase(),
+      sender: transfer.Transfer.Sender.toLowerCase(),
+      recipient: transfer.Transfer.Receiver.toLowerCase(),
+      amount: Math.round(parseFloat(transfer.Transfer.Amount) * multiplier),
+      block_timestamp: new Date(transfer.Block.Time),
+      tx_hash: transfer.Transaction.Hash.toLowerCase(),
+      log_index: transfer.Log.Index,
+      chain: config.chain,
+      provider: config.provider,
+      decimals: facilitatorConfig.token.decimals,
+      facilitator_id: facilitator.id,
+    };
+  });
+}

sync/transfers/trigger/chains/evm/base/bitquery/sync.ts

@@ -0,0 +1,5 @@
+import { createChainSyncTask } from '../../../../sync';
+import { baseBitqueryConfig } from './config';
+
+export const baseBitquerySyncTransfers =
+  createChainSyncTask(baseBitqueryConfig);

sync/transfers/trigger/chains/evm/base/cdp/config.ts

@@ -16,7 +16,7 @@ export const baseCdpConfig: SyncConfig = {
   facilitators: FACILITATORS_BY_CHAIN(Network.BASE),
   buildQuery,
   transformResponse,
-  enabled: true,
+  enabled: false,
   machine: 'medium-1x',
   splitSyncByFacilitator: true,
 };

sync/transfers/trigger/fetch/bitquery/fetch.ts

@@ -14,7 +14,8 @@ export async function fetchWithOffsetPagination(
   facilitator: Facilitator,
   facilitatorConfig: FacilitatorConfig,
   since: Date,
-  now: Date
+  now: Date,
+  onBatchFetched?: (batch: TransferEventData[]) => Promise<void>
 ): Promise<TransferEventData[]> {
   const allTransfers = [];
   let offset = 0;
@@ -39,6 +40,10 @@ export async function fetchWithOffsetPagination(
 
     allTransfers.push(...transfers);
 
+    if (onBatchFetched && transfers.length > 0) {
+      await onBatchFetched(transfers);
+    }
+
     if (transfers.length < config.limit) {
       hasMore = false;
     } else {

sync/transfers/trigger/fetch/fetch.ts

@@ -137,13 +137,10 @@ async function fetchWithOffset(
       facilitator,
       facilitatorConfig,
       since,
-      now
+      now,
+      onBatchFetched
     );
 
-    if (onBatchFetched && results.length > 0) {
-      await onBatchFetched(results);
-    }
-
     return { totalFetched: results.length };
   }
 

sync/transfers/trigger/sync.ts

@@ -5,6 +5,27 @@ import { fetchTransfers } from './fetch/fetch';
 
 import type { Facilitator, SyncConfig } from './types';
 
+function normalizeTransactionFrom(chain: string, address: string): string {
+  return chain === Network.SOLANA.toString() ? address : address.toLowerCase();
+}
+
+function getResumeSince(
+  latestTimestamp: Date | undefined,
+  syncConfig: SyncConfig,
+  syncStartDate: Date
+): Date {
+  if (!latestTimestamp) {
+    return syncStartDate;
+  }
+
+  const resumeTimestamp =
+    syncConfig.resumeOverlapMs !== undefined
+      ? latestTimestamp.getTime() - syncConfig.resumeOverlapMs
+      : latestTimestamp.getTime() + 1000;
+
+  return new Date(Math.max(resumeTimestamp, syncStartDate.getTime()));
+}
+
 async function syncFacilitator(
   syncConfig: SyncConfig,
   facilitator: Facilitator,
@@ -24,27 +45,32 @@ async function syncFacilitator(
       `[${syncConfig.chain}] Getting most recent transfer for ${facilitator.id}:${facilitatorConfig.address}`
     );
 
+    const resumeFromProviders = syncConfig.resumeFromProviders ?? [
+      syncConfig.provider,
+    ];
+
     const mostRecentTransfer = await getTransferEvents({
       orderBy: { block_timestamp: 'desc' },
       take: 1,
       where: {
         chain: syncConfig.chain,
-        transaction_from:
-          syncConfig.chain === Network.SOLANA.toString()
-            ? facilitatorConfig.address
-            : facilitatorConfig.address.toLowerCase(),
-        provider: syncConfig.provider,
+        transaction_from: normalizeTransactionFrom(
+          syncConfig.chain,
+          facilitatorConfig.address
+        ),
+        provider: { in: resumeFromProviders },
       },
     });
 
     logger.log(
-      `[${syncConfig.chain}] Most recent transfer: ${mostRecentTransfer[0]?.block_timestamp?.toISOString()}`
+      `[${syncConfig.chain}] Most recent transfer from ${resumeFromProviders.join(',')}: ${mostRecentTransfer[0]?.block_timestamp?.toISOString()}`
     );
 
-    // Start from 1 second after the most recent transfer to avoid re-fetching it
-    const since = mostRecentTransfer[0]?.block_timestamp
-      ? new Date(mostRecentTransfer[0].block_timestamp.getTime() + 1000)
-      : facilitatorConfig.syncStartDate;
+    const since = getResumeSince(
+      mostRecentTransfer[0]?.block_timestamp,
+      syncConfig,
+      facilitatorConfig.syncStartDate
+    );
 
     logger.log(
       `[${syncConfig.chain}] Syncing ${facilitator.id}:${facilitatorConfig.address} from ${since.toISOString()} to ${now.toISOString()}`

sync/transfers/trigger/types.ts

@@ -80,6 +80,8 @@ export type SyncConfig = QueryConfig & {
   enabled: boolean;
   machine: 'small-1x' | 'medium-1x' | 'large-2x';
   splitSyncByFacilitator?: boolean;
+  resumeFromProviders?: QueryProvider[];
+  resumeOverlapMs?: number;
 };
 
 export interface EvmChainConfig {
@@ -147,3 +149,35 @@ export interface BitQueryTransferRowStream {
     From: string;
   };
 }
+
+export interface EvmBitQueryTransferRow {
+  Transfer: {
+    Amount: string;
+    Sender: string;
+    Receiver: string;
+    Index: number;
+    Type: string;
+    Success: boolean;
+    Currency: {
+      Name: string;
+      SmartContract: string;
+      Symbol: string;
+      Decimals: number;
+    };
+  };
+  Block: {
+    Time: string;
+    Number: string;
+  };
+  Transaction: {
+    Hash: string;
+    From: string;
+    Index: number;
+  };
+  Log?: {
+    Index: number | null;
+  };
+  Call?: {
+    Index: number | null;
+  };
+}