agmission/server/scripts/migrate_app_aggregates.js

/* DEPRECATED — use scripts/migrate_applications.js instead.
 * This script is kept for reference only.
 */
'use strict';

/**
 * Application Aggregates Migration Script
 *
 * Backfills BOTH avgSpraySpeed AND totalSprLength/totalFlightLength on Application
 * (and totalSprLength/totalFlightLength on AppFile)
 * in a SINGLE traversal per application — no double-scanning.
 *
 * Fields updated:
 *   Application.avgSpraySpeed  — weighted avg ground speed (m/s) across all spray-on AppDetail records
 *   Application.totalSprLength   — cumulative geodesic spray distance (m) across all AppFile records
 *   Application.totalFlightLength — cumulative flight distance (m) across all AppFile records
 *   Application.avgXtError     — weighted avg abs cross-track error (m) across spray-on records
 *   Application.avgHdop        — weighted avg HDOP across spray-on records (lower = better)
 *   Application.flowAccuracyPct — (totalSprayMat/totalSprayed / appRate) × 100; backfilled in a second pass
 *   AppFile.totalSprLength       — per-file spray distance (m)
 *   AppFile.totalFlightLength    — per-file flight distance (m)
 *
 * Performance characteristics:
 *   - Applications processed most-recent-first (ObjectId descending = newest first)
 *   - AppDetail streamed via Mongoose cursor (never materialises all records for a file in memory)
 *   - Bulk writes to Application and AppFile flushed every --batch-size apps
 *   - Haversine (pure JS) replaces turf point objects in the inner loop — ~5× faster per record
 *   - `.lean()` on every Mongoose query
 *   - Tiered execution: --tier-days or --from-date limits the ObjectId range so you can target
 *     the most recent data first, then re-run without the flag for a full backfill
 *
 * Distance algorithm (matches readSatLogAsc in job_worker.js, corrected prevLonLat bug):
 *   For consecutive AppDetail rows sorted by gpsTime ASC:
 *     if ( (prevStat > 0 && curStat > 0) || curStat != prevStat )
 *       d = haversine(prevLon, prevLat, curLon, curLat)
 *       if d <= 1000m → totalSprLength += d
 *
 * Usage:
 *   node scripts/migrate_app_aggregates.js [options]
 *
 *   # Tier 1 — most recent 90 days first (fastest to see results)
 *   DEBUG=agm:migrate-app-aggregates node scripts/migrate_app_aggregates.js --tier-days=90
 *
 *   # Tier 2 — everything from 2023 onward
 *   DEBUG=agm:migrate-app-aggregates node scripts/migrate_app_aggregates.js --from-date=2023-01-01
 *
 *   # Full backfill (all-time)
 *   DEBUG=agm:migrate-app-aggregates node scripts/migrate_app_aggregates.js
 *
 *   # Dry-run (logs what would change, no writes)
 *   DEBUG=agm:migrate-app-aggregates node scripts/migrate_app_aggregates.js --dry-run --tier-days=7
 *
 *   # Recompute already-set fields (e.g. after algorithm fix)
 *   DEBUG=agm:migrate-app-aggregates node scripts/migrate_app_aggregates.js --force --tier-days=30
 *
 *   # Custom env file
 *   node scripts/migrate_app_aggregates.js --env ./environment_prod.env --tier-days=90
 *
 * Options:
 *   --env <path>        Environment file (default: ./environment.env)
 *   --dry-run           Report only; make no DB writes
 *   --force             Recompute even when both fields are already set
 *   --batch-size=N      Applications per bulkWrite flush (default: 50)
 *   --tier-days=N       Only process apps created in the last N days
 *   --from-date=YYYY-MM-DD  Only process apps created on/after this date (UTC)
 *   --concurrency=N     Parallel file-level cursors per application (default: 3)
 */

// ─── Environment bootstrap ────────────────────────────────────────────────────
const path = require('path');
const args = process.argv.slice(2);

let envFile = './environment.env';
for (let i = 0; i < args.length; i++) {
  if (args[i] === '--env' && args[i + 1]) { envFile = args[i + 1]; i++; }
  else if (args[i].startsWith('--env=')) { envFile = args[i].split('=')[1]; }
}
require('dotenv').config({ path: path.resolve(process.cwd(), envFile) });

// ─── Imports (after env is loaded) ───────────────────────────────────────────
const debug = require('debug')('agm:migrate-app-aggregates');
const mongoose = require('mongoose');
const { DBConnection } = require('../helpers/db/connect');
const Application = require('../model/application');
const AppFile    = require('../model/application_file');
const AppDetail  = require('../model/application_detail');

// ─── Argument parsing ─────────────────────────────────────────────────────────
const cfg = {
  dryRun:      false,
  force:       false,
  batchSize:   50,
  tierDays:    null,   // Number – only apps newer than N days
  fromDate:    null,   // ISO string – only apps on/after this date
  concurrency: 3,      // parallel file cursors per app
};

for (let i = 0; i < args.length; i++) {
  const a = args[i];
  if (a === '--dry-run')                    cfg.dryRun      = true;
  else if (a === '--force')                 cfg.force       = true;
  else if (a.startsWith('--batch-size='))   cfg.batchSize   = parseInt(a.split('=')[1], 10) || 50;
  else if (a.startsWith('--tier-days='))    cfg.tierDays    = parseInt(a.split('=')[1], 10) || null;
  else if (a.startsWith('--from-date='))    cfg.fromDate    = a.split('=')[1] || null;
  else if (a.startsWith('--concurrency='))  cfg.concurrency = parseInt(a.split('=')[1], 10) || 3;
  // --env already consumed above
}

// ─── Constants ────────────────────────────────────────────────────────────────
const MAX_SEGMENT_METERS = 1000;  // Sanity cap matching job_worker readSatLogAsc
const PROGRESS_LOG_INTERVAL = 100; // Log a line every N apps

// ─── Helpers ──────────────────────────────────────────────────────────────────

/**
 * Fast Haversine distance in metres between two WGS-84 points.
 * Avoids turf object allocation in tight loops.
 */
function haversineMeters(lon1, lat1, lon2, lat2) {
  const R  = 6371000;
  const φ1 = lat1 * Math.PI / 180;
  const φ2 = lat2 * Math.PI / 180;
  const Δφ = (lat2 - lat1) * Math.PI / 180;
  const Δλ = (lon2 - lon1) * Math.PI / 180;
  const a  = Math.sin(Δφ / 2) ** 2 + Math.cos(φ1) * Math.cos(φ2) * Math.sin(Δλ / 2) ** 2;
  return R * 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a));
}

/**
 * Create a minimal ObjectId whose timestamp equals `date`.
 */
function objectIdFromDate(date) {
  const ts  = Math.floor(new Date(date).getTime() / 1000);
  const hex = ts.toString(16).padStart(8, '0') + '0000000000000000';
  return new mongoose.Types.ObjectId(hex);
}

/**
 * Process all AppDetail records for one AppFile in a single streaming pass.
 * Returns { sprLength, flightLength, speedSum, speedCount, xtSum, xtCount, hdopSum, hdopCount }.
 *
 * Validity gates (matching job_worker conventions):
 *   sprLength    — spray-ON segment (prevStat > 0 OR curStat > 0) AND distance ≤ 1000 m AND time gap ≤ 120 s
 *   flightLength — distance ≤ 1000 m AND time gap ≤ 120 s (valid GPS segment, all movements)
 *   avgXtError   — spray-on (sprayStat 1 or 3) records within valid (both-gate) segments, xTrack ≠ 0
 *   avgHdop      — spray-on (sprayStat > 0) records within valid (both-gate) segments, stdHdop > 0
 */
async function processFile(fileId) {
  let sprLength    = 0;
  let flightLength = 0;
  let speedSum     = 0;
  let speedCount   = 0;
  let xtSum        = 0;
  let xtCount      = 0;
  let hdopSum      = 0;
  let hdopCount    = 0;

  let prevLon     = null;
  let prevLat     = null;
  let prevGpsTime = null;
  let prevStat    = -999;

  // Sort by gpsTime for correct consecutive-point distance calculation.
  // Tie-break by _id for determinism when gpsTime == 0 (legacy).
  const cursor = AppDetail
    .find({ fileId })
    .select('lat lon gpsTime sprayStat grSpeed xTrack stdHdop')
    .sort({ gpsTime: 1, _id: 1 })
    .lean()
    .cursor();

  for await (const rec of cursor) {
    const curLon     = rec.lon;
    const curLat     = rec.lat;
    const curGpsTime = rec.gpsTime;
    const curStat    = rec.sprayStat || 0;

    if (prevStat !== -999 &&
        (typeof curLon === 'number') && (typeof curLat === 'number') &&
        (typeof prevLon === 'number') && (typeof prevLat === 'number')) {

      const d = haversineMeters(prevLon, prevLat, curLon, curLat);

      // Time gap with midnight-rollover handling (gpsTime is seconds-of-day)
      let timeDif = (typeof curGpsTime === 'number' && typeof prevGpsTime === 'number')
        ? curGpsTime - prevGpsTime
        : Infinity;
      if (timeDif < 0 && Math.abs(timeDif) >= 80000) timeDif = 86400 - prevGpsTime + curGpsTime;

      // ── Valid segment: both distance AND time gate ───────────────────────────
      const segValid = d <= MAX_SEGMENT_METERS && timeDif > 0 && timeDif <= 120;

      // ── Spray length: spray-ON segment within valid (both-gate) window ────────
      // Matches job_worker _computeSprLength which applies both distance AND time gates.
      if (segValid && (prevStat > 0 || curStat > 0)) {
        sprLength += d;
      }

      // ── Flight length: all GPS movements within valid segments ───────────────
      if (segValid) {
        flightLength += d;
      }

      // ── XT error: spray-on records (stat 1 = on-swath, 3 = swath entry) ──────
      // Only within valid segments; exclude zero (no reading) and non-numeric values.
      if (segValid && (curStat === 1 || curStat === 3) &&
          typeof rec.xTrack === 'number' && rec.xTrack !== 0) {
        xtSum   += Math.abs(rec.xTrack);
        xtCount += 1;
      }

      // ── HDOP: spray-on records within valid segments, stdHdop > 0 ────────────
      if (segValid && curStat > 0 && typeof rec.stdHdop === 'number' && rec.stdHdop > 0) {
        hdopSum   += rec.stdHdop;
        hdopCount += 1;
      }
    }

    // ── Speed accumulation (all spray-on, no segment gate needed) ────────────
    if (curStat > 0 && typeof rec.grSpeed === 'number' && rec.grSpeed > 0) {
      speedSum   += rec.grSpeed;
      speedCount += 1;
    }

    prevLon     = curLon;
    prevLat     = curLat;
    prevGpsTime = curGpsTime;
    prevStat    = curStat;
  }

  return { sprLength, flightLength, speedSum, speedCount, xtSum, xtCount, hdopSum, hdopCount };
}

/**
 * Process all files for one Application with bounded parallelism.
 * Returns:
 *   appSprLength      – total spray length across all files (m)
 *   appFlightLength   – total flight length across all files (m)
 *   avgSpraySpeed     – weighted average ground speed across all files (m/s, null if no data)
 *   avgXtError        – weighted average abs cross-track error (m, null if no data)
 *   avgHdop           – weighted average HDOP over spray-on records (null if no data)
 *   fileOps           – array of { fileId, sprLength, flightLength } for per-file bulkWrite
 */
async function processApplication(app, appFiles) {
  if (!appFiles.length) {
    return { appSprLength: 0, appFlightLength: 0, avgSpraySpeed: null, avgXtError: null, avgHdop: null, fileOps: [] };
  }

  // Process files in small parallel batches (bounded concurrency)
  const fileResults = [];
  for (let i = 0; i < appFiles.length; i += cfg.concurrency) {
    const slice   = appFiles.slice(i, i + cfg.concurrency);
    const results = await Promise.all(slice.map(f => processFile(f._id)));
    for (let j = 0; j < slice.length; j++) {
      fileResults.push({ file: slice[j], result: results[j] });
    }
  }

  let appSprLength    = 0;
  let appFlightLength = 0;
  let totalSpeedSum   = 0;
  let totalSpeedCnt   = 0;
  let totalXtSum      = 0;
  let totalXtCnt      = 0;
  let totalHdopSum    = 0;
  let totalHdopCnt    = 0;
  const fileOps       = [];

  for (const { file, result } of fileResults) {
    appSprLength    += result.sprLength;
    appFlightLength += result.flightLength;
    totalSpeedSum   += result.speedSum;
    totalSpeedCnt   += result.speedCount;
    totalXtSum      += result.xtSum;
    totalXtCnt      += result.xtCount;
    totalHdopSum    += result.hdopSum;
    totalHdopCnt    += result.hdopCount;

    fileOps.push({ fileId: file._id, sprLength: result.sprLength, flightLength: result.flightLength });
  }

  const avgSpraySpeed = totalSpeedCnt > 0 ? totalSpeedSum / totalSpeedCnt : null;
  const avgXtError    = totalXtCnt > 0 ? totalXtSum / totalXtCnt : null;
  const avgHdop       = totalHdopCnt > 0 ? totalHdopSum / totalHdopCnt : null;

  return { appSprLength, appFlightLength, avgSpraySpeed, avgXtError, avgHdop, fileOps };
}

// ─── Flow accuracy backfill (Application-level only) ─────────────────────────

/**
 * Backfill Application.flowAccuracyPct for documents that already have
 * totalSprayMat > 0, totalSprayed > 0, and appRate > 0 but no flowAccuracyPct.
 * No AppDetail queries needed — all three source fields live on Application.
 */
async function backfillFlowAccuracy(dryRun, tierDays, fromDate) {
  debug('─'.repeat(60));
  debug('Second pass: backfill flowAccuracyPct …');

  const filter = {
    markedDelete: { $ne: true },
    flowAccuracyPct: { $exists: false },
    totalSprayed:  { $gt: 0 },
    totalSprayMat: { $gt: 0 },
    appRate:       { $gt: 0 },
  };

  if (tierDays) {
    const cutoff = new Date(Date.now() - tierDays * 86400 * 1000);
    filter._id = { $gte: objectIdFromDate(cutoff) };
  } else if (fromDate) {
    filter._id = { $gte: objectIdFromDate(fromDate + 'T00:00:00Z') };
  }

  debug('flowAccuracy filter: %o', filter);

  if (dryRun) {
    const count = await Application.countDocuments(filter);
    debug(`[dry-run] Would update ${count} Application docs with flowAccuracyPct`);
    return;
  }

  // MongoDB 4.2+ aggregation pipeline update — computes the field server-side
  const result = await Application.updateMany(filter, [
    {
      $set: {
        flowAccuracyPct: {
          $round: [
            { $multiply: [{ $divide: [{ $divide: ['$totalSprayMat', '$totalSprayed'] }, '$appRate'] }, 100] },
            2]
        }
      }
    }
  ]);

  debug(`flowAccuracyPct backfill: matched=${result.matchedCount} modified=${result.modifiedCount}`);
  debug('─'.repeat(60));
}

// ─── Main migration ───────────────────────────────────────────────────────────
async function migrate() {
  // Build Application filter
  const appFilter = { markedDelete: { $ne: true } };

  // Tier by recency
  if (cfg.tierDays) {
    const cutoff  = new Date(Date.now() - cfg.tierDays * 86400 * 1000);
    appFilter._id = { $gte: objectIdFromDate(cutoff) };
    debug(`Tier: apps from last ${cfg.tierDays} days (>= ${cutoff.toISOString()})`);
  } else if (cfg.fromDate) {
    appFilter._id = { $gte: objectIdFromDate(cfg.fromDate + 'T00:00:00Z') };
    debug(`Tier: apps from ${cfg.fromDate} onward`);
  }

  // Unless --force, revisit apps with missing/null fields and explicit zero values that may
  // have been written by earlier buggy imports.
  if (!cfg.force) {
    appFilter.$or = [
      { avgSpraySpeed:     { $exists: false } },
      { avgSpraySpeed:     null },
      { totalSprLength:    { $exists: false } },
      { totalSprLength:    null },
      { totalSprLength:    0 },
      { totalFlightLength: { $exists: false } },
      { totalFlightLength: null },
      { totalFlightLength: 0 },
      { avgXtError: { $exists: false } },
      { avgXtError: null },
      { avgHdop: { $exists: false } },
      { avgHdop: null },
    ];
  }

  debug('Config: %o', { ...cfg, envFile });
  debug('App filter: %o', appFilter);

  // Stats
  const stats = {
    examined:  0,
    updated:   0,
    skipped:   0,  // no AppFile / no AppDetail data
    errors:    0,
    startedAt: Date.now(),
  };

  // Pending bulk-write buffers
  let appBulk  = [];  // Application updateOne ops
  let fileBulk = [];  // AppFile updateOne ops

  async function flushBulk() {
    if (cfg.dryRun) {
      debug(`[dry-run] Would write ${appBulk.length} Application + ${fileBulk.length} AppFile ops`);
      appBulk  = [];
      fileBulk = [];
      return;
    }

    const writes = [];
    if (appBulk.length)  writes.push(Application.bulkWrite(appBulk,  { ordered: false }));
    if (fileBulk.length) writes.push(AppFile.bulkWrite(fileBulk,     { ordered: false }));
    await Promise.all(writes);

    appBulk  = [];
    fileBulk = [];
  }

  // Stream Applications — newest first
  const appCursor = Application
    .find(appFilter)
    .select('_id avgSpraySpeed totalSprLength')
    .sort({ _id: -1 })
    .lean()
    .cursor();

  for await (const app of appCursor) {
    stats.examined++;

    try {
      // Load AppFile docs for this application (not deleted)
      const appFiles = await AppFile
        .find({ appId: app._id, markedDelete: { $ne: true } })
        .select('_id')
        .lean();

      if (!appFiles.length) {
        stats.skipped++;
      } else {
        const { appSprLength, appFlightLength, avgSpraySpeed, avgXtError, avgHdop, fileOps } = await processApplication(app, appFiles);

        // Queue Application update
        const $set = {
          totalSprLength:    appSprLength,
          totalFlightLength: appFlightLength,
        };
        if (avgSpraySpeed !== null) $set.avgSpraySpeed = avgSpraySpeed;
        if (avgXtError !== null) $set.avgXtError = avgXtError;
        if (avgHdop !== null) $set.avgHdop = avgHdop;

        appBulk.push({
          updateOne: {
            filter: { _id: app._id },
            update: { $set },
          },
        });

        // Queue AppFile updates
        for (const op of fileOps) {
          fileBulk.push({
            updateOne: {
              filter: { _id: op.fileId },
              update: { $set: { totalSprLength: op.sprLength, totalFlightLength: op.flightLength } },
            },
          });
        }

        stats.updated++;
      }
    } catch (err) {
      debug(`Error processing app ${app._id}: ${err.message}`);
      stats.errors++;
    }

    // Flush when batch is full
    if (appBulk.length >= cfg.batchSize) {
      await flushBulk();
    }

    // Progress log
    if (stats.examined % PROGRESS_LOG_INTERVAL === 0) {
      const elapsed = ((Date.now() - stats.startedAt) / 1000).toFixed(1);
      const rate    = (stats.examined / parseFloat(elapsed)).toFixed(1);
      debug(
        `Progress: examined=${stats.examined} updated=${stats.updated} ` +
        `skipped=${stats.skipped} errors=${stats.errors} ` +
        `elapsed=${elapsed}s rate=${rate} apps/s`
      );
    }
  }

  // Final flush
  await flushBulk();

  const elapsed = ((Date.now() - stats.startedAt) / 1000).toFixed(1);
  debug('─'.repeat(60));
  debug(`Migration complete.`);
  debug(`  Examined : ${stats.examined}`);
  debug(`  Updated  : ${stats.updated}`);
  debug(`  Skipped  : ${stats.skipped}  (no files or no detail data)`);
  debug(`  Errors   : ${stats.errors}`);
  debug(`  Duration : ${elapsed}s`);
  if (!cfg.dryRun && stats.updated > 0) {
    debug(`  Fields set: Application.avgSpraySpeed, Application.totalSprLength, Application.totalFlightLength, Application.avgXtError, Application.avgHdop`);
    debug(`              AppFile.totalSprLength, AppFile.totalFlightLength`);
  }
  debug('─'.repeat(60));

  // ── Second pass: backfill flowAccuracyPct (Application-level only, no AppDetail needed) ──
  await backfillFlowAccuracy(cfg.dryRun, cfg.tierDays, cfg.fromDate);

  return stats;
}

// ─── Entry point ──────────────────────────────────────────────────────────────
process
  .on('uncaughtException',    err => { debug('Uncaught:', err); process.exit(1); })
  .on('unhandledRejection',   err => { debug('Unhandled rejection:', err); process.exit(1); });

async function main() {
  const dbConn = new DBConnection('migrate-app-aggregates');
  try {
    await dbConn.initialize({ setupExitHandlers: false });
    debug('DB connected');
    await migrate();
  } catch (err) {
    debug('Fatal error:', err);
  } finally {
    await dbConn.close();
    process.exit(0);
  }
}

main();