Merge pull request #305 from batphonghan/wAgEth_linea_kelp

kelp : wAgETH : tvl

adapters/wagETH_linea/package.json

@@ -0,0 +1,27 @@
+{
+  "name": "wageth-linea-adapter",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1",
+    "start": "node dist/index.js",
+    "compile": "tsc",
+    "watch": "tsc -w",
+    "clear": "rm -rf dist",
+    "dev-debug": "ts-node src/index.ts"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "@types/lodash": "^4.17.4",
+    "csv-parser": "^3.0.0",
+    "fast-csv": "^5.0.1",
+    "graphql-request": "^6.1.0"
+  },
+  "devDependencies": {
+    "@types/node": "^20.11.17",
+    "typescript": "^5.3.3"
+  }
+}

adapters/wagETH_linea/src/index.ts

@@ -0,0 +1,104 @@
+import fs from "fs";
+import { write } from "fast-csv";
+import csv from "csv-parser";
+
+import {
+  LINEA_WAGETH_SYMBOL,
+  LINEA_WAGETH_ADDR,
+  UserBalanceSubgraphEntry,
+  getAllWagEthHodlers
+} from "./query";
+
+export const MULTICALL_BATCH_SIZE = 1000;
+
+interface BlockData {
+  blockNumber: number;
+  blockTimestamp: number;
+}
+
+type OutputDataSchemaRow = {
+  block_number: number;
+  timestamp: number;
+  user_address: string;
+  token_address: string;
+  token_balance: bigint;
+  token_symbol: string; //token symbol should be empty string if it is not available
+  usd_price: number; //assign 0 if not available
+};
+
+export const getUserTVLByBlock = async (
+  blocks: BlockData
+): Promise<OutputDataSchemaRow[]> => {
+  const { blockNumber, blockTimestamp } = blocks;
+
+  const allUser = await getAllWagEthHodlers(blockNumber);
+  const csvRows: OutputDataSchemaRow[] = [];
+  allUser.forEach((item: UserBalanceSubgraphEntry) => {
+    csvRows.push({
+      block_number: blockNumber,
+      timestamp: blockTimestamp,
+      user_address: item.id,
+      token_address: LINEA_WAGETH_ADDR.toLowerCase(),
+      token_balance: BigInt(item.balance),
+      token_symbol: LINEA_WAGETH_SYMBOL,
+      usd_price: 0
+    });
+  });
+  return csvRows;
+};
+
+const readBlocksFromCSV = async (filePath: string): Promise<BlockData[]> => {
+  const blocks: BlockData[] = [];
+
+  await new Promise<void>((resolve, reject) => {
+    fs.createReadStream(filePath)
+      .pipe(csv()) // Specify the separator as '\t' for TSV files
+      .on("data", (row) => {
+        const blockNumber = parseInt(row.number, 10);
+        const blockTimestamp = parseInt(row.timestamp, 10);
+        if (!isNaN(blockNumber) && blockTimestamp) {
+          blocks.push({ blockNumber: blockNumber, blockTimestamp });
+        }
+      })
+      .on("end", () => {
+        resolve();
+      })
+      .on("error", (err) => {
+        reject(err);
+      });
+  });
+
+  return blocks;
+};
+
+readBlocksFromCSV("hourly_blocks.csv")
+  .then(async (blocks: any[]) => {
+    console.log(blocks);
+    const allCsvRows: any[] = []; // Array to accumulate CSV rows for all blocks
+
+    for (const block of blocks) {
+      try {
+        const result = await getUserTVLByBlock(block);
+        for (let i = 0; i < result.length; i++) {
+          allCsvRows.push(result[i]);
+        }
+      } catch (error) {
+        console.error(
+          `An error occurred for block ${JSON.stringify(block)}:`,
+          error
+        );
+      }
+    }
+    await new Promise((resolve, reject) => {
+      const ws = fs.createWriteStream("outputData.csv", { flags: "w" });
+      write(allCsvRows, { headers: true })
+        .pipe(ws)
+        .on("finish", () => {
+          console.log(`CSV file has been written.`);
+          resolve;
+        });
+    });
+  })
+  .catch((err) => {
+    console.error("Error reading CSV file:", err);
+  });

adapters/wagETH_linea/src/query.ts

@@ -0,0 +1,81 @@
+import { request } from "graphql-request";
+import { gql } from "graphql-request";
+
+const MULTICALL_BATCH_SIZE = 1000;
+
+const LINEA_WAGETH_SUBGRAPH =
+	"https://api.studio.thegraph.com/query/70817/linea-wageth/version/latest";
+export const LINEA_WAGETH_ADDR = "0x2a4f1dcc79b83608f9e3BC1F3F55fBEfCBFaE885";
+export const LINEA_WAGETH_SYMBOL = "wagETH";
+interface IDwise {
+	id: string;
+}
+
+async function subgraphFetchAllById<T extends IDwise>(
+	endpoint: string,
+	query: string,
+	collection: string,
+	variables: Record<string, unknown>
+): Promise<T[]> {
+	const data: T[] = [];
+	let lastId = "0x0000000000000000000000000000000000000000";
+	while (true) {
+		const resp: { [collection: string]: T[] } = await request(endpoint, query, {
+			...variables,
+			lastId
+		});
+
+		const batch: T[] = resp[collection];
+		if (batch.length == 0) {
+			break;
+		}
+
+		const last = batch[batch.length - 1];
+		lastId = last.id;
+
+		data.push(...batch);
+
+		if (batch.length < MULTICALL_BATCH_SIZE) {
+			break;
+		}
+	}
+	return data;
+}
+
+interface GraphQLQuery {
+	query: string;
+	collection: string;
+}
+
+export type UserBalanceSubgraphEntry = {
+	id: string;
+	balance: string;
+};
+
+export const USER_BALANCES_QUERY: GraphQLQuery = {
+    query: gql`
+        query PositionsQuery($block: Int, $lastId: ID!) {
+            userBalances(
+                where: { balance_gt: "0", id_gt: $lastId }
+                block: { number: $block }
+                orderBy: id
+                orderDirection: asc
+                first: 1000
+            ) {
+                id
+                balance
+            }
+        }
+		`,
+	collection: "userBalances"
+};
+
+export async function getAllWagEthHodlers(blockNumber: number) {
+	const positions = await subgraphFetchAllById<UserBalanceSubgraphEntry>(
+		LINEA_WAGETH_SUBGRAPH,
+		USER_BALANCES_QUERY.query,
+		USER_BALANCES_QUERY.collection,
+		{ block: blockNumber, lastId: "0x0000000000000000000000000000000000000000" }
+	);
+	return positions;
+}

adapters/wagETH_linea/test/sample_hourly_blocks.csv

@@ -0,0 +1,2 @@
+number,timestamp
+10550400,1728615585