package jnpf.audit.sdk;

import jnpf.audit.AuditDeliveryException;
import jnpf.audit.AuditEventApi;
import jnpf.audit.model.AuditEventDTO;
import jnpf.base.ActionResult;
import jnpf.base.ActionResultCode;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.ObjectProvider;

import java.util.List;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 两级降级异步发送器：Feign（AuditEventApi 同步）→ 本地 spool 落盘。
 *
 * <p><b>2026-08-11 变更</b>：原第一级 MQ（{@code streamBridge.send("AUDIT_EVENT", …)}）已移除。
 * 事件总线于 2026-08-10 切 Redis 后栈内不再有 RocketMQ binder，该级**每次必然失败**再由 Feign 接管——
 * 它已不是降级级次，只是每条事件多一次异常构造。移除后行为等价（此前本就每条都走 Feign）。
 * 若将来要恢复异步投递，正解是提供一个写 Redis 的 {@link AuditEventApi} 实现 bean 替代 Feign，
 * 本类一行不用改（见 tasks/todo.md 阶段 4，当前因 Redis 无持久化 + noeviction 暂不做）。
 *
 * <p>本类的并发/时序注释对应降级链路上经过验证的并发安全设计——重构时不要随手删改，
 * 每一处都对应一个真实存在过的竞态场景。
 * deliverOnce 只做 Feign 一级并给出明确成败信号（不写 spool），供普通路径与 replayer 各自决定失败处理。
 */
@Slf4j
public class AuditEventPublisher implements DisposableBean {
    private final ObjectProvider<AuditEventApi> feignApi;
    private final AuditSpoolWriter spool;
    private final ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 2, 60, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(1000),
            r -> { Thread t = new Thread(r, "audit-sender"); t.setDaemon(true); return t; },
            new ThreadPoolExecutor.AbortPolicy());   // 队列满走 catch → spool，不阻塞业务

    /** pending 集合：提交前在调用线程登记、run() finally 移除——统一覆盖"排队中/执行中/已出队未开跑"三态，
        无登记窗口（终验三轮④：原 run() 内 add 的写法在出队与登记之间有窗口，shutdownNow 撞上会双漏） */
    private final Set<AuditSendTask> pending = ConcurrentHashMap.newKeySet();

    public AuditEventPublisher(ObjectProvider<AuditEventApi> feignApi,
                               AuditSpoolWriter spool) {
        this.feignApi = feignApi;
        this.spool = spool;
    }

    /** 任务自携事件对象：异常/停机 drain 时能恢复出具体事件落 spool（评审#2，裸 lambda 恢复不出事件） */
    class AuditSendTask implements Runnable {
        final AuditEventDTO event;
        AuditSendTask(AuditEventDTO event) { this.event = event; }
        @Override public void run() {
            try {
                deliverOnce(event);
            } catch (Throwable t) {   // 最外层兜底：任何异常（含 JSON 序列化）都落 spool，绝不静默丢（评审#2）
                spool.write(event);
                log.error("AUDIT-LOST clientEventId={} type={} target={} 已落 spool",
                        event.getClientEventId(), event.getEventType(), event.getTargetTable(), t);
            } finally {
                pending.remove(this);
            }
        }
    }

    public void publishAsync(List<AuditEventDTO> events) {
        for (AuditEventDTO e : events) {
            AuditSendTask task = new AuditSendTask(e);
            pending.add(task);                    // 先登记再入池（终验三轮④）
            try {
                pool.execute(task);
            } catch (RejectedExecutionException full) {
                // 先写 spool 再移除登记（终验四轮②：反序会在 remove 与 write 之间留下双侧不可见窗口）
                try {
                    spool.write(e);   // 队列满直接落 spool（spec §5.5）
                } finally {
                    pending.remove(task);
                }
            }
        }
    }

    /**
     * 单次投递：Feign 同步，成功返回、失败抛异常——本方法不写 spool。
     * 终验#2/#3：普通路径与 replayer 都需要明确的成败信号各自决定失败处理，
     * 原 send() 失败时自己写 spool 正常返回，replayer 无法判断该不该保留待重试行。
     */
    void deliverOnce(AuditEventDTO e) {
        ActionResult<Integer> result = feignApi.getObject().record(e);
        if (result == null || !ActionResultCode.SUCCESS.getCode().equals(result.getCode())) {
            // HTTP 层失败由 fallbackFactory 抛出；业务层失败(HTTP 200 + fail body)靠本校验捕获（Codex 批次一 Critical#1）
            throw new AuditDeliveryException("Feign 投递返回非成功: " + (result == null ? "null" : result.getCode()), null);
        }
    }

    /**
     * Codex 批次二 #5：以 {@link DisposableBean#destroy()} 替代 jakarta {@code @PreDestroy}，
     * 跨 boot2(JDK8)/boot3 中立——{@code @PreDestroy} 在 boot2 走 {@code javax.annotation}、
     * boot3 走 {@code jakarta.annotation}，SDK 二进制无法同时兼容；DisposableBean 是 Spring 原生接口，
     * 两版一致。Spring 对 {@code @Bean} 会自动识别 DisposableBean，{@code AuditSdkAutoConfiguration} 无需改。
     */
    @Override
    public void destroy() {
        drain();
    }

    public void drain() {   // 停机 drain（spec §5.5）；不向外抛异常（终验三轮④：异常逃出会绕过全部兜底）
        pool.shutdown();
        try {
            if (!pool.awaitTermination(10, TimeUnit.SECONDS)) {
                pool.shutdownNow();
                spoolPending();
            }
        } catch (InterruptedException ie) {
            pool.shutdownNow();
            spoolPending();
            Thread.currentThread().interrupt();   // 兜底做完再恢复中断标记
        }
    }

    private void spoolPending() {
        // pending 统一覆盖排队中/执行中/出队未开跑三态；与任务自身 catch 落盘可能重复——
        // spool 重放与落库均按 client_event_id 幂等，重复无害
        for (AuditSendTask task : pending) {
            spool.write(task.event);
        }
    }
}