xo-stack/packages/stack/src/rates/rates-composite.ts

import { BaseRates } from './base-rates.js';
import { UUIDV7 } from '../utils/uuid.js';

/**
 * A composite rate adapter that combines rates from multiple rate sources
 * using a configurable policy.
 */
export class RatesComposite extends BaseRates {
  // List of the adapters with a unique ID for each adapter
  private adapters: { id: string; adapter: BaseRates }[] = [];

  // Track available pairs across all rate adapters
  private availablePairs: Set<string> = new Set();

  // Cache of the latest rates from each adapter
  private rateCache: Map<string, Map<string, Map<string, number>>> = new Map();

  // Off callbacks for unsubscribing from events
  private unsubscribeFunctions: Map<string, () => void> = new Map();

  /**
   * Creates a new RatesComposite instance
   *
   * @param adapters Array of BaseRates adapters to combine
   * @param policy Policy to determine how to combine rates (defaults to average)
   * @param refreshIntervalMs How often to emit combined rates (defaults to 60000ms)
   */
  constructor(
    adapters: BaseRates[],
    private policy: RatesCompositePolicy = ratesAveragePolicy,
    private refreshIntervalMs: number = 60000,
  ) {
    super();

    // Initialize rate cache
    this.rateCache = new Map();

    // Add all adapters to the composite
    adapters.forEach((adapter) => this.addAdapter(adapter, false));
  }

  /**
   * Start all rate adapters and setup event listeners
   */
  public async start(): Promise<void> {
    // Start all adapters
    await Promise.all(this.adapters.map(({ adapter }) => adapter.start()));

    // Start periodic refresh to emit combined rates
    this.startPeriodicRefresh();
  }

  /**
   * Stop all rate adapters and clean up event listeners
   */
  public async stop(): Promise<void> {
    // Stop periodic refresh
    if (this.refreshTimer) {
      clearInterval(this.refreshTimer);
      this.refreshTimer = undefined;
    }

    // Clean up all event listeners by calling their unsubscribe functions
    for (const unsubscribe of this.unsubscribeFunctions.values()) {
      unsubscribe();
    }
    this.unsubscribeFunctions.clear();

    // Stop all adapters
    await Promise.all(this.adapters.map(({ adapter }) => adapter.stop()));
  }

  /**
   * Adds a new rate adapter to the composite
   */
  public addAdapter(adapter: BaseRates, start: boolean = true): void {
    // Generate a random ID for the adapter
    const adapterId = UUIDV7.generateRandom().toString();

    // Add the adapter to the list
    this.adapters.push({ id: adapterId, adapter });

    const listener = (event: {
      denominatorUnitCode: string;
      numeratorUnitCode: string;
      price: number;
    }) => {
      this.handleRateUpdate(adapterId, event);
    };

    // Use the returned unsubscribe function from the custom EventEmitter
    const unsubscribe = adapter.on('rateUpdated', listener);

    // Store unsubscribe function associated with the adapter ID
    this.unsubscribeFunctions.set(adapterId, unsubscribe);

    // Start the adapter if requested
    if (start) {
      adapter.start();
    }
  }

  /**
   * Removes a rate adapter from the composite
   */
  public removeAdapter(adapter: BaseRates, stop: boolean = true): void {
    // Get the index of the adapter
    const adapterId = this.adapters.find(
      ({ adapter: storedAdapter }) => storedAdapter === adapter,
    )?.id;

    // If the adapter is not found, do nothing
    if (!adapterId) {
      return;
    }

    // Call the unsubscribe function for the adapter
    this.unsubscribeFunctions.get(adapterId)?.();

    // Remove the adapter's rates from the cache
    this.rateCache.forEach((numeratorMap) => {
      numeratorMap.forEach((rateMap) => {
        rateMap.delete(adapterId);
      });
    });

    // Remove the unsubscribe function for the adapter
    this.unsubscribeFunctions.delete(adapterId);

    // Remove the adapter from the list
    this.adapters = this.adapters.filter(({ id }) => id !== adapterId);

    // Stop the adapter if requested
    if (stop) {
      adapter.stop();
    }
  }

  /**
   * Sets the policy for how to combine rates
   */
  public setPolicy(policy: RatesCompositePolicy): void {
    this.policy = policy;
  }

  /**
   * Returns a set of all available currency pairs across all adapters
   */
  public async listPairs(): Promise<Set<string>> {
    // Get available pairs from all adapters
    const pairPromises = this.adapters.map(({ adapter }) =>
      adapter.listPairs(),
    );
    const pairSets = await Promise.all(pairPromises);

    // Combine all pairs
    const allPairs = new Set<string>();
    for (const pairSet of pairSets) {
      for (const pair of pairSet) {
        allPairs.add(pair);
      }
    }

    return allPairs;
  }

  // Timer for periodic refresh
  private refreshTimer?: ReturnType<typeof setInterval>;

  /**
   * Starts periodic refresh to emit combined rates
   */
  private startPeriodicRefresh(): void {
    // Clear any existing timer
    if (this.refreshTimer) {
      clearInterval(this.refreshTimer);
    }

    // Set up new timer
    this.refreshTimer = setInterval(() => {
      this.emitCombinedRates();
    }, this.refreshIntervalMs);

    // Emit initial rates immediately
    this.emitCombinedRates();
  }

  /**
   * Handles a rate update from an adapter
   */
  private handleRateUpdate(
    adapterId: string,
    event: {
      denominatorUnitCode: string;
      numeratorUnitCode: string;
      price: number;
    },
  ): void {
    const { denominatorUnitCode, numeratorUnitCode, price } = event;

    // Initialize maps if they don't exist
    if (!this.rateCache.has(denominatorUnitCode)) {
      this.rateCache.set(denominatorUnitCode, new Map());
    }

    const denomMap = this.rateCache.get(denominatorUnitCode)!;
    if (!denomMap.has(numeratorUnitCode)) {
      denomMap.set(numeratorUnitCode, new Map());
    }

    // Store the rate in the cache
    const rateMap = denomMap.get(numeratorUnitCode)!;
    rateMap.set(adapterId, price);

    // Add pair to available pairs
    this.availablePairs.add(`${numeratorUnitCode}/${denominatorUnitCode}`);
  }

  /**
   * Emits combined rates for all available pairs
   */
  private emitCombinedRates(): void {
    // Process each denominator currency
    for (const [denominator, numeratorMap] of this.rateCache.entries()) {
      // Process each numerator currency
      for (const [numerator, rateMap] of numeratorMap.entries()) {
        // Go through our adapters in order, plucking the rate for each adapter
        const rates = this.adapters
          .map(({ id }) => rateMap.get(id))
          .filter((rate) => rate !== undefined);

        // Apply policy to get combined rate
        const combinedRate = this.policy(rates);

        // Only emit if we have a valid rate
        if (combinedRate !== null) {
          this.emit('rateUpdated', {
            denominatorUnitCode: denominator,
            numeratorUnitCode: numerator,
            price: combinedRate,
          });
        }
      }
    }
  }
}

export type RatesCompositePolicy = (rates: number[]) => number | null;

/**
 * Average policy: returns the average of all valid rates
 * Returns null if no valid rates are available
 */
export const ratesAveragePolicy: RatesCompositePolicy = (rates) => {
  const validRates = rates.filter((rate) => !isNaN(rate) && isFinite(rate));
  return validRates.length > 0
    ? validRates.reduce((acc, rate) => acc + rate, 0) / validRates.length
    : null;
};

/**
 * Median policy: returns the median of all valid rates
 * Returns null if no valid rates are available
 */
export const ratesMedianPolicy: RatesCompositePolicy = (rates) => {
  const validRates = rates
    .filter((rate) => !isNaN(rate) && isFinite(rate))
    .sort((a, b) => a - b);
  if (validRates.length === 0) return null;

  const mid = Math.floor(validRates.length / 2);
  return validRates.length % 2 === 0
    ? (validRates[mid - 1] + validRates[mid]) / 2
    : validRates[mid];
};

/**
 * Min policy: returns the minimum of all valid rates
 * Returns null if no valid rates are available
 */
export const ratesMinPolicy: RatesCompositePolicy = (rates) => {
  const validRates = rates.filter((rate) => !isNaN(rate) && isFinite(rate));
  return validRates.length > 0 ? Math.min(...validRates) : null;
};

/**
 * Max policy: returns the maximum of all valid rates
 * Returns null if no valid rates are available
 */
export const ratesMaxPolicy: RatesCompositePolicy = (rates) => {
  const validRates = rates.filter((rate) => !isNaN(rate) && isFinite(rate));
  return validRates.length > 0 ? Math.max(...validRates) : null;
};

/**
 * First Available policy: returns the first valid rate found
 * Returns null if no valid rates are available
 */
export const ratesFirstAvailablePolicy: RatesCompositePolicy = (rates) => {
  const validRate = rates.find((rate) => !isNaN(rate) && isFinite(rate));
  return validRate !== undefined ? validRate : null;
};

/**
 * Priority policy factory: returns a policy that uses rates in order of priority
 * Returns the first valid rate from the adapters in the order they were provided
 */
export const ratesPriorityPolicyFactory = (): RatesCompositePolicy => {
  return (rates) => {
    for (const rate of rates) {
      if (!isNaN(rate) && isFinite(rate)) {
        return rate;
      }
    }
    return null;
  };
};