PerpsStreamManager.tsx

import React, { createContext, useContext } from 'react';
import { v4 as uuidv4 } from 'uuid';
import performance from 'react-native-performance';
import Engine from '../../../../core/Engine';
import { DevLogger } from '../../../../core/SDKConnect/utils/DevLogger';
import Logger from '../../../../util/Logger';
import { ensureError } from '../../../../util/errorUtils';
import {
  trace,
  endTrace,
  TraceName,
  TraceOperation,
} from '../../../../util/trace';
import PerpsConnectionManager from '../services/PerpsConnectionManager';
import {
  PERFORMANCE_CONFIG,
  PERPS_CONSTANTS,
  PerpsMeasurementName,
  type PriceUpdate,
  type Position,
  type Order,
  type OrderFill,
  type AccountState,
  type PerpsMarketData,
  findEvmAccount,
} from '@metamask/perps-controller';
import { PROVIDER_CONFIG } from '../constants/perpsConfig';
import { getE2EMockStreamManager } from '../utils/e2eBridgePerps';
import { CandleStreamChannel } from './channels/CandleStreamChannel';
import { getPreloadedData } from '../hooks/stream/hasCachedPerpsData';
import { InternalAccount } from '@metamask/keyring-internal-api';

/**
 * Gets the EVM account from the selected account group.
 * Mobile-specific helper using Engine context.
 * @returns EVM account or null if not found
 */
function getEvmAccountFromSelectedAccountGroup() {
  const { AccountTreeController } = Engine.context;
  const accounts = AccountTreeController.getAccountsFromSelectedAccountGroup();
  return findEvmAccount(accounts as InternalAccount[]);
}

// Generic subscription parameters
interface StreamSubscription<T> {
  id: string;
  callback: (data: T) => void;
  throttleMs?: number;
  timer?: NodeJS.Timeout;
  pendingUpdate?: T;
  hasReceivedFirstUpdate?: boolean; // Track if subscriber has received first update
}

// Base class for any stream type
abstract class StreamChannel<T> {
  protected cache = new Map<string, T>();
  protected subscribers = new Map<string, StreamSubscription<T>>();
  protected wsSubscription: (() => void) | null = null;
  // Track account context to prevent stale data across account switches
  protected accountAddress: string | null = null;
  // Track WebSocket connection timing for first data measurement
  protected wsConnectionStartTime: number | null = null;
  // Flag to pause emission during operations (keeps WebSocket alive)
  protected isPaused = false;
  // Retry counter for deferred connect() calls
  protected connectRetryCount = 0;
  // Timer handle for deferConnect so it can be cancelled on disconnect
  protected deferConnectTimer: ReturnType<typeof setTimeout> | null = null;
  private static readonly MAX_CONNECT_RETRIES = 150; // 30s at 200ms

  protected notifySubscribers(updates: T) {
    // Block emission if paused (WebSocket continues receiving updates)
    if (this.isPaused) {
      return;
    }

    this.subscribers.forEach((subscriber) => {
      // Check if this is the first update for this subscriber
      if (!subscriber.hasReceivedFirstUpdate) {
        subscriber.callback(updates);
        subscriber.hasReceivedFirstUpdate = true;
        return; // Don't set up throttle for the first update
      }

      // If no throttling (throttleMs is 0 or undefined), notify immediately
      if (!subscriber.throttleMs) {
        subscriber.callback(updates);
        return;
      }

      // For subsequent updates with throttling, use throttle logic
      // Store pending update
      subscriber.pendingUpdate = updates;

      // Throttle pattern: Only set timer if one isn't already running
      // This ensures callbacks fire at most once per throttleMs interval
      // WITHOUT resetting the countdown on every update (which would be debouncing)
      // The conditional check prevents timer accumulation - no memory leaks
      subscriber.timer ??= setTimeout(() => {
        if (subscriber.pendingUpdate) {
          subscriber.callback(subscriber.pendingUpdate);
          subscriber.pendingUpdate = undefined;
        }
        subscriber.timer = undefined;
      }, subscriber.throttleMs);
    });
  }

  subscribe(params: {
    callback: (data: T) => void;
    throttleMs?: number;
  }): () => void {
    const id = Math.random().toString(36);

    const subscription: StreamSubscription<T> = {
      id,
      ...params,
      hasReceivedFirstUpdate: false, // Initialize as false
    };
    this.subscribers.set(id, subscription);

    // Give immediate cached data if available
    const cached = this.getCachedData();
    if (cached != null) {
      params.callback(cached);
      // Mark as having received first update since we provided cached data
      subscription.hasReceivedFirstUpdate = true;
    }

    // Ensure WebSocket connected
    this.connect();

    // Return unsubscribe function
    return () => {
      const sub = this.subscribers.get(id);
      if (sub?.timer) {
        clearTimeout(sub.timer);
        sub.timer = undefined;
      }
      this.subscribers.delete(id);

      // Disconnect if no subscribers
      if (this.subscribers.size === 0) {
        this.disconnect();
      }
    };
  }

  protected connect() {
    // Override in subclasses
  }

  /**
   * Schedule a deferred connect() retry with safety checks.
   * Aborts if no subscribers remain or max retries exceeded.
   */
  protected deferConnect(delayMs: number): void {
    if (this.subscribers.size === 0) {
      this.connectRetryCount = 0;
      return;
    }

    if (this.connectRetryCount >= StreamChannel.MAX_CONNECT_RETRIES) {
      DevLogger.log(
        `${this.constructor.name}: Max connect retries exceeded (${StreamChannel.MAX_CONNECT_RETRIES}), giving up`,
      );
      this.connectRetryCount = 0;
      return;
    }

    this.connectRetryCount++;
    if (this.deferConnectTimer) {
      clearTimeout(this.deferConnectTimer);
    }
    this.deferConnectTimer = setTimeout(() => {
      this.deferConnectTimer = null;
      this.connect();
    }, delayMs);
  }

  /**
   * Common initialization guard for connect().
   * Returns true if the channel is ready to connect, false if deferred.
   * Resets connectRetryCount on success.
   *
   * When the connection manager is actively connecting, awaits the connection
   * promise and retries connect() once resolved, instead of blind 200ms polling.
   */
  protected ensureReady(): boolean {
    if (Engine.context.PerpsController.isCurrentlyReinitializing()) {
      this.deferConnect(PERPS_CONSTANTS.ReconnectionCleanupDelayMs);
      return false;
    }
    const connState = PerpsConnectionManager.getConnectionState();
    if (!connState.isInitialized) {
      // If actively connecting, await the connection promise instead of polling
      if (connState.isConnecting) {
        DevLogger.log(
          `${this.constructor.name}: ensureReady: awaiting active connection`,
        );
        this.awaitConnectionThenConnect();
        return false;
      }
      this.deferConnect(PERPS_CONSTANTS.ConnectRetryDelayMs);
      return false;
    }
    this.connectRetryCount = 0;
    return true;
  }

  /**
   * Await the PerpsConnectionManager connection promise, then retry connect().
   * This replaces blind 200ms polling when we know a connection is in progress.
   */
  private awaitConnectionThenConnect(): void {
    // Prevent duplicate awaits — only one outstanding wait at a time
    if (this.deferConnectTimer) {
      return;
    }
    // Use a sentinel timer value to signal that we're waiting on the promise
    // This prevents deferConnect from also scheduling a parallel timer
    const noop = () => {
      /* sentinel timer */
    };
    const sentinel = setTimeout(noop, 0);
    this.deferConnectTimer = sentinel;

    PerpsConnectionManager.waitForConnection()
      .then(() => {
        // Only clear if our sentinel is still the active timer; a disconnect()
        // followed by a new subscribe() may have replaced it with a real timer.
        if (this.deferConnectTimer === sentinel) {
          this.deferConnectTimer = null;
        }
        if (this.subscribers.size > 0) {
          this.connect();
        }
      })
      .catch(() => {
        if (this.deferConnectTimer === sentinel) {
          this.deferConnectTimer = null;
        }
        // Connection failed — fall back to normal defer polling
        if (this.subscribers.size > 0) {
          DevLogger.log(
            `${this.constructor.name}: awaitConnectionThenConnect: connection failed, falling back to polling`,
          );
          this.deferConnect(PERPS_CONSTANTS.ConnectRetryDelayMs);
        }
      });
  }

  /**
   * Reconnect the channel after WebSocket reconnection
   * Clears dead subscription and re-establishes if there are active subscribers
   */
  public reconnect() {
    this.disconnect();
    // Re-establish connection if there are active subscribers
    if (this.subscribers.size > 0) {
      this.connect();
    }
  }

  public disconnect() {
    this.connectRetryCount = 0;
    if (this.deferConnectTimer) {
      clearTimeout(this.deferConnectTimer);
      this.deferConnectTimer = null;
    }
    // This prevents orphaned timers from continuing to run after disconnect
    this.subscribers.forEach((subscriber) => {
      if (subscriber.timer) {
        clearTimeout(subscriber.timer);
        subscriber.timer = undefined;
      }
      subscriber.pendingUpdate = undefined;
    });

    if (this.wsSubscription) {
      this.wsSubscription();
      this.wsSubscription = null;
    }
    this.accountAddress = null;
    this.wsConnectionStartTime = null;
  }

  /**
   * Pause emission of updates to subscribers
   * WebSocket connection stays alive and continues receiving data
   * Used during batch operations to prevent UI re-renders from stale data
   */
  public pause(): void {
    this.isPaused = true;
  }

  /**
   * Resume emission of updates to subscribers
   * Subscribers will receive the next update from the WebSocket
   */
  public resume(): void {
    this.isPaused = false;
  }

  protected getCachedData(): T | null {
    // Override in subclasses to return null for no cache, or actual data
    return null;
  }

  public clearCache(): void {
    // This ensures no timers are orphaned during the disconnect/reconnect cycle
    this.subscribers.forEach((subscriber) => {
      // Clear any pending updates and timers
      if (subscriber.timer) {
        clearTimeout(subscriber.timer);
        subscriber.timer = undefined;
      }
      subscriber.pendingUpdate = undefined;
      subscriber.hasReceivedFirstUpdate = false;
    });

    // Disconnect the old WebSocket subscription to stop receiving old account data
    if (this.wsSubscription) {
      this.disconnect();
    }

    // Reset account context immediately
    this.accountAddress = null;
    this.wsConnectionStartTime = null;

    // Clear the cache
    this.cache.clear();

    // Notify subscribers with cleared data to trigger loading state
    // Using getClearedData() ensures type safety while maintaining loading semantics
    this.subscribers.forEach((subscriber) => {
      // Send cleared data to indicate "no data yet" (loading state)
      subscriber.callback(this.getClearedData());
    });

    // If we have active subscribers, they'll trigger reconnect in their next render
    // The connect() call will create a new WebSocket with the new account
  }

  protected abstract getClearedData(): T;
}

// Specific channel for prices
class PriceStreamChannel extends StreamChannel<Record<string, PriceUpdate>> {
  private readonly symbols = new Set<string>();
  private prewarmUnsubscribe?: () => void;
  private actualPriceUnsubscribe?: () => void;
  private allMarketSymbols: string[] = [];
  // Unique ID per prewarm cycle to detect stale promises and prevent subscription leaks
  private prewarmCycleId: number = 0;
  // Override cache to store individual PriceUpdate objects
  protected priceCache = new Map<string, PriceUpdate>();

  protected connect() {
    if (this.wsSubscription) {
      return;
    }

    if (!this.ensureReady()) return;

    // If we have a prewarm subscription, we're already subscribed to all markets
    // No need to create another subscription
    if (this.prewarmUnsubscribe) {
      // Just notify subscribers with cached data
      const cached = this.getCachedData();
      if (cached) {
        this.notifySubscribers(cached);
      }
      return;
    }

    // Collect all unique symbols from subscribers
    const allSymbols = Array.from(this.symbols);

    DevLogger.log(
      `PriceStreamChannel: allSymbols len=`,
      allSymbols.length,
      allSymbols,
    );

    if (allSymbols.length === 0) {
      return;
    }

    this.wsSubscription = Engine.context.PerpsController.subscribeToPrices({
      symbols: allSymbols,
      callback: (updates: PriceUpdate[]) => {
        // Update cache and build price map
        const priceMap: Record<string, PriceUpdate> = {};
        updates.forEach((update) => {
          // Map the update to PriceUpdate format
          const priceUpdate: PriceUpdate = {
            symbol: update.symbol,
            price: update.price,
            timestamp: Date.now(),
            percentChange24h: update.percentChange24h,
            bestBid: update.bestBid,
            bestAsk: update.bestAsk,
            spread: update.spread,
            markPrice: update.markPrice,
            funding: update.funding,
            openInterest: update.openInterest,
            volume24h: update.volume24h,
          };
          this.priceCache.set(update.symbol, priceUpdate);
          priceMap[update.symbol] = priceUpdate;
        });

        this.notifySubscribers(priceMap);
      },
    });
  }

  protected getCachedData(): Record<string, PriceUpdate> | null {
    if (this.priceCache.size === 0) return null;
    const cached: Record<string, PriceUpdate> = {};
    this.priceCache.forEach((value, key) => {
      cached[key] = value;
    });
    return cached;
  }

  protected getClearedData(): Record<string, PriceUpdate> {
    return {};
  }

  public clearCache(): void {
    // Clear the price-specific cache
    this.priceCache.clear();
    // Cleanup pre-warm subscription
    this.cleanupPrewarm();
    // Call parent clearCache
    super.clearCache();
  }

  subscribeToSymbols(params: {
    symbols: string[];
    callback: (prices: Record<string, PriceUpdate>) => void;
    throttleMs?: number;
  }): () => void {
    // Track symbols for filtering
    params.symbols.forEach((s) => {
      this.symbols.add(s);
    });

    // Ensure connection is established (allMids provides all symbols)
    // No need to reconnect when new symbols are added since allMids
    // already provides prices for all markets
    if (!this.wsSubscription && !this.prewarmUnsubscribe) {
      this.connect();
    }

    return this.subscribe({
      callback: (allPrices) => {
        // Filter to only requested symbols
        const filtered: Record<string, PriceUpdate> = {};
        params.symbols.forEach((symbol) => {
          if (allPrices?.[symbol]) {
            filtered[symbol] = allPrices[symbol];
          }
        });
        params.callback(filtered);
      },
      throttleMs: params.throttleMs,
    });
  }

  /**
   * Pre-warm the channel by subscribing to all market prices
   * This keeps a single WebSocket connection alive with all price updates
   * Non-blocking: Returns immediately while market fetch happens in background
   * @returns Cleanup function to call when leaving Perps environment
   */
  public async prewarm(): Promise<() => void> {
    if (this.prewarmUnsubscribe) {
      DevLogger.log('PriceStreamChannel: Already pre-warmed');
      return this.prewarmUnsubscribe;
    }

    const controller = Engine.context.PerpsController;
    if (!controller) return () => undefined;

    // Increment cycle ID to detect stale promises from previous prewarm cycles
    // This prevents subscription leaks when user navigates: Perps → away → back quickly
    this.prewarmCycleId++;
    const currentCycleId = this.prewarmCycleId;

    // Start market fetch in background (non-blocking)
    // We need the symbols to register subscribers, but we can return immediately
    controller
      .getMarkets()
      .then((markets) => {
        // If this promise is from a stale cycle, don't set up subscription
        // This prevents leaks when prewarm is called multiple times rapidly
        if (currentCycleId !== this.prewarmCycleId) {
          DevLogger.log('PriceStreamChannel: Skipping stale prewarm cycle', {
            currentCycleId,
            activeCycleId: this.prewarmCycleId,
          });
          return;
        }

        // If already cleaned up, don't set up subscription
        if (this.prewarmUnsubscribe === undefined) {
          return;
        }

        this.allMarketSymbols = markets.map((market) => market.name);

        DevLogger.log(
          'PriceStreamChannel: Pre-warming with all market symbols',
          {
            symbolCount: this.allMarketSymbols.length,
            symbols: this.allMarketSymbols.slice(0, 10),
          },
        );

        // WARNING: Do NOT set includeMarketData: true here. It triggers
        // per-symbol activeAssetCtx subscriptions (N symbols × N DEXs = N²
        // WebSocket connections). assetCtxs (1 per DEX) is always established
        // by the subscription service regardless of this flag.
        const unsub = controller.subscribeToPrices({
          symbols: this.allMarketSymbols,
          includeMarketData: false,
          callback: (updates: PriceUpdate[]) => {
            const priceMap: Record<string, PriceUpdate> = {};
            updates.forEach((update) => {
              const priceUpdate: PriceUpdate = {
                symbol: update.symbol,
                price: update.price,
                timestamp: Date.now(),
                percentChange24h: update.percentChange24h,
                bestBid: update.bestBid,
                bestAsk: update.bestAsk,
                spread: update.spread,
                markPrice: update.markPrice,
                funding: update.funding,
                openInterest: update.openInterest,
                volume24h: update.volume24h,
              };
              this.priceCache.set(update.symbol, priceUpdate);
              priceMap[update.symbol] = priceUpdate;
            });

            if (this.subscribers.size > 0) {
              this.notifySubscribers(priceMap);
            }
          },
        });

        // Store the actual unsubscribe function
        this.actualPriceUnsubscribe = unsub;
      })
      .catch((error) => {
        Logger.error(
          ensureError(error, 'PriceStreamChannel.prewarm.backgroundFetch'),
          {
            context: 'PriceStreamChannel.prewarm.backgroundFetch',
          },
        );
        // Reset state so subsequent prewarm/connect calls can recover
        this.prewarmUnsubscribe = undefined;
        this.allMarketSymbols = [];
        // Reconnect waiting subscribers that were skipped because prewarm was pending
        if (this.subscribers.size > 0) {
          this.connect();
        }
      });

    // Return cleanup function immediately (before markets load)
    this.prewarmUnsubscribe = () => {
      DevLogger.log('PriceStreamChannel: Cleaning up prewarm subscription');
      this.cleanupPrewarm();
    };

    return this.prewarmUnsubscribe;
  }

  /**
   * Cleanup pre-warm subscription
   */
  public cleanupPrewarm(): void {
    if (this.actualPriceUnsubscribe) {
      this.actualPriceUnsubscribe();
      this.actualPriceUnsubscribe = undefined;
    }
    this.prewarmUnsubscribe = undefined;
    this.allMarketSymbols = [];
  }
}

// Specific channel for orders (null = cleared on account switch; hooks show skeleton until next update)
class OrderStreamChannel extends StreamChannel<Order[] | null> {
  private prewarmUnsubscribe?: () => void;
  private firstDataTraceId?: string;

  protected connect() {
    if (this.wsSubscription) return;

    if (!this.ensureReady()) return;

    // Start trace for first data measurement (before subscription)
    this.firstDataTraceId = uuidv4();
    trace({
      name: TraceName.PerpsWebSocketFirstOrders,
      id: this.firstDataTraceId,
      op: TraceOperation.PerpsOperation,
    });

    // Track WebSocket connection start time for duration calculation
    this.wsConnectionStartTime = performance.now();

    this.wsSubscription = Engine.context.PerpsController.subscribeToOrders({
      callback: (orders: Order[]) => {
        // Validate account context
        const currentAccount =
          getEvmAccountFromSelectedAccountGroup()?.address || null;
        if (this.accountAddress && this.accountAddress !== currentAccount) {
          Logger.error(new Error('OrderStreamChannel: Wrong account context'), {
            expected: currentAccount,
            received: this.accountAddress,
          });
          return;
        }
        this.accountAddress = currentAccount;

        // Track first order data from WebSocket (only once per connection)
        if (this.wsConnectionStartTime !== null && this.firstDataTraceId) {
          const firstDataDuration =
            performance.now() - this.wsConnectionStartTime;

          // Log WebSocket performance measurement
          DevLogger.log(
            `${PERFORMANCE_CONFIG.LoggingMarkers.WebsocketPerformance} PerpsWS: First order data received`,
            {
              duration: `${firstDataDuration.toFixed(0)}ms`,
            },
          );

          // End trace with accurate duration
          endTrace({
            name: TraceName.PerpsWebSocketFirstOrders,
            id: this.firstDataTraceId,
            data: {
              success: true,
              duration: firstDataDuration,
            },
          });

          this.wsConnectionStartTime = null;
          this.firstDataTraceId = undefined;
        }

        this.cache.set('orders', orders);
        this.notifySubscribers(orders);
      },
    });
  }

  protected getCachedData() {
    const cached = this.cache.get('orders');
    if (cached !== undefined) return cached;
    return getPreloadedData<Order[]>('cachedOrders');
  }

  protected getClearedData(): Order[] | null {
    return null;
  }

  /**
   * Pre-warm the channel by creating a persistent subscription
   * This keeps the WebSocket connection alive and caches data continuously
   * @returns Cleanup function to call when leaving Perps environment
   */
  public prewarm(): () => void {
    if (this.prewarmUnsubscribe) {
      DevLogger.log('OrderStreamChannel: Already pre-warmed');
      return this.prewarmUnsubscribe;
    }

    // Create a real subscription with no-op callback to keep connection alive
    this.prewarmUnsubscribe = this.subscribe({
      callback: () => {
        // No-op callback - just keeps the connection alive for caching
      },
      throttleMs: 0, // No throttle for pre-warm
    });

    // Return cleanup function that clears internal state
    return () => {
      DevLogger.log('OrderStreamChannel: Cleaning up prewarm subscription');
      this.cleanupPrewarm();
    };
  }

  /**
   * Cleanup pre-warm subscription
   */
  public cleanupPrewarm(): void {
    if (this.prewarmUnsubscribe) {
      this.prewarmUnsubscribe();
      this.prewarmUnsubscribe = undefined;
    }
  }

  public disconnect() {
    this.firstDataTraceId = undefined;
    super.disconnect();
  }

  public clearCache(): void {
    // Cleanup pre-warm subscription
    this.cleanupPrewarm();
    this.firstDataTraceId = undefined;
    // Call parent clearCache
    super.clearCache();
  }
}

// Specific channel for positions (null = cleared on account switch; hooks show skeleton until next update)
class PositionStreamChannel extends StreamChannel<Position[] | null> {
  private prewarmUnsubscribe?: () => void;
  private firstDataTraceId?: string;

  protected connect() {
    if (this.wsSubscription) return;

    if (!this.ensureReady()) return;

    // Start trace for first data measurement (before subscription)
    this.firstDataTraceId = uuidv4();
    trace({
      name: TraceName.PerpsWebSocketFirstPositions,
      id: this.firstDataTraceId,
      op: TraceOperation.PerpsOperation,
    });

    // Track WebSocket connection start time for duration calculation
    this.wsConnectionStartTime = performance.now();

    this.wsSubscription = Engine.context.PerpsController.subscribeToPositions({
      callback: (positions: Position[]) => {
        // Validate account context
        const currentAccount =
          getEvmAccountFromSelectedAccountGroup()?.address || null;
        if (this.accountAddress && this.accountAddress !== currentAccount) {
          Logger.error(
            new Error('PositionStreamChannel: Wrong account context'),
            {
              expected: currentAccount,
              received: this.accountAddress,
            },
          );
          return;
        }
        this.accountAddress = currentAccount;

        // Track first position data from WebSocket (only once per connection)
        if (this.wsConnectionStartTime !== null && this.firstDataTraceId) {
          const firstDataDuration =
            performance.now() - this.wsConnectionStartTime;

          // Log WebSocket performance measurement
          DevLogger.log(
            `${PERFORMANCE_CONFIG.LoggingMarkers.WebsocketPerformance} PerpsWS: First position data received`,
            {
              metric: PerpsMeasurementName.PerpsWebsocketFirstPositionData,
              duration: `${firstDataDuration.toFixed(0)}ms`,
            },
          );

          // End trace with accurate duration
          endTrace({
            name: TraceName.PerpsWebSocketFirstPositions,
            id: this.firstDataTraceId,
            data: {
              success: true,
              duration: firstDataDuration,
            },
          });

          this.wsConnectionStartTime = null;
          this.firstDataTraceId = undefined;
        }

        this.cache.set('positions', positions);
        this.notifySubscribers(positions);
      },
    });
  }

  protected getCachedData() {
    const cached = this.cache.get('positions');
    if (cached !== undefined) return cached;
    return getPreloadedData<Position[]>('cachedPositions');
  }

  protected getClearedData(): Position[] | null {
    return null;
  }

  /**
   * Pre-warm the channel by creating a persistent subscription
   * This keeps the WebSocket connection alive and caches data continuously
   * @returns Cleanup function to call when leaving Perps environment
   */
  public prewarm(): () => void {
    if (this.prewarmUnsubscribe) {
      DevLogger.log('PositionStreamChannel: Already pre-warmed');
      return this.prewarmUnsubscribe;
    }

    // Create a real subscription with no-op callback to keep connection alive
    this.prewarmUnsubscribe = this.subscribe({
      callback: () => {
        // No-op callback - just keeps the connection alive for caching
      },
      throttleMs: 0, // No throttle for pre-warm
    });

    // Return cleanup function that clears internal state
    return () => {
      DevLogger.log('PositionStreamChannel: Cleaning up prewarm subscription');
      this.cleanupPrewarm();
    };
  }

  public disconnect() {
    this.firstDataTraceId = undefined;
    super.disconnect();
  }

  public clearCache(): void {
    // Cleanup pre-warm subscription
    this.cleanupPrewarm();
    this.firstDataTraceId = undefined;
    // Call parent clearCache
    super.clearCache();
  }

  /**
   * Cleanup pre-warm subscription
   */
  public cleanupPrewarm(): void {
    if (this.prewarmUnsubscribe) {
      this.prewarmUnsubscribe();
      this.prewarmUnsubscribe = undefined;
    }
  }

  /**
   * Apply optimistic update for TP/SL prices to a position
   * This immediately updates the UI before WebSocket confirms the change
   * @param coin - The coin/asset symbol
   * @param takeProfitPrice - The new take profit price (undefined to remove)
   * @param stopLossPrice - The new stop loss price (undefined to remove)
   */
  public updatePositionTPSLOptimistic(
    coin: string,
    takeProfitPrice: string | undefined,
    stopLossPrice: string | undefined,
  ): void {
    const cachedPositions = this.cache.get('positions');
    if (!cachedPositions) {
      DevLogger.log(
        'PositionStreamChannel: Cannot apply optimistic update - no cached positions',
      );
      return;
    }

    const positionIndex = cachedPositions.findIndex((p) => p.symbol === coin);
    if (positionIndex === -1) {
      DevLogger.log(
        `PositionStreamChannel: Cannot apply optimistic update - position not found for ${coin}`,
      );
      return;
    }

    // Create updated positions array with the optimistic TP/SL values
    const updatedPositions = cachedPositions.map((position, index) => {
      if (index === positionIndex) {
        return {
          ...position,
          takeProfitPrice,
          stopLossPrice,
          // Update counts based on whether TP/SL is set
          takeProfitCount: takeProfitPrice ? 1 : 0,
          stopLossCount: stopLossPrice ? 1 : 0,
        };
      }
      return position;
    });

    DevLogger.log('PositionStreamChannel: Applying optimistic TP/SL update', {
      coin,
      takeProfitPrice,
      stopLossPrice,
    });

    // Update cache and notify subscribers immediately
    this.cache.set('positions', updatedPositions);
    this.notifySubscribers(updatedPositions);
  }
}

// Specific channel for fills
class FillStreamChannel extends StreamChannel<OrderFill[]> {
  private prewarmUnsubscribe?: () => void;

  protected connect() {
    if (this.wsSubscription) return;

    if (!this.ensureReady()) return;

    this.wsSubscription = Engine.context.PerpsController.subscribeToOrderFills({
      callback: (fills: OrderFill[], isSnapshot?: boolean) => {
        let updated: OrderFill[];
        if (isSnapshot) {
          // Snapshot: replace cache with initial historical data
          // Sort by timestamp descending (newest first)
          updated = [...fills]
            .sort((a, b) => b.timestamp - a.timestamp)
            .slice(0, 100);
        } else {
          // Streaming: prepend new fills to existing (newest first)
          const existing = this.cache.get('fills') || [];
          // New fills go at the beginning since they're most recent
          updated = [...fills, ...existing].slice(0, 100);
        }
        this.cache.set('fills', updated);
        this.notifySubscribers(updated);
      },
    });
  }

  protected getCachedData() {
    return this.cache.get('fills') || [];
  }

  protected getClearedData(): OrderFill[] {
    return [];
  }

  /**
   * Pre-warm the channel by creating a persistent subscription
   * This keeps the WebSocket connection alive and caches fills data continuously
   * @returns Cleanup function to call when leaving Perps environment
   */
  public prewarm(): () => void {
    if (this.prewarmUnsubscribe) {
      DevLogger.log('FillStreamChannel: Already pre-warmed');
      return this.prewarmUnsubscribe;
    }

    // Create a real subscription with no-op callback to keep connection alive
    this.prewarmUnsubscribe = this.subscribe({
      callback: () => {
        // No-op callback - just keeps the connection alive for caching
      },
      throttleMs: 0, // No throttle for pre-warm
    });

    // Return cleanup function that clears internal state
    return () => {
      DevLogger.log('FillStreamChannel: Cleaning up prewarm subscription');
      this.cleanupPrewarm();
    };
  }

  /**
   * Cleanup pre-warm subscription
   */
  public cleanupPrewarm(): void {
    if (this.prewarmUnsubscribe) {
      this.prewarmUnsubscribe();
      this.prewarmUnsubscribe = undefined;
    }
  }

  public clearCache(): void {
    // Cleanup pre-warm subscription
    this.cleanupPrewarm();
    // Call parent clearCache
    super.clearCache();
  }
}

// Specific channel for account state
class AccountStreamChannel extends StreamChannel<AccountState | null> {
  private prewarmUnsubscribe?: () => void;
  private firstDataTraceId?: string;

  protected connect() {
    if (this.wsSubscription) return;

    if (!this.ensureReady()) return;

    // Start trace for first data measurement (before subscription)
    this.firstDataTraceId = uuidv4();
    trace({
      name: TraceName.PerpsWebSocketFirstAccount,
      id: this.firstDataTraceId,
      op: TraceOperation.PerpsOperation,
    });

    // Track WebSocket connection start time for duration calculation
    this.wsConnectionStartTime = performance.now();

    this.wsSubscription = Engine.context.PerpsController.subscribeToAccount({
      callback: (account: AccountState | null) => {
        // Validate account context
        const currentAccount =
          getEvmAccountFromSelectedAccountGroup()?.address || null;
        if (this.accountAddress && this.accountAddress !== currentAccount) {
          Logger.error(
            new Error('AccountStreamChannel: Wrong account context'),
            {
              expected: currentAccount,
              received: this.accountAddress,
            },
          );
          return;
        }
        this.accountAddress = currentAccount;

        // Track first account data from WebSocket (only once per connection)
        if (this.wsConnectionStartTime !== null && this.firstDataTraceId) {
          const firstDataDuration =
            performance.now() - this.wsConnectionStartTime;

          // Log WebSocket performance measurement
          DevLogger.log(
            `${PERFORMANCE_CONFIG.LoggingMarkers.WebsocketPerformance} PerpsWS: First account data received`,
            {
              duration: `${firstDataDuration.toFixed(0)}ms`,
            },
          );

          // End trace with accurate duration
          endTrace({
            name: TraceName.PerpsWebSocketFirstAccount,
            id: this.firstDataTraceId,
            data: {
              success: true,
              duration: firstDataDuration,
            },
          });

          this.wsConnectionStartTime = null;
          this.firstDataTraceId = undefined;
        }

        // Use base cache Map with consistent key
        this.cache.set('account', account);
        this.notifySubscribers(account as AccountState | null);
      },
    });
  }

  protected getCachedData(): AccountState | null {
    const cached = this.cache.get('account');
    if (cached !== undefined) return cached;
    return getPreloadedData<AccountState>('cachedAccountState');
  }

  protected getClearedData(): AccountState | null {
    return null;
  }

  public disconnect() {
    this.firstDataTraceId = undefined;
    super.disconnect();
  }

  public clearCache(): void {
    // Cleanup pre-warm subscription
    this.cleanupPrewarm();
    this.firstDataTraceId = undefined;
    // Call parent clearCache
    super.clearCache();
  }

  /**
   * Pre-warm the channel by creating a persistent subscription
   * This keeps the WebSocket connection alive and caches data continuously
   * @returns Cleanup function to call when leaving Perps environment
   */
  public prewarm(): () => void {
    if (this.prewarmUnsubscribe) {
      DevLogger.log('AccountStreamChannel: Already pre-warmed');
      return this.prewarmUnsubscribe;
    }

    // Create a real subscription with no-op callback to keep connection alive
    this.prewarmUnsubscribe = this.subscribe({
      callback: () => {
        // No-op callback - just keeps the connection alive for caching
      },
      throttleMs: 0, // No throttle for pre-warm
    });

    // Return cleanup function that clears internal state
    return () => {
      DevLogger.log('AccountStreamChannel: Cleaning up prewarm subscription');
      this.cleanupPrewarm();
    };
  }

  /**
   * Cleanup pre-warm subscription
   */
  public cleanupPrewarm(): void {
    if (this.prewarmUnsubscribe) {
      this.prewarmUnsubscribe();
      this.prewarmUnsubscribe = undefined;
    }
  }
}

// Open Interest Cap channel for tracking markets at capacity
class OICapStreamChannel extends StreamChannel<string[]> {
  private prewarmUnsubscribe?: () => void;

  protected connect() {
    if (this.wsSubscription) return;

    if (!this.ensureReady()) return;

    // Subscribe to OI cap updates (zero overhead - extracted from existing webData3)
    this.wsSubscription = Engine.context.PerpsController.subscribeToOICaps({
      callback: (caps: string[]) => {
        // Validate account context
        const currentAccount =
          getEvmAccountFromSelectedAccountGroup()?.address || null;
        if (this.accountAddress && this.accountAddress !== currentAccount) {
          Logger.error(new Error('OICapStreamChannel: Wrong account context'), {
            expected: currentAccount,
            received: this.accountAddress,
          });
          return;
        }
        this.accountAddress = currentAccount;

        this.cache.set('oiCaps', caps);
        this.notifySubscribers(caps);
      },
    });
  }

  protected getCachedData(): string[] | null {
    // Return null if no cache exists to distinguish from empty array
    const cached = this.cache.get('oiCaps');
    return cached ?? null;
  }

  protected getClearedData(): string[] {
    return [];
  }

  /**
   * Pre-warm the channel by creating a persistent subscription
   * This keeps the WebSocket connection alive and caches data continuously
   * @returns Cleanup function to call when leaving Perps environment
   */
  public prewarm(): () => void {
    if (this.prewarmUnsubscribe) {
      DevLogger.log('OICapStreamChannel: Already pre-warmed');
      return this.prewarmUnsubscribe;
    }

    // Create a real subscription with no-op callback to keep connection alive
    this.prewarmUnsubscribe = this.subscribe({
      callback: () => {
        // No-op callback - just keeps the connection alive for caching
      },
      throttleMs: 0, // No throttle for pre-warm
    });

    // Return cleanup function that clears internal state
    return () => {
      DevLogger.log('OICapStreamChannel: Cleaning up prewarm subscription');
      this.cleanupPrewarm();
    };
  }

  /**
   * Cleanup pre-warm subscription
   */
  public cleanupPrewarm(): void {
    if (this.prewarmUnsubscribe) {
      this.prewarmUnsubscribe();
      this.prewarmUnsubscribe = undefined;
    }
  }

  public clearCache(): void {
    // Cleanup pre-warm subscription
    this.cleanupPrewarm();
    // Call parent clearCache
    super.clearCache();
  }
}

// Top of book channel for best bid/ask data
class TopOfBookStreamChannel extends StreamChannel<
  { bestBid?: string; bestAsk?: string; spread?: string } | undefined
> {
  private currentSymbol: string | null = null;
  private cachedTopOfBook:
    | { bestBid?: string; bestAsk?: string; spread?: string }
    | undefined = undefined;

  protected connect() {
    if (!this.currentSymbol || this.wsSubscription) {
      return;
    }

    if (!this.ensureReady()) return;

    DevLogger.log(`TopOfBookStreamChannel: Subscribing to top of book`, {
      symbol: this.currentSymbol,
    });

    this.wsSubscription = Engine.context.PerpsController.subscribeToPrices({
      symbols: [this.currentSymbol],
      includeOrderBook: true,
      callback: (updates: PriceUpdate[]) => {
        const update = updates.find((u) => u.symbol === this.currentSymbol);
        if (update) {
          const topOfBook = {
            bestBid: update.bestBid,
            bestAsk: update.bestAsk,
            spread: update.spread,
          };
          this.cachedTopOfBook = topOfBook;
          this.notifySubscribers(topOfBook);
        }
      },
    });
  }

  protected getCachedData():
    | { bestBid?: string; bestAsk?: string; spread?: string }
    | undefined {
    return this.cachedTopOfBook;
  }

  protected getClearedData():
    | { bestBid?: string; bestAsk?: string; spread?: string }
    | undefined {
    return undefined;
  }

  public clearCache(): void {
    this.cachedTopOfBook = undefined;
    super.clearCache();
  }

  subscribeToSymbol(params: {
    symbol: string;
    callback: (
      orderBook:
        | { bestBid?: string; bestAsk?: string; spread?: string }
        | undefined,
    ) => void;
  }): () => void {
    if (this.currentSymbol && this.currentSymbol !== params.symbol) {
      DevLogger.log(
        'TopOfBookStreamChannel: Warning - different symbol requested, staying on current',
        {
          currentSymbol: this.currentSymbol,
          requestedSymbol: params.symbol,
        },
      );

      // Force disconnect to clear old symbol
      this.disconnect();

      // Set new symbol
      this.currentSymbol = params.symbol;
    } else if (!this.currentSymbol) {
      this.currentSymbol = params.symbol;
    }

    return this.subscribe({
      callback: params.callback,
    });
  }

  public disconnect() {
    this.currentSymbol = null;
    this.cachedTopOfBook = undefined;
    super.disconnect();
  }
}

// Market data channel for caching market list data
class MarketDataChannel extends StreamChannel<PerpsMarketData[]> {
  private lastFetchTime = 0;
  private fetchPromise: Promise<void> | null = null;
  private cachedProviderId: string | null = null;
  private readonly CACHE_DURATION =
    PERFORMANCE_CONFIG.MarketDataCacheDurationMs;

  protected connect() {
    if (!this.ensureReady()) return;

    // Check if connection manager is still connecting - retry later if so
    if (PerpsConnectionManager.isCurrentlyConnecting()) {
      this.deferConnect(PERPS_CONSTANTS.ConnectRetryDelayMs);
      return;
    }

    // Get current provider ID + network as a composite key.
    // Network changes (testnet toggle) must also invalidate the market cache.
    const controller = Engine.context.PerpsController;
    const currentProviderId =
      controller.state?.activeProvider || PROVIDER_CONFIG.DefaultProvider;
    // Note: uses state.isTestnet directly. If PROVIDER_CONFIG.MYX_TESTNET_ONLY is
    // ever re-enabled, this key would diverge from PerpsController.#providerIsTestnet().
    const currentNetworkKey = `${currentProviderId}:${controller.state?.isTestnet ? 'testnet' : 'mainnet'}`;

    // Invalidate cache if provider OR network changed
    if (this.cachedProviderId && this.cachedProviderId !== currentNetworkKey) {
      DevLogger.log(
        'PerpsStreamManager: Provider/network changed, invalidating cache',
        {
          from: this.cachedProviderId,
          to: currentNetworkKey,
        },
      );
      this.cache.delete('markets');
      this.lastFetchTime = 0;
    }

    // Fetch if cache is stale or empty
    const now = Date.now();
    const cached = this.cache.get('markets');
    const cacheAge = now - this.lastFetchTime;
    if (!cached || cacheAge > this.CACHE_DURATION) {
      DevLogger.log('PerpsStreamManager: Cache miss or stale', {
        hasCached: !!cached,
        cacheAgeMs: cached ? cacheAge : null,
        cacheExpired: cacheAge > this.CACHE_DURATION,
        cacheDurationMs: this.CACHE_DURATION,
        providerId: currentProviderId,
      });
      // Don't await - just trigger the fetch and handle errors
      this.fetchMarketData().catch((error) => {
        Logger.error(
          ensureError(error, 'PerpsStreamManager.fetchMarketData.background'),
          'PerpsStreamManager: Failed to fetch market data',
        );
      });
    } else {
      DevLogger.log('PerpsStreamManager: Using cached market data', {
        cacheAgeMs: cacheAge,
        cacheAgeSeconds: Math.round(cacheAge / 1000),
        marketCount: cached.length,
        cacheValidForMs: this.CACHE_DURATION - cacheAge,
        providerId: currentProviderId,
      });
      // Notify subscribers with cached data immediately
      this.notifySubscribers(cached);
    }
  }

  private async fetchMarketData(): Promise<void> {
    // Prevent concurrent fetches
    if (this.fetchPromise) {
      await this.fetchPromise;
      return;
    }

    this.fetchPromise = (async () => {
      const fetchStartTime = Date.now();
      try {
        const controller = Engine.context.PerpsController;

        // One-time read of controller-level preloaded cache (REST snapshot).
        // This avoids an HTTP round-trip when the controller already has fresh data.
        const controllerNetworkKey = `${controller.state?.activeProvider || PROVIDER_CONFIG.DefaultProvider}:${controller.state?.isTestnet ? 'testnet' : 'mainnet'}`;
        const cachedForProvider =
          controller.getCachedMarketDataForActiveProvider?.();
        if (
          cachedForProvider &&
          cachedForProvider.length > 0 &&
          (!this.cachedProviderId ||
            this.cachedProviderId === controllerNetworkKey)
        ) {
          DevLogger.log(
            'PerpsStreamManager: Using controller preloaded market data',
            {
              marketCount: cachedForProvider.length,
            },
          );
          this.cache.set('markets', cachedForProvider);
          this.lastFetchTime = Date.now();
          this.cachedProviderId = controllerNetworkKey;
          this.notifySubscribers(cachedForProvider);
          return;
        }

        DevLogger.log(
          'PerpsStreamManager: Fetching fresh market data from API',
        );

        // Snapshot provider + network BEFORE the async call to avoid race conditions.
        // If the user switches providers or toggles testnet while getMarketDataWithPrices()
        // is in-flight, we must not tag the returned data with the new network key.
        const preFetchNetworkKey = `${controller.state?.activeProvider || PROVIDER_CONFIG.DefaultProvider}:${controller.state?.isTestnet ? 'testnet' : 'mainnet'}`;

        const data = await controller.getMarketDataWithPrices();
        const fetchTime = Date.now() - fetchStartTime;

        // If provider or network changed during fetch, discard stale data
        const postFetchNetworkKey = `${controller.state?.activeProvider || PROVIDER_CONFIG.DefaultProvider}:${controller.state?.isTestnet ? 'testnet' : 'mainnet'}`;
        if (preFetchNetworkKey !== postFetchNetworkKey) {
          DevLogger.log(
            'PerpsStreamManager: Provider/network changed during fetch, discarding data',
            {
              fetchedFor: preFetchNetworkKey,
              current: postFetchNetworkKey,
            },
          );
          return;
        }

        // Update cache and track which provider+network this data came from
        this.cache.set('markets', data);
        this.lastFetchTime = Date.now();
        this.cachedProviderId = preFetchNetworkKey;

        // Notify all subscribers
        this.notifySubscribers(data);

        DevLogger.log('PerpsStreamManager: Market data fetched and cached', {
          marketCount: data.length,
          fetchTimeMs: fetchTime,
          providerId: this.cachedProviderId,
          cacheValidUntil: new Date(
            Date.now() + this.CACHE_DURATION,
          ).toISOString(),
        });
      } catch (error) {
        const fetchTime = Date.now() - fetchStartTime;
        const existing = this.cache.get('markets');
        const cacheStalenessMs = this.lastFetchTime
          ? Date.now() - this.lastFetchTime
          : null;
        Logger.error(ensureError(error, 'PerpsStreamManager.fetchMarketData'), {
          tags: {
            feature: PERPS_CONSTANTS.FeatureName,
          },
          context: {
            name: 'PerpsStreamManager',
            data: {
              method: 'fetchMarketData',
              fetchTimeMs: fetchTime,
              hadCachedData: !!existing,
              cachedMarketCount: existing?.length ?? 0,
              cacheStalenessMs,
            },
          },
        });
        // Keep existing cache if fetch fails
        if (existing) {
          DevLogger.log(
            'PerpsStreamManager: Using stale cache after fetch failure',
            {
              marketCount: existing.length,
            },
          );
          this.notifySubscribers(existing);
        }
      } finally {
        this.fetchPromise = null;
      }
    })();

    await this.fetchPromise;
  }

  /**
   * Force refresh market data
   */
  public async refresh(): Promise<void> {
    this.lastFetchTime = 0; // Force cache to be considered stale
    await this.fetchMarketData();
  }

  protected getCachedData(): PerpsMarketData[] | null {
    // Return channel cache if available (from previous fetch)
    const cached = this.cache.get('markets');
    if (cached !== undefined) return cached;

    // Fallback: read per-provider cache via helper
    const controller = Engine.context.PerpsController;
    return controller.getCachedMarketDataForActiveProvider?.() ?? null;
  }

  protected getClearedData(): PerpsMarketData[] {
    return [];
  }

  /**
   * Prewarm market data cache
   * @returns Cleanup function (no-op for REST data)
   */
  public prewarm(): () => void {
    // Fetch data immediately to populate cache
    this.fetchMarketData().catch((error) => {
      Logger.error(ensureError(error, 'MarketDataChannel.prewarm'), {
        context: 'MarketDataChannel.prewarm',
      });
    });

    // No cleanup needed for REST data
    return () => {
      // No-op
    };
  }

  /**
   * Clear cache and reset fetch time.
   *
   * @param skipNotify - When true, subscribers keep their current data (used on
   * account switches where market data is global and stays valid). When false
   * (default), subscribers are notified with `[]` and their throttle state is
   * reset so the next fetch result is delivered immediately.
   */
  public clearCache(skipNotify = false): void {
    this.cache.clear();
    this.lastFetchTime = 0;
    this.fetchPromise = null;
    this.cachedProviderId = null;

    this.subscribers.forEach((subscriber) => {
      if (subscriber.timer) {
        clearTimeout(subscriber.timer);
        subscriber.timer = undefined;
      }
      subscriber.pendingUpdate = undefined;

      if (!skipNotify) {
        subscriber.hasReceivedFirstUpdate = false;
        subscriber.callback([]);
      }
    });
  }
}

// Main manager class
export class PerpsStreamManager {
  public readonly prices = new PriceStreamChannel();
  public readonly orders = new OrderStreamChannel();
  public readonly positions = new PositionStreamChannel();
  public readonly fills = new FillStreamChannel();
  public readonly account = new AccountStreamChannel();
  public readonly marketData = new MarketDataChannel();
  public readonly oiCaps = new OICapStreamChannel();
  public readonly topOfBook = new TopOfBookStreamChannel();
  public readonly candles = new CandleStreamChannel(
    () => PerpsConnectionManager.getConnectionState().isInitialized,
  );

  // Future channels can be added here:
  // public readonly funding = new FundingStreamChannel();
  // public readonly trades = new TradeStreamChannel();

  /**
   * Force reconnection of all stream channels after WebSocket reconnection
   * Disconnects all channels and reconnects those with active subscribers
   */
  public clearAllChannels(): void {
    // Reconnect all channels - clears dead subscriptions and re-establishes
    // connections for channels that have active subscribers
    this.prices.reconnect();
    this.orders.reconnect();
    this.positions.reconnect();
    this.fills.reconnect();
    this.account.reconnect();
    this.marketData.reconnect();
    this.oiCaps.reconnect();
    this.topOfBook.reconnect();
    this.candles.reconnect();
  }
}

// Singleton instance
const streamManager = new PerpsStreamManager();

// Export singleton for pre-warming in PerpsConnectionManager
export const getStreamManagerInstance = () => streamManager;

// Context
const PerpsStreamContext = createContext<PerpsStreamManager | null>(null);

export const PerpsStreamProvider: React.FC<{
  children: React.ReactNode;
  testStreamManager?: PerpsStreamManager; // Only for testing
}> = ({ children, testStreamManager }) => {
  // Check for E2E mock stream manager
  const e2eMockStreamManager =
    getE2EMockStreamManager() as PerpsStreamManager | null;

  const selectedManager: PerpsStreamManager =
    testStreamManager || e2eMockStreamManager || streamManager;

  DevLogger.log('PerpsStreamProvider: Using stream manager:', {
    isTestManager: !!testStreamManager,
    isE2EMockManager: !!e2eMockStreamManager,
    isRealManager: selectedManager === streamManager,
  });

  return (
    <PerpsStreamContext.Provider value={selectedManager}>
      {children}
    </PerpsStreamContext.Provider>
  );
};

export const usePerpsStream = () => {
  const context = useContext(PerpsStreamContext);
  if (!context) {
    throw new Error('usePerpsStream must be used within PerpsStreamProvider');
  }
  return context;
};

// Type that only includes channel properties (excludes methods like clearAllChannels)
export type PerpsStreamChannelKey = {
  [K in keyof PerpsStreamManager]: PerpsStreamManager[K] extends {
    pause(): void;
  }
    ? K
    : never;
}[keyof PerpsStreamManager];

// Types are exported from controllers/types