loadCalculatorOutputIntoDB

#!/usr/bin/env node

/* eslint no-console: 0, no-await-in-loop: 0 */

const { isAbsolute, join } = require('path');
const { existsSync, readFileSync, createReadStream } = require('fs');

const _ = require('lodash');
const split2 = require('split2');
const through2 = require('through2');
const copyFrom = require('pg-copy-streams').from;

const { NDJSON } = require('./src/enums/outputFileFormats');
const { EAV } = require('./src/enums/outputFormats');

const DBService = require('./src/storage/services/DBService');

const calculatorOutputDir = process.argv[2];

if (!calculatorOutputDir) {
  console.error(
    'ERROR: The calculator output directory path must be specified as the 1st cli argument.'
  );
  process.exit(1);
}

const calculatorOutputDirPath = isAbsolute(calculatorOutputDir)
  ? calculatorOutputDir
  : join(process.cwd(), calculatorOutputDir);

if (!existsSync(calculatorOutputDirPath)) {
  console.error(
    `ERROR: The specified calculator output directory ${calculatorOutputDirPath} does not exist.`
  );
  process.exit(1);
}

const calculatorMetadataPath = join(
  calculatorOutputDirPath,
  'calculator_metadata.json'
);

if (!existsSync(calculatorMetadataPath)) {
  console.error(
    `ERROR: ${calculatorOutputDirPath} does not contain a calculator_metadata.json file.`
  );
  process.exit(1);
}

const calculatorMetadata = JSON.parse(
  readFileSync(calculatorMetadataPath, {
    encoding: 'utf8'
  })
);

if (calculatorMetadata.calculatorSettings.outputFileFormat !== NDJSON) {
  console.error(
    'ERROR: The loader requires the calculator was run with --outputFileFormat=NDJSON'
  );
  process.exit(1);
}

if (calculatorMetadata.calculatorSettings.outputFormat !== EAV) {
  console.error(
    'ERROR: The loader requires the calculator was run with --outputFormat=EAV'
  );
  process.exit(1);
}

const { calculators: pm3MeasureCalculatorsMetadata } = calculatorMetadata;

const insertPM3CalculatorMetadata = async () => {
  const metadata = _.cloneDeep(calculatorMetadata);
  metadata.calculators.forEach(cMeta => {
    /* eslint-disable-next-line no-param-reassign */
    delete cMeta.outputFileName;
  });

  const {
    rows: [{ id }]
  } = await DBService.query(
    'INSERT INTO public.pm3_calculator_metadata (metadata) VALUES ($1) RETURNING id',
    [metadata]
  );

  return id;
};

// NOTE: Will cascade and clean up pm3_measure_calculator_metadata and pm3_eav_append_only as well.
const deletePM3CalculatorMetadata = pm3calc_id =>
  DBService.query('DELETE FROM public.pm3_calculator_metadata WHERE id = $1', [
    pm3calc_id
  ]);

const insertPM3MeasureCalculatorsMetadata = async pm3calc_id => {
  const pm3meacalc_ids = [];

  const client = await DBService.pool.connect();
  try {
    await client.query('BEGIN');

    for (let i = 0; i < pm3MeasureCalculatorsMetadata.length; ++i) {
      const metadata = pm3MeasureCalculatorsMetadata[i];

      const {
        rows: [{ id }]
      } = await client.query(
        'INSERT INTO public.pm3_measure_calculator_metadata (pm3calc_id, metadata) VALUES ($1, $2) RETURNING id',
        [pm3calc_id, _.omit(metadata, 'outputFileName')]
      );

      pm3meacalc_ids.push(id);
    }

    await client.query('COMMIT');

    return pm3meacalc_ids;
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    await client.release();
  }
};

const loadMetadata = async (metadataType, pm3calc_id, metadataFileName) => {
  if (!metadataFileName) {
    return;
  }

  const outputFilePath = join(calculatorOutputDir, metadataFileName);

  if (existsSync(outputFilePath)) {
    const client = await DBService.pool.connect();

    const pseudoMetadata = {
      measure: metadataType,
      year: calculatorMetadata.calculatorSettings.year,
      isCanonical: true
    };

    try {
      const {
        rows: [{ id }]
      } = await client.query(
        'INSERT INTO public.pm3_measure_calculator_metadata (pm3calc_id, metadata) VALUES ($1, $2) RETURNING id',
        [pm3calc_id, pseudoMetadata]
      );

      await new Promise(async (resolve, reject) => {
        const dataStream = createReadStream(outputFilePath)
          .on('error', reject)
          .pipe(split2(JSON.parse)) // https://github.com/mcollina/split2#ndj---newline-delimited-json
          .pipe(
            through2.obj(function ndjsonTransformer(
              { tmc, attribute, value },
              $,
              cb
            ) {
              const v = typeof value === 'string' ? `"""${value}"""` : value;

              const row = `${id},"${tmc}","${attribute}",${v}`;
              this.push(Buffer.from(`${row}\n`));
              cb();
            })
          );

        const copyStream = client.query(
          copyFrom(
            'COPY public.pm3_eav_append_only (pm3meacalc_id, tmc, attribute, value) FROM STDIN WITH CSV'
          )
        );

        dataStream.on('error', reject);

        copyStream.on('error', reject);
        copyStream.on('end', resolve);

        dataStream.pipe(copyStream);
      });
    } catch (err) {
      throw err;
    } finally {
      await client.release();
    }
  } else {
    throw new Error(
      `ERROR: measure file ${outputFilePath} is missing from the outputDir`
    );
  }
};

const loadTmcMetadata = loadMetadata.bind(null, 'TMC_METADATA');
const loadRisMetadata = loadMetadata.bind(null, 'RIS_METADATA');

const loadCalculatorsEAVOutput = async pm3meacalc_ids =>
  Promise.all(
    pm3meacalc_ids.map(async (pm3meacalc_id, i) => {
      const metadata = pm3MeasureCalculatorsMetadata[i];
      const { outputFileName } = metadata;
      const outputFilePath = join(calculatorOutputDir, outputFileName);

      if (existsSync(outputFilePath)) {
        const client = await DBService.pool.connect();

        try {
          await new Promise(async (resolve, reject) => {
            const dataStream = createReadStream(outputFilePath)
              .on('error', reject)
              .pipe(split2(JSON.parse)) // https://github.com/mcollina/split2#ndj---newline-delimited-json
              .pipe(
                through2.obj(function ndjsonTransformer(
                  { tmc, attribute, value },
                  $,
                  cb
                ) {
                  const v =
                    typeof value === 'string' ? `"""${value}"""` : value;

                  const row = `${pm3meacalc_id},"${tmc}","${attribute}",${v}`;
                  this.push(Buffer.from(`${row}\n`));
                  cb();
                })
              );

            const copyStream = client.query(
              copyFrom(
                'COPY public.pm3_eav_append_only (pm3meacalc_id, tmc, attribute, value) FROM STDIN WITH CSV'
              )
            );

            dataStream.on('error', reject);

            copyStream.on('error', reject);
            copyStream.on('end', resolve);

            dataStream.pipe(copyStream);
          });
        } catch (err) {
          throw err;
        } finally {
          await client.release();
        }
      } else {
        throw new Error(
          `ERROR: measure file ${outputFilePath} is missing from the outputDir`
        );
      }
    })
  );

(async () => {
  let pm3calc_id;
  let pm3meacalc_ids;

  const { tmcMetadataFileName, risMetadataFileName } = calculatorMetadata;

  try {
    pm3calc_id = await insertPM3CalculatorMetadata();
    pm3meacalc_ids = await insertPM3MeasureCalculatorsMetadata(pm3calc_id);

    await loadTmcMetadata(pm3calc_id, tmcMetadataFileName);

    if (!_.isEmpty(risMetadataFileName)) {
      await loadRisMetadata(pm3calc_id, risMetadataFileName);
    }

    await loadCalculatorsEAVOutput(pm3meacalc_ids);
  } catch (err) {
    // Clean up failed load. DELETE cascades.
    await deletePM3CalculatorMetadata(pm3calc_id);
    console.error(err.message);
  } finally {
    DBService.end();
  }
})();