Scheduler and worker

.env.example

@@ -4,21 +4,28 @@ PORT=3000
 HUB_ADDRESS=0xCfEB869F69431e42cdB54A4F4f105C19C080A601
 PROXY_FACTORY_ADDRESS=0xD833215cBcc3f914bD1C9ece3EE7BF8B14f841bb
 SAFE_ADDRESS=0xC89Ce4735882C9F0f0FE26686c53074E09B0D550
+TX_SENDER_ADDRESS=0xA6452911d5274e2717BC1Fa793b21aB600EC9133
 
+# Circles Services
 API_SERVICE_ENDPOINT=http://api.circles.local
-GRAPH_NODE_ENDPOINT=http://graph.circles.local
 RELAY_SERVICE_ENDPOINT=http://relay.circles.local
 
+# Ethereum RPC Node
 ETHEREUM_NODE_WS=ws://localhost:8545
 
+# Graph Node
+GRAPH_NODE_ENDPOINT=http://graph.circles.local
 SUBGRAPH_NAME=CirclesUBI/circles-subgraph
 
 # Database
-DATABASE_URL=postgres://username:password@localhost:5432/dbname
-DATABASE_DIALECT=postgres
+DATABASE_URL=postgres://postgres:postgres@localhost:5432/api
+
+# Redis
+REDIS_URL=redis://localhost:6379
 
 # AWS S3
 AWS_REGION=eu-central-1
-AWS_S3_BUCKET=circles-ubi
+AWS_S3_BUCKET=circles-ubi-development
+AWS_S3_BUCKET_TRUST_NETWORK=circles-ubi-network-development
 AWS_ACCESS_KEY_ID=
 AWS_SECRET_ACCESS_KEY=

README.md

@@ -67,6 +67,7 @@ An offchain API service to safely store and resolve [`Circles`] user data from p
 
 - NodeJS environment (tested with v12 and v14)
 - PostgreSQL database
+* Redis
 
 ## Usage
 
@@ -95,13 +96,14 @@ npm run test:watch
 npm run lint
 
 # Start local server and watch changes
-npm run serve
+npm run watch:all
 
 # Build for production
 npm run build
 
 # Run production server
 npm start
+npm worker:start
 ```
 
 ## License

package.json

@@ -19,11 +19,13 @@
     "db:seed": "sequelize db:seed:all",
     "lint": "eslint --ignore-path .gitignore .",
     "serve": "cross-env NODE_ENV=development babel-node ./src/index.js --source-maps inline",
+    "serve:all": "concurrently --prefix '{pid}-{name}' --names 'API,WORKERS' --kill-others 'npm run serve' 'npm run worker:serve'",
     "start": "cross-env NODE_ENV=production node ./build/index.js",
-    "test": "jest --forceExit --detectOpenHandles",
+    "test": "cross-env NODE_ENV=test jest --forceExit --detectOpenHandles",
     "test:watch": "npm run test -- --watch",
     "watch": "nodemon --watch ./src --exec npm run serve",
-    "worker:serve": "cross-env NODE_ENV=development babel-node ./src/worker.js --source-maps inline",
+    "watch:all": "concurrently --prefix '{pid}-{name}' --names 'API,WORKERS' --kill-others 'npm run watch' 'npm run worker:watch'",
+    "worker:serve": "NODE_ENV=development babel-node -r dotenv/config ./src/worker.js --source-maps inline",
     "worker:start": "cross-env NODE_ENV=production node ./build/worker.js",
     "worker:watch": "nodemon --watch ./src --exec npm run worker:serve"
   },
@@ -37,6 +39,7 @@
     "@babel/preset-env": "^7.12.1",
     "babel-eslint": "^10.1.0",
     "babel-plugin-add-module-exports": "^1.0.4",
+    "concurrently": "^5.3.0",
     "cross-env": "^7.0.2",
     "eslint": "^7.12.1",
     "eslint-config-prettier": "^6.15.0",
@@ -52,10 +55,11 @@
   },
   "dependencies": {
     "@babel/runtime": "^7.12.1",
-    "@circles/core": "^2.10.2",
+    "@circles/core": "^2.10.3",
     "@circles/transfer": "^1.1.0",
     "aws-sdk": "^2.781.0",
     "body-parser": "^1.19.0",
+    "bull": "^3.18.0",
     "celebrate": "^12.2.0",
     "compression": "^1.7.4",
     "cors": "^2.8.5",

src/constants.js

@@ -4,10 +4,7 @@ export const ZERO_ADDRESS = '0x0000000000000000000000000000000000000000';
 
 export const PATHFINDER_BASE_PATH = path.join(__dirname, '..', 'pathfinder');
 export const EDGES_FILE_PATH = path.join(PATHFINDER_BASE_PATH, 'edges.json');
-export const EDGES_TMP_FILE_PATH = path.join(
-  PATHFINDER_BASE_PATH,
-  'edges.json-tmp',
-);
+
 export const PATHFINDER_FILE_PATH = path.join(
   PATHFINDER_BASE_PATH,
   'pathfinder',

src/controllers/transfers.js

@@ -2,10 +2,11 @@ import httpStatus from 'http-status';
 
 import APIError from '../helpers/errors';
 import Transfer from '../models/transfers';
-import core from '../services/core';
+import transferSteps from '../services/findTransferSteps';
+import { checkFileExists } from '../services/edgesFile';
 import { checkSignature } from '../helpers/signature';
+import { requestGraph } from '../services/graph';
 import { respondWithSuccess } from '../helpers/responses';
-import { transferSteps, getTransferMetrics } from '../services/transfer';
 
 function prepareTransferResult(response) {
   return {
@@ -79,7 +80,7 @@ export default {
         }
       }`;
 
-      const data = await core.utils.requestGraph({ query });
+      const data = await requestGraph(query);
 
       if (!data || !data.user) {
         throw new APIError(httpStatus.FORBIDDEN, 'Not allowed');
@@ -114,6 +115,15 @@ export default {
   },
 
   findTransferSteps: async (req, res, next) => {
+    if (!checkFileExists()) {
+      next(
+        new APIError(
+          httpStatus.SERVICE_UNAVAILABLE,
+          'Trust network file does not exist',
+        ),
+      );
+    }
+
     try {
       const result = await transferSteps({
         ...req.body,
@@ -126,7 +136,7 @@ export default {
   },
 
   getMetrics: async (req, res) => {
-    const result = await getTransferMetrics();
-    respondWithSuccess(res, result);
+    // @DEPRECATED
+    respondWithSuccess(res);
   },
 };

src/index.js

@@ -39,11 +39,24 @@ app.use(helmet());
 
 // Log HTTP requests and route them to winston
 app.use(
-  morgan('dev', {
-    stream: {
-      write: (message) => logger.verbose(message.replace('\n', '')),
+  morgan(
+    (tokens, req, res) => {
+      return [
+        tokens.method(req, res),
+        tokens.url(req, res),
+        tokens.status(req, res),
+        tokens.res(req, res, 'content-length'),
+        '-',
+        tokens['response-time'](req, res),
+        'ms',
+      ].join(' ');
     },
-  }),
+    {
+      stream: {
+        write: (message) => logger.info(message.replace('\n', '')),
+      },
+    },
+  ),
 );
 
 // Mount all API routes

src/models/edges.js

@@ -13,14 +13,17 @@ const Edge = db.define(
     from: {
       type: Sequelize.STRING(42),
       allowNull: false,
+      unique: 'edges_unique',
     },
     to: {
       type: Sequelize.STRING(42),
       allowNull: false,
+      unique: 'edges_unique',
     },
     token: {
       type: Sequelize.STRING(42),
       allowNull: false,
+      unique: 'edges_unique',
     },
     capacity: {
       type: Sequelize.STRING,

src/services/edgesDatabase.js

@@ -0,0 +1,41 @@
+import Edge from '../models/edges';
+
+export async function upsertEdge(edge) {
+  if (edge.capacity.toString() === '0') {
+    return destroyEdge(edge);
+  } else {
+    return Edge.upsert(edge, {
+      where: {
+        token: edge.token,
+        from: edge.from,
+        to: edge.to,
+      },
+    });
+  }
+}
+
+export async function destroyEdge(edge) {
+  return Edge.destroy({
+    where: {
+      token: edge.token,
+      from: edge.from,
+      to: edge.to,
+    },
+  });
+}
+
+export async function queryEdges(where) {
+  return await Edge.findAll({
+    where,
+    order: [['from', 'ASC']],
+    raw: true,
+  });
+}
+
+export async function getStoredEdges({ hasOnlyFileFields = false } = {}) {
+  return await Edge.findAll({
+    attributes: hasOnlyFileFields ? ['from', 'to', 'token', 'capacity'] : null,
+    order: [['from', 'ASC']],
+    raw: true,
+  });
+}

src/services/edgesFile.js

@@ -0,0 +1,53 @@
+import fastJsonStringify from 'fast-json-stringify';
+import fs from 'fs';
+import path from 'path';
+
+import web3 from './web3';
+import { EDGES_FILE_PATH, PATHFINDER_BASE_PATH } from '../constants';
+
+const stringify = fastJsonStringify({
+  title: 'Circles Edges Schema',
+  type: 'array',
+  properties: {
+    from: {
+      type: 'string',
+    },
+    to: {
+      type: 'string',
+    },
+    token: {
+      type: 'string',
+    },
+    capacity: {
+      type: 'string',
+    },
+  },
+});
+
+export function checkFileExists() {
+  return fs.existsSync(EDGES_FILE_PATH);
+}
+
+// Store edges into .json file for pathfinder executable
+export async function writeToFile(
+  edges,
+  tmpFileKey = web3.utils.randomHex(16).slice(2),
+) {
+  return new Promise((resolve, reject) => {
+    // Write to temporary file first
+    const tmpFilePath = path.join(
+      PATHFINDER_BASE_PATH,
+      `edges.json-tmp-${tmpFileKey}`,
+    );
+
+    fs.writeFile(tmpFilePath, stringify(edges), (error) => {
+      if (error) {
+        reject(new Error(`Could not write to ${tmpFilePath} file: ${error}`));
+      } else {
+        // Finally rename it to destination file
+        fs.renameSync(tmpFilePath, EDGES_FILE_PATH);
+        resolve();
+      }
+    });
+  });
+}