pointsight/main.ts

/***
 * Copyright © 2021 Sören W. Oesterwind (TheGreydiamond). All rights reserved.
 */

// Requires (Imports)
const express = require("express");
const fs = require("fs");
const mysql = require('mysql');
const autobahnAPI = require("./apiHandler/autobahn/main.js");
/*const ffhAPI = require("./apiHandler/ffh/main.js");
const stadtkoelnAPI = require("./apiHandler/ffh/main.js");*/
const customWebcamAPI = require("./apiHandler/customWebcams/main.js");
const airspyAPI = require("./apiHandler/airspydirectory/main");
const ttncommsAPI = require("./apiHandler/ttncomms/main");
const freifunkAPI = require("./apiHandler/freifunk/main");
const senseboxAPI = require("./apiHandler/opensensemap/main.js");
const opentopiaAPI = require("./apiHandler/opentopia/main.js");
const ttnGateway = require("./apiHandler/ttnantennas/main");
const stubMod = require("./apiHandler/stub/main");
const _ = require("underscore");
const chalk = require('chalk');
const func = require("./functions.js")
const minify = require('express-minify');
const compression = require('compression');
const cookieParser = require('cookie-parser')
const bodyParser = require('body-parser');
const Sentry = require('@sentry/node');
const Tracing = require('@sentry/tracing');
const bent = require("bent");
const Influx = require("influx");


function padLeadingZeros(num, size) {
  let s = num + "";
  while (s.length < size) s = "0" + s;
  return s;
}

const enabledModules = [airspyAPI, autobahnAPI, customWebcamAPI, freifunkAPI, senseboxAPI, opentopiaAPI, ttnGateway, ttncommsAPI];

// Webserver init
const app = express();


const allTags = {};
const metaGlobals = { "desc": "Pointsight is a centralized geolocalized API aggragator.", "titlePrefx": "Pointsight - " }

// Skeleton Variables
let jsonConfigGlobal = {
  fontAwesome: undefined,
  mapboxAccessToken: undefined,
  cookieSecret: undefined,
  mapquest: undefined,
  here: undefined,
  sentryDsn: undefined,
  database: {
    host: "127.0.0.1",
    user: "pointsight",
    password: "",
    database: "pointsight",
    customSettings: {
      wait_timout: 28800
    }
  },
  env: "PROD",
  maint: true,
  betaMode: false,
  port: 3000,
  adress: '127.0.0.1',
  taxonomyCacheInterval: 5,
  metrics: {
    influx: {
      enable: false,
      host: "127.0.0.1",
      database: "pointsight",
      user: "pointsight",
      password: ""
    },
    writeMetricsToMySQL: true
  }
}

// Load config
try {
  const data = fs.readFileSync("config/default.json", "utf8");
  jsonConfigGlobal = _.extend(jsonConfigGlobal, JSON.parse(data));
} catch (error) {
  console.error(
    "While reading the config an error occured. The error was: " + error
  );
}
console.log(jsonConfigGlobal.metrics.influx)

Sentry.init({
  dsn: jsonConfigGlobal.sentryDsn,
  integrations: [
    // enable HTTP calls tracing
    new Sentry.Integrations.Http({ tracing: true }),
    // enable Express.js middleware tracing
    new Tracing.Integrations.Express({ app }),
    new Tracing.Integrations.Mysql({ useMysql: true })
  ],
  // Set tracesSampleRate to 1.0 to capture 100%
  // of transactions for performance monitoring.
  // We recommend adjusting this value in production
  tracesSampleRate: 0.8,
});
app.use(Sentry.Handlers.requestHandler());
app.use(Sentry.Handlers.tracingHandler());

const con = mysql.createConnection(jsonConfigGlobal.database);
const connectionPool = mysql.createPool(jsonConfigGlobal.database);
const allPoolConnections = [];
let influxD = undefined;
let isInfluxReady = false;
let lastHttpRequest = Math.floor(Date.now() / 1000)

if(jsonConfigGlobal.metrics.influx.enable){
  console.log("[Influx] Connecting to InfluxDB...");
    influxD = new Influx.InfluxDB({
    host: jsonConfigGlobal.metrics.influx.host,
    database: jsonConfigGlobal.metrics.influx.database,
    user: jsonConfigGlobal.metrics.influx.user,
    password: jsonConfigGlobal.metrics.influx.password,
    schema: [
      {
        measurement: 'apitaxonomy',
        fields: { calls: Influx.FieldType.INTEGER },
        tags: []
      }
    ]
  });
  console.log("[Influx] Checking if database exists...");
  influxD.getDatabaseNames()
  .then(names => {
    if (!names.includes('pointsight')) {
      return influxD.createDatabase('pointsight');
    }
  })
  .then(() => {
    console.log("[Influx] Database ready.");
    isInfluxReady = true;
  })
  .catch(error => console.log({ error }));
}

connectionPool.getConnection((err, conn) => {
  conn.on('error', function (err) {
    console.log("----[MYSQL ERROR]----")
    console.log(err); // 'ER_BAD_DB_ERROR'
    console.log("[MYSQL ERROR] Connection pool error");
    console.log("----]MYSQL ERROR[----")
  });

  func.handleMysqlErrors(err)
  //Create system tabels
  const queries = ["CREATE TABLE IF NOT EXISTS `apikeys` ( `id` BIGINT NOT NULL AUTO_INCREMENT , `apikey` VARCHAR(64) NOT NULL , `owner` VARCHAR(64) NOT NULL , `hosts` JSON NOT NULL , `created` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP , `expire` DATETIME NOT NULL , PRIMARY KEY (`id`));", "CREATE TABLE IF NOT EXISTS `betatokens` ( `id` BIGINT NOT NULL AUTO_INCREMENT , `token` VARCHAR(64) NOT NULL , `owner` VARCHAR(64) NOT NULL, `created` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP , `expire` DATETIME NOT NULL , PRIMARY KEY (`id`));",
    "CREATE TABLE IF NOT EXISTS `reports` ( `id` INT NOT NULL AUTO_INCREMENT , `point_id` VARCHAR(512) NOT NULL , `mail` VARCHAR(512) NOT NULL , `comment` VARCHAR(2048) NOT NULL , PRIMARY KEY (`id`)) ENGINE = InnoDB;",
    "CREATE TABLE IF NOT EXISTS `apitaxonomy` ( `id` INT NOT NULL AUTO_INCREMENT , `apiKey` VARCHAR(255) NOT NULL , `date` DATE NOT NULL DEFAULT CURRENT_TIMESTAMP , `calls` INT NOT NULL DEFAULT '0' , PRIMARY KEY (`id`)) ENGINE = InnoDB;"]
  queries.forEach(element => {
    conn.query(element, function (err) {
      if (err) throw err;
      console.log("Table created for Main system");
    });
  });
  conn.release()
})

for (let i = 0; i < enabledModules.length; i++) {
  connectionPool.getConnection((err, conn) => {
    if (err) {
      if (err.code == "ECONNREFUSED") {
        console.log(chalk.red("⚠ Connection to database failed. Is the database server running? ⚠"))
        server.close()
        process.exit(5)
      } else {
        throw err;
      }
    }

    allPoolConnections.push(conn);
    conn.on('error', function (err) {
      console.log("----[MYSQL ERROR]----")
      console.log("[MYSQL ERROR] Module connection error");
      console.log(err); // 'ER_BAD_DB_ERROR'
      if (err.code == "ECONNRESET") {
        connectionPool.getConnection((err, connP) => {
          if (err) throw err;
          conn = connP // Trying to mitigate ECONNRESET issues, even tho this didn't fix it lol
        })
      }
      console.log("----]MYSQL ERROR[----")
    });
    _.each(allPoolConnections, function (arg) { setInterval(function () { arg.query("SELECT 1") }, jsonConfigGlobal.database.customSettings.wait_timout - 50) })
    try {
      enabledModules[i].initialize(conn, jsonConfigGlobal)
      const currTags = enabledModules[i].getModuleMeta().tags
      for (const tagy in currTags) {
        const tag = currTags[tagy]
        if (_.isFinite(allTags[tag])) {
          allTags[tag] = allTags[tag] + 1;
        } else {
          allTags[tag] = 1;
        }
      }
    } catch (error) {
      console.warn(chalk.red("⚠ Module " + enabledModules[i].getModuleMeta().title + " failed to initialize() ⚠" + "\n Error: " + error + "\n Stacktrace: " + error.stack))
      enabledModules[i] = stubMod;
    }
  });

  setInterval(function () {
    for (const conyT in allPoolConnections) {
      allPoolConnections[conyT].query("SELECT 1;", function (err) {
        if (err) throw err;
        console.log("Querrying heartbeat for connection pool number: " + conyT);
      });
    }
  }, jsonConfigGlobal.database.customSettings.wait_timout * 1000)
}


app.use(express.static("static"));

// A crappy logging middleware, please don't look at this
app.use(function (req, res, next) {
  lastHttpRequest = Math.floor(Date.now() / 1000)
  const date_ob = new Date();
  const date = ("0" + date_ob.getDate()).slice(-2);
  // current month
  const month = ("0" + (date_ob.getMonth() + 1)).slice(-2);
  // current year
  const year = date_ob.getFullYear();
  // current hours
  const hours = padLeadingZeros(date_ob.getHours(), 2)
  // current minutes
  const minutes = padLeadingZeros(date_ob.getMinutes(), 2)
  // current seconds
  const seconds = padLeadingZeros(date_ob.getSeconds(), 2)
  // prints date & time in YYYY-MM-DD HH:MM:SS format
  console.log("[" + year + "-" + month + "-" + date + " " + hours + ":" + minutes + ":" + seconds + "] " + req.method + " " + req.path);
  next();
});
app.use(bodyParser.urlencoded({ extended: true }));
app.use(cookieParser(jsonConfigGlobal.cookieSecret));
if (jsonConfigGlobal.betaMode) {
  require('./middleware/beta.middleware')(app, con, jsonConfigGlobal);
}

// API Taxonomy 
const apiTaxonomyCache = {};
function flushApiTaxonomyCache():void {
  if(jsonConfigGlobal.metrics.writeMetricsToMySQL){
    const updateSql = "UPDATE apitaxonomy SET calls = calls+? WHERE apikey LIKE ? AND DATE(date) = CURDATE()";
    const insertSql = "INSERT INTO apitaxonomy (apikey, calls) VALUES(?, ?);"
    console.log( Object.entries(apiTaxonomyCache).length + " entries in apiTaxonomyCache")
    for (const [key, value] of Object.entries(apiTaxonomyCache)) {
      con.query(updateSql, [value, key], function (err, result) {
        if (err) {
          console.error(err);
        }
        if(!jsonConfigGlobal.metrics.influx.enable){
          apiTaxonomyCache[key] = 0;
        }
        
        if (result.affectedRows == 0) {
          con.query(insertSql, [key, value], function (err, result) {
            if (err) {
              console.error(err);
            }
          });
        }
      });
    }
  }else{
    if(!jsonConfigGlobal.metrics.influx.enable){
    console.log(chalk.red("⚠ MySQL Metrics writing has been disabled, but Influx is not setup! ⚠"));}
  }
  if(jsonConfigGlobal.metrics.influx.enable){
    const toWritePoints = [];
    console.log( Object.entries(apiTaxonomyCache).length + " entries in apiTaxonomyCache in Influx")
    for (const [key, value] of Object.entries(apiTaxonomyCache)) {
      const val = apiTaxonomyCache[key]
      toWritePoints.push({
        measurement: 'apitaxonomy',
        tags: {},
        fields: { calls: val },
        timestamp: new Date,
      })
      apiTaxonomyCache[key] = 0;
    }
    influxD.writePoints(toWritePoints, {
      database: jsonConfigGlobal.metrics.influx.database,
      precision: 's',
    })
    .catch(error => {
      console.error(`[InfluxDB] Error saving data to InfluxDB! ${error.stack}`)
    });
  } 
}

if(jsonConfigGlobal.env == "DEV"){
  jsonConfigGlobal.taxonomyCacheInterval = 0.5;
}

setInterval(function () {
  flushApiTaxonomyCache()
  console.log("[API TAXONOMY] Wrote taxonomy to database");
}, jsonConfigGlobal.taxonomyCacheInterval * 60 * 1000); // Every `taxonomyCacheInterval` minutes


connectionPool.getConnection((err, conn) => {
  conn.on('error', function (err) {
    console.log("----[MYSQL ERROR]----")
    console.log("[MYSQL ERROR] System connection error");
    console.log(err);
    console.log("----]MYSQL ERROR[----")
  });

  setInterval(function () {
    conn.query("SELECT 1;", function (err) {
      if (err) throw err;
      console.log("Querrying heartbeat for system connection");
    });
  }, jsonConfigGlobal.database.customSettings.wait_timout * 1000) // Every 20 Minutes
  func.handleMysqlErrors(err)
  require('./middleware/maint.middleware')(app, jsonConfigGlobal, metaGlobals);
  require('./middleware/apitoken.middleware')(app, con, apiTaxonomyCache);
  require('./routes/api.route.ts')(app, enabledModules, conn, { tags: allTags });
  require('./routes/index.route.js')(app, metaGlobals, jsonConfigGlobal.fontAwesome, jsonConfigGlobal.mapboxAccessToken, jsonConfigGlobal);
  app.use(Sentry.Handlers.errorHandler());
  require('./routes/error.route.js')(app, jsonConfigGlobal, metaGlobals); // Make sure this is always last
  app.use(minify());
  app.use(compression());
})

let server = undefined;

setInterval(function () {
  if (Math.floor(Date.now() / 1000) - lastHttpRequest >= 60 * 60) { // One hour after the last http request
    const request = bent("https://pointsight.project-name-here.de/api/getPOI?boundingEast=55.21649013168979;10.696563720703127&boundingWest=54.59354980818523;9.825897216796877&key=b03f8aaf-1f32-4d9e-914a-9a50f904833d", "GET", "json", 200);
    request().then(function (body) {
      console.log("Tried to pull POI from pointsight.project-name-here.de")
    })
  }
}, 60000);

process.on('SIGINT', function () {
  console.log("Caught interrupt signal and shutting down gracefully");
  flushApiTaxonomyCache(); // Save the api taxonomy cache before quitting
  server.close(); // Make th express server stop
  _.each(allPoolConnections, function (arg) { arg.release(); }) // Let go off all mysql pool connections
  con.end(); // Close the system's mysql connection
  Sentry.close(2000).then(function () { // Wait for sentry to quit
    console.log(chalk.blue("👋 Goodbye! 👋"))
    process.exit(); // Quit the application
  });
});

// Start server
server = app.listen(jsonConfigGlobal.port, jsonConfigGlobal.adress, () => {
  console.log(`The Pointsight is running at http://localhost:${jsonConfigGlobal.port}`)
  setTimeout(function () {
    if (jsonConfigGlobal.env == "DEV") {
      console.log(allTags);
    }

  }, 1000)
})