package gc.mda.signal_batch.global.websocket.service;

import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * WebSocket 세션별 진행 중인 쿼리 추적 및 글로벌 동시 쿼리 제한 관리
 * Phase 5: 대기열 기반 관리 (거부 대신 순번 대기)
 */
@Slf4j
@Service
public class ActiveQueryManager {

    // 세션ID -> 활성 쿼리 정보 매핑
    private final Map<String, ActiveQuery> activeQueries = new ConcurrentHashMap<>();

    // 글로벌 동시 쿼리 제한 (Semaphore, fair=true로 FIFO 대기)
    private Semaphore globalQuerySemaphore;
    private final AtomicInteger waitingCount = new AtomicInteger(0);

    // 대기열 추적 (FIFO 순서 보장)
    private final ConcurrentLinkedQueue<String> waitingQueue = new ConcurrentLinkedQueue<>();

    @Value("${websocket.query.max-concurrent-global:60}")
    private int maxConcurrentGlobal;

    @Value("${websocket.query.queue-timeout-seconds:30}")
    private int queueTimeoutSeconds;

    @Value("${websocket.memory.heap-reject-threshold:0.85}")
    private double heapRejectThreshold;

    @PostConstruct
    public void init() {
        this.globalQuerySemaphore = new Semaphore(maxConcurrentGlobal, true);
        log.info("Global query semaphore initialized: maxConcurrent={}, queueTimeout={}s",
                maxConcurrentGlobal, queueTimeoutSeconds);
    }

    /**
     * 활성 쿼리 정보
     */
    public static class ActiveQuery {
        private final String sessionId;
        private final String queryId;
        private final LocalDateTime startTime;
        private final AtomicBoolean cancelled = new AtomicBoolean(false);
        private volatile String currentTable;
        private volatile int processedChunks;

        public ActiveQuery(String sessionId, String queryId) {
            this.sessionId = sessionId;
            this.queryId = queryId;
            this.startTime = LocalDateTime.now();
        }

        public boolean isCancelled() {
            return cancelled.get();
        }

        public void cancel() {
            cancelled.set(true);
        }

        public void updateProgress(String tableName, int chunks) {
            this.currentTable = tableName;
            this.processedChunks = chunks;
        }

        public String getSessionId() {
            return sessionId;
        }

        public String getQueryId() {
            return queryId;
        }

        public LocalDateTime getStartTime() {
            return startTime;
        }

        public String getCurrentTable() {
            return currentTable;
        }

        public int getProcessedChunks() {
            return processedChunks;
        }
    }

    /**
     * 새 쿼리 등록
     */
    public ActiveQuery registerQuery(String sessionId, String queryId) {
        ActiveQuery query = new ActiveQuery(sessionId, queryId);
        activeQueries.put(sessionId, query);
        log.debug("Query registered: sessionId={}, queryId={}", sessionId, queryId);
        return query;
    }

    /**
     * 쿼리 완료 처리
     */
    public void completeQuery(String sessionId) {
        ActiveQuery query = activeQueries.remove(sessionId);
        if (query != null) {
            log.debug("Query completed: sessionId={}, queryId={}, duration={}s",
                    sessionId, query.getQueryId(),
                    java.time.Duration.between(query.getStartTime(), LocalDateTime.now()).toSeconds());
        }
    }

    /**
     * 세션의 진행 중인 쿼리 취소
     */
    public void cancelQuery(String sessionId) {
        ActiveQuery query = activeQueries.get(sessionId);
        if (query != null) {
            query.cancel();
            log.info("Query cancelled: sessionId={}, queryId={}, table={}, chunks={}",
                    sessionId, query.getQueryId(), query.getCurrentTable(), query.getProcessedChunks());
        }
    }

    /**
     * 세션의 활성 쿼리 조회
     */
    public ActiveQuery getQuery(String sessionId) {
        return activeQueries.get(sessionId);
    }

    /**
     * 활성 쿼리 존재 여부 확인
     */
    public boolean hasActiveQuery(String sessionId) {
        return activeQueries.containsKey(sessionId);
    }

    /**
     * 모든 활성 쿼리 조회 (모니터링용)
     */
    public Map<String, ActiveQuery> getAllActiveQueries() {
        return new ConcurrentHashMap<>(activeQueries);
    }

    /**
     * 전체 활성 쿼리 수
     */
    public int getActiveQueryCount() {
        return activeQueries.size();
    }

    // ── 힙 메모리 체크 ──

    private volatile long lastHeapWarningTime = 0;

    /**
     * 현재 힙 사용률이 임계값을 초과하는지 확인
     * 외부(ChunkedTrackStreamingService 등)에서도 호출 가능
     */
    public boolean isHeapPressureHigh() {
        Runtime rt = Runtime.getRuntime();
        long used = rt.totalMemory() - rt.freeMemory();
        long max = rt.maxMemory();
        double usage = (double) used / max;
        if (usage > heapRejectThreshold) {
            long now = System.currentTimeMillis();
            if (now - lastHeapWarningTime > 5000) {  // 5초마다만 로그
                lastHeapWarningTime = now;
                log.warn("[MEMORY] Heap pressure high: {}% ({} MB / {} MB)",
                        String.format("%.1f", usage * 100),
                        used / (1024 * 1024), max / (1024 * 1024));
            }
            return true;
        }
        return false;
    }

    // ── 글로벌 동시 쿼리 제한 ──

    /**
     * 글로벌 쿼리 슬롯 획득 (즉시 시도, 대기 없음)
     * 힙 메모리 압박 시 슬롯 획득을 거부하여 대기열로 전환
     * @return true: 슬롯 획득 성공, false: 슬롯 없음 (대기열 진입 필요)
     */
    public boolean tryAcquireQuerySlotImmediate(String queryId) {
        if (isHeapPressureHigh()) {
            log.warn("Query {} deferred to queue: heap memory pressure", queryId);
            return false;
        }
        boolean acquired = globalQuerySemaphore.tryAcquire();
        if (acquired) {
            log.info("Query {} acquired global slot immediately: active={}/{}",
                    queryId, getGlobalActiveQueryCount(), maxConcurrentGlobal);
        }
        return acquired;
    }

    /**
     * 글로벌 쿼리 슬롯 획득 (대기 큐에서 FIFO 순서로 대기) - 기존 호환
     * @return true: 슬롯 획득 성공, false: 타임아웃으로 획득 실패
     */
    public boolean tryAcquireQuerySlot(String queryId) throws InterruptedException {
        int waiting = waitingCount.incrementAndGet();
        int active = getGlobalActiveQueryCount();
        log.info("Query {} waiting for global slot: waiting={}, active={}/{}",
                queryId, waiting, active, maxConcurrentGlobal);
        try {
            boolean acquired = globalQuerySemaphore.tryAcquire(queueTimeoutSeconds, TimeUnit.SECONDS);
            if (acquired) {
                log.info("Query {} acquired global slot: active={}/{}",
                        queryId, getGlobalActiveQueryCount(), maxConcurrentGlobal);
            } else {
                log.warn("Query {} timed out waiting for global slot after {}s (active={}/{})",
                        queryId, queueTimeoutSeconds, active, maxConcurrentGlobal);
            }
            return acquired;
        } finally {
            waitingCount.decrementAndGet();
        }
    }

    /**
     * 2초 간격으로 슬롯 대기 (대기열 기반)
     * 호출자가 상태 콜백을 통해 QUEUED 상태를 클라이언트에 전송할 수 있음
     * @param queryId 쿼리 ID
     * @param maxWaitSeconds 최대 대기 시간 (초)
     * @return true: 슬롯 획득, false: 타임아웃
     */
    public boolean waitForSlotWithQueue(String queryId, int maxWaitSeconds) throws InterruptedException {
        waitingQueue.offer(queryId);
        waitingCount.incrementAndGet();
        try {
            long deadline = System.currentTimeMillis() + (maxWaitSeconds * 1000L);
            while (System.currentTimeMillis() < deadline) {
                boolean acquired = globalQuerySemaphore.tryAcquire(2, TimeUnit.SECONDS);
                if (acquired) {
                    waitingQueue.remove(queryId);
                    log.info("Query {} acquired slot after queuing: active={}/{}",
                            queryId, getGlobalActiveQueryCount(), maxConcurrentGlobal);
                    return true;
                }
                // 슬롯 못 얻으면 계속 대기 (호출자에서 QUEUED 상태 전송)
            }
            waitingQueue.remove(queryId);
            log.warn("Query {} timed out in queue after {}s", queryId, maxWaitSeconds);
            return false;
        } finally {
            waitingCount.decrementAndGet();
        }
    }

    /**
     * 대기열에서의 순번 조회
     * @return 1-based 순번, 대기열에 없으면 0
     */
    public int getQueuePosition(String queryId) {
        int position = 1;
        for (String id : waitingQueue) {
            if (id.equals(queryId)) {
                return position;
            }
            position++;
        }
        return 0;
    }

    /**
     * 현재 대기열 크기
     */
    public int getQueueSize() {
        return waitingQueue.size();
    }

    /**
     * 대기열 직접 접근 (서비스에서 offer/remove 가능)
     */
    public ConcurrentLinkedQueue<String> getWaitingQueue() {
        return waitingQueue;
    }

    /**
     * 글로벌 쿼리 슬롯 반환
     */
    public void releaseQuerySlot(String queryId) {
        globalQuerySemaphore.release();
        log.debug("Query {} released global slot: active={}/{}",
                queryId, getGlobalActiveQueryCount(), maxConcurrentGlobal);
    }

    /**
     * 글로벌 활성 쿼리 수 (Semaphore 기준)
     */
    public int getGlobalActiveQueryCount() {
        return maxConcurrentGlobal - globalQuerySemaphore.availablePermits();
    }

    /**
     * 대기 중인 쿼리 수
     */
    public int getWaitingCount() {
        return waitingCount.get();
    }

    /**
     * 최대 동시 쿼리 수
     */
    public int getMaxConcurrentGlobal() {
        return maxConcurrentGlobal;
    }
}
