Iterator based Generator to do left-join + aggregate behavior against two typescript/javascript iterables.

Iterator-Based-Left-Join-Aggregate-Metric.ts

// This generator is used when you have an two iterables (array,etc.) that you want to zip together so the first set of timed-metrics is 
// associated with the seconds listed in the second. It will aggregate all the points using the function passed and works as a generator
// so no new collection is allocated, rather the aggregated (group by second) values are returned as the input iterators are consumed.
// WARNING: The input sets _must be_ ordered by time ascending.

type Metric = [number, number]; // [second, value]
type ResultPoint = [number, number | null]; // [second, value | null]
type Accumulator = (current: number | undefined, next: number) => number;

// Generator function to process metrics, yielding points for specified seconds, requires both metrics and timeWindow to be preordered ascending 
function* processMetrics(
  metrics: Iterable<Metric>,
  timeWindow: Iterable<number>,
  outer: boolean = false,
  accumulate: Accumulator = (current, next) => (current ?? 0) + next,
  divideAccumulationBy: number | undefined = undefined
): Generator<ResultPoint> {
  // Get iterator for metrics and time window
  const metricIterator = metrics[Symbol.iterator]();
  const timeIterator = timeWindow[Symbol.iterator]();
  let currentMetric = metricIterator.next();
  let currentTime = timeIterator.next();

  // Process each second in the time window
  while (!currentTime.done) {
    const second = currentTime.value;

    // Skip metrics before the current second
    while (!currentMetric.done && currentMetric.value[0] < second) {
      currentMetric = metricIterator.next();
    }

    // Accumulate values for the current second
    let accumulated: number | undefined;
    let count = 0;
    while (!currentMetric.done && currentMetric.value[0] === second) {
      accumulated = accumulate(accumulated, currentMetric.value[1]); // possibly skip undefined and null
      count++;
      currentMetric = metricIterator.next();
    }

    // Yield result: value if present, or null if outer is true
    if (accumulated !== undefined && count > 0) {
      // Compute final value with intentional division
      let finalValue = accumulated;
      if (divideAccumulationBy === undefined) {
        finalValue = accumulated / count;
      } else if (divideAccumulationBy !== undefined) {
        finalValue = accumulated / divideAccumulationBy;
      }
      yield [second, finalValue];
    } else if (outer) {
      yield [second, null];
    }

    currentTime = timeIterator.next();
  }
}

/*
 * Example Accumulator Functions:
 *
 * Note: For accumulators that require division by count (e.g., average), use the sum accumulator
 * with divideAccumulationBy = undefined. For others, use divideAccumulationBy = 1 (no effective division).
 *
 * 1. Running Average (use sum accumulator with divideAccumulationBy = undefined):
 *    Averages values by summing and dividing by count.
 *    Accumulator: (current, next) => (current ?? 0) + next
 *    Example: processMetrics(..., (current, next) => (current ?? 0) + next, undefined, outer)
 *    For [20, 30]: (20 + 30) / 2 = 25
 *
 * 2. Minimum (divideAccumulationBy = 1):
 *    Takes the smallest value.
 *    const minValue: Accumulator = (current, next) => Math.min(current ?? next, next);
 *
 * 3. Maximum (divideAccumulationBy = 1):
 *    Takes the largest value.
 *    const maxValue: Accumulator = (current, next) => Math.max(current ?? next, next);
 *
 * 4. First (divideAccumulationBy = 1):
 *    Takes the first value.
 *    const firstValue: Accumulator = (current, next) => current ?? next;
 *
 * 5. Last (divideAccumulationBy = 1):
 *    Takes the last value.
 *    const lastValue: Accumulator = (current, next) => next;
 *
 * 6. Cumulative Sum (divideAccumulationBy = 1):
 *    Sums all values without dividing.
 *    const cumulativeSum: Accumulator = (current, next) => (current ?? 0) + next;
 *
 * 7. Exponential Smoothing (divideAccumulationBy = 1):
 *    Applies exponential moving average with smoothing factor alpha.
 *    const exponentialSmoothing: Accumulator = (current, next) => {
 *      const alpha = 0.5;
 *      return current === undefined ? next : alpha * next + (1 - alpha) * current;
 *    };
 */