feat: control api

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@kryptokrauts/event-processor-node-lib",
-  "version": "1.0.5",
+  "version": "1.0.6",
   "description": "Wrapper for ease listening on antelope blockchain based on @blockmatic/antelope-ship-reader",
   "main": "dist/index.js",
   "types": "dist/index.d.ts",
@@ -9,7 +9,8 @@
     "src"
   ],
   "scripts": {
-    "dev-example": "env-cmd -f ./.env_local tsnd examples/atomicmarket.ts | npx pino-pretty --colorize",
+    "dev-aa": "env-cmd -f ./.env_local tsnd examples/atomicassets.ts | npx pino-pretty --colorize",
+    "dev-am": "env-cmd -f ./.env_local_am tsnd examples/atomicmarket.ts | npx pino-pretty --colorize",
     "lint": "eslint --ignore-path .eslintignore \"**/*.+(js|ts|tsx)\"",
     "format": "prettier --ignore-path .gitignore --write \"**/*.+(js|json|ts|tsx)\"",
     "format:check": "prettier -c .",
@@ -38,6 +39,7 @@
     "@types/node-fetch": "2.6.9",
     "dotenv": "16.3.1",
     "env-cmd": "10.1.0",
+    "express": "4.18.3",
     "kafkajs": "2.2.4",
     "pino": "8.16.2"
   },

src/common/config.ts

@@ -4,6 +4,8 @@ import { Logger, pino } from 'pino';
 
 dotenv.config();
 
+export const control_api_port = process.env.PORT || 8000;
+
 export const EOSIO_CONFIG = {
   start_block: Number(process.env.EOSIO_START_BLOCK),
   eosio_node_api: process.env.EOSIO_NODE_API,

src/control-api.ts

@@ -0,0 +1,30 @@
+import express from 'express';
+import { control_api_port, getLogger } from './common/config';
+
+const api = express();
+const logger = getLogger('reset-api');
+
+export async function startControlApi(reset_handler) {
+  api.get('/reset', (request, response) => {
+    const restart_at_block = request.query.blocknum;
+    const reset_database = request.query.reset_db || false;
+
+    if (restart_at_block === undefined) {
+      response.status(404).send('no blocknum defined, aborting');
+    } else {
+      logger.info(
+        `Parameters provided: blocknum: ${restart_at_block}, reset_db: ${reset_database}`,
+      );
+      reset_handler(restart_at_block, reset_database);
+      response
+        .status(200)
+        .send(
+          `event to restart at block ${restart_at_block} and reset database = ${reset_database} emitted`,
+        );
+    }
+  });
+
+  api.listen(control_api_port, (): void => {
+    logger.info(`processor-node control api started, running on port ${control_api_port}`);
+  });
+}

src/eosio/ship-reader-wrapper.ts

@@ -13,6 +13,7 @@ import {
   ResetEvent,
   ShipReaderWrapperConfig,
 } from '../common/types';
+import { startControlApi } from '../control-api';
 import KafkaWrapper from '../kafka/kafka-wrapper';
 import { fetchAbi, getHeadBlockNum } from './chain-api';
 
@@ -63,8 +64,10 @@ export class ShipReaderWrapper {
     logger.info(KAFKA_CONFIG);
     logger.info(KAFKA_TOPIC_CONFIG);
 
-    // connect to kafka and retrieve last processed message data
+    // start control api for sending reset message
+    startControlApi(this.handleExternalReset);
 
+    // connect to kafka and retrieve last processed message data
     this.kafka_wrapper = new KafkaWrapper({ header_prefix: this.config.message_header_prefix });
 
     const { last_blocknum: last_blocknum, type } = await this.kafka_wrapper.connect();
@@ -376,4 +379,20 @@ export class ShipReaderWrapper {
       process.kill(process.pid, type);
     }
   }
+
+  private async handleExternalReset(restart_at_block: number, reset_db: boolean) {
+    try {
+      logger.warn(`Got external reset event to restart at blockum ${restart_at_block}`);
+      const resetEvent: string = this.createResetEvent(
+        'external_restart',
+        `caused by control api call`,
+        restart_at_block,
+        reset_db,
+      );
+      resetEvent && (await this.kafka_wrapper.sendEvent(resetEvent, 'reset_event'));
+      await this.gracefulShutdown();
+    } finally {
+      process.kill(process.pid, -1);
+    }
+  }
 }