package com.agent.management.external;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.time.Instant;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * SSE 事件总线：按 runId 缓存事件，支持多订阅者 fan-out 与 Last-Event-ID 续传。
 *
 * 设计：
 * - 引擎层在 safeSend 中调用 {@link #publish} 发布事件，事件被缓存并推送给所有活跃订阅者。
 * - 客户端通过 GET /stream 调用 {@link #subscribe}，可携带上次收到的事件 id 进行断线重连。
 * - 终止事件（workflow_complete / workflow_error）后，订阅者自动关闭，缓存保留 1 小时供后续查询。
 *
 * 同时被内部 API（/api/workflows/{id}/run）和外部 API（/api/v1/...）共用：
 * 内部 API 不调用 subscribe，事件依然会缓存（但不被消费），运行结束后定时清理。
 */
@Slf4j
@Component
public class SseEventBus {

    /**
     * 单个 runId 缓存的最大事件数（超出后丢弃最旧事件，断线重连可能拿不到完整历史）。
     * <p>v1.1：从 1000 调到 5000。clarify 长等待期间可能持续累积 thinking / tool_progress 事件，
     * 用户几小时后回来重连需看到完整思考过程。clarify_request 永不丢弃（在 publish 中特殊处理）。</p>
     */
    private static final int MAX_EVENTS_PER_RUN = 5000;

    /**
     * 运行结束后缓存保留时长。
     * <p>v1.1：从 1 小时调到 25 小时。配合 24h 兜底超时，确保用户离开一天回来后还能看到完整重放。</p>
     */
    private static final long TTL_AFTER_COMPLETION_MS = 90_000_000L; // 25h

    /**
     * SSE 连接超时。
     * <p>v1.1：从 30 分钟调到 24 小时（兜底）。Askfor 场景用户可能几小时后才回来，
     * 真正的"无超时"由前端断线重连 + Last-Event-ID 续传机制保证。</p>
     */
    private static final long SSE_TIMEOUT_MS = 86_400_000L; // 24h

    private final Map<String, RunStream> streams = new ConcurrentHashMap<>();
    private final ScheduledExecutorService cleaner = Executors.newSingleThreadScheduledExecutor(r -> {
        Thread t = new Thread(r, "sse-event-bus-cleaner");
        t.setDaemon(true);
        return t;
    });

    /**
     * 提前注册 runId（POST /runs 创建阶段调用）。
     * 调用是幂等的，重复注册不会清空已有事件。
     */
    public void register(String runId) {
        streams.computeIfAbsent(runId, k -> new RunStream());
    }

    /**
     * 判断 runId 是否已注册（用于校验 stream 接口的路径参数）。
     */
    public boolean exists(String runId) {
        return runId != null && streams.containsKey(runId);
    }

    /**
     * 发布事件到指定 runId 的流。所有订阅者实时收到，事件同时进入缓存。
     */
    public void publish(String runId, String eventName, Object payload) {
        if (runId == null) return;
        RunStream stream = streams.computeIfAbsent(runId, k -> new RunStream());
        RecordedEvent rec = new RecordedEvent(stream.idGen.incrementAndGet(), eventName, payload, Instant.now());

        // M-6：写入缓存（超出窗口则丢弃最旧事件）
        // 原 CopyOnWriteArrayList 每次 add/remove(0) 都是 O(n) 数组复制，5000 元素高频 publish 浪费 CPU。
        // 改用 ArrayDeque + synchronized：addLast/removeFirst 均 O(1)，持锁时间是常数，不影响订阅者 fan-out。
        // v1.1：MAX_EVENTS_PER_RUN 调到 5000，长等待期间一般不会超限；若超限丢失最旧的 thinking 片段可接受，
        // clarify_request 持续保存在 WorkflowPause 表中，前端断线重连时优先通过 GET /state 恢复 Askfor UI。
        synchronized (stream.events) {
            stream.events.addLast(rec);
            while (stream.events.size() > MAX_EVENTS_PER_RUN) {
                stream.events.removeFirst();
            }
        }

        // fan-out：广播到所有订阅者（无需持锁，subscribers 是 ConcurrentHashMap.newKeySet）
        for (SseEmitter sub : stream.subscribers) {
            try {
                sub.send(SseEmitter.event()
                        .id(String.valueOf(rec.id))
                        .name(rec.name)
                        .data(rec.payload));
            } catch (IOException | IllegalStateException e) {
                log.debug("[SseEventBus] 推送订阅者失败 runId={} subscriber 已移除: {}", runId, e.getMessage());
                stream.subscribers.remove(sub);
            }
        }

        // 终止事件处理：关闭订阅者并安排清理
        if (isTerminal(eventName)) {
            stream.completedAt = Instant.now();
            stream.failed = "workflow_error".equals(eventName);
            for (SseEmitter sub : stream.subscribers) {
                try { sub.complete(); } catch (Exception ignore) { /* 已关闭 */ }
            }
            stream.subscribers.clear();
            cleaner.schedule(() -> streams.remove(runId), TTL_AFTER_COMPLETION_MS, TimeUnit.MILLISECONDS);
        }
    }

    /**
     * 订阅指定 runId 的事件流。
     *
     * @param lastEventId 客户端最后收到的事件 id（来自 Last-Event-ID 头部），null 表示从头开始
     * @return SseEmitter；若 runId 不存在返回 null
     */
    public SseEmitter subscribe(String runId, Long lastEventId) {
        RunStream stream = streams.get(runId);
        if (stream == null) return null;

        SseEmitter emitter = new SseEmitter(SSE_TIMEOUT_MS);

        // 补发历史事件（lastEventId 之后的所有事件）
        // 快照必须在 synchronized 块内取，否则 ArrayDeque 迭代期间被 publish 修改会抛 ConcurrentModificationException
        // 持锁仅用于复制数组（O(n) 内存拷贝），之后的 I/O 不占锁，避免阻塞 publish
        RecordedEvent[] snapshot;
        synchronized (stream.events) {
            snapshot = stream.events.toArray(new RecordedEvent[0]);
        }
        for (RecordedEvent rec : snapshot) {
            if (lastEventId != null && rec.id <= lastEventId) continue;
            try {
                emitter.send(SseEmitter.event()
                        .id(String.valueOf(rec.id))
                        .name(rec.name)
                        .data(rec.payload));
            } catch (IOException e) {
                log.debug("[SseEventBus] 补发历史失败 runId={}: {}", runId, e.getMessage());
            }
        }

        // 若运行已结束，补发完即关闭
        if (stream.completedAt != null) {
            try { emitter.complete(); } catch (Exception ignore) { /* 已关闭 */ }
            return emitter;
        }

        // 注册为活跃订阅者
        stream.subscribers.add(emitter);
        emitter.onCompletion(() -> stream.subscribers.remove(emitter));
        emitter.onTimeout(() -> stream.subscribers.remove(emitter));
        emitter.onError(ex -> stream.subscribers.remove(emitter));
        return emitter;
    }

    /**
     * 查询 runId 当前状态（用于 GET /runs/{runId} 与校验）。
     * @return null 表示未注册；否则返回 [状态, 已完成时间]
     */
    public RunStatus getStatus(String runId) {
        RunStream stream = streams.get(runId);
        if (stream == null) return null;
        int count;
        synchronized (stream.events) {
            count = stream.events.size();
        }
        return new RunStatus(
                stream.completedAt == null ? "RUNNING" : (stream.failed ? "FAILED" : "SUCCESS"),
                stream.completedAt,
                count
        );
    }

    private boolean isTerminal(String eventName) {
        return "workflow_complete".equals(eventName) || "workflow_error".equals(eventName);
    }

    // ========== 内部数据结构 ==========

    private static final class RunStream {
        final AtomicLong idGen = new AtomicLong(0);
        final Deque<RecordedEvent> events = new ArrayDeque<>();
        final Set<SseEmitter> subscribers = ConcurrentHashMap.newKeySet();
        volatile Instant completedAt;
        volatile boolean failed;
    }

    private static final class RecordedEvent {
        final long id;
        final String name;
        final Object payload;
        final Instant timestamp;

        RecordedEvent(long id, String name, Object payload, Instant timestamp) {
            this.id = id;
            this.name = name;
            this.payload = payload;
            this.timestamp = timestamp;
        }
    }

    /** 对外暴露的运行状态快照 */
    public record RunStatus(String status, Instant completedAt, int eventCount) {}
}
