package jnpf.audit.sdk;

import jnpf.audit.model.AuditEventDTO;
import jnpf.util.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.StandardCopyOption;
import java.nio.file.StandardOpenOption;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * spool 重放器（Task 7 Step 3；评审#3：spool 必须可恢复才算兜底）。
 *
 * <p>周期性扫描 spool 目录的 {@code .jsonl} / {@code .retry} 文件 → 原子重命名加
 * {@code .replaying.<owner-token>} 后缀（认领，token 每次新生成，见"#B+#C"段）→ 逐行反序列化走
 * {@link AuditEventPublisher#deliverOnce(AuditEventDTO)}
 * （成功丢弃该行、抛异常保留）→ 全部成功删 {@code .replaying}；部分失败把剩余行写入
 * {@code <稳定 base 文件名>.<UUID>.retry} 新文件（base+UUID 保证同轮多文件失败不撞名，终验五轮⑤；
 * base 名的取法见下文"内审 Important 修复"段 {@link #stableBaseName(String)}），
 * {@code .retry} 写成功后删对应 {@code .replaying}（不留孤儿）；<b>绝不写回活动原名</b>（终验四轮③：
 * 重命名期间写入端可能已在原名新建文件追加新事件，写回原名会覆盖丢数据）。落库幂等（client_event_id）
 * 使重放天然安全。
 *
 * <p>调度手段：自建单线程 daemon {@link ScheduledExecutorService}（非 @Scheduled），不依赖宿主
 * 是否开启 @EnableScheduling，生命周期由本 bean 自控（见 task-7-report 决策说明）。
 * 由 {@code audit.spool-replay-enabled}（默认 true）开关。
 *
 * <p><b>双层锁协议（Codex 批次二复验 #1/#2 终局修复）</b>：与 {@link AuditSpoolWriter} 共享同一
 * {@link AuditSpoolLock}（进程内 monitor + 跨进程 {@code spool.lock} FileLock）。快速文件操作在锁内、
 * 慢速网络重放在锁外，临界区划分：
 * <ol>
 *   <li><b>认领</b>（{@link #replayFile}）：{@code rename→.replaying.<owner-token>} + {@code setLastModified}
 *       刷新租约在<b>同一临界区</b>——#1 与 writer 的 append 互斥（消除"取走文件瞬间"与"正在 append"的窗口
 *       竞态）；#2 rename 与刷新原子成对（消除"rename 成功后、刷新 mtime 前被孤儿回收误抢"的窗口）。
 *       临界区内<b>先锁内重验</b>（#B）再 rename。replayer 路径 {@code tryLock} 拿不到就跳过本轮，下轮再来。</li>
 *   <li><b>readAllLines</b>：锁外——认领后 {@code .replaying} 已不在任何写入端/扫描集视线内（孤儿回收
 *       10 分钟阈值 + 刚刷新的 mtime 保护），读取安全且不阻塞 writer。</li>
 *   <li><b>逐行 deliverOnce</b>：锁外（网络慢操作）。</li>
 *   <li><b>收尾</b>（{@link #finishReplay}）：写 {@code .bad} / 写 {@code .retry} / 删 {@code .replaying}
 *       在锁内（各自快操作）。</li>
 *   <li><b>孤儿回收 rename</b>（{@link #recoverOrphanedReplaying}）：检查+rename 在锁内。</li>
 * </ol>
 * 60 秒静默窗口（{@link #QUIET_WINDOW_MS}）<b>保留为纯优化</b>（减少活跃文件搅动），非正确性依赖——
 * 正确性已由锁协议保证。<b>残余</b>：单文件重放超 10 分钟仍可能被其它进程当孤儿抢走 → 后果=重复重放，
 * 落库按 client_event_id 幂等保证不丢不重入库，可接受；{@code spool.lock} 永不参与扫描/重放/删除
 * （扫描过滤已按 {@link AuditSpoolLock#LOCK_FILE_NAME} 排除）。
 *
 * <p><b>裁定修复 D（孤儿 .replaying 恢复）</b>：进程若在"改名成 .replaying"之后、"处理完"之前崩溃，
 * 会留下孤儿——扫描集固定为 {@code {.jsonl,.retry}}，永远不会认领它，事件永久滞留。每轮扫描时，
 * 额外把<b>修改时间早于（当前时刻 - 10 分钟，即两个调度周期）</b>的 .replaying 文件原子改名为
 * {@code <稳定 base 文件名>.<UUID>.retry}（锁内），回归 retry 通道由下一轮自然消化；不在回收动作
 * 内直接处理文件内容，避免与"本实例/并发实例正在重放中的文件"混淆。年龄阈值即为防止误抢：正常一轮
 * 处理远快于一个调度周期，若 .replaying 存活跨越两个周期才可判定为孤儿。落库按 client_event_id
 * 幂等，重复重放安全。
 *
 * <p><b>Codex 批次二三轮 #B+#C（认领名 owner-token 化 + 锁内重验）</b>：认领 rename 目标由固定后缀
 * {@code .replaying} 改为 owner-token 化的 {@code .replaying.<ownerUUID>}（token 每次认领新生成，
 * {@link #REPLAYING_TOKEN_SUFFIX} 识别）。收尾/删除只操作本 owner token 的路径——陈旧 owner（被抢后收尾）
 * 与新 owner 永不同名，{@code deleteIfExists} 物理上不可能删掉新 owner 刚认领的文件（消除 #C 固定名跨
 * owner 冲突）。同时认领（{@link #replayFile}）与孤儿回收（{@link #recoverOrphanedReplaying}）都改为
 * <b>锁内先重验</b>（exists + mtime 仍超阈/仍静默）再 rename——锁外 {@code listFiles} 预筛只作候选收集，
 * 消除"预筛后、锁内 rename 前，路径被删除并复用为新文件 → 旧检查结果误搬新文件"的窗口（#B）。
 * {@code setLastModified} 失败保留 warn+继续：owner-token 唯一名下租约失败的最坏后果=被孤儿回收→重复重放
 * （幂等吸收），不再有丢失路径。
 *
 * <p><b>Codex 批次二三轮 #D（优雅退出）</b>：{@link #destroy()} 在 {@code shutdownNow} 后
 * {@code awaitTermination}({@link #SHUTDOWN_AWAIT_MS}) 等待重放线程真正退出（超时 warn），与
 * {@link AuditSpoolLock} 的终止位配合，避免"停机返回但重放线程仍在锁内收尾"。
 *
 * <p><b>内审 Important 修复（retry 文件名收敛，防 NAME_MAX 无界增长）</b>：部分失败写 {@code .retry}、
 * 孤儿回收改名、坏行隔离写 {@code .bad}，前缀都先经 {@link #stableBaseName(String)} 从文件名剥离<b>所有</b>
 * {@code .<UUID>.retry} 层与 {@code .replaying.<UUID>} token 后缀还原出稳定 base（如 {@code audit-events-20260723.jsonl}），
 * 再拼<b>恰好一层</b>新 UUID，文件名长度从此有界，不随失败轮次增长。UUID 段用 {@link #RETRY_LAYER_SUFFIX}
 * 严格匹配标准 8-4-4-4-12 十六进制格式并锚定字符串末尾，不会误剥业务文件名自身携带的点段（如 {@code .jsonl}）。
 * （Codex 五轮起：{@link #recoverOrphanedWriting()} 的隔离改名不再走本机制——直接在原名后追加
 * {@code .stale}，见其专属段；{@code .retry.writing} 层剥离分支保留作通用防御，当前无调用方会触发。）
 *
 * <p><b>坏行隔离与脱敏（Codex 批次二 #3/#4）</b>：反序列化失败的坏行不静默丢、不进 remaining，收尾时批量
 * 隔离进 {@code <稳定 base>.<UUID>.bad}（不参与任何自动重放，留人工修复补录，见 {@link #finishReplay}）；
 * 坏行 ERROR 日志只记文件名/行号/行长/内容 SHA-256 前 12 位（{@link #sha256Prefix12}），绝不输出行内容正文。
 *
 * <p><b>写后原子发布孤儿回收（Codex 四轮 Critical#1）</b>：{@link AuditSpoolWriter} 降级路径改为「先写
 * 唯一名 {@code .writing} 临时文件、再原子改名发布为 {@code .retry}」（消灭跨进程半写窗口，见其类
 * Javadoc）。若 writer 恰好崩溃于两步之间（或改名本身因极端 IO 异常失败），会遗留
 * {@code <稳定base>.<UUID>.retry.writing} 孤儿——扫描集固定排除 {@code .writing}（见
 * {@link #replayOnce()}），主流程永远不会认领它。每轮额外调用 {@link #recoverOrphanedWriting()}
 * 处理这类孤儿；具体动作见下一段——Codex 五轮复验推翻了本段最初"回归 .retry 通道当正常事件重放"的方案。
 *
 * <p><b>Codex 五轮修复（.writing 隔离区，不能仅凭年龄自动发布）</b>：四轮版本曾把超龄
 * {@code .retry.writing} 直接改名回归 {@code .retry} 通道当正常事件重放——Codex 五轮复验指出这不安全：
 * degraded writer（未持 FileLock 的降级路径）若恰好在其唯一一次 {@link Files#write} 调用<b>内部</b>
 * 被操作系统/磁盘 IO 阻塞超过 10 分钟，该 writer 线程仍然存活、仍持有该文件的打开 FD，只是尚未返回；
 * 此时它的 {@code .writing} 文件在孤儿回收眼中已"超龄"，若照旧改名发布为 {@code .retry}，重放器会把它
 * 当正常事件认领（rename→{@code .replaying}）、读到空文件（内容还没真正落盘）、判定"处理完毕"并删除
 * {@code .replaying}；随后卡住的 write 调用恢复，向着已被 unlink 的 inode 追加——数据静默消失，不留
 * 任何错误痕迹。半行截断时同理：坏行隔离机制只能吸收"取走时刻"已经写入的前缀，写调用尚未完成的后半段
 * 一样会丢。
 *
 * <p>结论：<b>超龄不足以断言 writer 已死</b>，{@link #recoverOrphanedWriting()} 因此只做<b>隔离</b>、
 * 绝不发布、绝不删除——锁内重验通过后把 {@code .retry.writing} 原子改名为 {@code <原名>.stale}（同
 * 目录），{@code log.error} 告警（脱敏：仅记文件名/大小/mtime，不含内容）后即止步：不进 {@code .retry}
 * 通道（不会被重放器认领）、不进 {@link #finishReplay} 收尾（不会被删除）。{@code .stale} 排除在一切
 * 扫描/认领之外——主扫描（{@link #replayOnce()}）只认 {@code .jsonl}/{@code .retry}，
 * {@link #recoverOrphanedReplaying()} 只认 {@code .replaying.<uuid>}，{@link #recoverOrphanedWriting()}
 * 自身只认 {@code .retry.writing}，三者的后缀判定都不会匹配以 {@code .stale} 收尾的文件名，故隔离后的
 * 文件物理上不可能被本类任何逻辑再次触碰；也无需纳入 {@link #stableBaseName(String)} 的剥层范围——
 * {@code .stale} 从不参与任何"新建 retry/bad 文件"的前缀推导。
 *
 * <p>安全性论证的核心是 POSIX {@code rename(2)} 的语义：<b>rename 不 unlink</b>——它只改变目录项指向的
 * 名字，被卡住的 write 调用持有的是对 inode 的 FD，与路径名无关，恢复后仍会把内容完整写入<b>同一个
 * inode</b>（现在挂在 {@code .stale} 这个名字下），不会因改名而丢失或写偏。事件因此保全在隔离区，等待
 * 人工核查后手工补录（对照 {@code clientEventId} 去重）；隔离区本身永不被任何自动化流程删除，因此
 * "重放器把仍在写的文件当孤儿抢走→空读→删除→静默丢失"这条路径被<b>物理消除</b>（不是靠时序窗口收窄，
 * 而是代码上根本不存在能删除 {@code .stale} 的路径）。writer 侧针对"发布源被隔离"的自愈见
 * {@link AuditSpoolWriter} 类 Javadoc"Codex 五轮修复（自愈重发）"段——move 目标源被隔离后不会让事件
 * 真的丢，只会在 {@code .stale} 里留一份至多被人工补录时重复的副本（落库按 {@code clientEventId} 幂等
 * 吸收）。
 */
@Slf4j
public class AuditSpoolReplayer implements InitializingBean, DisposableBean {

    private static final long DEFAULT_REPLAY_INTERVAL_MS = 300_000L;   // spec: fixedDelay 5 分钟（默认，可经 audit.spool-replay-interval-ms 覆盖）
    private static final String REPLAYING_SUFFIX = ".replaying";
    // 裁定修复 D：孤儿 .replaying.<uuid> 年龄阈值 = 两个默认调度周期（10 分钟），防止误抢并发实例正在处理的文件。
    // 刻意锚定 DEFAULT_REPLAY_INTERVAL_MS 而非实例调度周期：孤儿年龄阈值是"多久才敢判定文件被崩溃遗弃"的
    // 崩溃恢复安全底线，与"多久轮询一次"是两个关注点——把轮询周期调短（如测试环境 15s 加速验证）不应连带
    // 缩小这条安全窗口，否则一次略慢的正常重放就可能被误判为孤儿并被并发实例抢走（幂等吸收但徒增搅动）。
    private static final long ORPHAN_REPLAYING_AGE_MS = 2 * DEFAULT_REPLAY_INTERVAL_MS;
    // Codex 批次二 #1：活动 .jsonl 静默窗口——距今不足 60 秒本轮跳过。锁协议落地后此窗口降级为纯优化
    // （减少活跃文件搅动），非正确性依赖；正确性由 AuditSpoolLock 双层锁保证
    private static final long QUIET_WINDOW_MS = 60_000L;
    // 内审 Important 修复：精确匹配"尾部恰好一层 <UUID>.retry"，UUID 严格按 8-4-4-4-12 十六进制格式
    // 锚定字符串末尾（$），避免误剥业务文件名自身携带的点段（如 audit-events-20260723.jsonl 的 .jsonl）
    private static final Pattern RETRY_LAYER_SUFFIX = Pattern.compile(
            "\\.[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\\.retry$");
    // Codex 批次二三轮 #B+#C：认领名 owner-token 化——rename 目标 <src原名>.replaying.<ownerUUID>，
    // 每次认领新生成 owner token。孤儿匹配/stableBaseName 剥离都以本模式识别 .replaying.<uuid> 层，
    // 锚定字符串末尾（$），UUID 严格 8-4-4-4-12 十六进制格式
    private static final Pattern REPLAYING_TOKEN_SUFFIX = Pattern.compile(
            "\\.replaying\\.[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$");
    // Codex 四轮 Critical#1：AuditSpoolWriter 降级路径写后原子发布的临时名 <稳定base>.<UUID>.retry.writing——
    // 崩溃/rename 失败会遗留孤儿，本模式供扫描排除与 recoverOrphanedWriting() 孤儿回收识别，
    // 锚定字符串末尾（$），UUID 严格 8-4-4-4-12 十六进制格式
    private static final Pattern RETRY_WRITING_SUFFIX = Pattern.compile(
            "\\.[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\\.retry\\.writing$");
    // Codex 五轮：recoverOrphanedWriting() 隔离目标后缀——只隔离不发布不删除。任何以此结尾的文件名都不再
    // 匹配 RETRY_WRITING_SUFFIX/REPLAYING_TOKEN_SUFFIX（二者均要求恰好以 .retry.writing / .replaying.<uuid>
    // 收尾），也不以 .jsonl/.retry 结尾，故隔离后天然不会被本类任何扫描逻辑再次命中，见类 Javadoc"Codex 五轮修复"段
    private static final String STALE_SUFFIX = ".stale";
    // Codex 批次二三轮 #D：destroy 时等待重放线程退出的上限
    private static final long SHUTDOWN_AWAIT_MS = 5_000L;

    private final AuditEventPublisher publisher;
    private final File spoolDir;
    private final boolean enabled;
    private final AuditSpoolLock spoolLock;
    // 调度周期：默认 DEFAULT_REPLAY_INTERVAL_MS（5 分钟），可经 audit.spool-replay-interval-ms 覆盖。
    // 仅影响"多久轮询一次 spool"，不影响孤儿年龄阈值（ORPHAN_REPLAYING_AGE_MS，固定安全底线，见其注释）。
    // 生产默认不动；测试环境可调短（如 15s）加速降级链恢复重放验证。
    private final long replayIntervalMs;
    private ScheduledExecutorService scheduler;

    public AuditSpoolReplayer(AuditEventPublisher publisher, String spoolDir, boolean enabled,
                              AuditSpoolLock spoolLock) {
        this(publisher, spoolDir, enabled, spoolLock, DEFAULT_REPLAY_INTERVAL_MS);
    }

    public AuditSpoolReplayer(AuditEventPublisher publisher, String spoolDir, boolean enabled,
                              AuditSpoolLock spoolLock, long replayIntervalMs) {
        this.publisher = publisher;
        this.spoolDir = new File(spoolDir);
        this.enabled = enabled;
        this.spoolLock = spoolLock;
        // 兜底：非正数（配置笔误/0）回落默认周期，绝不让 scheduleWithFixedDelay 收到非法周期
        this.replayIntervalMs = replayIntervalMs > 0 ? replayIntervalMs : DEFAULT_REPLAY_INTERVAL_MS;
    }

    @Override
    public void afterPropertiesSet() {
        if (!enabled) {
            log.info("[audit] spool replayer 已禁用 (audit.spool-replay-enabled=false)");
            return;
        }
        scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
            Thread t = new Thread(r, "audit-spool-replayer");
            t.setDaemon(true);
            return t;
        });
        // initialDelay 取与周期一致：避免启动瞬间与业务争 IO；遗留 spool 至多延迟一个周期被重放
        scheduler.scheduleWithFixedDelay(this::replaySafely,
                replayIntervalMs, replayIntervalMs, TimeUnit.MILLISECONDS);
    }

    @Override
    public void destroy() {
        if (scheduler == null) {
            return;
        }
        // Codex 批次二三轮 #D：shutdownNow 后等待重放线程真正退出，避免"停机返回但重放线程仍在锁内/
        // 正调 AuditSpoolLock 收尾"的窗口——与 AuditSpoolLock.destroy 的终止位配合优雅退出。
        scheduler.shutdownNow();
        try {
            if (!scheduler.awaitTermination(SHUTDOWN_AWAIT_MS, TimeUnit.MILLISECONDS)) {
                log.warn("[audit] spool 重放线程 {}ms 内未退出（可能卡在 IO/网络重放中）", SHUTDOWN_AWAIT_MS);
            }
        } catch (InterruptedException ie) {
            Thread.currentThread().interrupt();   // 恢复中断标记后返回
        }
    }

    private void replaySafely() {
        try {
            replayOnce();
        } catch (Throwable t) {
            // 调度线程抛异常会导致 scheduleWithFixedDelay 停摆，后续轮次不再执行——必须吞住
            log.error("[audit] spool 重放轮次异常", t);
        }
    }

    void replayOnce() {
        if (!spoolDir.isDirectory()) {
            return;
        }
        recoverOrphanedReplaying();
        // Codex 四轮 Critical#1：孤儿 .retry.writing 回收——AuditSpoolWriter 降级路径写后原子发布中断
        // （崩溃/rename 失败）遗留的临时文件，扫描集固定排除 .writing，永远不会被下面的主流程认领
        recoverOrphanedWriting();
        // 扫描集只含 .jsonl/.retry，显式排除 spool.lock（锁文件永不参与扫描/重放）与 .writing
        // （Codex 四轮 Critical#1：写后原子发布的临时文件，内容可能不完整，绝不能被本扫描直接命中；
        // Codex 五轮起，超龄 .writing 孤儿只会被 recoverOrphanedWriting() 隔离为 .stale，不再回归本扫描集）。
        // .stale（隔离区，见类 Javadoc"Codex 五轮修复"段）天然不以 .jsonl/.retry 结尾，被本谓词排除，
        // 无需额外条件
        File[] targets = spoolDir.listFiles((dir, name) ->
                !name.equals(AuditSpoolLock.LOCK_FILE_NAME)
                        && !name.endsWith(".writing")
                        && (name.endsWith(".jsonl") || name.endsWith(".retry")));
        if (targets == null) {
            return;
        }
        long now = System.currentTimeMillis();
        for (File src : targets) {
            // Codex 批次二 #1：60 秒静默窗口——距今不足 60 秒的 .jsonl 视为写入端可能活跃，本轮跳过，
            // 下一轮再认领（<0.5 QPS，延迟一周期无害）。锁协议落地后此跳过仅为优化（减少活跃文件搅动），
            // 即便不跳过、正确性也由 AuditSpoolLock 的 append↔rename 互斥保证。.retry 由本重放器自建、
            // 无外部写入端，不受此限；.bad 本就不在扫描集（且从不重放）。
            if (src.getName().endsWith(".jsonl")
                    && now - src.lastModified() < QUIET_WINDOW_MS) {
                continue;
            }
            replayFile(src);
        }
    }

    /**
     * 内审 Important 修复：从文件名剥离<b>所有</b> {@code .<UUID>.retry} 层与 {@code .replaying.<UUID>} token 后缀，
     * 还原出稳定 base name（如 {@code audit-events-20260723.jsonl}），见类注释"内审 Important 修复"段。
     *
     * <p>剥离顺序：先去掉末尾 owner-token 化的 {@code .replaying.<UUID>} 后缀（认领时最后追加，
     * 是最外层，最多一层，用 {@link #REPLAYING_TOKEN_SUFFIX} 匹配），再去掉 {@link AuditSpoolWriter}
     * 降级路径写后原子发布的临时层 {@code .<UUID>.retry.writing}（Codex 四轮 Critical#1，
     * {@link #RETRY_WRITING_SUFFIX} 匹配，同样最多一层、与 replaying token 互斥不共存），最后循环剥离
     * 末尾的 {@code .<UUID>.retry} 层，直到不再匹配为止（病态场景下可能有多层，历史累积的每一层都会被剥掉）。
     * UUID 段用 {@link #RETRY_LAYER_SUFFIX}/{@link #REPLAYING_TOKEN_SUFFIX}/{@link #RETRY_WRITING_SUFFIX}
     * 严格匹配标准 8-4-4-4-12 十六进制格式并锚定字符串末尾，不会误剥业务文件名本身自带的点段。
     *
     * @param name 原始文件名（可能已带若干层 retry 后缀 + 一层 replaying token 或一层 writing 临时层，
     *             也可能是从未重放过的原始文件名）
     * @return 剥离干净的稳定 base name
     */
    private String stableBaseName(String name) {
        String base = name;
        Matcher replayingMatcher = REPLAYING_TOKEN_SUFFIX.matcher(base);
        if (replayingMatcher.find()) {
            base = base.substring(0, replayingMatcher.start());
        }
        // Codex 四轮 Critical#1：.retry.writing 只会是 AuditSpoolWriter 降级路径追加的最外层（写后原子
        // 发布的临时名，从不嵌套），与 .replaying.<uuid> 互斥同时出现，故也只剥最多一层
        Matcher writingMatcher = RETRY_WRITING_SUFFIX.matcher(base);
        if (writingMatcher.find()) {
            base = base.substring(0, writingMatcher.start());
        }
        Matcher matcher;
        while ((matcher = RETRY_LAYER_SUFFIX.matcher(base)).find()) {
            base = base.substring(0, matcher.start());
        }
        return base;
    }

    /**
     * 裁定修复 D：回收崩溃遗留的 .replaying 孤儿文件，见类注释。
     *
     * <p>只对"修改时间早于 now - {@link #ORPHAN_REPLAYING_AGE_MS}"的 {@code .replaying.<uuid>} 生效——
     * 年龄阈值防止误抢并发实例（或本实例本轮稍早）正在处理中的文件；检查+rename 在
     * {@link AuditSpoolLock} 临界区（锁协议临界区 5）完成，交回 {@code .retry} 通道由本轮或下一轮
     * {@link #replayFile(File)} 正常消化，本方法不直接读取/处理文件内容。
     *
     * <p><b>Codex 批次二三轮 #B（锁内重验）</b>：锁外 {@code listFiles} 预筛只作候选收集；预筛与锁内
     * rename 之间，该路径可能已被删除并复用为一个<b>新</b>文件（同名巧合），若沿用预筛结果直接搬走会把
     * 新文件误当旧孤儿。故锁内 rename 前<b>重新确认</b> orphan 仍存在且 mtime 仍超阈值，才执行 rename。
     */
    private void recoverOrphanedReplaying() {
        File[] orphans = spoolDir.listFiles((dir, name) ->
                !name.equals(AuditSpoolLock.LOCK_FILE_NAME)
                        && REPLAYING_TOKEN_SUFFIX.matcher(name).find());
        if (orphans == null || orphans.length == 0) {
            return;
        }
        long now = System.currentTimeMillis();
        for (final File orphan : orphans) {
            if (now - orphan.lastModified() < ORPHAN_REPLAYING_AGE_MS) {
                continue;   // 锁外预筛：未超龄，可能正被处理中，本轮不碰
            }
            String orphanName = orphan.getName();
            // 内审 Important 修复：先归一到稳定 base 再拼恰好一层新 UUID，防止孤儿在已累积多层
            // 后缀的名字上继续追加、文件名无界增长
            final File retry = new File(orphan.getParentFile(),
                    stableBaseName(orphanName) + "." + UUID.randomUUID() + ".retry");
            final boolean[] moved = {false};
            try {
                // 锁协议临界区 5：检查+rename 在锁内，与并发实例/本进程认领互斥；tryLock 拿不到则本轮跳过该孤儿
                boolean lockRun = spoolLock.tryLockAndRun(() -> {
                    // Codex#B 锁内重验：预筛后路径可能已被删除并复用为新文件，重新确认仍存在且仍超龄才搬走
                    if (!orphan.exists()) {
                        return;
                    }
                    if (System.currentTimeMillis() - orphan.lastModified() < ORPHAN_REPLAYING_AGE_MS) {
                        return;
                    }
                    Files.move(orphan.toPath(), retry.toPath(), StandardCopyOption.ATOMIC_MOVE);
                    moved[0] = true;
                });
                if (lockRun && moved[0]) {
                    log.warn("[audit] 回收孤儿 .replaying 文件: {} -> {}", orphanName, retry.getName());
                }
                // lockRun=false（他进程持锁）或 moved=false（锁内重验未过）：本轮跳过该孤儿，下轮再判
            } catch (IOException moveFail) {
                // 改名失败：可能已被其他实例/本轮稍早处理，下一轮再判一次
                log.warn("[audit] 孤儿 .replaying 回收改名失败，下轮再试: {}", orphanName);
            }
        }
    }

    /**
     * Codex 五轮修复：回收 {@link AuditSpoolWriter} 降级路径「写后原子发布」中断遗留的
     * {@code <稳定base>.<UUID>.retry.writing} 孤儿——<b>只隔离，永不发布为 .retry，永不删除</b>。
     *
     * <p>四轮版本曾把超龄孤儿直接改名回归 {@code .retry} 通道当正常事件重放，Codex 五轮复验指出这不安全：
     * 超龄（mtime 早于 {@link #ORPHAN_REPLAYING_AGE_MS}）只能说明"这个文件很久没有被更新"，不能证明
     * writer 已经死亡——它可能仍卡在那唯一一次 {@link java.nio.file.Files#write} 调用内部（磁盘/网络
     * 存储 IO 阻塞），FD 仍打开、进程仍存活。若照旧回归 {@code .retry} 通道，重放器会认领→读到空文件
     * （或崩溃语义下的半行前缀）→判定处理完毕→删除；卡住的 write 恢复后写向已 unlink 的 inode，数据
     * 静默消失，且不留任何错误日志。完整论证见类 Javadoc"Codex 五轮修复（.writing 隔离区）"段。
     *
     * <p>修复后动作：锁内重验（exists + 仍超龄，防止误抢正在写入中的文件）通过后，把 {@code .retry.writing}
     * <b>原子改名为 {@code <原名>.stale}</b>（同目录）并 {@code log.error} 告警（脱敏：仅文件名/大小/mtime，
     * 不含内容）——不解析内容、不写回任何可被扫描/认领的名字、不删除。{@code rename} 不 unlink：即便该
     * writer 随后真的恢复并把内容写完，也是写进同一个 inode（现名 {@code .stale}），内容不会因改名而
     * 丢失，只是滞留隔离区待人工核查补录（对照 {@code clientEventId} 去重，落库天然幂等）。writer 侧针对
     * "发布源被隔离"的自愈见 {@link AuditSpoolWriter} 类 Javadoc。
     */
    private void recoverOrphanedWriting() {
        // RETRY_WRITING_SUFFIX 只匹配以 ".<uuid>.retry.writing" 结尾的文件名——本方法产出的 .stale
        // 文件不再以此结尾，天然不会被下一轮扫描重新命中（见类 Javadoc"Codex 五轮修复"段）
        File[] orphans = spoolDir.listFiles((dir, name) ->
                !name.equals(AuditSpoolLock.LOCK_FILE_NAME)
                        && RETRY_WRITING_SUFFIX.matcher(name).find());
        if (orphans == null || orphans.length == 0) {
            return;
        }
        long now = System.currentTimeMillis();
        for (final File orphan : orphans) {
            if (now - orphan.lastModified() < ORPHAN_REPLAYING_AGE_MS) {
                continue;   // 锁外预筛：未超龄，可能正被 writer 写入中（或卡在 Files.write 内部），本轮不碰
            }
            String orphanName = orphan.getName();
            // Codex 五轮：只隔离，原名后追加 .stale——不发布为 .retry（不会被重放器认领）、不删除
            // （.stale 排除在一切扫描之外，见类 Javadoc"Codex 五轮修复"段的完整论证）
            final File stale = new File(orphan.getParentFile(), orphanName + STALE_SUFFIX);
            final long[] quarantinedSize = {-1L};
            final long[] quarantinedMtime = {-1L};
            final boolean[] moved = {false};
            try {
                // 锁内重验（同 recoverOrphanedReplaying 的 #B 模式）：预筛后、锁内 rename 前，路径可能
                // 已被 writer 完成发布并删除、或被别的实例先一步隔离，重新确认仍存在且仍超龄才动手
                boolean lockRun = spoolLock.tryLockAndRun(() -> {
                    if (!orphan.exists()) {
                        return;
                    }
                    if (System.currentTimeMillis() - orphan.lastModified() < ORPHAN_REPLAYING_AGE_MS) {
                        return;
                    }
                    quarantinedSize[0] = orphan.length();
                    quarantinedMtime[0] = orphan.lastModified();
                    Files.move(orphan.toPath(), stale.toPath(), StandardCopyOption.ATOMIC_MOVE);
                    moved[0] = true;
                });
                if (lockRun && moved[0]) {
                    // log.error 而非 warn：这是需要人工核查补录的隔离事件，不是可自愈的瞬时状况
                    log.error("[audit] 超龄 .retry.writing 隔离至 .stale（不自动发布/不自动删除，"
                                    + "需人工核查补录，见 AuditSpoolReplayer 类 Javadoc"
                                    + "\".writing 隔离区处置\"说明段）: file={} size={}bytes mtime={}",
                            stale.getName(), quarantinedSize[0], quarantinedMtime[0]);
                }
                // lockRun=false（他进程持锁）或 moved=false（锁内重验未过）：本轮跳过该孤儿，下轮再判
            } catch (IOException moveFail) {
                // 改名失败：可能已被其他实例/本轮稍早处理，下一轮再判一次
                log.warn("[audit] 孤儿 .retry.writing 隔离改名失败，下轮再试: {}", orphanName);
            }
        }
    }

    private void replayFile(final File src) {
        // 锁协议临界区 2（认领）：Codex 批次二 #1/#2/#B/#C。
        // #B+#C 认领名 owner-token 化——每次认领新生成 ownerToken，rename 目标 <src原名>.replaying.<ownerToken>。
        //   收尾/删除只操作本 owner token 的路径：陈旧 owner（被抢后收尾）与新 owner 永不同名，
        //   deleteIfExists 物理上不可能误删他 owner 刚认领的同名文件（消除 Codex#C 固定名跨 owner 冲突）。
        final File replaying = new File(src.getParentFile(),
                src.getName() + REPLAYING_SUFFIX + "." + UUID.randomUUID());
        final boolean isJsonl = src.getName().endsWith(".jsonl");
        final boolean[] renamed = {false};
        // rename→.replaying.<owner> 与 setLastModified 刷新租约在同一临界区（进程内 monitor + 跨进程 FileLock）：
        //   #1：与 writer 的 append 互斥，消除"取走文件瞬间"与"正在 append"的窗口竞态；
        //   #2：rename 与刷新原子成对，消除"rename 成功后、刷新 mtime 前被孤儿回收误抢"的窗口。
        try {
            boolean lockRun = spoolLock.tryLockAndRun(() -> {
                // Codex#B 锁内重验：锁外预筛（quiet window）与锁内之间，src 可能已被他 owner 取走/删除。
                // 锁内重新确认 src 仍存在（.jsonl 且仍在静默窗口内说明写入端可能活跃，让给下轮）才认领。
                if (!src.exists()) {
                    return;
                }
                if (isJsonl && System.currentTimeMillis() - src.lastModified() < QUIET_WINDOW_MS) {
                    return;
                }
                Files.move(src.toPath(), replaying.toPath(), StandardCopyOption.ATOMIC_MOVE);
                // setLastModified 返回 false 仅告警不中断。Codex#C：owner-token 唯一名下，租约建立失败的最坏
                // 后果=本文件超 10 分钟后被孤儿回收→重复重放（落库按 client_event_id 幂等吸收），不再有丢失路径。
                if (!replaying.setLastModified(System.currentTimeMillis())) {
                    log.warn("[audit] 认领后刷新 spool 租约 mtime 失败（最坏后果=重复重放，幂等吸收）: {}",
                            replaying.getName());
                }
                renamed[0] = true;
            });
            if (!lockRun) {
                // FileLock 被别的进程持有 → 本轮跳过该文件，下轮再认领
                return;
            }
            if (!renamed[0]) {
                // 锁内重验未过（已被他 owner 取走/删除，或 .jsonl 仍活跃）→ 本轮跳过，下轮再判
                return;
            }
        } catch (IOException moveFail) {
            // 改名失败（正被他进程/他轮取走）→ 跳过，下轮再试
            log.warn("[audit] spool 文件占用改名失败，跳过: {}", src.getName());
            return;
        }

        // 锁协议临界区外（readAllLines）：认领后 .replaying 已被本进程独占，读取不阻塞 writer
        List<String> lines;
        try {
            lines = Files.readAllLines(replaying.toPath(), StandardCharsets.UTF_8);
        } catch (IOException readFail) {
            // 读失败不删 .replaying（避免读失败即丢数据），留待人工/下次进程处理
            log.error("[audit] spool 读取失败，保留 .replaying: {}", replaying.getName(), readFail);
            return;
        }

        // 锁协议临界区外（逐行 deliverOnce=网络慢操作）：把行分类为 remaining（投递失败待重试）
        // 与 badLines（反序列化失败待隔离），此循环不做任何 spool 文件写操作
        List<String> remaining = new ArrayList<>();
        List<String> badLines = new ArrayList<>();
        for (int i = 0; i < lines.size(); i++) {
            String line = lines.get(i);
            if (line == null || line.trim().isEmpty()) {
                continue;
            }
            AuditEventDTO event;
            try {
                event = JsonUtil.getJsonToBean(line, AuditEventDTO.class);
            } catch (Exception parseFail) {
                // Codex 批次二 #3：坏行留待收尾批量隔离到 .bad（不进 remaining、不静默丢）。
                // Codex 批次二 #4：日志只记文件名/行号/行长/内容 SHA-256 前 12 位 + 异常类名，绝不输出行正文
                // （parseFail 只取类名——Jackson 异常正文可能内嵌源片段，故不整体打印）。
                badLines.add(line);
                log.error("[audit] spool 坏行检出待隔离: file={} lineNo={} lineLen={} sha256_12={} cause={}",
                        replaying.getName(), i + 1, line.length(), sha256Prefix12(line),
                        parseFail.getClass().getName());
                continue;
            }
            try {
                publisher.deliverOnce(event);   // 成功→丢弃该行；失败→保留待下轮
            } catch (Exception deliverFail) {
                // 好行投递失败（AuditDeliveryException）→ 保留待下轮，行为不变
                remaining.add(line);
            }
        }

        // 锁协议临界区 4（收尾）：写 .bad / 写 .retry / 删 .replaying 在锁内（各自快操作）。
        // 用 writer 语义 lockAndRun（始终执行）：投递已完成，收尾必须落地，避免留 .replaying 被重复重放
        try {
            spoolLock.lockAndRun(() -> finishReplay(src, replaying, remaining, badLines));
        } catch (IOException writeFail) {
            log.error("[audit] spool 重放收尾写盘失败，保留 .replaying: {}", replaying.getName(), writeFail);
        }
    }

    /**
     * 锁协议临界区 4（收尾）：批量写坏行 {@code .bad}、写剩余行 {@code .retry}、删 {@code .replaying}。
     * 全程由 {@link AuditSpoolLock#lockAndRun} 保护，仅做快速文件操作。
     *
     * <p>Codex 批次二 #3：坏行批量隔离到 {@code <稳定 base>.<UUID>.bad}（不重放、不静默丢）；写 {@code .bad}
     * 失败则该批坏行退回 remaining（宁可重试也不丢）。剩余行（含投递失败 + 隔离失败退回的坏行）非空则写
     * {@code <稳定 base>.<UUID>.retry}（终验四轮③：绝不写回活动原名），否则删 {@code .replaying}；有 remaining
     * 时也在 {@code .retry} 写成功后删 {@code .replaying}（不留孤儿）。
     *
     * @throws IOException 写 {@code .retry} 或删 {@code .replaying} 失败——调用方据此保留 {@code .replaying} 待下轮
     */
    private void finishReplay(File src, File replaying, List<String> remaining, List<String> badLines)
            throws IOException {
        List<String> retryLines = new ArrayList<>(remaining);
        if (!badLines.isEmpty()) {
            try {
                writeNewSpoolFile(src, ".bad", badLines);
            } catch (IOException badFail) {
                // 写 .bad 失败：坏行退回重试（宁可重试也不丢）；日志不含正文，只记文件名/条数
                retryLines.addAll(badLines);
                log.warn("[audit] spool 坏行隔离写盘失败，退回重试: file={} count={}",
                        replaying.getName(), badLines.size(), badFail);
            }
        }
        if (retryLines.isEmpty()) {
            Files.deleteIfExists(replaying.toPath());   // 全部成功：删 .replaying
        } else {
            // 部分失败：剩余行写 <稳定 base>.<UUID>.retry 新文件，绝不写回活动原名（终验四轮③）
            writeNewSpoolFile(src, ".retry", retryLines);
            Files.deleteIfExists(replaying.toPath());   // .retry 写成功后删 .replaying，不留孤儿
        }
    }

    /**
     * 把 {@code lines} 写入一个新建的 {@code <稳定 base>.<UUID><suffix>} 文件（{@code .retry} 或 {@code .bad}）。
     * 前缀先经 {@link #stableBaseName(String)} 归一，再拼恰好一层新 UUID，文件名长度有界（内审 Important 修复）；
     * {@link StandardOpenOption#CREATE_NEW} 保证不覆盖既有文件，UUID 使撞名概率可忽略。
     */
    private void writeNewSpoolFile(File src, String suffix, List<String> lines) throws IOException {
        File out = new File(src.getParentFile(),
                stableBaseName(src.getName()) + "." + UUID.randomUUID() + suffix);
        StringBuilder sb = new StringBuilder();
        for (String line : lines) {
            sb.append(line).append('\n');
        }
        Files.write(out.toPath(), sb.toString().getBytes(StandardCharsets.UTF_8),
                StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE);
    }

    /**
     * Codex 批次二 #4：坏行内容的 SHA-256 十六进制<b>前 12 位</b>，供日志定位坏行而不泄露正文
     * （spec "ERROR 日志不含字段值正文"）。SHA-256 为 JDK 标配算法，{@link NoSuchAlgorithmException}
     * 理论不可达；万一不可达则回落到固定占位串，绝不回退成打印正文。
     */
    private static String sha256Prefix12(String line) {
        try {
            MessageDigest md = MessageDigest.getInstance("SHA-256");
            byte[] digest = md.digest(line.getBytes(StandardCharsets.UTF_8));
            StringBuilder hex = new StringBuilder(12);
            for (byte b : digest) {
                hex.append(Character.forDigit((b >> 4) & 0xF, 16));
                hex.append(Character.forDigit(b & 0xF, 16));
                if (hex.length() >= 12) {
                    break;
                }
            }
            return hex.substring(0, 12);
        } catch (NoSuchAlgorithmException e) {
            return "sha256-unavailable";
        }
    }
}