feat(watcher): create a route to healthcheck /version

watcher/package.json

@@ -33,9 +33,14 @@
     "debug": "^4.3.4",
     "dotenv": "^16.0.3",
     "ethers": "^5.7.2",
+    "kcors": "^2.2.2",
+    "koa": "^2.15.3",
+    "koa-router": "^12.0.1",
     "mongoose": "^6.8.3",
     "redis": "^4.5.1",
-    "socket.io": "^4.5.4"
+    "socket.io": "^4.5.4",
+    "supertest": "^7.0.0",
+    "yup": "^1.4.0"
   },
   "devDependencies": {
     "eslint": "^8.31.0",

watcher/src/app.js

@@ -1,3 +1,9 @@
+const http = require('http');
+const Koa = require('koa');
+const corsMiddleware = require('kcors')();
+
+const { errorMiddleware } = require('./controllers/error');
+const { router } = require('./controllers/router');
 const config = require('./config');
 const ethereum = require('./loaders/ethereum');
 const socket = require('./loaders/socket');
@@ -29,7 +35,16 @@ if (!wsHost) throw Error('missing wsHost');
 if (!httpHost) throw Error('missing httpHost');
 if (!hubAddress) throw Error('missing hubAddress');
 
-socket.init();
+const koa = new Koa();
+const app = http.createServer(koa.callback());
+
+socket.init(app);
+
+koa
+  .use(errorMiddleware)
+  .use(corsMiddleware)
+  .use(router.routes())
+  .use(router.allowedMethods());
 
 const start = async ({ replayer = true, syncWatcher = true } = {}) => {
   try {
@@ -106,4 +121,5 @@ const stop = async () => {
 module.exports = {
   stop,
   start,
+  app,
 };

watcher/src/config.js

@@ -13,6 +13,7 @@ const { getLogger } = require('./utils/logger');
 const logger = getLogger('config');
 
 const {
+  PORT,
   MONGO_HOST,
   REDIS_HOST,
   FLAVOUR,
@@ -113,11 +114,14 @@ if (!chain.httpHost) {
   throw Error('missing ethereum RPC endpoint ETH_RPC_HOST');
 }
 
+const serverPort = parseInt(PORT, 10) || 3000;
+
 logger.log('chain', chain);
 logger.log('flavour', flavour);
 logger.log('mongo', mongo);
 logger.log('redis', redis);
 logger.log('runtime', runtime);
+logger.log('serverPort', serverPort);
 
 module.exports = {
   abi,
@@ -126,4 +130,5 @@ module.exports = {
   mongo,
   redis,
   runtime,
+  serverPort,
 };

watcher/src/controllers/error.js

@@ -0,0 +1,47 @@
+const { getLogger, APP_NAMESPACE } = require('../utils/logger');
+const logger = getLogger('controllers:error');
+const {
+  ValidationError,
+  InternalError,
+  ObjectNotFoundError,
+  errorHandler,
+} = require('../utils/error');
+
+const errorMiddleware = async (ctx, next) => {
+  try {
+    await next();
+  } catch (error) {
+    errorHandler(error, { type: 'request', request: ctx.request });
+    if (error instanceof InternalError) {
+      logger.log('InternalError:', error.message);
+      ctx.status = 500;
+      ctx.body = {
+        ok: false,
+        error: 'Something went wrong, you should retry later.',
+      };
+      return;
+    }
+    if (error instanceof ValidationError) {
+      logger.log('ValidationError:', error.message);
+      ctx.status = 400;
+      ctx.body = { ok: false, error: error.message };
+      return;
+    }
+    if (error instanceof ObjectNotFoundError) {
+      logger.log('ObjectNotFoundError:', error.message);
+      ctx.status = 404;
+      ctx.body = { ok: false, error: error.message };
+      return;
+    }
+    logger.log('Error:', error.message);
+    ctx.status = 500;
+    ctx.body = {
+      ok: false,
+      error: 'Internal error',
+    };
+  }
+};
+
+module.exports = {
+  errorMiddleware,
+};

watcher/src/controllers/router.js

@@ -0,0 +1,19 @@
+const Router = require('koa-router');
+const { getVersion } = require('../services/version');
+const { getLogger } = require('../utils/logger');
+const logger = getLogger('controllers:router');
+
+
+const router = new Router();
+
+// version
+router.get('/version', async (ctx) => {
+  logger.log('GET /version');
+  const version = await getVersion();
+  ctx.body = { ok: true, version };
+});
+
+
+module.exports = {
+  router,
+};

watcher/src/index.js

@@ -1,3 +1,9 @@
-const { start } = require('./app');
+const { start, app } = require('./app');
+const { serverPort } = require('./config');
+const { getLogger, APP_NAMESPACE } = require('./utils/logger');
+const logger = getLogger(APP_NAMESPACE);
+
+app.listen(serverPort);
+logger.log(`Server listening on port ${serverPort}`);
 
 start();

watcher/src/loaders/socket.js

@@ -1,4 +1,3 @@
-const http = require('http');
 const socketIo = require('socket.io');
 const { createClient } = require('redis');
 const { createAdapter } = require('@socket.io/redis-adapter');
@@ -8,16 +7,14 @@ const { throwIfMissing, errorHandler } = require('../utils/error');
 
 const logger = getLogger('socket');
 
-const server = http.createServer();
-
 let ws = null;
 
 const getWs = () => {
   if (ws === null) throw new Error('socket not initialized');
   return ws;
 };
 
-const init = async () => {
+const init = async (server = throwIfMissing()) => {
   try {
     logger.log('init socket');
     const redisConfig = config.redis;
@@ -31,7 +28,7 @@ const init = async () => {
     subClient.on('end', () => logger.log('subClient end'));
     await Promise.all[(pubClient.connect(), subClient.connect())];
     const redisAdapter = createAdapter(pubClient, subClient);
-    ws = socketIo(server);
+    ws = socketIo(server, { path: '/ws', cors: { origin: '*' } });
     ws.adapter(redisAdapter);
     logger.log('socket initialized');
   } catch (error) {

watcher/src/services/version.js

@@ -0,0 +1,7 @@
+const { version } = require('../../package.json');
+
+const getVersion = () => version;
+
+module.exports = {
+  getVersion,
+};

watcher/src/utils/error.js

@@ -1,3 +1,4 @@
+const { ValidationError } = require('yup');
 const { getLogger, APP_NAMESPACE } = require('./logger');
 const { sleep } = require('./utils');
 
@@ -38,6 +39,24 @@ class Web3ProviderError extends InternalError {
   }
 }
 
+class OperationalError extends Error {
+  constructor(message) {
+    super(message);
+    this.name = this.constructor.name;
+  }
+}
+
+class ObjectNotFoundError extends OperationalError {
+  constructor(message, originalError) {
+    super(message);
+    this.name = this.constructor.name;
+    this.originalError = originalError;
+    if (originalError && typeof originalError === 'object') {
+      Object.assign(this, getPropsToCopy(originalError));
+    }
+  }
+}
+
 const wrapEthCall = async (promise) => {
   try {
     return await promise;
@@ -75,6 +94,8 @@ const errorHandler = async (error, context) => {
 };
 
 module.exports = {
+  ValidationError,
+  ObjectNotFoundError,
   InternalError,
   Web3ProviderError,
   wrapEthCall,

watcher/test/test-utils.js

@@ -338,6 +338,8 @@ const setCheckpointToLastBlock = async (dbName) => {
   );
 };
 
+const parseResult = (res) => ({ ...res, data: JSON.parse(res.text) });
+
 module.exports = {
   addApporders,
   addDatasetorders,
@@ -354,6 +356,7 @@ module.exports = {
   getMatchableRequestorder,
   transferResourceERC721,
   timestampRegex,
+  parseResult,
   bytes32Regex,
   addressRegex,
   APPORDERS_COLLECTION,

watcher/test/watcher-common.test.js

@@ -1,9 +1,10 @@
+const supertest = require('supertest');
 const ethers = require('ethers');
 const { utils, IExec } = require('iexec');
 const socket = require('../src/loaders/socket');
 // jest spies
 const socketEmitSpy = jest.spyOn(socket, 'emit');
-const { start, stop } = require('../src/app');
+const { start, stop, app } = require('../src/app');
 const { replayPastOnly } = require('../src/controllers/replayer');
 const { chain } = require('../src/config');
 const { sleep } = require('../src/utils/utils');
@@ -21,6 +22,7 @@ const {
   getMatchableRequestorder,
   transferResourceERC721,
   timestampRegex,
+  parseResult,
   bytes32Regex,
   APPORDERS_COLLECTION,
   DATASETORDERS_COLLECTION,
@@ -36,6 +38,8 @@ const { init: ethereumInit } = require('../src/loaders/ethereum');
 jest.setTimeout(120000);
 
 const PROCESS_TRIGGERED_EVENT_TIMEOUT = 1000;
+const NOT_FOUND_ERROR_STATUS = 404;
+const OK_STATUS = 200;
 
 let chainId;
 const chainUrl = chain.httpHost;
@@ -46,6 +50,8 @@ const rpc = new ethers.providers.JsonRpcProvider(chainUrl);
 const wallet = new ethers.Wallet(PRIVATE_KEY, rpc);
 
 let iexec;
+let serverHttp;
+let request;
 const signer = utils.getSignerFromPrivateKey(chainUrl, PRIVATE_KEY);
 
 beforeAll(async () => {
@@ -67,6 +73,28 @@ beforeAll(async () => {
     await iexec.wallet.getAddress(),
   );
   await iexec.account.withdraw(stake);
+  serverHttp = app.listen();
+  request = supertest(serverHttp);
+});
+
+afterAll(async () => {
+  serverHttp.close();
+});
+
+describe('API', () => {
+  describe('Common', () => {
+    test('GET /version', async () => {
+      const { data, status } = await request.get('/version').then(parseResult);
+      expect(status).toBe(OK_STATUS);
+      expect(data.ok).toBe(true);
+      expect(data.version).toBeDefined();
+    });
+    test('GET /foo (not found)', async () => {
+      const res = await request.get('/foo');
+      expect(res.status).toBe(NOT_FOUND_ERROR_STATUS);
+      expect(res.text).toBe("Not Found");
+    });
+  });
 });
 
 describe('Watcher', () => {