core: Onchain Price observation (#4)

* onchain price state initialisation

* onchain price update listener added

* added ticker routes to api

* linter fixes

* socket support added

* added price listeners on app state init

package.json

@@ -31,8 +31,10 @@
     "dotenv": "^16.3.1",
     "ethers": "^6.9.0",
     "express": "^4.18.2",
+    "express-validator": "^7.0.1",
     "log4js": "^6.9.1",
     "node-cron": "^3.0.3",
-    "redux": "^5.0.0"
+    "redux": "^5.0.0",
+    "socket.io": "^4.7.2"
   }
 }

src/api/v1/definitions/README.md

@@ -0,0 +1,31 @@
+# Route definition declarations guide
+
+In order to better and/or more strictly type the codebase, every new route
+will have to follow these rules:
+
+1. Every route in /routes folder should have corresponding file inside /definitions
+   named the same only with .d (definitions) before the .ts file type ending
+
+2. Inside the routes definitions file, there should be a namespace named the
+   same as the class inside the routes folder which is being typed. The same
+   namespace is the default export of that file.
+   There should also be an enum in which every route/method is defined
+
+3. Every namespace should implement next generics:
+  + `ResponseBody<T extends ENameOfTheRoute>` - Returns specified routes response body
+  + `RequestBody<T extends ENameOfTheRoute>` - Returns specified routes request body
+  + `RequestQueries<T extends ENameOfTheRoute>` - Returns specified routes query params
+  + `RequestParams<T extends ENameOfTheRoute>` - Returns specified routes request params
+  + `Response<T extends ENameOfTheRoute>` - Returns expressjs response object
+  + `Request<T extends ENameOfTheRoute>` - Returns expressjs request object
+  + `RouteMethod<T extends ENameOfTheRoute>` - Returns typization for next function
+
+   Every generic should return default empty object if route wasn't implemented
+
+4. The naming convention is PascalCase/UpperCamelCase and it should look something like this:
+  - For namespace name
+    + `{ROUTE_NAME}Definitions` - AuthRouteDefinitions
+  - For enum name
+    + `E{ROUTE_NAME}` - EAuthRoute
+  - For enum values
+    + `{HTTP_METHOD}{ROUTE_NAME}` - GetMe or PostAuthGithub

src/api/v1/definitions/tickers.route.d.ts

@@ -0,0 +1,48 @@
+import { EmptyObject, ParamsDictionary } from "../../../types/util.types";
+import { TickerEntity } from "../../../entities/ticker.entity";
+import { IResponseSuccess } from "../../../utils/response.util";
+import { NextFunction, Request as ExpressRequest, Response as ExpressResponse } from "express";
+
+
+export enum ETickersRoute {
+  GetTickerList = "GetTickerList",
+  GetTicker = "GetTicker",
+  GetTickerHistory = "GetTickerHistory"
+}
+
+declare namespace TickersRouteDefinitions {
+  type ResponseBody<T extends ETickersRoute> =
+    // GET /tickers/[ticker]
+    T extends ETickersRoute.GetTicker ? TickerEntity :
+    // GET /tickers
+    T extends ETickersRoute.GetTickerList ? TickerEntity[] :
+    // GET /tickers/[ticker]/history
+    T extends ETickersRoute.GetTickerHistory ? EmptyObject :
+    EmptyObject
+
+  type RequestBody<T extends ETickersRoute> = // eslint-disable-line @typescript-eslint/no-unused-vars
+    EmptyObject;
+
+  type RequestQueries<T extends ETickersRoute> = // eslint-disable-line @typescript-eslint/no-unused-vars
+    EmptyObject
+
+  type RequestParams<T extends ETickersRoute> =
+    // GET /tickers/[ticker]
+    T extends ETickersRoute.GetTicker ? TickerParams :
+    // GET /tickers/[ticker]/history
+    T extends ETickersRoute.GetTickerHistory ? TickerParams :
+    EmptyObject;
+
+  type Response<T extends ETickersRoute> = ExpressResponse<IResponseSuccess<ResponseBody<T>>>
+
+  type Request<T extends ETickersRoute> = ExpressRequest<RequestParams<T> & ParamsDictionary, IResponseSuccess<ResponseBody<T>>, RequestBody<T>, RequestQueries<T>>
+
+  type RouteMethod<T extends ETickersRoute> = (request: Request<T>, response: Response<T>, next: NextFunction) => Promise<any>;
+
+  // PARAMS
+  type TickerParams = {
+    tickerSymbol: string;
+  }
+}
+
+export default TickersRouteDefinitions;

src/api/v1/index.ts

@@ -1,8 +1,16 @@
 import { Router } from "express";
+import { error } from "./middlewares/error.middleware";
 import StatusRoute from "./routes/status.route";
+import TickersRoute from "./routes/tickers.route";
+import TickersValidator from "./validators/tickers.validator";
 
 const v1 = Router();
 
+v1.get("/tickers/", TickersRoute.getTickerList);
+v1.get("/tickers/:tickerSymbol", TickersValidator.validateGetTicker, TickersRoute.getTicker);
+
 v1.get("/status", StatusRoute.getStatus);
 
+v1.use(error);
+
 export default v1;

src/api/v1/middlewares/error.middleware.ts

@@ -0,0 +1,31 @@
+import { NextFunction, Request, Response } from "express";
+import {
+  CustomValidationError, NotFoundError, InvalidRequestError
+} from "../../../utils/errors.util";
+import { APIResponse } from "../../../utils/response.util";
+import logger from "../../../utils/logger.util";
+
+export async function error(error: Error, request: Request, response: Response, next: NextFunction) { // eslint-disable-line @typescript-eslint/no-unused-vars
+  logger.error(error);
+
+  let status: number;
+  let message: string;
+  let errors: any | undefined;
+
+  if (error instanceof NotFoundError) {
+    status = 404;
+    message = error.message || "Not found";
+  } else if (error instanceof InvalidRequestError) {
+    status = 400;
+    message = error.message || "Invalid request";
+  } else if (error instanceof CustomValidationError) {
+    status = 422;
+    message = error.message || "Validation error";
+    errors = error.err;
+  } else {
+    status = 500;
+    message = "Server error";
+  }
+
+  return response.status(status).json(APIResponse.error(status, message, errors));
+}

src/api/v1/middlewares/validate.middleware.ts

@@ -0,0 +1,19 @@
+import { ValidationChain, validationResult } from "express-validator";
+import { NextFunction, RequestHandler } from "express";
+import { CustomValidationError } from "../../../utils/errors.util";
+
+async function validateRequest(request?: Request, response?: Response, next?: NextFunction) {
+  const result = validationResult(request);
+  const errors = result.mapped();
+
+  if (!result.isEmpty()) {
+    return next(new CustomValidationError(errors));
+  } else {
+    return next();
+  }
+}
+
+export function val(validationChain: ValidationChain[]): RequestHandler[] {
+  // @ts-expect-error Type mismatch for validationChain, but actually is fitting
+  return [...validationChain, validateRequest];
+}

src/api/v1/routes/tickers.route.ts

@@ -0,0 +1,41 @@
+import TickersRouteDefinitions from "../definitions/tickers.route";
+import { ETickersRoute } from "../definitions/tickers.route";
+import { tickerEntityFromReduxState } from "../../../entities/ticker.entity";
+import store from "../../../redux/store";
+import { NotFoundError } from "../../../utils/errors.util";
+import { APIResponse } from "../../../utils/response.util";
+
+class TickersRoute {
+  public static getTicker: TickersRouteDefinitions.RouteMethod<ETickersRoute.GetTicker> = async (request, response, next) => {
+    try {
+      const {
+        tickerSymbol
+      } = request.params;
+
+      const tickerData = tickerEntityFromReduxState(tickerSymbol, store.getState());
+
+      if (!tickerData) {
+        throw new NotFoundError();
+      }
+
+      return response.status(200).json(APIResponse.success(tickerData));
+    } catch (error) {
+      next(error);
+    }
+  };
+
+  public static getTickerList: TickersRouteDefinitions.RouteMethod<ETickersRoute.GetTickerList> = async (request, response, next) => {
+    try {
+      const rootState = store.getState();
+      const symbols = rootState.tickers.symbols;
+
+      const tickerList = symbols.map(symbol => tickerEntityFromReduxState(symbol, rootState));
+
+      return response.status(200).json(APIResponse.success(tickerList));
+    } catch (error) {
+      next(error);
+    }
+  };
+}
+
+export default TickersRoute;

src/api/v1/validators/tickers.validator.ts

@@ -0,0 +1,19 @@
+import TickersRouteDefinitions from "../definitions/tickers.route";
+import { val } from "../middlewares/validate.middleware";
+import { checkSchema } from "express-validator";
+import { ValidatorFields } from "../../../types/util.types";
+
+type GetTickerFields =
+  (keyof TickersRouteDefinitions.TickerParams)
+
+const getTickerSchema: ValidatorFields<GetTickerFields> = {
+  tickerSymbol: {
+    in: ["params"],
+    errorMessage: "Ticker must be a string",
+    isString: true
+  }
+};
+
+export default class TickersValidator {
+  public static validateGetTicker = val(checkSchema(getTickerSchema));
+}

src/entities/ticker.entity.ts

@@ -0,0 +1,22 @@
+import { RootState } from "../redux/redux.types";
+import { Nullable } from "../types/util.types";
+
+export type TickerEntity =  {
+  symbol: string;
+  onchainPrice?: number;
+  offchainPrice?: number;
+}
+
+export function tickerEntityFromReduxState(tickerSymbol: string, state: RootState): Nullable<TickerEntity> {
+  const tickerPrices = state.prices.current[tickerSymbol];
+
+  if (tickerPrices) {
+    return {
+      symbol: tickerSymbol,
+      onchainPrice: tickerPrices.onchain,
+      offchainPrice: tickerPrices.offchain
+    };
+  } else {
+    return undefined;
+  }
+}

src/index.ts

@@ -1,8 +1,10 @@
 import router from "./app";
 import { createServer, Server } from "http";
 import logger from "./utils/logger.util";
+import SocketService from "./services/socket.service";
 
 const server: Server = createServer(router);
+export const RootSocket = new SocketService(server);
 
 server.listen(router.get("port"), router.get("host"), async () => {
   process.on("SIGINT", async () => {

src/jobs/app.jobs.ts

@@ -1,10 +1,14 @@
 import { initTickerState, setupTickerFetchingJob } from "./ticker.jobs";
-import { setupOffchainPriceFetchingJob } from "./price.jobs";
+import { initOnchainPriceState, setupOffchainPriceFetchingJob } from "./price.jobs";
 import logger from "../utils/logger.util";
+import store from "../redux/store";
 
 const initAppState = async () => {
   logger.log("Initialising the app state");
   await initTickerState();
+
+  const symbols = store.getState().tickers.symbols;
+  await initOnchainPriceState(symbols);
 };
 
 export const initApp = async () => {

src/jobs/price.jobs.ts

@@ -2,13 +2,15 @@ import CoinGeckoService from "../services/coingecko/coingecko.service";
 import CronService from "../services/cron.service";
 import store from "../redux/store";
 import { getCoingeckoIdByChainlinkTicker } from "../constants/coingecko";
-import { setCurrentOffchainPrice } from "../redux/prices/prices.redux.actions";
+import { setCurrentOffchainPrice, setCurrentOnchainPrice } from "../redux/prices/prices.redux.actions";
 import OracleService from "../services/oracle/oracle.service";
+import logger from "../utils/logger.util";
+import { RootSocket } from "../index";
 
-export const setupOffchainPriceFetchingJob = async () => {
-  const coinGecko = new CoinGeckoService();
-  const oracleService = new OracleService();
+const coinGecko = new CoinGeckoService();
+const oracleService = new OracleService();
 
+export const setupOffchainPriceFetchingJob = async () => {
   CronService.scheduleRecurringJob(async () => {
     const tickerSymbols = store.getState().tickers.symbols;
     const tickerCoinGeckoIds = tickerSymbols.map(getCoingeckoIdByChainlinkTicker);
@@ -27,4 +29,31 @@ export const setupOffchainPriceFetchingJob = async () => {
       } catch (err) { console.log(err);}
     }
   });
+};
+
+const updateReduxTickerOnchainPrice = (tickerSymbol: string, tickerPrice: number) => {
+  store.dispatch(setCurrentOnchainPrice(tickerSymbol, tickerPrice));
+};
+
+export const initOnchainPriceState = async (symbols: string[]) => {
+  for (let i = 0; i < symbols.length; i++) {
+    const tickerSymbol = symbols[i];
+    const newPrice = await oracleService.getOnchainPrice(tickerSymbol);
+    updateReduxTickerOnchainPrice(tickerSymbol, newPrice);
+  }
+};
+
+const isJobActive: Map<string, boolean> = new Map();
+export const setupOnchainPriceFetchingJobFor = async (tickerSymbol: string) => {
+  if (isJobActive.get(tickerSymbol)) {
+    return;
+  } else {
+    isJobActive.set(tickerSymbol, true);
+  }
+
+  logger.log("Setting up the onchain Price fetching observer for ", tickerSymbol);
+  await oracleService.listenForOnchainPriceUpdates(tickerSymbol, (tickerPriceData) => {
+    updateReduxTickerOnchainPrice(tickerSymbol, tickerPriceData.newPrice);
+    RootSocket.emitEvent("TickerPriceUpdated", [tickerSymbol, tickerPriceData.newPrice]);
+  });
 };

src/jobs/ticker.jobs.ts

@@ -3,6 +3,7 @@ import store from "../redux/store";
 import { addTicker } from "../redux/tickers/tickers.redux.actions";
 import { TickerRegistryData } from "../services/tickers/tickers.service.types";
 import logger from "../utils/logger.util";
+import { setupOnchainPriceFetchingJobFor } from "./price.jobs";
 
 const tickersService = new TickersService();
 
@@ -17,11 +18,16 @@ export const initTickerState = async () => {
   tickers.forEach(tickerData => {
     updateReduxTickerState(tickerData);
     symbols.push(tickerData.tickerSymbol);
+    setupOnchainPriceFetchingJobFor(tickerData.tickerSymbol);
   });
+
   logger.log("Tickers found: ", symbols);
 };
 
 export const setupTickerFetchingJob = async () => {
   logger.log("Setting up the Ticker Fetching observer");
-  tickersService.listenForTickerUpdates(updateReduxTickerState).then();
+  tickersService.listenForTickerUpdates(tickerData => {
+    updateReduxTickerState(tickerData);
+    setupOnchainPriceFetchingJobFor(tickerData.tickerSymbol);
+  }).then();
 };