agmission/Development/server/workers/obstacle_worker.js

'use strict';

const debug = require('debug')('agm:obstacle_worker'),
  cron = require('node-cron'),
  util = require('util'),
  path = require('path'),
  fs = require('fs-extra'),
  moment = require('moment'),
  cheerio = require('cheerio'),
  puppeteer = require('puppeteer'),
  env = require('../helpers/env'),
  isProd = env.PRODUCTION,
  obsUtil = require('../helpers/file_obstacle'),
  utils = require('../helpers/utils'),
  unzip = require('extract-zip'),
  Obstacle = require('../model/obstacles'),
  { DBConnection } = require('../helpers/db/connect');

// Initialize database connection
const workerDB = new DBConnection('Obstacle Worker');

debug("Is Prod", isProd);
const runUTC = moment.utc();

process
  .on('uncaughtException', function (err) {
    debug(err);
    debug("Node NOT Exiting...");
    process.exit(1);
  })
  .on('unhandledRejection', (reason, p) => {
    debug(reason, 'Unhandled Rejection at Promise', p);
  });

// Initialize the database connection
workerDB.initialize({ setupExitHandlers: false });

const stateFileName = path.join(__dirname, 'obstacle_wkr_state.json');

const readObstaclesASync = util.promisify(obsUtil.readObstacles);
const downloadAsync = util.promisify(utils.download);

/*
* * * * * *
| | | | | |
| | | | | day of week
| | | | month
| | | day of month
| | hour
| minute
second ( optional )

field	        value
second	      0-59
minute	      0-59
hour	        0-23
day of month	1-31
month	        1-12 (or names)
day of week	  0-7 (or names, 0 or 7 are Sunday)
*/

const cleanTempFilesTiming = isProd ? '5 1 * * 7' : `*/30 * * * * *`;
// In Production mode, run “At 01:05 on every 7th day-of-week.”
cron.schedule(cleanTempFilesTiming, async () => {
  debug("Start cleaning temp. files ...", moment.utc().toISOString());
  const tempPaths = isProd ?
    ['/media/ssd1/agmission/.tmp/', '/media/ssd1/agmission/reports/dat/'] :
    ['/media/data/trung/work/AgMission/trunk/Development/server/.tmp/', '/media/data/trung/work/AgMission/trunk/Development/server/reports/dat/'];

  const numOfDays = isProd ? 7 : 1;
  let delCmd = tempPaths.map(p => `find ${p} -mindepth 1 -mtime +${numOfDays} -daystart -prune -exec rm -f -R {} \\;`);
  delCmd = delCmd.join(' ; ');

  try {
    await utils.execAsync(delCmd);
    debug("Cleaning temp files is Done.");
  } catch (err) {
    debug(err.stack);
  }
},
  {
    scheduled: false,
    timezone: "Etc/UTC"
  });

const updateObstaclesData = {
  schedule: isProd ? '30 1 1 */1 *' : `*/30 * * * * *`, //`${runUTC.minute() + 1} ${runUTC.hour()} ${runUTC.date()} * *`,
  status: 0
}
// In Production mode, run “At 01:30 on day-of-month 1 in every month.”
cron.schedule(updateObstaclesData.schedule, async () => {
  // Check and only proceed when is idle and the db connection is connected
  if (!workerDB.isReady() || updateObstaclesData.status)
    return;

  const task = 'Updating obstacles';
  debug(`Start ${task} ...`, moment.utc().toISOString());

  try {
    updateObstaclesData.status = 1;
    await updateFAAObstacles();
  } catch (err) {
    debug(isProd ? err.message : err.stack);
  } finally {
    debug(`${task} is Done.`);
    updateObstaclesData.status = 0;
  }
},
  {
    scheduled: true,
    timezone: "Etc/UTC"
  });

const getDOFLinks = html => {
  var data = [], a;
  const $ = cheerio.load(html);
  $("table caption:contains('Digital Obstacle Files'), tbody > tr > td").each((i, td) => {
    a = $(td).find('a');
    if (a.length && (a.text() && a.text().toUpperCase() === "DOF")) {
      data.push({ product: a.text(), link: a.attr('href') });
    }
  });
  return data;
}

async function getLatestFAAObsZip(lastState) {
  try {
    const browser = await puppeteer.launch({
      headless: "new",
      args: ['--incognito'],
      ignoreHTTPSErrors: true,
      ignoreDefaultArgs: ['--disable-dev-shm-usage'],
    });
    const page = await browser.newPage();
    await page.goto(env.FAA_DOF_URL, { waitUntil: 'networkidle2' });
    const response = await page.evaluate(() => document.querySelector('body').innerHTML);
    await browser.close();

    let dofs = getDOFLinks(response);
    if (dofs.length) {
      let dof = dofs[dofs.length - 1];
      if (!dof.link) throw new Error('no_obs_file');

      if (lastState.dofFile && path.basename(lastState.dofFile.toLowerCase()) === path.basename(dof.link).toLowerCase())
        throw new Error('obs_file_already_updated');

      return dof.link;
    }
    return null;
  } catch (err) {
    throw err;
  }
}

async function updateFAAObstacles() {
  const unzipPath = path.join(env.TEMP_DIR, '/obs/');
  const dofFilePath = path.join(unzipPath, 'DOF.DAT');
  const lastState = fs.readJsonSync(stateFileName, { throws: false }) || {};

  let fileURL, zipFile;
  let obstacles = [];

  try {
    fileURL = await getLatestFAAObsZip(lastState);
    if (fileURL) {
      zipFile = path.join(unzipPath, path.basename(fileURL));
      fs.ensureDirSync(unzipPath);
      // Check if the file is already downloaded
      if (!fs.existsSync(zipFile)) {
        debug("Downloading FAA Obstacle file :" + fileURL);
        await downloadAsync(fileURL, zipFile);
      } else {
        debug("Skipped. FAA Obstacle file already downloaded :" + zipFile);
      }
    } else {
      throw new Error('no_obs_file');
    }

    debug("Unzipping :" + zipFile);
    await unzip(zipFile, { dir: path.resolve(unzipPath) });

    if (!fs.pathExistsSync(dofFilePath)) throw new Error('no_obs_dof_file');

    obstacles = await readObstaclesASync(dofFilePath);

    debug("Removing old obstacles !");
    await Obstacle.deleteMany({ "properties.type": { $ne: "USER" } });

    if (obstacles && obstacles.length) {
      debug(`Importing ${obstacles.length} new records ...`);
      const chunks = utils.chunkArray(obstacles, 1000);
      for (const chunk of chunks) {
        await Obstacle.insertMany(chunk);
      }
    }

    debug(`Imported Obstacle file ${dofFilePath} with ${obstacles.length ? obstacles.length : 0} records`);

    lastState['dofFile'] = fileURL;
    lastState['total'] = obstacles.length;
    lastState['date'] = moment.utc().toISOString();
    fs.writeJsonSync(stateFileName, lastState);

    return obstacles.length;
  } catch (err) {
    if (err.message && !err.message.includes('_')) {
      debug('Error when updating FAA Obstacles', err);
    }
    throw err;
  }
}