feature: Indexer

tools/indexer/.env.example

@@ -0,0 +1,12 @@
+ETHEREUM_RPC=https://eth-mainnet.alchemyapi.io/v2/xxx
+MOONBEAM_RPC=https://moonbeam.api.onfinality.io/public
+ENVIRONMENT=development # or production or staging
+
+PORT=3000
+BATCH_SIZE=5000
+
+PGHOST=pg1
+PGUSER=postgres
+PGDATABASE=postgres
+PGPASSWORD=postgres
+PGPORT=5432

tools/indexer/.gitignore

@@ -0,0 +1,4 @@
+node_modules
+# Keep environment variables out of version control
+.env
+postgres-data*

tools/indexer/Dockerfile

@@ -0,0 +1,29 @@
+FROM node:16
+ENV NODE_ENV=production
+
+WORKDIR /app
+
+COPY package.json ./package.json
+COPY package-lock.json ./package-lock.json
+
+
+# START should delete later
+COPY nomad-xyz-sdk-1.2.4.tgz ./nomad-xyz-sdk-1.2.4.tgz
+RUN npm i
+
+
+RUN npm i ./nomad-xyz-sdk-1.2.4.tgz
+
+# RUN rm ./nomad-xyz-sdk-1.2.4.tgz
+# END should delete later
+
+
+COPY api ./api
+COPY main.ts ./main.ts
+
+COPY prisma ./prisma
+RUN npm run build
+
+COPY core ./core
+
+CMD [ "npm", "run", "start" ]

tools/indexer/api/index.ts

@@ -0,0 +1,80 @@
+import express from "express";
+import { DB, MsgRequest } from "../core/db";
+import * as dotenv from "dotenv";
+import Logger from "bunyan";
+import promBundle from 'express-prom-bundle';
+import { prefix } from "../core/metrics";
+
+dotenv.config({});
+
+function fail(res: any, code: number, reason: string) {
+  return res.status(code).json({ error: reason });
+}
+
+const PORT = process.env.PORT;
+
+export async function run(db: DB, logger: Logger) {
+  const app = express();
+
+  const log = (
+    req: express.Request,
+    res: express.Response,
+    next: express.NextFunction
+  ) => {
+    logger.info(`request to ${req.url}`);
+    next();
+  };
+
+  const metricsMiddleware = promBundle({
+    httpDurationMetricName: prefix + '_api',
+    buckets: [0.1, 0.3, 0.6, 1, 1.5, 2.5, 5],
+    includeMethod: true,
+    includePath: true,
+    metricsPath: '/metrics',
+  });
+  metricsMiddleware
+
+  app.use(metricsMiddleware)
+
+  // app.use(promMid({
+  //   metricsPath: '/metrics',
+  //   collectDefaultMetrics: true,
+  //   requestDurationBuckets: [0.1, 0.5, 1, 1.5],
+  //   requestLengthBuckets: [512, 1024, 5120, 10240, 51200, 102400],
+  //   responseLengthBuckets: [512, 1024, 5120, 10240, 51200, 102400],
+  //   prefix: prefix + '_api',
+  // }));
+
+  app.get("/healthcheck", log, (req, res) => {
+    res.send("OK!");
+  });
+
+  app.get("/tx/:tx", log, async (req, res) => {
+    const messages = await db.getMessageByEvm(req.params.tx);
+    return res.json(messages.map(m => m.serialize()));
+  });
+
+  app.get("/hash/:hash", log, async (req, res) => {
+    const message = await db.getMessageByHash(req.params.hash);
+    if (!message) return res.status(404).json({});
+    return res.json(message.serialize());
+  });
+
+  app.get(
+    "/tx",
+    log,
+    async (req: express.Request<{}, {}, {}, MsgRequest>, res) => {
+      const { size } = req.query;
+
+      if (size && size > 30) return fail(res, 403, "maximum page size is 30");
+
+      const messages = await db.getMessages(req.query);
+
+      return res.json(messages.map(m => m.serialize()));
+    }
+  );
+
+  app.listen(PORT, () => {
+    logger.info(`Server is running at https://localhost:${PORT}`);
+  });
+}

tools/indexer/core/api.ts

@@ -0,0 +1,102 @@
+import express from "express";
+import { DB, MsgRequest } from "./db";
+import * as dotenv from "dotenv";
+import Logger from "bunyan";
+import { Orchestrator } from "./orchestrator";
+import { Processor } from "./consumer";
+import { replacer } from "./utils";
+dotenv.config({});
+
+function fail(res: any, code: number, reason: string) {
+  return res.status(code).json({ error: reason });
+}
+
+const PORT = process.env.DEBUG_PORT || '1337';
+
+export async function run(o: Orchestrator, logger: Logger) {
+  const app = express();
+
+  const log = (
+    req: express.Request,
+    res: express.Response,
+    next: express.NextFunction
+  ) => {
+    logger.info(`request to ${req.url}`);
+    next();
+  };
+
+  app.get("/healthcheck", log, (req, res) => {
+    res.send("OK!");
+  });
+
+  app.get("/hash/:hash", log, async (req, res) => {
+    const p = (o.consumer as Processor);
+    const message = p.getMsg(req.params.hash);
+    if (message) {
+        return res.json(message.serialize());
+    } else {
+        return res.status(404).json({});
+    }
+  });
+
+  app.get("/tx/:tx", log, async (req, res) => {
+    const p = (o.consumer as Processor);
+    const message = Array.from(p.messages).find(m => m.tx && m.tx! === req.params.tx);
+    if (message) {
+        return res.json(message.serialize());
+    } else {
+        return res.status(404).json({});
+    }
+  });
+
+  app.get("/status", log, async (req, res) => {
+    const promises: Promise<[number, {
+      lastIndexed: number;
+      numMessages: number;
+      numRpcFailures: number;
+    }]>[] = Array.from(o.indexers.entries()).map(async ([domain, indexer]): Promise<[number, {
+      lastIndexed: number;
+      numMessages: number;
+      numRpcFailures: number;
+    }]> => {
+      return [domain, {
+        lastIndexed: indexer.lastIndexed.valueOf(),
+        numMessages: await o.db.getMessageCount(domain),
+        numRpcFailures: indexer.failureCounter.num()
+      }]
+    });
+    const entries = await Promise.all(
+      promises
+    );
+
+    const x = new Map(
+      entries
+    );
+    return res.json(JSON.stringify(x, replacer));
+  })
+
+  app.get("/msg/:origin/:state", log, async (req, res) => {
+
+    const {origin: originStr, state: stateStr} = req.params;
+    let origin: number, state: number;
+
+    try {
+        origin = parseInt(originStr);
+        state = parseInt(stateStr);
+    } catch(e) {
+        return res.status(407).json({error: `One of the params (origin or stage) is invalid`});
+    }
+
+    const p = (o.consumer as Processor);
+    const messages = Array.from(p.messages).filter(m => m.origin === origin && m.state === state);
+    if (messages.length) {
+        return res.json(messages.map(m => m.serialize()));
+    } else {
+        return res.status(404).json({});
+    }
+  });
+
+  app.listen(PORT, () => {
+    logger.info(`Server is running at https://localhost:${PORT}`);
+  });
+}