Build a Nightly Data Sync

async function runNightlySync(customerId) {
  const runId = generateRunId();
  console.log(`Starting nightly sync run ${runId} for customer ${customerId}`);

  try {
    // Phase 1: load checkpoint and confirm prerequisites
    const checkpoint = await loadCheckpoint(customerId);
    const { sourceToken, virtuousToken } = await loadCredentials(customerId);

    // Phase 2: read changes from source
    const sourceChanges = await readSourceChanges(sourceToken, checkpoint.lastSourceTimestamp);
    console.log(`Found ${sourceChanges.length} source changes`);

    // Phase 3: apply changes to Virtuous (with throttling)
    const results = await applyChangesToVirtuous(virtuousToken, sourceChanges);

    // Phase 4: write checkpoint and emit run report
    await persistCheckpoint(customerId, {
      runId,
      completedAt: new Date(),
      lastSourceTimestamp: results.highestSourceTimestamp,
      successCount: results.successes,
      failureCount: results.failures,
    });

    await emitRunReport(customerId, runId, results);
  } catch (err) {
    console.error(`Nightly sync run ${runId} failed:`, err);
    await emitFailureReport(customerId, runId, err);
    throw err;
  }
}

async function readSourceChanges(sourceToken, sinceTimestamp) {
  const changes = [];
  let cursor = null;

  do {
    const url = buildPageUrl(sinceTimestamp, cursor);
    const response = await fetch(url, { headers: { Authorization: `Bearer ${sourceToken}` } });
    const page = await response.json();
    changes.push(...page.records);
    cursor = page.nextCursor;
  } while (cursor);

  return changes;
}

async function confirmPrerequisites(customerId) {
  // 1. Source credentials valid
  const sourceToken = await loadSourceToken(customerId);
  if (!sourceToken || isExpired(sourceToken)) {
    throw new Error('Source credentials missing or expired');
  }

  // 2. Virtuous credentials valid
  const virtuousToken = await loadVirtuousToken(customerId);
  if (!virtuousToken || isExpired(virtuousToken)) {
    throw new Error('Virtuous credentials missing or expired');
  }

  // 3. Circuit breaker check
  const checkpoint = await loadCheckpoint(customerId);
  if (checkpoint.paused_until && new Date(checkpoint.paused_until) > new Date()) {
    throw new Error(`Sync paused until ${checkpoint.paused_until}`);
  }

  // 4. Virtuous reachable
  const test = await fetch('https://api.virtuoussoftware.com/api/Health', {
    headers: { Authorization: `Bearer ${virtuousToken}` },
  });
  if (!test.ok) {
    throw new Error('Virtuous API not reachable');
  }
}

async function readChangesViaFullSnapshot(customerId) {
  const sourceToken = await loadSourceToken(customerId);
  const currentSnapshot = await fetchFullSourceSnapshot(sourceToken);
  const previousSnapshot = await loadPreviousSnapshot(customerId);

  const changes = [];
  const previousById = new Map(previousSnapshot.map((r) => [r.id, r]));

  for (const current of currentSnapshot) {
    const previous = previousById.get(current.id);
    if (!previous || diffRecord(current, previous)) {
      changes.push({ type: previous ? 'update' : 'create', record: current });
    }
    previousById.delete(current.id);
  }

  // Anything left in previousById is a deletion
  for (const deleted of previousById.values()) {
    changes.push({ type: 'delete', record: deleted });
  }

  // Save the current snapshot for next run
  await persistSnapshot(customerId, currentSnapshot);

  return changes;
}

async function applyChangesWithThrottling(virtuousToken, changes) {
  const REQUESTS_PER_HOUR = 1200;             // Conservative — leave 20% headroom
  const MS_BETWEEN_REQUESTS = (60 * 60 * 1000) / REQUESTS_PER_HOUR;

  let lastRequestAt = 0;
  const results = { successes: 0, failures: 0, failureDetails: [] };

  for (const change of changes) {
    // Pace the requests
    const elapsed = Date.now() - lastRequestAt;
    if (elapsed < MS_BETWEEN_REQUESTS) {
      await sleep(MS_BETWEEN_REQUESTS - elapsed);
    }
    lastRequestAt = Date.now();

    try {
      await applyChange(virtuousToken, change);
      results.successes++;
    } catch (err) {
      if (isRetryable(err)) {
        // Re-queue for the next run
        await persistDeferredChange(change, err);
      } else {
        results.failures++;
        results.failureDetails.push({ change, error: err.message });
      }
    }
  }

  return results;
}

async function persistCheckpoint(customerId, runMetadata) {
  await db.nightly_sync_checkpoints.upsert({
    customer_id: customerId,
    last_run_id: runMetadata.runId,
    last_run_completed_at: runMetadata.completedAt,
    last_source_timestamp: runMetadata.lastSourceTimestamp,
    consecutive_failures: 0,                 // reset on success
  });
}

async function emitRunReport(customerId, runId, results) {
  const report = {
    customer_id: customerId,
    run_id: runId,
    completed_at: new Date(),
    total_changes: results.successes + results.failures,
    successes: results.successes,
    failures: results.failures,
    duration_seconds: results.durationSeconds,
    rate_limit_pauses: results.rateLimitPauseCount,
    failure_details: results.failureDetails.slice(0, 50), // truncate for log readability
  };

  await db.nightly_sync_runs.insert(report);

  if (results.failures > results.successes * 0.05) {
    // Failure rate above 5% — alert
    await alertOps(`Nightly sync for ${customerId} had ${results.failures} failures`);
  }
}

async function applyChangesResumable(virtuousToken, changes, runId) {
  // Check if a previous attempt for this run exists
  const prevProgress = await db.partial_run_progress.find({ run_id: runId });
  const startIndex = prevProgress?.last_completed_index ?? 0;

  for (let i = startIndex; i < changes.length; i++) {
    await applyChange(virtuousToken, changes[i]);

    // Persist progress every N records
    if (i % 100 === 0) {
      await db.partial_run_progress.upsert({
        run_id: runId,
        last_completed_index: i,
        last_persisted_at: new Date(),
      });
    }
  }

  // Clean up partial-progress record after successful completion
  await db.partial_run_progress.delete({ run_id: runId });
}

async function applyCircuitBreaker(customerId, runFailed) {
  const checkpoint = await loadCheckpoint(customerId);

  if (runFailed) {
    const newFailureCount = checkpoint.consecutive_failures + 1;

    if (newFailureCount >= 3) {
      // Pause for 24 hours
      const pauseUntil = new Date(Date.now() + 24 * 60 * 60 * 1000);
      await db.nightly_sync_checkpoints.update({ customer_id: customerId }, {
        consecutive_failures: newFailureCount,
        paused_until: pauseUntil,
      });
      await alertOps(`Sync for ${customerId} paused after ${newFailureCount} failures`);
    } else {
      await db.nightly_sync_checkpoints.update({ customer_id: customerId }, {
        consecutive_failures: newFailureCount,
      });
    }
  } else {
    // Reset on success
    await db.nightly_sync_checkpoints.update({ customer_id: customerId }, {
      consecutive_failures: 0,
      paused_until: null,
    });
  }
}