Merge pull request #120 from paraswap/fix/epoch-46-and-48-retrospecti…

…ve-patch Fix/epoch 46 and 48 retrospective patch
paraswap · Jul 8, 2024 · 306dac3 · 306dac3
2 parents 48ae8b3 + 31c6629
commit 306dac3
Show file tree

Hide file tree

Showing 14 changed files with 649 additions and 142 deletions.
diff --git a/scripts/gas-refund-program/computeGasRefund.ts b/scripts/gas-refund-program/computeGasRefund.ts
@@ -38,6 +38,11 @@ startComputingGasRefundAllChains()
     process.exit(0);
   })
   .catch(err => {
-    logger.error('startComputingGasRefundAllChains exited with error:', err, err.response?.data);
+    logger.error(
+      'startComputingGasRefundAllChains exited with error:',
+      err,
+      err.response?.data,
+      err.request?.path,
+    );
     process.exit(1);
   });
diff --git a/scripts/gas-refund-program/transactions-indexing/fetchRefundableTransactions.ts b/scripts/gas-refund-program/transactions-indexing/fetchRefundableTransactions.ts
@@ -8,58 +8,37 @@ import {
 } from '../persistance/db-persistance';
 import { getAllTXs, getContractAddresses } from './transaction-resolver';
 import {
-  GasRefundTransactionData,
   TransactionStatus,
   GasRefundV2EpochFlip,
   getRefundPercent,
   getMinStake,
 } from '../../../src/lib/gas-refund/gas-refund';
 import { ONE_HOUR_SEC } from '../../../src/lib/utils/helpers';
 import { PriceResolverFn } from '../token-pricing/psp-chaincurrency-pricing';
-import StakesTracker, {
-  StakedScoreV1,
-  StakedScoreV2,
-} from '../staking/stakes-tracker';
+import StakesTracker from '../staking/stakes-tracker';
 import { MIGRATION_SEPSP2_100_PERCENT_KEY } from '../staking/2.0/utils';
 import { isTruthy } from '../../../src/lib/utils';
 import { AUGUSTUS_SWAPPERS_V6_OMNICHAIN } from '../../../src/lib/constants';
 import { fetchParaswapV6StakersTransactions } from '../../../src/lib/paraswap-v6-stakers-transactions';
 import { ExtendedCovalentGasRefundTransaction } from '../../../src/types-from-scripts';
+import { GasRefundTransactionDataWithStakeScore, TxProcessorFn } from './types';
+import { applyEpoch46Patch } from '../../per-epoch-patches/epoch-46';
+import { applyEpoch48Patch } from '../../per-epoch-patches/epoch-48';
+import { PatchInput } from '../../per-epoch-patches/types';
+import type { Logger } from 'log4js';
 
-// empirically set to maximise on processing time without penalising memory and fetching constraigns
-const SLICE_DURATION = 4 * ONE_HOUR_SEC;
-
-type GasRefundTransactionDataWithStakeScore = GasRefundTransactionData & {
-  stakeScore: StakedScoreV2 | StakedScoreV1;
-};
-
-export async function fetchRefundableTransactions({
+function constructTransactionsProcessor({
   chainId,
-  startTimestamp,
   endTimestamp,
   epoch,
   resolvePrice,
 }: {
   chainId: number;
-  startTimestamp: number;
   endTimestamp: number;
   epoch: number;
   resolvePrice: PriceResolverFn;
-}): Promise<GasRefundTransactionDataWithStakeScore[]> {
-  const logger = global.LOGGER(
-    `GRP:fetchRefundableTransactions: epoch=${epoch}, chainId=${chainId}`,
-  );
-
-  logger.info(`start indexing between ${startTimestamp} and ${endTimestamp}`);
-
-  const lastTimestampTxByContract = await fetchLastTimestampTxByContract({
-    chainId,
-    epoch,
-  });
-
-  const allButV6ContractAddresses = getContractAddresses({ epoch, chainId });
-
-  async function filterFormatAndStoreRefundableTransactions(
+}): TxProcessorFn {
+  return async function filterAndFormatRefundableTransactions(
     transactions: ExtendedCovalentGasRefundTransaction[],
     computeRefundPercent: (
       epoch: number,
@@ -155,104 +134,195 @@ export async function fetchRefundableTransactions({
         })
         .filter(isTruthy);
 
-    if (refundableTransactions.length > 0) {
-      // TODO
-      // logger.info(
-      //   `updating ${refundableTransactions.length} transactions for chainId=${chainId} epoch=${epoch} _startTimestampSlice=${_startTimestampSlice} _endTimestampSlice=${_endTimestampSlice}`,
-      // );
-      await writeTransactions(refundableTransactions);
-
-      const stakeScoreEntries = refundableTransactions
-        .map(({ stakeScore, ...transaction }) =>
-          composeGasRefundTransactionStakeSnapshots(transaction, stakeScore),
-        )
-        .flat();
-
-      await writeStakeScoreSnapshots(stakeScoreEntries);
-    }
     return refundableTransactions;
-  }
+  };
+}
 
-  return (
-    await Promise.all([
-      ...allButV6ContractAddresses.map(async contractAddress => {
-        assert(contractAddress, 'contractAddress should be defined');
-        const lastTimestampProcessed =
-          lastTimestampTxByContract[contractAddress] || 0;
+// empirically set to maximise on processing time without penalising memory and fetching constraigns
+const SLICE_DURATION = 4 * ONE_HOUR_SEC;
 
-        const _startTimestamp = Math.max(
-          startTimestamp,
-          lastTimestampProcessed + 1,
-        );
+export async function fetchRefundableTransactions({
+  chainId,
+  startTimestamp,
+  endTimestamp,
+  epoch,
+  resolvePrice,
+}: {
+  chainId: number;
+  startTimestamp: number;
+  endTimestamp: number;
+  epoch: number;
+  resolvePrice: PriceResolverFn;
+}): Promise<GasRefundTransactionDataWithStakeScore[]> {
+  const logger = global.LOGGER(
+    `GRP:fetchRefundableTransactions: epoch=${epoch}, chainId=${chainId}`,
+  );
 
-        const slicedBatches: GasRefundTransactionDataWithStakeScore[][] = [];
-        for (
-          let _startTimestampSlice = _startTimestamp;
-          _startTimestampSlice < endTimestamp;
-          _startTimestampSlice += SLICE_DURATION
-        ) {
-          const _endTimestampSlice = Math.min(
-            _startTimestampSlice + SLICE_DURATION,
-            endTimestamp,
-          );
+  logger.info(`start indexing between ${startTimestamp} and ${endTimestamp}`);
+
+  const lastTimestampTxByContract = await fetchLastTimestampTxByContract({
+    chainId,
+    epoch,
+  });
+
+  const allButV6ContractAddresses = getContractAddresses({ epoch, chainId });
+
+  const processRawTxs = constructTransactionsProcessor({
+    chainId,
+    endTimestamp,
+    epoch,
+    resolvePrice,
+  });
 
+  const allTxsV5AndV6Merged = await Promise.all([
+    ...allButV6ContractAddresses.map(async contractAddress => {
+      assert(contractAddress, 'contractAddress should be defined');
+      const lastTimestampProcessed =
+        lastTimestampTxByContract[contractAddress] || 0;
+
+      const _startTimestamp = Math.max(
+        startTimestamp,
+        lastTimestampProcessed + 1,
+      );
+
+      // Step 1: Create an array of time slices
+      const timeSlices = [];
+      for (
+        let _startTimestampSlice = _startTimestamp;
+        _startTimestampSlice < endTimestamp;
+        _startTimestampSlice += SLICE_DURATION
+      ) {
+        const _endTimestampSlice = Math.min(
+          _startTimestampSlice + SLICE_DURATION,
+          endTimestamp,
+        );
+        timeSlices.push({
+          start: _startTimestampSlice,
+          end: _endTimestampSlice,
+        });
+      }
+
+      // Step 2: Map each slice to a promise
+      const promises = timeSlices.map(({ start, end }) =>
+        (async () => {
           logger.info(
-            `fetching transactions between ${_startTimestampSlice} and ${_endTimestampSlice} for contract=${contractAddress}...`,
+            `fetching transactions between ${start} and ${end} for contract=${contractAddress}...`,
           );
 
           const transactions = await getAllTXs({
             epoch,
-            startTimestamp: _startTimestampSlice,
-            endTimestamp: _endTimestampSlice,
+            startTimestamp: start,
+            endTimestamp: end,
             chainId,
             epochEndTimestamp: endTimestamp,
             contractAddress,
           });
 
           logger.info(
-            `fetched ${transactions.length} txs between ${_startTimestampSlice} and ${_endTimestampSlice} for contract=${contractAddress}`,
+            `fetched ${transactions.length} txs between ${start} and ${end} for contract=${contractAddress}`,
           );
 
-          const refundableTransactions =
-            await filterFormatAndStoreRefundableTransactions(
-              transactions,
-              (epoch, totalScore) => {
-                const result =
-                  contractAddress === MIGRATION_SEPSP2_100_PERCENT_KEY
-                    ? 1 // 100%
-                    : getRefundPercent(epoch, totalScore);
-
-                return result;
-              },
-            );
-          if (refundableTransactions.length > 0) {
-            slicedBatches.push(refundableTransactions);
-          }
-        }
-        return slicedBatches.flat();
-      }),
-
-      ...Array.from(AUGUSTUS_SWAPPERS_V6_OMNICHAIN).map(
-        async contractAddress => {
-          const epochNewStyle = epoch - GasRefundV2EpochFlip;
-
-          const lastTimestampProcessed =
-            lastTimestampTxByContract[contractAddress];
+          const refundableTransactions = await processRawTxs(
+            transactions,
+            (epoch, totalScore) => {
+              const result =
+                contractAddress === MIGRATION_SEPSP2_100_PERCENT_KEY
+                  ? 1 // 100% for migration tx
+                  : getRefundPercent(epoch, totalScore);
+              return result;
+            },
+          );
 
-          const allStakersTransactionsDuringEpoch =
-            await fetchParaswapV6StakersTransactions({
-              epoch: epochNewStyle,
-              timestampGreaterThan: lastTimestampProcessed,
-              chainId,
-              address: contractAddress,
-            });
+          return refundableTransactions.length > 0
+            ? refundableTransactions
+            : [];
+        })(),
+      );
+
+      // Step 3: Use Promise.all to execute all promises concurrently
+      const result = await Promise.all(promises);
+
+      // Step 4: Flatten the result and return
+      return result.flat();
+    }),
+
+    ...Array.from(AUGUSTUS_SWAPPERS_V6_OMNICHAIN).map(async contractAddress => {
+      const epochNewStyle = epoch - GasRefundV2EpochFlip;
+
+      const lastTimestampProcessed = lastTimestampTxByContract[contractAddress];
+
+      const allStakersTransactionsDuringEpoch =
+        await fetchParaswapV6StakersTransactions({
+          epoch: epochNewStyle,
+          timestampGreaterThan: lastTimestampProcessed,
+          chainId,
+          address: contractAddress,
+        });
+
+      return await processRawTxs(
+        allStakersTransactionsDuringEpoch,
+        (epoch, totalUserScore) => getRefundPercent(epoch, totalUserScore),
+      );
+    }),
+  ]);
+
+  const flattened = allTxsV5AndV6Merged.flat();
+  const withPatches = await addPatches({
+    txs: flattened,
+    epoch,
+    processRawTxs,
+    chainId,
+  });
+  await storeTxs({
+    txsWithScores: withPatches,
+    logger,
+  });
+  return withPatches;
+}
 
-          return await filterFormatAndStoreRefundableTransactions(
-            allStakersTransactionsDuringEpoch,
-            (epoch, totalUserScore) => getRefundPercent(epoch, totalUserScore),
-          );
-        },
-      ),
-    ])
-  ).flat();
+async function storeTxs({
+  txsWithScores: refundableTransactions,
+  logger,
+}: {
+  txsWithScores: GasRefundTransactionDataWithStakeScore[];
+  logger: Logger;
+}) {
+  if (refundableTransactions.length > 0) {
+    logger.info(
+      `updating total of ${refundableTransactions.length} for this chan and epoch`,
+    );
+    await writeTransactions(refundableTransactions);
+
+    const stakeScoreEntries = refundableTransactions
+      .map(({ stakeScore, ...transaction }) =>
+        composeGasRefundTransactionStakeSnapshots(transaction, stakeScore),
+      )
+      .flat();
+
+    await writeStakeScoreSnapshots(stakeScoreEntries);
+  }
+}
+async function addPatches({
+  txs,
+  epoch,
+  processRawTxs,
+  chainId,
+}: PatchInput & { epoch: number }): Promise<
+  GasRefundTransactionDataWithStakeScore[]
+> {
+  if (epoch === 46)
+    return applyEpoch46Patch({
+      txs,
+      processRawTxs,
+      chainId,
+    });
+
+  if (epoch === 48)
+    return applyEpoch48Patch({
+      txs,
+      processRawTxs,
+      chainId,
+    });
+
+  return txs;
 }
diff --git a/scripts/gas-refund-program/transactions-indexing/swaps-subgraph.ts b/scripts/gas-refund-program/transactions-indexing/swaps-subgraph.ts
@@ -19,6 +19,7 @@ import { thegraphClient } from '../../../src/lib/utils/data-providers-clients';
 import { createSubgraphURL } from '../../../src/lib/utils/subgraphs';
 
 const REORGS_BLOCKHASH_BY_CHAIN_ID: Record<string, string[]> = {
+  // TODO: this extra condition should apply to only specific epoch. Or better get rid of it at all (not sure what epoch this belongs to?), because it makes fetching too sloow
   [CHAIN_ID_POLYGON]: [
     '0x2019b19233191f463805ce55f5aaedb139cff358408da5e3d145c20dab47dab5',
     '0x4c48a4abde9207bcde996f3aa48741114d2eb8a0fea8ccecab9583ee5f6da235',
@@ -139,10 +140,26 @@ export async function getSuccessfulSwaps({
         : {},
     );
 
-    const { data } = await thegraphClient.post<SwapsGQLRespose>(subgraphURL, {
+    const postData = {
       query: regorgBlockHashes ? SwapsQueryBlockHash : SwapsQuery,
       variables,
-    });
+    };
+
+    const result = await thegraphClient.post<SwapsGQLRespose>(
+      subgraphURL,
+      postData,
+    );
+
+    const { data } = result;
+
+    if (!data.data) {
+      throw new Error(
+        'No data in the response chainId: ' +
+          chainId +
+          ' data ' +
+          JSON.stringify(postData),
+      );
+    }
 
     const swaps = data.data.swaps;