---

import { BehaviorSubject, combineLatest, Observable, Subject, Subscription } from 'rxjs';
import { map } from 'rxjs/operators';

import { EventManager } from '@core/EventManager';
import { SymbolSource } from '@src/core/SymbolSource';
import { Candle } from '@src/types';
import { Timeframes } from '@src/types/timeframes';
import { normalizeSymbol } from '@src/utils';

export interface DataSourceParams {
  getData: (timeframe: Timeframes, symbol: string, until?: Candle) => Promise<Candle[] | null>;
  eventManager: EventManager;
}

export interface RealtimeEvent {
  symbol: string;
  candle: Candle;
}

const EMPTY_CANDLES$ = new BehaviorSubject<Candle[]>([]);
const EMPTY_LAST_CANDLE$ = new BehaviorSubject<Candle | null>(null);
const EMPTY_REALTIME$ = new Subject<Candle>();

export class DataSource {
  private readonly getData: DataSourceParams['getData'];
  private readonly eventManager: EventManager;

  private readonly states = new Map<string, SymbolSource>();
  private readonly activeSymbols$ = new BehaviorSubject<Set<string>>(new Set());

  private realtimeSub: Subscription | null = null;
  private readonly subscriptions = new Subscription();

  constructor({ getData, eventManager }: DataSourceParams) {
    this.getData = getData;
    this.eventManager = eventManager;
    this.initSubscriptions();
  }

  public setSymbols(symbols: string[]): void {
    const nextSet = new Set<string>();

    for (const raw of symbols) {
      const s = normalizeSymbol(raw);
      if (s) nextSet.add(s);
    }

    for (const s of nextSet) {
      if (!this.states.has(s)) {
        this.ensureState(s);
      }
    }

    for (const existingKey of this.states.keys()) {
      if (!nextSet.has(existingKey)) {
        this.dropState(existingKey);
      }
    }

    this.activeSymbols$.next(nextSet);
  }

  public symbolsObs(): Observable<string[]> {
    return this.activeSymbols$.pipe(map((set) => Array.from(set)));
  }

  public bindRealtime(stream$: Observable<RealtimeEvent>): void {
    this.unbindRealtime();

    this.realtimeSub = stream$.subscribe(({ symbol, candle }) => {
      const s = normalizeSymbol(symbol);
      if (s && this.activeSymbols$.value.has(s)) {
        this.states.get(s)?.pushRealtime(candle);
      }
    });
  }

  public unbindRealtime(): void {
    if (this.realtimeSub) {
      this.realtimeSub.unsubscribe();
      this.realtimeSub = null;
    }
  }

  public subscribe(symbolRaw: string, cb: (next: Candle[]) => void): Subscription {
    const symbol = normalizeSymbol(symbolRaw);
    if (!symbol) return new Subscription();
    return this.ensureState(symbol).data$().subscribe(cb);
  }

  public subscribeRealtime(symbolRaw: string, cb: (next: Candle) => void): Subscription {
    const symbol = normalizeSymbol(symbolRaw);
    if (!symbol) return new Subscription();
    return this.ensureState(symbol).realtime$().subscribe(cb);
  }

  public data$(symbolRaw: string): Observable<Candle[]> {
    const symbol = normalizeSymbol(symbolRaw);
    return symbol ? this.ensureState(symbol).data$() : EMPTY_CANDLES$.asObservable();
  }

  public realtime$(symbolRaw: string): Observable<Candle> {
    const symbol = normalizeSymbol(symbolRaw);
    return symbol ? this.ensureState(symbol).realtime$() : EMPTY_REALTIME$.asObservable();
  }

  public lastCandle$(symbolRaw: string): Observable<Candle | null> {
    const symbol = normalizeSymbol(symbolRaw);
    return symbol ? this.ensureState(symbol).lastCandle$() : EMPTY_LAST_CANDLE$.asObservable();
  }

  public getLastCandle(symbolRaw: string): Candle | null {
    const symbol = normalizeSymbol(symbolRaw);
    return symbol ? (this.states.get(symbol)?.getLastValue() ?? null) : null;
  }

  public waitUntilReady = async (symbolRaw: string): Promise<void> => {
    const symbol = normalizeSymbol(symbolRaw);

    if (!symbol) {
      return;
    }

    await this.ensureState(symbol).waitUntilReady();
  };

  public updateRealtime(symbolRaw: string, next: Candle): void {
    const symbol = normalizeSymbol(symbolRaw);
    if (symbol && this.activeSymbols$.value.has(symbol)) {
      this.ensureState(symbol).pushRealtime(next);
    }
  }

  public async loadTill(symbolRaw: string, time: number): Promise<void> {
    const symbol = normalizeSymbol(symbolRaw);
    if (symbol && this.activeSymbols$.value.has(symbol)) {
      await this.ensureState(symbol).loadTill(time);
    }
  }

  public loadMoreHistory = async (symbolRaw: string): Promise<void> => {
    const symbol = normalizeSymbol(symbolRaw);
    if (symbol && this.activeSymbols$.value.has(symbol)) {
      await this.ensureState(symbol).loadMoreHistory();
    }
  };

  public loadAllHistory = async (symbolRaw: string): Promise<void> => {
    const symbol = normalizeSymbol(symbolRaw);
    if (!symbol || !this.activeSymbols$.value.has(symbol)) return;

    const st = this.ensureState(symbol);
    await st.loadAllHistory();
  };

  public getIsLoading(symbolRaw: string): boolean {
    const symbol = normalizeSymbol(symbolRaw);
    return symbol ? (this.states.get(symbol)?.isLoadingValue() ?? false) : false;
  }

  public getOldestTime(symbolRaw: string): number | null {
    const symbol = normalizeSymbol(symbolRaw);
    return symbol ? (this.states.get(symbol)?.getOldestTime() ?? null) : null;
  }

  public destroy(): void {
    this.unbindRealtime();
    this.subscriptions.unsubscribe();

    for (const key of this.states.keys()) {
      this.dropState(key);
    }

    this.activeSymbols$.complete();
  }

  private initSubscriptions(): void {
    this.subscriptions.add(
      combineLatest([this.eventManager.getSelectedSeries()]).subscribe(() => {
        this.states.forEach((st) => st.saveRealtimeCache());
      }),
    );

    this.subscriptions.add(
      this.eventManager.timeframe().subscribe((tf) => {
        const symbols = Array.from(this.activeSymbols$.value);

        Promise.all(
          symbols.map((s) => {
            const st = this.states.get(s);
            return st ? st.reload(tf) : Promise.resolve();
          }),
        ).catch((error) => console.error('[DataSource] Global timeframe reload error:', error));
      }),
    );
  }

  private ensureState(symbol: string): SymbolSource {
    let st = this.states.get(symbol);

    if (!st) {
      st = new SymbolSource({
        symbol,
        getData: this.getData,
        getTimeframe: () => this.eventManager.getTimeframe(),
      });

      this.states.set(symbol, st);
      st.init();
    }

    return st;
  }

  private dropState(symbol: string): void {
    const st = this.states.get(symbol);

    if (st) {
      st.destroy();
      this.states.delete(symbol);
    }
  }
}


import dayjs from 'dayjs';

import { BehaviorSubject, Observable, Subject } from 'rxjs';

import { Candle } from '@src/types';
import { Timeframes } from '@src/types/timeframes';
import { normalizeSeriesData, parseTimeframe } from '@src/utils';

const FUTURE_WHITESPACES = 5000;

export interface SymbolSourceParams {
  symbol: string;
  getData: (timeframe: Timeframes, symbol: string, until?: Candle) => Promise<Candle[] | null>;
  getTimeframe: () => Timeframes;
}

export class SymbolSource {
  public readonly symbol: string;

  private readonly getData: SymbolSourceParams['getData'];
  private readonly getTimeframe: SymbolSourceParams['getTimeframe'];

  private readonly currentDataSubject = new BehaviorSubject<Candle[]>([]);
  private readonly realtimeSubject = new Subject<Candle>();
  private readonly lastCandleSubject = new BehaviorSubject<Candle | null>(null);
  private readonly isLoadingSubject = new BehaviorSubject<boolean>(false);
  private readonly isInitializedSubject = new BehaviorSubject<boolean>(false);

  private realtimeCache: Candle[] = [];
  private realtimeBuffer: Candle[] = [];

  private oldestCandle: Candle | null = null;
  private newestCandle: Candle | null = null;

  private loadSeq = 0;
  private loadingPromise: Promise<void> | null = null;

  private isEndOfData = false;

  constructor({ symbol, getData, getTimeframe }: SymbolSourceParams) {
    this.symbol = symbol;
    this.getData = getData;
    this.getTimeframe = getTimeframe;
  }

  public init(): void {
    this.reload(this.getTimeframe());
  }

  public data$(): Observable<Candle[]> {
    return this.currentDataSubject.asObservable();
  }

  public realtime$(): Observable<Candle> {
    return this.realtimeSubject.asObservable();
  }

  public lastCandle$(): Observable<Candle | null> {
    return this.lastCandleSubject.asObservable();
  }

  public isInitialized$(): Observable<boolean> {
    return this.isInitializedSubject.asObservable();
  }

  public async waitUntilReady(): Promise<void> {
    while (!this.isInitializedSubject.value) {
      const loadingPromise = this.loadingPromise;

      if (!loadingPromise) {
        return;
      }

      await loadingPromise;
    }
  }

  public isLoadingValue(): boolean {
    return this.isLoadingSubject.value;
  }

  public getOldestTime(): number | null {
    return this.oldestCandle?.time ?? null;
  }

  public getLastValue(): Candle | null {
    return this.lastCandleSubject.value;
  }

  public destroy(): void {
    this.loadSeq += 1;
    this.loadingPromise = null;

    this.currentDataSubject.complete();
    this.realtimeSubject.complete();
    this.lastCandleSubject.complete();
    this.isLoadingSubject.complete();
    this.isInitializedSubject.complete();

    this.realtimeCache = [];
    this.realtimeBuffer = [];
    this.oldestCandle = null;
    this.newestCandle = null;
  }

  public pushRealtime(next: Candle): void {
    const candle = this.normalizeCandle(next);

    this.realtimeCache = this.normalizeList(this.realtimeCache.concat(candle));

    if (!this.newestCandle || candle.time >= this.newestCandle.time) {
      this.newestCandle = candle;
      this.lastCandleSubject.next(candle);
    }

    if (!this.isInitializedSubject.value) {
      this.realtimeBuffer.push(candle);
      return;
    }

    this.realtimeSubject.next(candle);
  }

  public saveRealtimeCache(): void {
    if (this.realtimeCache.length === 0) return;

    const timeframe = this.getTimeframe();

    const current = this.currentDataSubject.value.filter((candle) => candle.open !== undefined);
    const next = this.normalizeList(current.concat(this.realtimeCache));

    this.newestCandle = next[next.length - 1] ?? null;
    this.realtimeCache = [];

    this.currentDataSubject.next(applyWhitespacesToFuture(next, timeframe));
    this.lastCandleSubject.next(this.newestCandle);
  }

  public async loadMoreHistory(): Promise<void> {
    await this.waitUntilReady();

    if (this.isEndOfData) return;

    if (this.loadingPromise) {
      return this.loadingPromise;
    }

    this.loadSeq += 1;

    if (!this.oldestCandle) return;

    const timeframe = this.getTimeframe();

    const task = (async () => {
      this.isLoadingSubject.next(true);

      try {
        if (!this.oldestCandle) return;

        const olderData = await this.getData(timeframe, this.symbol, this.oldestCandle);

        if (olderData === null) {
          this.isEndOfData = true;
          return;
        }

        const older = this.normalizeList(olderData);

        if (older.length === 0) return;

        const current = this.currentDataSubject.value.filter((candle) => candle.open !== undefined);
        const combined = this.normalizeList(older.concat(current));

        const nextOldest = combined[0] ?? null;

        if (!nextOldest || !this.oldestCandle || nextOldest.time >= this.oldestCandle.time) {
          return;
        }

        this.oldestCandle = nextOldest;
        this.newestCandle = combined[combined.length - 1] ?? this.newestCandle;

        this.currentDataSubject.next(applyWhitespacesToFuture(combined, timeframe));
        this.lastCandleSubject.next(this.newestCandle);

        this.saveRealtimeCache();
      } catch (error) {
        console.error('[DataSource] Ошибка при догрузке истории:', error);
      } finally {
        this.isLoadingSubject.next(false);
      }
    })();

    this.loadingPromise = task;

    task.finally(() => {
      if (this.loadingPromise === task) {
        this.loadingPromise = null;
      }
    });

    await task;
  }

  public async loadAllHistory(): Promise<void> {
    await this.waitUntilReady();

    if (this.isEndOfData) return;

    if (this.loadingPromise) {
      await this.loadingPromise;
    }

    const seq = ++this.loadSeq;
    const timeframe = this.getTimeframe();

    const task = (async () => {
      this.isLoadingSubject.next(true);

      try {
        let current = this.currentDataSubject.value.filter((candle) => candle.open !== undefined);
        let oldest = current[0] ?? null;

        if (!oldest) return;

        const seenOldestTimes = new Set<number>();

        while (oldest) {
          if (seq !== this.loadSeq) return;

          if (seenOldestTimes.has(oldest.time)) {
            break;
          }

          seenOldestTimes.add(oldest.time);

          // eslint-disable-next-line no-await-in-loop
          const olderData = await this.getData(timeframe, this.symbol, oldest);

          if (seq !== this.loadSeq) return;

          if (olderData === null) {
            this.isEndOfData = true;
            break;
          }

          const older = this.normalizeList(olderData);

          if (older.length === 0) break;

          const combined = this.normalizeList(older.concat(current));
          const nextOldest = combined[0] ?? null;

          if (!nextOldest || nextOldest.time >= oldest.time) {
            break;
          }

          oldest = nextOldest;
          current = combined;
        }

        this.oldestCandle = current[0] ?? null;
        this.newestCandle = current[current.length - 1] ?? null;

        this.currentDataSubject.next(applyWhitespacesToFuture(current, timeframe));
        this.lastCandleSubject.next(this.newestCandle);

        this.saveRealtimeCache();
      } catch (error) {
        console.error('[DataSource] Ошибка при полной загрузке истории:', error);
      } finally {
        if (seq === this.loadSeq) {
          this.isLoadingSubject.next(false);
        }
      }
    })();

    this.loadingPromise = task;

    task.finally(() => {
      if (this.loadingPromise === task) {
        this.loadingPromise = null;
      }
    });

    await task;
  }

  public async loadTill(time: number): Promise<void> {
    await this.waitUntilReady();

    while (this.oldestCandle && this.oldestCandle.time >= time) {
      const before = this.oldestCandle.time;

      // eslint-disable-next-line no-await-in-loop
      await this.loadMoreHistory();

      if (!this.oldestCandle) break;
      if (this.oldestCandle.time === before) break;
    }
  }

  public async reload(timeframe: Timeframes): Promise<void> {
    this.loadSeq += 1;

    const seq = this.loadSeq;

    this.isInitializedSubject.next(false);
    this.isLoadingSubject.next(true);

    this.currentDataSubject.next([]);
    this.realtimeCache = [];
    this.realtimeBuffer = [];
    this.oldestCandle = null;
    this.newestCandle = null;
    this.isEndOfData = false;
    this.lastCandleSubject.next(null);

    const task = (async () => {
      try {
        const loaded = (await this.getData(timeframe, this.symbol)) ?? [];

        if (seq !== this.loadSeq) return;

        const normalized = this.normalizeList(loaded);

        this.oldestCandle = normalized[0] ?? null;
        this.newestCandle = normalized[normalized.length - 1] ?? null;

        this.currentDataSubject.next(applyWhitespacesToFuture(normalized, timeframe));
        this.lastCandleSubject.next(this.newestCandle);

        this.isInitializedSubject.next(true);
        this.flushRealtimeBuffer();
      } catch (error) {
        console.error('[DataSource] Ошибка при загрузке данных:', error);
      } finally {
        if (seq === this.loadSeq) {
          this.isLoadingSubject.next(false);
        }
      }
    })();

    this.loadingPromise = task;

    task.finally(() => {
      if (this.loadingPromise === task) {
        this.loadingPromise = null;
      }
    });

    await task;
  }

  private flushRealtimeBuffer(): void {
    if (this.realtimeBuffer.length === 0) return;

    const newestTime = this.newestCandle?.time ?? Number.NEGATIVE_INFINITY;

    const bufferedCandles = this.normalizeList(this.realtimeBuffer.filter((candle) => candle.time >= newestTime));

    for (const candle of bufferedCandles) {
      this.realtimeSubject.next(candle);
    }

    this.realtimeBuffer = [];
  }

  private normalizeCandle(candle: Candle): Candle {
    const time = candle.time > 1e10 ? Math.floor(candle.time / 1000) : Math.floor(candle.time);

    if (time === candle.time) {
      return candle;
    }

    return {
      ...candle,
      time,
    };
  }

  private normalizeList(candles: Candle[]): Candle[] {
    return normalizeSeriesData(candles.map((candle) => this.normalizeCandle(candle)));
  }
}

export function applyWhitespacesToFuture(data: Candle[], timeframe: Timeframes): Candle[] {
  const lastCandle = data[data.length - 1];

  if (!lastCandle) {
    return data;
  }

  const { candleWidth, dayjsUnit } = parseTimeframe(timeframe);
  const startTime = dayjs.unix(lastCandle.time).utc();
  const result = data.slice();

  for (let index = 1; index <= FUTURE_WHITESPACES; index += 1) {
    result.push({
      time: startTime.add(candleWidth * index, dayjsUnit).unix(),
    } as Candle);
  }

  return result;
}