trunk-server/backend/controllers/uploads.js

var db = require('../db');
var mkdirp = require('mkdirp');
var path = require('path');
var fs = require('fs');
var sysStats = require("../sys_stats");
var probe = require('node-ffprobe');
var mongoose = require("mongoose");
var System = require("../models/system");
var Talkgroup = require("../models/talkgroup");
var Group = require("../models/group");
var {callModel:Call} = require("../models/call");
var secrets = require("../config/secrets");
var util = require("util")
var config = require('../config/config.json');

var s3_endpoint = process.env['S3_ENDPOINT'] != null ? process.env['S3_ENDPOINT'] : 's3.us-west-1.wasabisys.com';
var s3_bucket = process.env['S3_BUCKET'] != null ? process.env['S3_BUCKET'] : 'openmhz-west';
var s3_profile = process.env['S3_PROFILE'] != null ? process.env['S3_PROFILE'] : 'wasabi-account';

const wasabi = require('aws-sdk');

const wasabiEndpoint = new wasabi.Endpoint(s3_endpoint);
var wasabiCredentials = new wasabi.SharedIniFileCredentials({
  profile: s3_profile
});
wasabi.config.credentials = wasabiCredentials;
const wasabiS3 = new wasabi.S3({
  endpoint: wasabiEndpoint,
  maxRetries: 0
});

async function callLength(file) {

  return new Promise((resolve, reject) => {
    probe(file, function (err, probeData) {
      if (err) {
        reject(err);
      } else {
        resolve(Math.round(probeData.format.duration));
      }
    });
  })
}

exports.upload = function (req, res, next) {
  process.nextTick(function () {



    /** When using the "single"
        data come in "req.file" regardless of the attribute "name". **/
    var tmp_path = req.file.path;

    /** The original name of the uploaded file
        stored in the variable "originalname". **/
    var target_path = 'uploads/' + req.file.originalname;

    if (!req.file && (path.extname(req.file.originalname)) != '.m4a') {
      console.warn("[" + req.params.shortName + "] Error file name is wrong or file does not exist");
      res.status(500);
      res.send("Error, invalid filename:\n");
      return;
    }
    var shortName = req.params.shortName.toLowerCase();
    var apiKey = req.body.api_key;



    System.findOne({
      'shortName': shortName
    }, ["key", "ignoreUnknownTalkgroup"], async function (err, item) {

      if (err) {
        console.warn("[" + req.params.shortName + "] Error /:shortName/upload - Error: " + err);
        res.status(500);
        res.send("Error, invalid shotName:\n" + err);
        return;
      }
      if (item === null) {
        console.warn("[" + req.params.shortName + "] Error /:shortName/upload ShortName does not exist");
        res.status(500);
        res.send("ShortName does not exist: " + shortName + "\n");
        return;
      }
      if (apiKey != item.key) {
        console.warn("[" + req.params.shortName + "] Error /:shortName/upload API Key Mismatch - Provided key: " + apiKey);
        res.status(500);
        res.send("API Keys do not match!\n");
        return;
      }

      var talkgroupNum = parseInt(req.body.talkgroup_num);
      var freq = parseFloat(req.body.freq);
      var time = new Date(parseInt(req.body.start_time) * 1000);
      var stopTime = new Date();
      if (req.body.stop_time) {
        var stopTime = new Date(parseInt(req.body.stop_time) * 1000);
      }
      var startTime = req.body.start_time;
      var emergency = parseInt(req.body.emergency);


      if (item.ignoreUnknownTalkgroup == true) {
        //console.log("ignore unknown talkgroup ");
        //await Talkgroup.findOne({ answer: 42 }).select({ _id: 1 }).lean().then(doc => !!doc)
        talkgroupExists = await Talkgroup.exists({
          'shortName': shortName,
          'num': talkgroupNum
        });
        //console.log("In talkgroup: " + talkgroupExists);
        //talkgroupExists = await Talkgroup.findOne({ 'shortName': shortName,  'num': talkgroupNum }).select({ _id: 1 }).lean().then(doc => !!doc).exec();
        if (!talkgroupExists) {
          try {
            fs.unlinkSync(req.file.path)
            //file removed
          } catch (err) {
            console.log("[" + call.shortName + "] error deleting: " + req.file.path);
          }

          res.status(500);
          res.send("Talkgroup does not exist, skipping.\n");
          return;

        }
      }
      // Add in an API Key check

      try {

        var srcList = JSON.parse(req.body.source_list);

        // Mongoose does not allow 'errors' to be used in the Schema for call. need to rename to error.
        var badFreqList = JSON.parse(req.body.freq_list);
        var freqList = badFreqList.map(obj => {
          obj.error = obj.errors;
          delete obj.errors;
          return obj;
        });

      } catch (err) {
        var srcList = [];
        var freqList = [];
        console.warn("[" + req.params.shortName + "] Error /:shortName/upload Parsing Source/Freq List -  Error: " + err);

        res.status(500);
        res.send("Error parsing sourcelist " + err);
      }

      var base_path = config.mediaDirectory;
      var local_path = "/" + shortName + "/" + time.getFullYear() + "/" + (time.getMonth() + 1) + "/" + time.getDate() + "/";
      var target_file = base_path + local_path + path.basename(req.file.originalname);
      var object_key = "media/" + shortName + "-" + talkgroupNum + "-" + startTime + ".m4a";

      var endpoint = s3_endpoint;
      var bucket = s3_bucket;


      // If not a pro plan, use S3 storage
      var url = 'https://' + s3_endpoint + "/" + s3_bucket + "/" + object_key;
      var objectStore = true;


      var call = new Call({
        shortName: shortName,
        talkgroupNum: talkgroupNum,
        objectKey: object_key,
        endpoint: endpoint,
        bucket: bucket,
        time: time,
        name: talkgroupNum + "-" + startTime + ".m4a",
        freq: freq,
        url: url,
        emergency: emergency,
        path: local_path,
        srcList: srcList,
        freqList: freqList,
        len: -1
      });

      try {
        if (req.body.call_length) {
          call.len = parseInt(req.body.call_length);
        } else {
          call.len = (stopTime - time) / 1000; //await callLength(req.file.path);
        }
        res.status(200).end();


        if (!objectStore) {
          mkdirp.sync(base_path + local_path, function (err) {
            if (err) console.error(err);
          });
          var src = fs.createReadStream(req.file.path);
          var dest = fs.createWriteStream(target_file);
          dest.on('unpipe', (src) => {
            src.end();
            dest.end();
          });
          src.pipe(dest);
          src.unpipe()
          src.on('error', function (err) {
            console.warn("[" + call.shortName + "] Error /:shortName/upload copying to location - Error: " + err + " Read: " + req.file.path + " Target: " + target_file);
            fs.unlink(req.file.path, (err) => {
              if (err)
                console.log("[" + call.shortName + "] error deleting: " + req.file.path);
            });
            res.contentType('json');
            res.send(JSON.stringify({
              success: false,
              error: "Upload failed"
            }));
          });
          src.on('end', function () {
            fs.unlink(req.file.path);
          });
        } else {
          //var spacesSrc = fs.createReadStream(req.file.path);
          //var awsSrc = fs.createReadStream(req.file.path);
          var wasabiSrc = fs.createReadStream(req.file.path);

          // call S3 to retrieve upload file to specified bucket

          /*  var spacesParams = {Bucket: 'openmhz', Key: object_key, Body: spacesSrc, ACL: 'public-read'};
            // call S3 to retrieve upload file to specified bucket
            spacesS3.upload (spacesParams, function (err, data) {
              //fs.unlink(req.file.path);
              if (err) {
                console.log("Spaces Error", err);
              }
            });*/
          /*
          var awsParams = {Bucket: 'openmhz', Key: object_key, Body: awsSrc, ACL: 'public-read'};

          awsS3.upload (awsParams, function (err, data) {
            fs.unlink(req.file.path);
            if (err) {
              console.log("AWS Error", err);
            }
          });*/

          var wasabiParams = {
            Bucket: s3_bucket,
            Key: object_key,
            Body: wasabiSrc,
            ACL: 'public-read'
          };


          //var options = {partSize: 10 * 1024 * 1024, queueSize: 1};
          /*var upload = new wasabiS3.ManagedUpload({
            leavePartsOnError: true,
            partSize: 10 * 1024 * 1024, queueSize: 1,
            params: wasabiParams
          });*/
          //wasabiS3.upload(wasabiParams, options, function(err, data) {
          wasabiS3.upload(wasabiParams, function (err, data) {
            wasabiSrc.destroy();
            fs.unlink(req.file.path, (err) => {
              if (err) {
                console.log("[" + call.shortName + "]error deleting: " + req.file.path);
              }
              call.save();
              sysStats.addCall(call.toObject());

              // we only want to notify clients if the clip is longer than 1 second.
              if (call.len >= 1) {
                req.call = call.toObject();
                next();
              }
            });
            if (err) {
              console.log("[" + call.shortName + "] " + call.name + " -   content-length: " + req.headers['content-length'] + " Wasabi Error", err);
            }
          });
        }
      } catch (err) {
        console.warn("[" + call.shortName + "] Upload Error: " + err + " Filename: " + call.name + " content-length: " + req.headers['content-length']);
        res.status(500);
        sysStats.addError(call.toObject());
        res.contentType('json');
        res.send(JSON.stringify({
          success: false,
          error: "FFProbe"
        }));
        fs.unlink(req.file.path, (err) => {
          if (err)
            console.log("error deleting: " + req.file.path);
        });
      }



    });
  });
}