'use strict'; /** * Async Export controller — /api/v1/jobs/:jobId/export and /api/v1/exports/:exportId * * Flow: * 1. POST /api/v1/jobs/:jobId/export → creates ExportJob record (status=pending), * kicks off async generation, returns { exportId, status: 'pending' }. * 2. GET /api/v1/exports/:exportId → poll status; when ready returns { status: 'ready', downloadUrl }. * 3. GET /api/v1/exports/:exportId/download → streams the file, schedules cleanup. * * FE / integration notes: * - For the daily 17:00 batch: POST export after previous day's jobs are confirmed sprayed, * poll every 10–30 s, then download the CSV when ready. * - interval param works identically to the records endpoint (GPS point thinning). * - CSV has all raw trace fields + job/session header columns repeated per row for * direct Power BI / data-warehouse import without joins. */ const path = require('path'); const fs = require('fs'); const { Transform, pipeline } = require('stream'); const { promisify } = require('util'); const pipelineAsync = promisify(pipeline); const ObjectId = require('mongodb').ObjectId; const moment = require('moment'); const { Job, App, AppFile, AppDetail } = require('../model'); const ExportJob = require('../model/export_job'); const { AppParamError, AppAuthError } = require('../helpers/app_error'); const { Errors, HttpStatus, ExportUnits } = require('../helpers/constants'); const utils = require('../helpers/utils'); const env = require('../helpers/env'); const EXPORT_TTL_HOURS = parseInt(process.env.EXPORT_TTL_HOURS) || 24; // Re-use the same helpers from api_pub (inline to avoid a shared helper module for now) function parseInterval(raw) { if (raw == null || raw === '') return null; const v = parseFloat(raw); return isFinite(v) && v > 0 ? v : null; } function decodeSatsIn(raw) { return utils.isNumber(raw) ? (raw > 99 ? raw - 100 : raw) : null; } function decodeCorrectionFields(tslu, calcodeFreq) { const correctionId = utils.isNumber(tslu) ? (tslu > 100 ? tslu - 100 : tslu) : null; let waasId = null; if (utils.isNumber(calcodeFreq) && calcodeFreq >= 20001 && calcodeFreq <= 29999) waasId = calcodeFreq - 20000; return { correctionId, waasId }; } function computeAppRateApplied(lminApp, grSpeed, swath) { if (!utils.isNumber(lminApp) || !utils.isNumber(grSpeed) || !utils.isNumber(swath)) return null; if (grSpeed === 0 || swath === 0) return null; return lminApp / (grSpeed * swath) * 10000; } /** Verify job ownership — throws on mismatch. */ async function ownerJob(jobId, ownerId) { const job = await Job.findOne({ _id: jobId, markedDelete: { $ne: true } }).lean(); if (!job) AppParamError.throw(Errors.JOB_NOT_FOUND); if (!job.byPuid || job.byPuid.toString() !== ownerId.toString()) AppAuthError.throw(); return job; } // ─── Unit conversion helpers ───────────────────────────────────────────────── // All raw AppDetail values are stored in SI/metric units. // When units='us', these factors convert to US customary equivalents. const CONV = { msToMph: v => +(v * 2.23694).toFixed(4), // m/s → mph mToFt: v => +(v * 3.28084).toFixed(3), // m → ft cToF: v => +(v * 9 / 5 + 32).toFixed(2), // °C → °F LminToGmin: v => +(v * 0.264172).toFixed(4), // L/min → gal/min LhaToGac: v => +(v * 0.10694).toFixed(4), // L/ha → gal/ac }; function applyConv(v, fn) { return (v != null && v !== '') ? fn(Number(v)) : v; } /** * Returns CSV column definitions for the requested unit system. * Each entry: { key (row-object property), header (CSV column name) }. */ function getCsvColumns(units) { const us = units === ExportUnits.US; return [ // Job / session metadata — no unit conversion { key: 'jobId' }, { key: 'orderNumber' }, { key: 'jobName' }, { key: 'sessionId' }, { key: 'fileName' }, { key: 'pilotName' }, // GPS data { key: 'timestampUtc' }, { key: 'gpsTime' }, { key: 'lat' }, { key: 'lon' }, { key: 'utmX' }, { key: 'utmY' }, { key: 'alt', header: us ? 'alt_ft' : 'alt_m' }, { key: 'groundSpeed', header: us ? 'groundSpeed_mph' : 'groundSpeed_ms' }, { key: 'heading' }, { key: 'crossTrackError', header: us ? 'crossTrackError_ft' : 'crossTrackError_m' }, { key: 'lockedLine' }, { key: 'hdop' }, { key: 'satsInView' }, { key: 'correctionId' }, { key: 'waasId' }, { key: 'sprayStat' }, // Application data { key: 'flowRateApplied', header: us ? 'flowRateApplied_galMin' : 'flowRateApplied_Lmin' }, { key: 'flowRateRequired', header: us ? 'flowRateRequired_galMin' : 'flowRateRequired_Lmin' }, { key: 'appRateRequired', header: us ? 'appRateRequired_galAc' : 'appRateRequired_Lha' }, { key: 'appRateApplied', header: us ? 'appRateApplied_galAc' : 'appRateApplied_Lha' }, { key: 'swathWidth', header: us ? 'swathWidth_ft' : 'swathWidth_m' }, { key: 'boomPressure_psi' }, // PSI already; no conversion { key: 'sprayOnLag_s' }, { key: 'sprayOffLag_s' }, { key: 'pulsesPerLitre' }, // MET { key: 'windSpeed', header: us ? 'windSpeed_mph' : 'windSpeed_ms' }, { key: 'windDir_deg' }, { key: 'temp', header: us ? 'temp_f' : 'temp_c' }, { key: 'humidity_pct' }, ]; } function escapeCsv(val) { if (val == null) return ''; const s = String(val); if (s.includes(',') || s.includes('"') || s.includes('\n')) return `"${s.replace(/"/g, '""')}"`; return s; } function recordToRow(d, sessionMeta, jobHeader, units) { const us = units === ExportUnits.US; const { correctionId, waasId } = decodeCorrectionFields(d.tslu, d.calcodeFreq); const appRateApplied = computeAppRateApplied(d.lminApp, d.grSpeed, d.swath); const row = { ...jobHeader, sessionId: sessionMeta.appId, fileName: sessionMeta.fileName, pilotName: sessionMeta.operator ?? '', timestampUtc: d.gpsTime ? moment.unix(d.gpsTime).utc().toISOString() : '', gpsTime: d.gpsTime ?? '', lat: d.lat ?? '', lon: d.lon ?? '', utmX: d.utmX ?? '', utmY: d.utmY ?? '', alt: us ? applyConv(d.alt, CONV.mToFt) : (d.alt ?? ''), groundSpeed: us ? applyConv(d.grSpeed, CONV.msToMph) : (d.grSpeed ?? ''), heading: d.head ?? '', crossTrackError: us ? applyConv(d.xTrack, CONV.mToFt) : (d.xTrack ?? ''), lockedLine: d.llnum ?? '', hdop: d.stdHdop ?? '', satsInView: decodeSatsIn(d.satsIn) ?? '', correctionId: correctionId ?? '', waasId: waasId ?? '', sprayStat: d.sprayStat ?? '', flowRateApplied: us ? applyConv(d.lminApp, CONV.LminToGmin) : (d.lminApp ?? ''), flowRateRequired: us ? applyConv(d.lminReq, CONV.LminToGmin) : (d.lminReq ?? ''), appRateRequired: us ? applyConv(d.lhaReq, CONV.LhaToGac) : (d.lhaReq ?? ''), appRateApplied: us ? applyConv(appRateApplied, CONV.LhaToGac) : (appRateApplied ?? ''), swathWidth: us ? applyConv(d.swath, CONV.mToFt) : (d.swath ?? ''), boomPressure_psi: d.psi ?? '', sprayOnLag_s: sessionMeta.meta?.sprOnLag ?? '', sprayOffLag_s: sessionMeta.meta?.sprOffLag ?? '', pulsesPerLitre: sessionMeta.meta?.pulsesPerLit ?? '', windSpeed: us ? applyConv(d.windSpd, CONV.msToMph) : (d.windSpd ?? ''), windDir_deg: d.windDir ?? '', temp: us ? applyConv(d.temp, CONV.cToF) : (d.temp ?? ''), humidity_pct: d.humid ?? '' }; const cols = getCsvColumns(units); return cols.map(c => escapeCsv(row[c.key])).join(',') + '\n'; } // ─── Async generation ───────────────────────────────────────────────────────── async function generateExport(exportJobId) { const exportJob = await ExportJob.findById(exportJobId); if (!exportJob) return; try { exportJob.status = 'processing'; await exportJob.save(); const job = await Job.findById(exportJob.jobId, 'name orderNumber').lean(); const jobHeader = { jobId: exportJob.jobId, orderNumber: job?.orderNumber ?? '', jobName: job?.name ?? '' }; const apps = await App.find({ jobId: exportJob.jobId, markedDelete: { $ne: true } }).lean(); const appFiles = await AppFile.find( { appId: { $in: apps.map(a => a._id) }, markedDelete: { $ne: true } } ).lean(); const filesByAppId = {}; for (const f of appFiles) { const key = f.appId.toString(); if (!filesByAppId[key]) filesByAppId[key] = []; filesByAppId[key].push(f); } const interval = exportJob.interval; const outPath = path.join(env.TEMP_DIR, `export_${exportJobId}.${exportJob.format}`); const writeStream = fs.createWriteStream(outPath); const units = exportJob.units || 'metric'; if (exportJob.format === 'csv') { // Write header row (unit-aware column names) const cols = getCsvColumns(units); writeStream.write(cols.map(c => c.header || c.key).join(',') + '\n'); for (const app of apps) { const files = filesByAppId[app._id.toString()] || []; for (const appFile of files) { const sessionMeta = { appId: app._id, fileName: app.fileName, operator: appFile.meta?.operator, meta: appFile.meta }; // Stream AppDetail records for this file using a cursor (memory-efficient). // Exclude sprayStat=3 (spray segment START marker — stores anchor position for // the next area calculation; not an application-data record for consumers). const cursor = AppDetail.find( { fileId: appFile._id, sprayStat: { $ne: 3 } }, null, { sort: { _id: 1 }, lean: true } ).cursor(); let prevGpsTime = null; for await (const record of cursor) { if (interval) { if (prevGpsTime !== null && (record.gpsTime - prevGpsTime) < interval) continue; prevGpsTime = record.gpsTime; } writeStream.write(recordToRow(record, sessionMeta, jobHeader, units)); } } } } else if (exportJob.format === 'geojson') { // GeoJSON FeatureCollection — one Feature per GPS point. // sprayStat=3 excluded: it is a spray segment START marker, not application data. writeStream.write('{"type":"FeatureCollection","features":[\n'); let first = true; for (const app of apps) { const files = filesByAppId[app._id.toString()] || []; for (const appFile of files) { const cursor = AppDetail.find( { fileId: appFile._id, sprayStat: { $ne: 3 } }, null, { sort: { _id: 1 }, lean: true } ).cursor(); let prevGpsTime = null; for await (const d of cursor) { if (interval) { if (prevGpsTime !== null && (d.gpsTime - prevGpsTime) < interval) continue; prevGpsTime = d.gpsTime; } if (!utils.isNumber(d.lon) || !utils.isNumber(d.lat)) continue; const feature = { type: 'Feature', geometry: { type: 'Point', coordinates: [d.lon, d.lat, d.alt ?? 0] }, properties: { jobId: exportJob.jobId, sessionId: String(app._id), fileName: app.fileName, timestampUtc: d.gpsTime ? moment.unix(d.gpsTime).utc().toISOString() : null, sprayStat: d.sprayStat, groundSpeed: d.grSpeed } }; writeStream.write((first ? '' : ',\n') + JSON.stringify(feature)); first = false; } } } writeStream.write('\n]}'); } await new Promise((resolve, reject) => { writeStream.end(); writeStream.on('finish', resolve); writeStream.on('error', reject); }); const expiresAt = new Date(Date.now() + EXPORT_TTL_HOURS * 3600 * 1000); exportJob.status = 'ready'; exportJob.filePath = outPath; exportJob.expiresAt = expiresAt; await exportJob.save(); } catch (err) { exportJob.status = 'error'; exportJob.errorMsg = err.message; await exportJob.save(); console.error('[export] generation failed', err); } } // ─── Route handlers ─────────────────────────────────────────────────────────── /** * POST /api/v1/jobs/:jobId/export * Body: { format: 'csv' | 'geojson', interval?: number } */ async function triggerExport(req, res) { const jobId = parseInt(req.params.jobId, 10); if (!isFinite(jobId)) AppParamError.throw('invalid jobId'); await ownerJob(jobId, req.uid); const format = req.body?.format; if (!['csv', 'geojson'].includes(format)) { return res.status(HttpStatus.BAD_REQUEST).json({ error: 'format must be csv or geojson' }); } const interval = parseInterval(req.body?.interval); const rawUnits = req.body?.units; const units = rawUnits === ExportUnits.US ? ExportUnits.US : ExportUnits.METRIC; // default metric const exportJob = await ExportJob.create({ owner: ObjectId(req.uid), jobId, format, interval, units, status: 'pending' }); // Kick off async generation — do not await setImmediate(() => generateExport(exportJob._id)); res.status(HttpStatus.CREATED).json({ exportId: exportJob._id, status: exportJob.status, format: exportJob.format, units: exportJob.units, createdAt: exportJob.createdAt }); } /** * GET /api/v1/exports/:exportId * Poll for export status. When ready, includes downloadUrl. */ async function getExportStatus(req, res) { const exportId = req.params.exportId; if (!ObjectId.isValid(exportId)) AppParamError.throw('invalid exportId'); const exportJob = await ExportJob.findOne({ _id: ObjectId(exportId), owner: ObjectId(req.uid) }).lean(); if (!exportJob) return res.status(HttpStatus.NOT_FOUND).json({ error: Errors.NOT_FOUND }); const payload = { exportId: exportJob._id, status: exportJob.status, format: exportJob.format, units: exportJob.units, createdAt: exportJob.createdAt, expiresAt: exportJob.expiresAt ?? null, error: exportJob.errorMsg ?? null }; if (exportJob.status === 'ready') { // Provide a download URL — the frontend calls this to stream the file payload.downloadUrl = `/api/v1/exports/${exportId}/download`; } res.json(payload); } /** * GET /api/v1/exports/:exportId/download * Streams the generated export file. Schedules file deletion after streaming. */ async function downloadExport(req, res) { const exportId = req.params.exportId; if (!ObjectId.isValid(exportId)) AppParamError.throw('invalid exportId'); const exportJob = await ExportJob.findOne({ _id: ObjectId(exportId), owner: ObjectId(req.uid), status: 'ready' }).lean(); if (!exportJob || !exportJob.filePath) { return res.status(HttpStatus.NOT_FOUND).json({ error: Errors.NOT_FOUND }); } const ext = exportJob.format === 'geojson' ? 'geojson' : 'csv'; const contentType = exportJob.format === 'geojson' ? 'application/geo+json' : 'text/csv'; const filename = `export_job${exportJob.jobId}_${exportJob._id}.${ext}`; res.setHeader('Content-Type', contentType); res.setHeader('Content-Disposition', `attachment; filename="${filename}"`); const readStream = fs.createReadStream(exportJob.filePath); readStream.pipe(res); readStream.on('end', () => { // Clean up file after streaming (fire-and-forget) fs.unlink(exportJob.filePath, () => {}); ExportJob.updateOne({ _id: exportJob._id }, { $set: { status: 'pending', filePath: null } }).catch(() => {}); }); readStream.on('error', (err) => { console.error('[export] stream error', err); res.end(); }); } module.exports = { triggerExport, getExportStatus, downloadExport };