const path = require('path'); const fs = require('fs'); const opentelemetry = require('@opentelemetry/api'); const sysStats = require("../sys_stats"); const System = require("../models/system"); const Talkgroup = require("../models/talkgroup"); const { callModel: Call } = require("../models/call"); const { trace, context } = opentelemetry; 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'); const agent = new https.Agent({ 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 client = new S3Client({ requestHandler: new NodeHttpHandler({ httpsAgent: agent, }), credentials: fromIni({ profile: s3_profile }), endpoint: s3_endpoint, region: s3_region, maxAttempts: 2, }); exports.upload = async function (req, res, next) { const tracer = trace.getTracer("upload-service"); return context.with(context.active(), async () => { const parentSpan = trace.getActiveSpan(context.active()); await tracer.startActiveSpan('upload_handler', { parent: parentSpan }, async (span) => { try { span.setAttribute('call.shortName', req.params.shortName.toLowerCase()); span.setAttribute('call.talkgroup_num', req.body.talkgroup_num); span.setAttribute('call.start_time', req.body.start_time); 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`); span.setStatus({ code: opentelemetry.SpanStatusCode.ERROR, message: "Invalid file or filename", }); 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; } let item = null; await tracer.startActiveSpan('validate_system', { parent: trace.getActiveSpan(context.active()) }, async (validateSpan) => { try { item = await System.findOne({ shortName }, ["key", "ignoreUnknownTalkgroup"]); validateSpan.setAttribute('system.exists', !!item); } catch (err) { validateSpan.recordException(err); validateSpan.setStatus({ code: opentelemetry.SpanStatusCode.ERROR, message: "Error validating system: " + err.message, }); } finally { validateSpan.end(); } }); 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; } if (item.ignoreUnknownTalkgroup) { const talkgroupExists = await Talkgroup.exists({ shortName, num: talkgroupNum, }); if (!talkgroupExists) { try { fs.unlinkSync(req.file.path); } catch (err) { console.error(`[${call.shortName}] error deleting: ${req.file.path}`); } 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}-${startTime}${path.extname(req.file.originalname)}`; const url = `${s3_public_url}/${object_key}`; const call = new 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, srcList, len: req.body.call_length ? parseFloat(req.body.call_length) : (stopTime - time) / 1000, }); let fileContent; await tracer.startActiveSpan('upload_to_s3', { parent: trace.getActiveSpan(context.active()) }, async (uploadSpan) => { try { fileContent = fs.readFileSync(req.file.path); 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) { uploadSpan.recordException(err); uploadSpan.setStatus({ code: opentelemetry.SpanStatusCode.ERROR, message: "Upload Error: " + err.message, }); console.warn(`[${call.shortName}] Upload Error: ${err}`); } finally { uploadSpan.end(); } }); await tracer.startActiveSpan('save_call', { parent: trace.getActiveSpan(context.active()) }, async (saveSpan) => { try { await call.save(); } catch (err) { saveSpan.recordException(err); saveSpan.setStatus({ code: opentelemetry.SpanStatusCode.ERROR, message: "Error saving call: " + err.message, }); console.warn(`[${call.shortName}] Error saving call: ${err}`); } finally { saveSpan.end(); } }); sysStats.addCall(call.toObject()); if (call.len >= 1) { req.call = call.toObject(); next(); } await tracer.startActiveSpan('cleanup_temp_file', { parent: trace.getActiveSpan(context.active()) }, async (cleanupSpan) => { try { fs.unlinkSync(req.file.path); cleanupSpan.setAttribute('file.deleted', true); } catch (err) { cleanupSpan.recordException(err); console.warn("There was an Error deleting: " + req.file.path); } finally { cleanupSpan.end(); } }); } catch (error) { console.error("Error processing call upload: " + error); span.recordException(error); span.setStatus({ code: opentelemetry.SpanStatusCode.ERROR, message: error.message, }); } finally { span.end(); } }); }); };