trunk-server/backend/controllers/uploads.js

const path = require('path');
const fs = require('fs');
const sysStats = require("../sys_stats");
const  systemSchema  = require("../models/systemSchema");
const  talkgroupSchema  = require("../models/talkgroupSchema");
const  callSchema  = require("../models/callSchema");
const mongoose = require("mongoose");
const { S3Client, PutObjectCommand } = require("@aws-sdk/client-s3");
const { fromIni } = require("@aws-sdk/credential-providers");
const { NodeHttpHandler } = require('@smithy/node-http-handler');
const https = require('https');

// keepAlive matters: without it every upload does a full TCP + TLS handshake to S3.
const agent = new https.Agent({
  keepAlive: true,
  maxSockets: 250,
});

const s3_endpoint = process.env['S3_ENDPOINT'] ?? 'https://s3.us-west-1.wasabisys.com';
const s3_region = process.env['S3_REGION'] ?? 'us-west-1';
const s3_bucket = process.env['S3_BUCKET'] ?? 'openmhz-west';
const s3_profile = process.env['S3_PROFILE'] ?? 'wasabi-account';
const s3_public_url = process.env['S3_PUBLIC_URL'] ?? `${s3_endpoint}/${s3_bucket}`;
const host = process.env['MONGO_NODE_DRIVER_HOST'] != null ? process.env['MONGO_NODE_DRIVER_HOST'] : 'mongo';
const port = process.env['MONGO_NODE_DRIVER_PORT'] != null ? process.env['MONGO_NODE_DRIVER_PORT'] : 27017;
const mongoUrl = 'mongodb://' + host + ':' + port + '/scanner';


// `asPromise()` returns a promise that resolves to the connection
// once the connection succeeds, or rejects if connection failed.
const mongo_conn_slow = mongoose.createConnection(mongoUrl,  { maxPoolSize: 150 });
const mongo_conn_fast = mongoose.createConnection(mongoUrl,  { maxPoolSize: 150 });
mongo_conn_slow.on('error', console.error);
mongo_conn_fast.on('error', console.error);
mongo_conn_slow.on('disconnected', () => {
  console.log('Mongo Slow Disconnected');
  mongo_conn_slow.openUri(mongoUrl, { maxPoolSize: 150 }).catch(console.error);
});
mongo_conn_fast.on('disconnected', () => {
  console.log('Mongo Fast Disconnected');
  mongo_conn_fast.openUri(mongoUrl, { maxPoolSize: 150 }).catch(console.error);
});

mongo_conn_slow.model('Call', callSchema);
mongo_conn_fast.model('System', systemSchema);
mongo_conn_fast.model('Talkgroup', talkgroupSchema);



const client = new S3Client({
  requestHandler: new NodeHttpHandler({
    httpsAgent: agent,
  }),
  credentials: fromIni({ profile: s3_profile }),
  requestChecksumCalculation: 'WHEN_REQUIRED',
 	responseChecksumValidation: 'WHEN_REQUIRED',
  endpoint: s3_endpoint,
  region: s3_region,
  maxAttempts: 2,
});

// Every upload needs the system's key, so cache the lookup. A changed key or
// ignoreUnknownTalkgroup setting takes up to the TTL to apply. Only found systems
// are cached, so a newly created system works immediately.
const SYSTEM_CACHE_TTL_MS = 60 * 1000;
const systemCache = new Map();

async function getSystem(shortName) {
  const cached = systemCache.get(shortName);
  if (cached && cached.expires > Date.now()) {
    return cached.system;
  }
  const system = await mongo_conn_fast.model("System").findOne({ shortName }, ["key", "ignoreUnknownTalkgroup"]).lean();
  if (system) {
    systemCache.set(shortName, { system, expires: Date.now() + SYSTEM_CACHE_TTL_MS });
  } else {
    systemCache.delete(shortName);
  }
  return system;
}

exports.upload = async function (req, res, next) {
  let start_time = Date.now();
  let validateSystemTime = 0;
  let readFileTime = 0;
  let uploadFileTime = 0;
  let saveCallTime = 0;
  let cleanupTime = 0;
  let statsTime = 0;
  try {
    if (!req.file || !['.m4a', '.mp3'].includes(path.extname(req.file.originalname))) {
      console.warn(`[${req.params.shortName}] Error file name is wrong or file does not exist`);
      res.status(500).send("Error, invalid filename:\n");
      return;
    }

    const shortName = req.params.shortName.toLowerCase();
    const apiKey = req.body.api_key;
    const talkgroupNum = parseInt(req.body.talkgroup_num);
    const freq = parseFloat(req.body.freq);
    const time = new Date(parseInt(req.body.start_time) * 1000);
    let stopTime = new Date();
    if (req.body.stop_time) {
      stopTime = new Date(parseInt(req.body.stop_time) * 1000);
    }
    const startTime = req.body.start_time;
    const emergency = parseInt(req.body.emergency);

    let errorCount = parseInt(req.body.error_count) || 0;
    let spikeCount = parseInt(req.body.spike_count) || 0;

    let srcList = [];
    try {
      srcList = JSON.parse(req.body.source_list);
    } catch (err) {
      console.warn(`[${req.params.shortName}] Error /:shortName/upload Parsing Source/Freq List - Error: ${err}`);
      res.status(500).send("Error parsing sourcelist " + err);
      return;
    }

    patches = [];

    let req_patches;
    req_patches = req.body.patch_list;

    if(typeof req_patches != "undefined"){
      var split_patches = req_patches.replace("[","").replace("]","").split(",");

      for (var patch in split_patches){
        patches.push(split_patches[patch]);
      }
    }

    let item = null;

    try {
      item = await getSystem(shortName);
    } catch (err) {
      console.warn(`[${req.params.shortName}] Error validating system: ${err}`);
    }

    if (!item) {
      console.warn(`[${req.params.shortName}] Error /:shortName/upload ShortName does not exist`);
      res.status(500).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).send("API Keys do not match!\n");
      return;
    }

    if (shortName === "hennearmer" && [3421, 3423].includes(talkgroupNum)) {
      res.status(200).end();
      return;
    }
    validateSystemTime = Date.now();
    if (item.ignoreUnknownTalkgroup) {
      const talkgroupExists = await mongo_conn_fast.model("Talkgroup").exists({
        shortName,
        num: talkgroupNum,
      });

      if (!talkgroupExists) {
        res.status(500).send("Talkgroup does not exist, skipping.\n");
        return;
      }
    }

    res.status(200).end();

    const local_path = `/${shortName}/${time.getFullYear()}/${time.getMonth() + 1}/${time.getDate()}/`;
    const object_key = `media/${shortName}/${talkgroupNum}/${shortName}-${talkgroupNum}-${startTime}${path.extname(req.file.originalname)}`;
    const url = `${s3_public_url}/${object_key}`;

    const call = new (mongo_conn_slow.model("Call"))({
      shortName,
      talkgroupNum,
      objectKey: object_key,
      endpoint: s3_endpoint,
      bucket: s3_bucket,
      time,
      name: `${talkgroupNum}-${startTime}.m4a`,
      freq,
      errorCount,
      spikeCount,
      url,
      emergency,
      path: local_path,
      patches: patches,
      srcList,
      len: req.body.call_length ? parseFloat(req.body.call_length) : (stopTime - time) / 1000,
    });

    let fileContent;

    try {
      fileContent = fs.readFileSync(req.file.path);
      readFileTime = Date.now();
      const command = new PutObjectCommand({
        Bucket: s3_bucket,
        Key: object_key,
        Body: fileContent,
        ACL: 'public-read',
      });
      var result = await client.send(command);
      if (result && result.$metadata.httpStatusCode !== 200) {
        console.error(`[${shortName}] Upload Error status code: ${result.$metadata.httpStatusCode}`);
        console.error(result);
      }
    } catch (err) {
      console.warn(`[${call.shortName}] Upload Error: ${err}`);
    }
    uploadFileTime = Date.now();

    try {
      await call.save();
    } catch (err) {
      console.warn(`[${call.shortName}] Error saving call: ${err}`);
    }

    saveCallTime = Date.now();
    sysStats.addCall(call.toObject());
    statsTime = Date.now();

    if (call.len >= 1) {
      req.call = call.toObject();
      next();
    }

    //console.log(`[${call.shortName}] \t Verify System: ${validateSystemTime - start_time}  \t Read file: ${readFileTime - validateSystemTime} \t Upload: ${ uploadFileTime - validateSystemTime} \t Save: ${saveCallTime - uploadFileTime} \tStats: ${statsTime - saveCallTime}\tCleanup: ${cleanupTime - statsTime} \t\t Total: ${cleanupTime - start_time}`);
  } catch (error) {
    console.error("Error processing call upload: " + error);
  } finally {
    // multer has already written the temp file, so remove it on every exit path.
    if (req.file) {
      try {
        fs.unlinkSync(req.file.path);
      } catch (err) {
        console.warn("There was an Error deleting: " + req.file.path);
      }
    }
    cleanupTime = Date.now();
    const totalTime = Date.now() - start_time;
    if (totalTime > 10000) {
      console.warn(`[${req.params.shortName}] Slow Upload - Size: ${call.len} Verify System: ${validateSystemTime - start_time}  Read file: ${readFileTime - validateSystemTime} Upload: ${ uploadFileTime - validateSystemTime} Save: ${saveCallTime - uploadFileTime} Stats: ${statsTime - saveCallTime} Cleanup: ${cleanupTime - statsTime} Total: ${totalTime}`);
    }
  }
};