Full Pipeline (#261)

1. Implement the method `internalizedTokenImbalance`

Which takes a transaction, database and transaction simulator to perform
the full pipeline process outlined in this [spec
doc](https://docs.google.com/document/d/15mByvoI7EQ-wFAM9QkDCMnjA4AVjbG0_G0GdirziGBM/edit#)
and illustrated in the screenshot in PR description

2. Write tests
3. Update the index.ts file (i.e. the tenderly web-hook) to make use of
this function.


nearly concludes Phase 1 of part #177 (excluding a few follow ups).

---------

Co-authored-by: Gent Rexha <[email protected]>

internal_transfers/actions/src/accounting.ts

@@ -71,8 +71,8 @@ export async function simulateSolverSolution(
       contractAddress: SETTLEMENT_CONTRACT_ADDRESS,
       sender: solverAddress,
       value: "0",
-      blockNumber: competition.simulationBlock,
     },
+    startBlock: competition.simulationBlock,
   });
 
   return {
@@ -87,13 +87,13 @@ interface commonSimulationParams {
   contractAddress: string;
   sender: string;
   value: string;
-  blockNumber: number;
 }
 
 interface SettlementSimulationParams {
   full: string;
   reduced: string;
   common: commonSimulationParams;
+  startBlock: number;
 }
 
 interface SimulationPair {
@@ -104,28 +104,31 @@ interface SimulationPair {
 async function simulateBoth(
   simulator: TransactionSimulator,
   params: SettlementSimulationParams,
-  numAttempts: number = 2
+  numAttempts: number = 3
 ): Promise<SimulationPair> {
-  let attempts = 0;
-  while (attempts < numAttempts) {
+  let attemptNumber = 0;
+  while (attemptNumber < numAttempts) {
     try {
       return {
         full: await simulator.simulate({
           ...params.common,
+          blockNumber: params.startBlock + attemptNumber,
           callData: params.full,
         }),
         reduced: await simulator.simulate({
           ...params.common,
+          blockNumber: params.startBlock + attemptNumber,
           callData: params.reduced,
         }),
       };
     } catch (error) {
-      attempts += 1;
-      // Increment blockNumber on Failed Simulation!
-      params.common.blockNumber += 1;
+      attemptNumber += 1;
+      console.warn(`Failed simulation attempt ${attemptNumber}`);
     }
   }
-  throw new Error(`failed simulations on ${numAttempts} blocks from `);
+  throw new Error(
+    `failed simulations on ${numAttempts} blocks beginning from ${params.startBlock}`
+  );
 }
 
 export function getInternalizedImbalance(

internal_transfers/actions/src/pipeline.ts

@@ -0,0 +1,52 @@
+import { partitionEventLogs } from "./parse";
+import {
+  insertPipelineResults,
+  insertSettlementEvent,
+  recordExists,
+} from "./database";
+import {
+  getInternalizedImbalance,
+  MinimalTxData,
+  simulateSolverSolution,
+} from "./accounting";
+import { Queryable } from "@databases/pg";
+import { TransactionSimulator } from "./simulate/interface";
+
+export async function internalizedTokenImbalance(
+  txData: MinimalTxData,
+  db: Queryable,
+  simulator: TransactionSimulator
+): Promise<void> {
+  const txHash = txData.hash;
+  console.log(`processing settlement transaction with hash: ${txData.hash}`);
+  // Duplication Guard!
+  if (await recordExists(db, txHash)) {
+    console.warn(`record exists for tx: ${txHash}`);
+    return;
+  }
+
+  // There are other events being returned here, but we only need the settlement(s)
+  const { settlements } = partitionEventLogs(txData.logs);
+
+  // It's annoying to have to handle the possibility of multiple settlements
+  // in the same transaction, but it could happen.
+  for (const settlement of settlements) {
+    const settlementSimulations = await simulateSolverSolution(
+      txData,
+      simulator
+    );
+    const eventMeta = { txHash, blockNumber: txData.blockNumber };
+    const settlementEvent = settlement;
+    // If there is a simulation, get imbalances otherwise assume none.
+    if (settlementSimulations) {
+      await insertPipelineResults(db, {
+        settlementSimulations,
+        imbalances: getInternalizedImbalance(settlementSimulations),
+        eventMeta,
+        settlementEvent,
+      });
+    } else {
+      await insertSettlementEvent(db, eventMeta, settlementEvent);
+    }
+  }
+}

internal_transfers/actions/tests/e2e/pipeline.spec.ts

@@ -0,0 +1,42 @@
+import { TenderlySimulator } from "../../src/simulate/tenderly";
+import { internalizedTokenImbalance } from "../../src/pipeline";
+import { getDB } from "../../src/database";
+import { getTxData } from "./helper";
+
+const { TENDERLY_USER, TENDERLY_PROJECT, TENDERLY_ACCESS_KEY } = process.env;
+
+const simulator = new TenderlySimulator(
+  TENDERLY_USER || "INVALID_USER",
+  TENDERLY_PROJECT || "TENDERLY_PROJECT",
+  TENDERLY_ACCESS_KEY || "TENDERLY_ACCESS_KEY"
+);
+
+const db = getDB("postgresql://postgres:postgres@localhost:5432/postgres");
+
+describe.skip("Run Full Pipeline", () => {
+  test("run pipeline on non-internalized transaction", async () => {
+    const notInternalized = await getTxData(
+      "0x0f86c06d9ace6a88644db6b654a904aa62c82305023e094ce49650467c91bd6e"
+    );
+    await expect(
+      internalizedTokenImbalance(notInternalized, db, simulator)
+    ).resolves.not.toThrowError();
+  }, 300000);
+  test("run pipeline on batch with internalized transfers", async () => {
+    const internalized = await getTxData(
+      "0xDCD5CF12340B50ACC04DBE7E14A903BE373456C81E4DB20DD84CF0301F6AB869"
+    );
+    await expect(
+      internalizedTokenImbalance(internalized, db, simulator)
+    ).resolves.not.toThrowError();
+  }, 300000);
+
+  test("run pipeline on transaction with unavailable competition data", async () => {
+    const txHash =
+      "0xe6a0fbad3f9571e7614dbbc1d65d523cbeb6929b59bd20cde80ac791899fccfb";
+    const unavailable = await getTxData(txHash);
+    await expect(
+      internalizedTokenImbalance(unavailable, db, simulator)
+    ).rejects.toThrow(`No competition found for ${txHash}`);
+  }, 300000);
+});