Support query tx input by blob hash

Signed-off-by: KayleCoder <kalecoder@outlook.com>

package.json

@@ -17,11 +17,16 @@
     "body-parser": "^1.20.2",
     "c-kzg": "^2.1.2",
     "cors": "^2.8.5",
-    "express": "^4.18.2"
+    "express": "^4.18.2",
+    "lodash": "^4.17.21",
+    "pg": "^8.11.3",
+    "pg-hstore": "^2.3.4",
+    "sequelize": "^6.35.2"
   },
   "devDependencies": {
     "@types/cors": "^2.8.17",
     "@types/express": "^4.17.21",
+    "@types/lodash": "^4.14.202",
     "gts": "^5.2.0",
     "ts-node": "^10.9.2",
     "typescript": "^5.3.3"

src/chain/index.ts

@@ -0,0 +1,12 @@
+import axios from "axios";
+import {CONFIGS} from "../config";
+const rpcUrl = CONFIGS.chain.rpcUrl as string;
+export async function queryTxByHash(hash: string) {
+    const result = await axios.post(rpcUrl, {
+        method: "eth_getTransactionByHash",
+        params: [hash],
+        id: 1,
+        jsonrpc: "2.0"
+    });
+    return result;
+}

src/config/index.ts

@@ -0,0 +1,23 @@
+export const CONFIGS = {
+    common: {
+        env: getEnv('ENV', 'dev'),
+    },
+    server: {
+        port: getEnv('PORT', 3000)
+    },
+    db: {
+        host: getEnv('DB_HOST', 'localhost'),
+        port: getEnv('DB_PORT', 15432),
+        database: 'blobscan_dev',
+        user: getEnv('DB_USER', 'ethda'),
+        password: getEnv('DB_PASSWORD', ''),
+    },
+    chain: {
+        rpcUrl: getEnv('RPC_URL', 'https://rpc-devnet.ethda.io')
+    }
+}
+
+export function getEnv(key: string, defaultValue: string | number): string | number {
+    const result = process.env[key] || defaultValue;
+    return typeof defaultValue === 'string' ? result : Number(result);
+}

src/db/index.ts

@@ -0,0 +1,6 @@
+import {Sequelize} from "sequelize";
+import {CONFIGS} from "../config";
+
+export const sequelize = new Sequelize(`postgresql://${CONFIGS.db.user}:${CONFIGS.db.password}@${CONFIGS.db.host}:${CONFIGS.db.port}/${CONFIGS.db.database}?schema=public`, {
+    logging: CONFIGS.common.env === 'dev'
+});

src/index.ts

@@ -5,10 +5,14 @@ import { resolve } from 'path';
 import {commitmentsToVersionedHashes, EncodeBlobs} from '@ethda/blobs';
 import {blobToKzgCommitment, computeBlobKzgProof, loadTrustedSetup} from "c-kzg";
 const app = express();
-const port = process.env.PORT ?? 3000;
 const SETUP_FILE_PATH = resolve(__dirname, 'trusted.txt');
-import axios, {HttpStatusCode} from 'axios'
-const rpcUrl = process.env.RPC_URL??'https://rpc-devnet.ethda.io'
+import {HttpStatusCode} from 'axios'
+import {queryTxByHash} from "./chain";
+import {queryForArray} from "./util/dbUtils";
+import {indexTransactions} from "./job";
+import {CONFIGS} from "./config";
+const port = CONFIGS.server.port ?? 3000;
+
 loadTrustedSetup(SETUP_FILE_PATH);
 app.use(cors({
     origin:true,
@@ -40,25 +44,40 @@ app.post('/convert/blob', (req: any, res: any) => {
 app.get('/transaction/:hash', async (req: any, res: any) => {
     const hash = req.params.hash;
     if (hash) {
-        const result = await axios.post(rpcUrl, {
-            method: "eth_getTransactionByHash",
-            params: [hash],
-            id: 1,
-            jsonrpc: "2.0"
-        });
+        const result = await queryTxByHash(hash);
         if (result.status === HttpStatusCode.Ok) {
             res.json({
                 data: result.data.result.input
             })
         } else {
             res.status(500).send('query tx failed');
         }
-
     } else {
         res.status(400).send('invalid tx hash');
     }
-})
+});
+
+app.get('/blob/:hash/txData', async (req: any, res: any) => {
+    const hash = req.params.hash;
+    if (hash) {
+        const result = await queryForArray(`SELECT
+\tbt.tx_hash,
+\ttr.input_data
+FROM
+\tblobs_on_transactions AS bt
+\tJOIN "transaction" AS tr ON bt.tx_hash = tr.hash 
+WHERE
+\tbt.blob_hash = ? AND re_indexed = 0`, [hash]);
+        res.json(result);
+    } else {
+        res.json([]);
+    }
+});
 
 app.listen(port, () => {
     console.log(`server start at: ${port}`);
 });
+
+indexTransactions().catch(e => {
+    console.error(e);
+});

src/job/index.ts

@@ -0,0 +1,32 @@
+import {Transaction} from "../model/Transcation";
+import * as _ from "lodash";
+import {queryTxByHash} from "../chain";
+import {sleep} from "../util/commonUtils";
+import {HttpStatusCode} from "axios";
+
+export async function indexTransactions() {
+    while (true) {
+        const transactions = await Transaction.findAll({
+            where: {
+                re_indexed: 1
+            },
+            limit: 100
+        });
+        if (!_.isEmpty(transactions)) {
+            for (const t of transactions) {
+                const result = await queryTxByHash(t.getDataValue('hash'));
+                if (result.status === HttpStatusCode.Ok) {
+                    await Transaction.update({
+                        input_data: result.data.result.input,
+                        re_indexed: 0,
+                    }, {
+                        where: {
+                            hash: t.getDataValue('hash')
+                        }
+                    });
+                }
+            }
+        }
+        await sleep(3 * 1000);
+    }
+}

src/model/BlobsOnTransactions.ts

@@ -0,0 +1,12 @@
+import {sequelize} from "../db";
+import {DataTypes} from "sequelize";
+
+export const BlobsOnTransactions = sequelize.define("blobs_on_transactions",
+    {
+        blob_hash: { type: DataTypes.STRING},
+        tx_hash: { type: DataTypes.STRING},
+        index: { type: DataTypes.INTEGER},
+    }, {
+    timestamps: false,
+    freezeTableName: true
+});

src/model/Transcation.ts

@@ -0,0 +1,12 @@
+import {sequelize} from "../db";
+import {DataTypes} from "sequelize";
+
+export const Transaction = sequelize.define("transaction",
+    {
+        hash: { type: DataTypes.STRING, primaryKey: true},
+        input_data: { type: DataTypes.STRING},
+        re_indexed: { type: DataTypes.INTEGER},
+    }, {
+        timestamps: false,
+        freezeTableName: true
+    });

src/util/commonUtils.ts

@@ -0,0 +1,7 @@
+export function sleep(time: number) {
+    return new Promise<void>((resolve, reject) => {
+        setTimeout(() => {
+            resolve();
+        }, time);
+    });
+}

src/util/dbUtils.ts

@@ -0,0 +1,110 @@
+import {QueryTypes, Sequelize, Transaction} from 'sequelize';
+
+import * as _ from 'lodash';
+import {sequelize} from "../db";
+
+export function queryForCount(
+  sql: string,
+  replace: any[],
+  transaction?: Transaction
+): Promise<number> {
+  return sequelize
+    .query(sql, {
+      replacements: replace,
+      type: QueryTypes.SELECT,
+      raw: true,
+      transaction,
+    })
+    .then((r: any[]) => {
+      if (!_.isEmpty(r)) {
+        const res = r[0];
+        return res[Object.keys(res)[0]];
+      }
+    });
+}
+
+export function queryForArray(
+  sql: string,
+  replace: any[],
+  transaction?: Transaction
+): Promise<any[]> {
+  return sequelize
+    .query(sql, {
+      replacements: replace,
+      type: QueryTypes.SELECT,
+      transaction,
+    })
+    .then((r: any[]) => {
+      if (!_.isEmpty(r)) {
+        return r;
+      }
+      return [];
+    });
+}
+
+export function queryForObj(
+  sql: string,
+  replace: any[],
+  transaction?: Transaction
+): Promise<any> {
+  return sequelize
+    .query(sql, {
+      replacements: replace,
+      type: QueryTypes.SELECT,
+      transaction,
+    })
+    .then((r: any[]) => {
+      if (!_.isEmpty(r)) {
+        return r[0];
+      }
+      return {};
+    });
+}
+
+export function queryForUpdate(
+  sql: string,
+  replace: any[],
+  transaction?: Transaction
+): Promise<number> {
+  return sequelize
+    .query(sql, {
+      replacements: replace,
+      type: QueryTypes.UPDATE,
+      transaction,
+    })
+    .then((r: any) => {
+      return 0;
+    });
+}
+
+export function queryForInsert(
+  sql: string,
+  replace: any[],
+  transaction?: Transaction
+): Promise<number> {
+  return sequelize
+    .query(sql, {
+      replacements: replace,
+      type: QueryTypes.INSERT,
+      transaction,
+    })
+    .then((r: any) => {
+      return 0;
+    });
+}
+
+export function queryForDelete(
+  sql: string,
+  replace: any[],
+  transaction?: Transaction
+): Promise<number> {
+  return sequelize
+    .query(sql, {
+      replacements: replace,
+      type: QueryTypes.DELETE,
+      transaction,
+    })
+    .then((r: any) => {
+      return 0;
+    });
+}