package com.yumu.noveltranslator.util;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * SSE 发射器工具类。
 * 提供静态方法用于基本的 SSE 操作，同时作为 Spring Bean 提供 Redis Stream 支持以实现事件重放。
 */
@Component
public class SseEmitterUtil {

    private static final Logger log = LoggerFactory.getLogger(SseEmitterUtil.class);
    private static final long DEFAULT_TIMEOUT_MS = 300_000L; // 5 分钟，对齐 application.yaml
    private static final int MAX_ACTIVE_SSE = Integer.getInteger("noveltrans.max.sse", 100);

    /**
     * Emitter 注册表：key=唯一ID，value=可取消的 future
     */
    private static final Map<String, CompletableFuture<Void>> EMITTER_REGISTRY = new ConcurrentHashMap<>();
    private static final AtomicInteger ACTIVE_COUNT = new AtomicInteger(0);

    private final StringRedisTemplate stringRedisTemplate;

    public SseEmitterUtil(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    // ==================== 静态方法（保持向后兼容） ====================

    /**
     * 注册 emitter，返回唯一 ID（用于心跳循环）
     * @throws IllegalStateException 如果超过最大连接数
     */
    public static String registerEmitter(SseEmitter emitter) {
        int count = ACTIVE_COUNT.get();
        if (count >= MAX_ACTIVE_SSE) {
            throw new IllegalStateException("SSE 连接数已达上限: " + MAX_ACTIVE_SSE);
        }

        String emitterId = UUID.randomUUID().toString();
        CompletableFuture<Void> cancellationSignal = new CompletableFuture<>();
        EMITTER_REGISTRY.put(emitterId, cancellationSignal);
        ACTIVE_COUNT.incrementAndGet();

        emitter.onCompletion(() -> {
            unregisterEmitter(emitterId);
        });
        emitter.onTimeout(() -> {
            unregisterEmitter(emitterId);
            emitter.complete();
        });
        emitter.onError((ex) -> {
            if (ex != null) {
                String msg = ex.getMessage();
                if (msg != null &&
                    !(msg.contains("aborted") ||
                      msg.contains("中止") ||
                      msg.contains("reset") ||
                      msg.contains("broken pipe") ||
                      msg.contains("client abort"))) {
                    log.warn("SSE emitter error: {}", msg);
                }
            }
            unregisterEmitter(emitterId);
            emitter.complete();
        });

        return emitterId;
    }

    /**
     * 检查 emitter 是否仍然活跃（心跳循环用）
     */
    public static boolean isEmitterActive(String emitterId) {
        CompletableFuture<Void> future = EMITTER_REGISTRY.get(emitterId);
        return future != null && !future.isCancelled() && !future.isDone();
    }

    /**
     * 注销 emitter（内部调用）
     */
    private static void unregisterEmitter(String emitterId) {
        CompletableFuture<Void> future = EMITTER_REGISTRY.remove(emitterId);
        if (future != null) {
            future.complete(null);
        }
        ACTIVE_COUNT.decrementAndGet();
    }

    /**
     * 获取当前活跃的 SSE 连接数（监控用）
     */
    public static int getActiveCount() {
        return ACTIVE_COUNT.get();
    }

    /**
     * 创建 SSE emitter（简单版，不注册到注册表）
     * 适用于不需要心跳管理的场景
     */
    public static SseEmitter createSseEmitter(Long timeout) {
        SseEmitter emitter = new SseEmitter(timeout != null ? timeout : DEFAULT_TIMEOUT_MS);
        emitter.onCompletion(() -> {});
        emitter.onTimeout(emitter::complete);
        emitter.onError((ex) -> {
            if (ex != null) {
                String msg = ex.getMessage();
                if (msg != null &&
                    !(msg.contains("aborted") ||
                      msg.contains("中止") ||
                      msg.contains("reset") ||
                      msg.contains("broken pipe") ||
                      msg.contains("client abort"))) {
                    log.warn("SSE emitter error: {}", msg);
                }
            }
            emitter.complete();
        });
        return emitter;
    }

    public static void sendError(SseEmitter emitter, String msg) {
        try {
            emitter.send(SseEmitter.event().data("ERROR: " + msg));
        } catch (Exception ignored) {}
    }

    public static void sendDone(SseEmitter emitter) {
        try {
            emitter.send(SseEmitter.event().data("[DONE]"));
        } catch (Exception ignored) {}
    }

    public static void sendData(SseEmitter emitter, String data) {
        try {
            emitter.send(SseEmitter.event().data(data));
        } catch (Exception ignored) {}
    }

    /**
     * 发送心跳事件，防止长翻译过程中 SSE 连接因超时而断开。
     * 使用 SSE comment 格式（冒号前缀），前端会自动忽略此类注释行。
     */
    public static void sendHeartbeat(SseEmitter emitter) {
        try {
            emitter.send(SseEmitter.event().comment("heartbeat"));
        } catch (Exception ignored) {}
    }

    public static void complete(SseEmitter emitter) {
        emitter.complete();
    }

    public static void completeWithError(SseEmitter emitter, Exception ex) {
        emitter.completeWithError(ex);
    }

    // ==================== 实例方法（Redis Stream 支持） ====================

    /**
     * XADD Lua 脚本：向 Redis Stream 添加事件，带 MAXLEN 限制防止无限增长
     */
    private static final String XADD_LUA =
            "redis.call('XADD', KEYS[1], 'MAXLEN', '~', '10000', '*', 'event', ARGV[1]); return 'OK'";

    private static final DefaultRedisScript<String> XADD_SCRIPT =
            new DefaultRedisScript<>(XADD_LUA, String.class);

    /**
     * XRANGE Lua 脚本：从 Redis Stream 获取 lastEventId 之后的所有事件
     */
    private static final String XRANGE_LUA =
            "local results = redis.call('XRANGE', KEYS[1], '(' .. ARGV[1], '+'); " +
            "local out = {}; " +
            "for i = 1, #results, 1 do " +
            "  out[i] = results[i][2][2]; " +
            "end; " +
            "return out";

    private static final DefaultRedisScript<List> XRANGE_SCRIPT =
            new DefaultRedisScript<>(XRANGE_LUA, List.class);

    /**
     * 发布协作事件到 Redis Stream，支持断线重连后补发。
     *
     * @param projectId 项目ID
     * @param eventType 事件类型（如 chapter.updated, comment.added）
     * @param payload   事件载荷（JSON 字符串）
     * @return 生成的事件 ID
     */
    public String publishCollabEvent(String projectId, String eventType, String payload) {
        String streamKey = "collab:events:" + projectId;
        String eventId = UUID.randomUUID().toString();
        long timestamp = System.currentTimeMillis();

        String eventJson = String.format(
                "{\"eventId\":\"%s\",\"type\":\"%s\",\"payload\":%s,\"timestamp\":%d}",
                eventId, eventType, payload, timestamp);

        stringRedisTemplate.execute(XADD_SCRIPT, List.of(streamKey), eventJson);

        log.debug("发布协作事件: projectId={}, type={}, eventId={}", projectId, eventType, eventId);
        return eventId;
    }

    /**
     * 从 Redis Stream 重放指定项目自 lastEventId 之后的遗漏事件。
     *
     * @param projectId   项目ID
     * @param lastEventId 上次收到的事件 ID（Redis Stream 条目 ID），为 null 时不重放
     * @param emitter     SSE emitter，用于发送事件
     */
    public void replayMissedEvents(String projectId, String lastEventId, SseEmitter emitter) {
        if (lastEventId == null || lastEventId.isBlank()) {
            log.debug("无 lastEventId，跳过事件重放: projectId={}", projectId);
            return;
        }

        String streamKey = "collab:events:" + projectId;
        @SuppressWarnings("unchecked")
        List<String> events = (List<String>) stringRedisTemplate.execute(XRANGE_SCRIPT, List.of(streamKey), lastEventId);

        if (events == null || events.isEmpty()) {
            log.debug("无遗漏事件需要重放: projectId={}, lastEventId={}", projectId, lastEventId);
            return;
        }

        int sent = 0;
        for (String eventJson : events) {
            if (eventJson == null || eventJson.isEmpty()) {
                continue;
            }
            try {
                emitter.send(SseEmitter.event()
                        .name("replay")
                        .data(eventJson));
                sent++;
            } catch (Exception e) {
                log.warn("重放事件失败: projectId={}, error={}", projectId, e.getMessage());
                break;
            }
        }

        log.info("重放完成: projectId={}, sent={} events", projectId, sent);
    }
}
