package com.goldenchart.websocket;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.goldenchart.dto.CandleBarDto;
import com.goldenchart.storage.Ta4jStorage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.socket.messaging.SessionSubscribeEvent;
import org.springframework.web.socket.messaging.SessionUnsubscribeEvent;
import org.ta4j.core.Bar;
import org.ta4j.core.BarSeries;
import org.ta4j.core.BaseBarSeriesBuilder;
import org.ta4j.core.bars.TimeBarBuilderFactory;
import org.ta4j.core.num.DoubleNumFactory;

import java.time.Instant;

import java.time.Duration;
import java.time.ZonedDateTime;
import java.time.ZoneOffset;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * 동적 STOMP 구독 관리자 (명세서 3.3).
 *
 * 프론트엔드가 /sub/charts/{market}/{type} 채널을 구독하면:
 *  1. Ta4jStorage 에 해당 종목 BarSeries 가 없으면 업비트 REST API Warm-up
 *  2. UpbitWebSocketClient 구독 목록에 market 추가
 *
 * 구독자 수가 0이 되면:
 *  1. UpbitWebSocketClient 에서 market 제거
 *  2. Ta4jStorage 에서 BarSeries 삭제 (메모리 방어)
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class DynamicSubscriptionManager {

    // /sub/charts/{market}/{type} 패턴 파서
    private static final Pattern TOPIC_PATTERN =
            Pattern.compile("^/sub/charts/([^/]+)/([^/]+)$");

    private final Ta4jStorage ta4jStorage;
    private final UpbitWebSocketClient upbitWsClient;
    private final WebClient.Builder webClientBuilder;
    private final ObjectMapper objectMapper;

    @Value("${upbit.api.base-url:https://api.upbit.com}")
    private String upbitBaseUrl;

    @Value("${upbit.api.candle-limit:200}")
    private int candlePageSize;

    /** EMA 수렴을 위한 warm-up 목표 봉 수 (페이지네이션으로 달성) */
    @Value("${upbit.api.warmup-count:600}")
    private int warmupTargetCount;

    @Value("${upbit.api.request-delay-ms:110}")
    private long requestDelayMs;

    /** 구독자 수 추적: "KRW-BTC:1m" → count */
    private final ConcurrentHashMap<String, AtomicInteger> subscriberCount =
            new ConcurrentHashMap<>();

    /** 워밍업 중인 키 추적 (중복 요청 방지) */
    private final ConcurrentHashMap<String, Boolean> warmingUp =
            new ConcurrentHashMap<>();

    /** 워밍업 전용 단일 스레드 — 업비트 API 레이트리밋 보호용 직렬 실행 */
    private final ExecutorService warmUpExecutor = Executors.newSingleThreadExecutor(r -> {
        Thread t = new Thread(r, "warmup-thread");
        t.setDaemon(true);
        return t;
    });

    // ── STOMP 구독/구독해제 이벤트 리스너 ────────────────────────────────────

    @EventListener
    public void handleSubscribe(SessionSubscribeEvent event) {
        String dest = getDestination(event.getMessage().getHeaders());
        if (dest == null) return;
        Matcher m = TOPIC_PATTERN.matcher(dest);
        if (!m.matches()) return;

        String market     = m.group(1);
        String candleType = m.group(2);
        String key        = market + ":" + candleType;

        int count = subscriberCount.computeIfAbsent(key, k -> new AtomicInteger(0))
                .incrementAndGet();
        log.info("[DynamicSubscriptionManager] 구독: {} (총 {}명)", key, count);

        // Warm-up: 메모리에 없고 현재 워밍업 중이 아니면 비동기 초기화
        if (!ta4jStorage.exists(market, candleType)
                && warmingUp.putIfAbsent(key, Boolean.TRUE) == null) {
            warmUpExecutor.submit(() -> {
                try {
                    warmUp(market, candleType);
                } finally {
                    warmingUp.remove(key);
                }
            });
        }
        // UpbitWebSocket 구독 추가
        upbitWsClient.addMarket(market);
        if ("1m".equals(candleType)) {
            upbitWsClient.resubscribeAll();
        }
    }

    @EventListener
    public void handleUnsubscribe(SessionUnsubscribeEvent event) {
        // unsubscribe 이벤트는 구독 ID 기반이므로 별도 매핑 필요 (간소화: 0 도달 시 GC)
        // 실제 구현에서는 sessionId + subscriptionId → topic 역매핑 테이블 필요
        // 여기서는 구독 수 감소만 처리
        String dest = getDestination(event.getMessage().getHeaders());
        if (dest == null) return;
        Matcher m = TOPIC_PATTERN.matcher(dest);
        if (!m.matches()) return;

        String market     = m.group(1);
        String candleType = m.group(2);
        String key        = market + ":" + candleType;

        AtomicInteger counter = subscriberCount.get(key);
        if (counter == null) return;

        int remaining = counter.decrementAndGet();
        log.info("[DynamicSubscriptionManager] 구독 해제: {} (남은 {}명)", key, remaining);

        if (remaining <= 0) {
            subscriberCount.remove(key);
            upbitWsClient.removeMarket(market);
            ta4jStorage.remove(market, candleType);
            log.info("[DynamicSubscriptionManager] GC: {} 메모리 해제", key);
        }
    }

    /**
     * live-conditions API 등 — 최소 봉 수 확보까지 동기 warm-up (종목별 독립).
     */
    public void ensureBarsReadySync(String market, String candleType, int minBars) {
        upbitWsClient.addMarket(market);
        if (ta4jStorage.getBarCount(market, candleType) >= minBars) {
            return;
        }
        String key = market + ":" + candleType;
        if (warmingUp.putIfAbsent(key, Boolean.TRUE) == null) {
            try {
                warmUp(market, candleType);
            } finally {
                warmingUp.remove(key);
            }
        } else {
            // 다른 스레드 warm-up 중 — 짧게 대기 후 재확인
            for (int i = 0; i < 30 && ta4jStorage.getBarCount(market, candleType) < minBars; i++) {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }

    /**
     * 실시간 전략 체크 등 STOMP 클라이언트 없이도 백엔드가 틱·캔들을 수집하도록 마켓을 고정 구독한다.
     */
    public void ensureMarketPinned(String market, String candleType) {
        String key = market + ":" + candleType;
        if (!ta4jStorage.exists(market, candleType)
                && warmingUp.putIfAbsent(key, Boolean.TRUE) == null) {
            warmUpExecutor.submit(() -> {
                try {
                    warmUp(market, candleType);
                } finally {
                    warmingUp.remove(key);
                }
            });
        }
        upbitWsClient.addMarket(market);
        log.info("[DynamicSubscriptionManager] 실시간 전략용 마켓 고정: {}:{}", market, candleType);
    }

    // ── Warm-up ────────────────────────────────────────────────────────────────

    /**
     * 업비트 REST API 로 최근 N개 캔들을 페이지네이션으로 가져와 Ta4jStorage 를 초기화한다.
     * EMA(200) 수렴에는 약 1000봉이 필요하므로 warmupTargetCount(기본 600)봉을 목표로 한다.
     */
    private void warmUp(String market, String candleType) {
        String path = Ta4jStorage.CANDLE_TYPE_MAP.getOrDefault(candleType, "minutes/1");
        Duration duration = Ta4jStorage.parseDuration(candleType);
        List<Map<String, Object>> allRaw = new ArrayList<>();
        String toParam = null;  // null = 최신부터
        int fetched = 0;

        try {
            while (fetched < warmupTargetCount) {
                int pageSize = Math.min(candlePageSize, warmupTargetCount - fetched);
                String url = upbitBaseUrl + "/v1/candles/" + path
                        + "?market=" + market + "&count=" + pageSize
                        + (toParam != null ? "&to=" + toParam : "");

                Thread.sleep(requestDelayMs);
                String json = webClientBuilder.build()
                        .get().uri(url)
                        .retrieve().bodyToMono(String.class).block();
                if (json == null || json.isBlank()) break;

                List<Map<String, Object>> page =
                        objectMapper.readValue(json, new TypeReference<>() {});
                if (page.isEmpty()) break;

                allRaw.addAll(page);
                fetched += page.size();

                // 다음 페이지: 가장 오래된 봉 시각보다 1초 이전을 to 로 설정.
                // 업비트 to 파라미터는 exclusive 이지만, 동일 시각 중복 방지를 위해 1초 뺀다.
                String oldest = (String) page.get(page.size() - 1).get("candle_date_time_utc");
                if (oldest == null) break;
                long oldestEpoch = parseEpoch(oldest);
                toParam = Instant.ofEpochSecond(oldestEpoch - 1)
                        .atZone(ZoneOffset.UTC)
                        .format(java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss"))
                        + "Z";

                if (page.size() < pageSize) break;  // 더 이상 데이터 없음
            }

            // 업비트 응답은 최신→과거 순 → 오름차순 정렬 후 Ta4jStorage 에 적재
            allRaw.sort(Comparator.comparingLong(r -> parseEpoch((String) r.get("candle_date_time_utc"))));

            for (Map<String, Object> r : allRaw) {
                long epochSec = parseEpoch((String) r.get("candle_date_time_utc"));
                Instant endInstant = Instant.ofEpochSecond(epochSec).plus(duration);
                try {
                    BarSeries tmp = new BaseBarSeriesBuilder()
                            .withBarBuilderFactory(new TimeBarBuilderFactory())
                            .withNumFactory(DoubleNumFactory.getInstance()).build();
                    Bar bar = tmp.barBuilder()
                            .timePeriod(duration).endTime(endInstant)
                            .openPrice(toDouble(r.get("opening_price")))
                            .highPrice(toDouble(r.get("high_price")))
                            .lowPrice(toDouble(r.get("low_price")))
                            .closePrice(toDouble(r.get("trade_price")))
                            .volume(toDouble(r.get("candle_acc_trade_volume")))
                            .build();
                    ta4jStorage.addBar(market, candleType, bar);
                } catch (Exception e) {
                    log.trace("[DynamicSubscriptionManager] warmUp bar 스킵: {}", e.getMessage());
                }
            }
            log.info("[DynamicSubscriptionManager] Warm-up 완료: {} / {} ({} bars)", market, candleType, allRaw.size());
        } catch (Exception e) {
            log.error("[DynamicSubscriptionManager] Warm-up 실패 {}/{}: {}", market, candleType, e.getMessage());
        }
    }

    // ── 유틸 ──────────────────────────────────────────────────────────────────

    private String getDestination(org.springframework.messaging.MessageHeaders headers) {
        Object dest = headers.get("simpDestination");
        return dest instanceof String s ? s : null;
    }

    private static long parseEpoch(String utcStr) {
        if (utcStr == null) return 0L;
        return ZonedDateTime.parse(utcStr + "Z",
                java.time.format.DateTimeFormatter.ISO_DATE_TIME).toEpochSecond();
    }

    private static double toDouble(Object v) {
        return v == null ? 0.0 : ((Number) v).doubleValue();
    }
}