package com.haobang.wal;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.haobang.config.AppProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.Charset;
import java.nio.file.AtomicMoveNotSupportedException;
import java.nio.file.DirectoryStream;
import java.nio.file.FileStore;
import java.nio.file.FileAlreadyExistsException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardCopyOption;
import java.nio.file.StandardOpenOption;
import java.nio.file.attribute.PosixFilePermission;
import java.security.MessageDigest;
import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.EnumSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;

/**
 * Kafka 接管前的本地持久化 WAL。
 *
 * <p>网络线程只向有界队列提交事件；单写线程顺序追加并批量 force。Dispatcher 只读取已经
 * force 的记录，Kafka ACK 后按连续 sequence 推进持久化游标。崩溃恢复可能重复投递游标后
 * 的少量记录，但 event_id 不变，由下游幂等约束消除业务重复。</p>
 */
@Component
public class SyslogWalService {
    private static final Logger log = LoggerFactory.getLogger(SyslogWalService.class);
    private static final DateTimeFormatter RECORD_TIME = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS");
    private static final DateTimeFormatter FILE_TIME = DateTimeFormatter.ofPattern("yyyyMMdd-HHmmssSSS");
    private static final String SEGMENT_GLOB = "syslog-wal-*.jsonl";

    public enum AdmissionResult { ACCEPTED, BACKPRESSURE, UNAVAILABLE }

    private final AppProperties properties;
    private final ObjectMapper objectMapper;
    private final Charset syslogCharset;
    private final PointTimeAllocator pointTimeAllocator = new PointTimeAllocator();
    private final AtomicBoolean initialized = new AtomicBoolean();
    private final AtomicLong nextSequence = new AtomicLong();
    private final AtomicLong durableSequence = new AtomicLong();
    private final LongAdder appended = new LongAdder();
    private final LongAdder admissionBackpressure = new LongAdder();
    private final LongAdder quarantined = new LongAdder();
    private final LongAdder terminalRecorded = new LongAdder();
    private final Object durableMonitor = new Object();
    private final Object checkpointMonitor = new Object();

    private volatile boolean running;
    private volatile Throwable fatalFailure;
    private volatile Checkpoint persistedCheckpoint = new Checkpoint();
    private volatile Checkpoint pendingCheckpoint = new Checkpoint();
    private volatile long checkpointDirtyRecords;
    private volatile long lastCheckpointFlushNanos;
    private volatile DiskState diskState = DiskState.NORMAL;
    private volatile long lastDiskCheckNanos;

    private ArrayBlockingQueue<SyslogIngressEvent> appendQueue;
    private Thread writerThread;
    private Path rootDirectory;
    private Path activeDirectory;
    private Path completedDirectory;
    private Path quarantineDirectory;
    private Path terminalDirectory;
    private Path checkpointDirectory;
    private Path checkpointFile;
    private FileChannel writerChannel;
    private Path writerPath;
    private LocalDate writerDate;
    private long writerBytes;
    private int recordsSinceForce;
    private long lastWrittenSequence;
    private long lastForceNanos;

    public SyslogWalService(AppProperties properties, ObjectMapper objectMapper) {
        this.properties = properties;
        /*
         * WAL 是内部、固定 schema 的持久化协议，不能直接复用 Redis 缓存的多态 ObjectMapper。
         *
         * CacheConfig 为 Redis 的 LinkedHashMap/实体反序列化开启了 default typing。直接复用它会在
         * WAL JSON 中写入 "@class"，但它的安全白名单又不允许 com.haobang.wal.SyslogWalRecord，
         * 最终造成“自己写出的第一条 WAL 自己无法读取”，Dispatcher 永久停在 offset=0。
         *
         * copy() 保留 JavaTimeModule 等通用模块，但 deactivateDefaultTyping() 明确移除 Redis 专用的
         * 多态协议。这样新 WAL 不再依赖 Java 类名，也不会通过磁盘 JSON 触发任意类型解析。
         * SyslogWalRecord 对历史 "@class" 字段做了精确忽略，因此现场已经落盘的旧记录仍可恢复。
         */
        this.objectMapper = objectMapper.copy().deactivateDefaultTyping();
        this.syslogCharset = resolveSyslogCharset(properties.getSyslogCharset());
    }

    @PostConstruct
    public void initialize() {
        if (!initialized.compareAndSet(false, true)) return;
        validateConfiguration();
        try {
            rootDirectory = Paths.get(properties.getSyslogWalDirectory()).toAbsolutePath().normalize();
            activeDirectory = rootDirectory.resolve("active");
            completedDirectory = rootDirectory.resolve("completed");
            quarantineDirectory = rootDirectory.resolve("quarantine");
            terminalDirectory = rootDirectory.resolve("terminal");
            checkpointDirectory = rootDirectory.resolve("checkpoint");
            checkpointFile = checkpointDirectory.resolve("dispatch-checkpoint.json");
            Files.createDirectories(activeDirectory);
            Files.createDirectories(completedDirectory);
            Files.createDirectories(quarantineDirectory);
            Files.createDirectories(terminalDirectory);
            Files.createDirectories(checkpointDirectory);
            applyDirectoryPermissions(rootDirectory);
            applyDirectoryPermissions(activeDirectory);
            applyDirectoryPermissions(completedDirectory);
            applyDirectoryPermissions(quarantineDirectory);
            applyDirectoryPermissions(terminalDirectory);
            applyDirectoryPermissions(checkpointDirectory);

            repairIncompleteTails();
            persistedCheckpoint = loadCheckpoint();
            pendingCheckpoint = persistedCheckpoint.copy();
            long maxSequence = Math.max(persistedCheckpoint.sequence, scanMaxSequence());
            nextSequence.set(maxSequence);
            durableSequence.set(maxSequence);
            cleanupCompletedSegments();
            refreshDiskState(true);
        } catch (Exception failure) {
            fatalFailure = failure;
            throw new IllegalStateException("Syslog WAL 初始化失败，拒绝开放接入端口: "
                    + properties.getSyslogWalDirectory(), failure);
        }

        appendQueue = new ArrayBlockingQueue<>(properties.getSyslogWalQueueCapacity());
        running = true;
        lastForceNanos = System.nanoTime();
        lastCheckpointFlushNanos = System.nanoTime();
        writerThread = new Thread(this::writerLoop, "syslog-wal-writer");
        writerThread.setDaemon(true);
        writerThread.start();
        log.info("Syslog WAL 已启用: directory={}, charset={}, queueCapacity={}, highWatermark={}%, lowWatermark={}%, "
                        + "flushRecords={}, flushIntervalMs={}, checkpointSequence={}",
                rootDirectory, syslogCharset.name(), properties.getSyslogWalQueueCapacity(),
                properties.getSyslogWalHighWatermarkPercent(), properties.getSyslogWalLowWatermarkPercent(),
                properties.getSyslogWalFlushRecords(), properties.getSyslogWalFlushIntervalMs(),
                persistedCheckpoint.sequence);
    }

    public AdmissionResult tryAppend(SyslogIngressEvent event) {
        if (properties.isSyslogPointTimeWriteEnabled() && event.getPointTimeNs() == null) {
            throw new IllegalArgumentException("point_time 写入已开启，SyslogIngressEvent 必须先固化 point_time_ns");
        }
        refreshDiskState(false);
        if (!running || fatalFailure != null || diskState == DiskState.CRITICAL) {
            return AdmissionResult.UNAVAILABLE;
        }
        if (!appendQueue.offer(event)) {
            admissionBackpressure.increment();
            return AdmissionResult.BACKPRESSURE;
        }
        return AdmissionResult.ACCEPTED;
    }

    public boolean shouldPauseTcp() {
        refreshDiskState(false);
        if (fatalFailure != null || diskState != DiskState.NORMAL) return true;
        int high = percentageCount(properties.getSyslogWalHighWatermarkPercent());
        return appendQueue != null && appendQueue.size() >= high;
    }

    public boolean canResumeTcp() {
        refreshDiskState(false);
        if (fatalFailure != null || diskState != DiskState.NORMAL) return false;
        int low = percentageCount(properties.getSyslogWalLowWatermarkPercent());
        return appendQueue != null && appendQueue.size() <= low;
    }

    public long getRetryIntervalMs() {
        refreshDiskState(false);
        return diskState == DiskState.NORMAL && fatalFailure == null
                ? properties.getSyslogWalRetryIntervalMs()
                : properties.getSyslogWalDiskRetryIntervalMs();
    }
    public int getQueueSize() { return appendQueue == null ? 0 : appendQueue.size(); }
    public int getQueueCapacity() { return properties.getSyslogWalQueueCapacity(); }
    public long getAppendedCount() { return appended.sum(); }
    public long getBackpressureCount() { return admissionBackpressure.sum(); }
    public long getQuarantinedCount() { return quarantined.sum(); }
    public long getTerminalRecordedCount() { return terminalRecorded.sum(); }
    public long getIngressPointTimeAssignedCount() { return pointTimeAllocator.getAssignedCount(); }
    public long getDurableSequence() { return durableSequence.get(); }
    public long getCheckpointSequence() { return persistedCheckpoint.sequence; }
    public long getPendingRecordsApprox() {
        return Math.max(0L, nextSequence.get() - persistedCheckpoint.sequence);
    }
    public Throwable getFatalFailure() { return fatalFailure; }
    public boolean isDiskCritical() { refreshDiskState(false); return diskState == DiskState.CRITICAL; }

    /** 每个事件只调用一次；TCP 背压重试必须复用返回对象，不能重新分配 Point 身份。 */
    public SyslogIngressEvent prepareForWal(SyslogIngressEvent event) {
        if (!properties.isSyslogPointTimeWriteEnabled() || event.getPointTimeNs() != null) {
            return event;
        }
        long receiveTimeMillis = event.getReceivedAt().atZone(ZoneId.of("Asia/Shanghai"))
                .toInstant().toEpochMilli();
        return event.withPointTimeNs(pointTimeAllocator.allocate(receiveTimeMillis));
    }

    public WalCursor openCursor() throws IOException {
        return new WalCursor(persistedCheckpoint.copy());
    }

    /** 仅由 Dispatcher 在连续 ACK 区间推进后调用。 */
    public void markCheckpoint(SyslogWalRecord record) {
        synchronized (checkpointMonitor) {
            if (record.getSequence() <= pendingCheckpoint.sequence) return;
            pendingCheckpoint = new Checkpoint(record.getSequence(), record.getSegment(), record.getEndOffset());
            checkpointDirtyRecords++;
            long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - lastCheckpointFlushNanos);
            if (checkpointDirtyRecords >= properties.getSyslogWalCheckpointFlushRecords()
                    || elapsedMs >= properties.getSyslogWalCheckpointFlushIntervalMs()) {
                flushCheckpointLocked();
            }
        }
    }

    /** 永久异常必须先持久化隔离记录，调用方才能把对应 sequence 视为完成。 */
    public void quarantine(SyslogWalRecord record, String reason, Throwable failure) throws IOException {
        Map<String, Object> envelope = new LinkedHashMap<>();
        envelope.put("quarantined_at", Instant.now().toString());
        envelope.put("reason", reason);
        envelope.put("error_type", failure == null ? null : failure.getClass().getName());
        envelope.put("error_message", failure == null ? null : failure.getMessage());
        envelope.put("wal_segment", record.getSegment());
        envelope.put("wal_offset", record.getStartOffset());
        envelope.put("record", record);
        byte[] json = objectMapper.writeValueAsBytes(envelope);
        Path target = quarantineDirectory.resolve(String.format("quarantine-%020d-%s.json",
                record.getSequence(), record.getEventId()));
        try {
            try (FileChannel channel = FileChannel.open(target, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE)) {
                writeFully(channel, json);
                channel.force(true);
            }
            applyFilePermissions(target);
        } catch (FileAlreadyExistsException alreadyPersisted) {
            // 进程可能在隔离文件 force 后、游标推进前退出；重复恢复视为已持久化。
            return;
        }
        quarantined.increment();
    }

    /** 业务终态先持久化原因，随后才允许 Dispatcher 推进连续 ACK 游标。 */
    public void recordTerminal(SyslogWalRecord record, String type, String reason) throws IOException {
        Map<String, Object> envelope = new LinkedHashMap<>();
        envelope.put("completed_at", Instant.now().toString());
        envelope.put("type", type);
        envelope.put("reason", reason);
        envelope.put("event_id", record.getEventId());
        envelope.put("sequence", record.getSequence());
        envelope.put("source_ip", record.getSourceIp());
        envelope.put("wal_segment", record.getSegment());
        envelope.put("wal_offset", record.getStartOffset());
        byte[] json = objectMapper.writeValueAsBytes(envelope);
        Path target = terminalDirectory.resolve(String.format("terminal-%020d-%s.json",
                record.getSequence(), record.getEventId()));
        try {
            try (FileChannel channel = FileChannel.open(target, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE)) {
                writeFully(channel, json);
                channel.force(true);
            }
            applyFilePermissions(target);
        } catch (FileAlreadyExistsException alreadyPersisted) {
            return;
        }
        terminalRecorded.increment();
    }

    private void writerLoop() {
        try {
            while (running || !appendQueue.isEmpty()) {
                SyslogIngressEvent event = appendQueue.poll(properties.getSyslogWalFlushIntervalMs(), TimeUnit.MILLISECONDS);
                if (event != null) appendEvent(event);
                long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - lastForceNanos);
                if (recordsSinceForce > 0 && (recordsSinceForce >= properties.getSyslogWalFlushRecords()
                        || elapsedMs >= properties.getSyslogWalFlushIntervalMs())) {
                    forceWriter();
                }
                flushCheckpointIfDue();
            }
            forceWriter();
            synchronized (checkpointMonitor) { flushCheckpointLocked(); }
        } catch (InterruptedException interrupted) {
            Thread.currentThread().interrupt();
            failWal(interrupted);
        } catch (Throwable failure) {
            failWal(failure);
        } finally {
            closeWriter();
        }
    }

    private void appendEvent(SyslogIngressEvent event) throws Exception {
        long sequence = nextSequence.incrementAndGet();
        SyslogWalRecord record = toRecord(sequence, event);
        byte[] json = objectMapper.writeValueAsBytes(record);
        byte[] line = new byte[json.length + 1];
        System.arraycopy(json, 0, line, 0, json.length);
        line[line.length - 1] = (byte) '\n';

        LocalDate receiveDate = event.getReceivedAt().toLocalDate();
        if (writerChannel == null) openWriter(sequence, receiveDate);
        if (writerBytes > 0 && (writerBytes + line.length > properties.getSyslogWalSegmentMaxBytes()
                || !receiveDate.equals(writerDate))) {
            forceWriter();
            closeWriter();
            openWriter(sequence, receiveDate);
        }
        writeFully(writerChannel, line);
        writerBytes += line.length;
        recordsSinceForce++;
        lastWrittenSequence = sequence;
    }

    private SyslogWalRecord toRecord(long sequence, SyslogIngressEvent event) throws Exception {
        byte[] raw = event.getRawMessage().getBytes(syslogCharset);
        SyslogWalRecord record = new SyslogWalRecord();
        if (properties.isSyslogPointTimeWriteEnabled()) {
            record.setSchemaVersion(3);
            record.setPointTimeNs(event.getPointTimeNs());
        } else {
            record.setSchemaVersion(2);
        }
        record.setSequence(sequence);
        record.setEventId(event.getEventId());
        record.setReceiveTime(event.getReceivedAt().format(RECORD_TIME));
        record.setSourceIp(event.getSourceIp());
        record.setSourcePort(event.getSourcePort());
        record.setProtocol(event.getProtocol());
        record.setCharset(syslogCharset.name());
        record.setRawLength(raw.length);
        record.setRawSha256(sha256(raw));
        record.setRawMessage(event.getRawMessage());
        SyslogWalRecordValidator.validateRecord(record);
        return record;
    }

    private void forceWriter() throws IOException {
        if (writerChannel == null || recordsSinceForce == 0) return;
        writerChannel.force(false);
        durableSequence.set(lastWrittenSequence);
        appended.add(recordsSinceForce);
        recordsSinceForce = 0;
        lastForceNanos = System.nanoTime();
        synchronized (durableMonitor) { durableMonitor.notifyAll(); }
    }

    private void openWriter(long startSequence, LocalDate date) throws IOException {
        String name = String.format("syslog-wal-%020d-%s.jsonl", startSequence,
                LocalDateTime.now().format(FILE_TIME));
        writerPath = activeDirectory.resolve(name);
        writerChannel = FileChannel.open(writerPath, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE);
        applyFilePermissions(writerPath);
        writerDate = date;
        writerBytes = 0L;
    }

    private void closeWriter() {
        if (writerChannel == null) return;
        try { writerChannel.close(); }
        catch (IOException failure) { log.warn("关闭 WAL 分段失败: file={}", writerPath, failure); }
        finally { writerChannel = null; }
    }

    private void failWal(Throwable failure) {
        fatalFailure = failure;
        running = false;
        synchronized (durableMonitor) { durableMonitor.notifyAll(); }
        log.error("Syslog WAL 发生致命故障，接入已进入 fail-closed 状态", failure);
    }

    private int percentageCount(int percent) {
        return Math.max(1, (int) Math.floor(properties.getSyslogWalQueueCapacity() * percent / 100.0));
    }

    private void validateConfiguration() {
        if (!properties.isSyslogWalEnabled()) {
            throw new IllegalArgumentException("syslog.wal.enabled=false 会破坏可靠接入边界，正式链路禁止关闭 WAL");
        }
        if (properties.getSyslogWalQueueCapacity() <= 0 || properties.getSyslogWalFlushRecords() <= 0
                || properties.getSyslogWalFlushIntervalMs() <= 0 || properties.getSyslogWalSegmentMaxBytes() <= 0
                || properties.getSyslogWalShutdownTimeoutMs() <= 0
                || properties.getSyslogWalCheckpointFlushRecords() <= 0
                || properties.getSyslogWalCheckpointFlushIntervalMs() <= 0) {
            throw new IllegalArgumentException("WAL 队列、刷盘、分段、游标和停机参数必须大于 0");
        }
        if (properties.getSyslogWalRetryIntervalMs() <= 0
                || properties.getSyslogWalDiskRetryIntervalMs() <= 0) {
            throw new IllegalArgumentException("WAL 队列和磁盘重试间隔必须大于 0");
        }
        if (properties.getSyslogWalLowWatermarkPercent() < 0
                || properties.getSyslogWalHighWatermarkPercent() <= properties.getSyslogWalLowWatermarkPercent()
                || properties.getSyslogWalHighWatermarkPercent() >= 100) {
            throw new IllegalArgumentException("WAL 水位必须满足 0 <= low < high < 100");
        }
        if (properties.getSyslogWalDiskHighWatermarkPercent() <= 0
                || properties.getSyslogWalDiskCriticalWatermarkPercent()
                <= properties.getSyslogWalDiskHighWatermarkPercent()
                || properties.getSyslogWalDiskCriticalWatermarkPercent() >= 100) {
            throw new IllegalArgumentException("WAL 磁盘水位必须满足 0 < high < critical < 100");
        }
        if (properties.getSyslogWalRetentionDays() < 0 || properties.getSyslogWalMinFreeBytes() < 0) {
            throw new IllegalArgumentException("WAL 保留天数和最小剩余字节不能小于 0");
        }
    }

    public Charset getSyslogCharset() {
        return syslogCharset;
    }

    private Charset resolveSyslogCharset(String configuredCharset) {
        if (configuredCharset == null || configuredCharset.trim().isEmpty()) {
            throw new IllegalArgumentException("syslog.charset 不能为空");
        }
        try {
            return Charset.forName(configuredCharset.trim());
        } catch (Exception invalidCharset) {
            throw new IllegalArgumentException("不支持的 syslog.charset: " + configuredCharset, invalidCharset);
        }
    }

    private void repairIncompleteTails() throws IOException {
        for (Path path : listSegments(activeDirectory)) {
            long size = Files.size(path);
            if (size == 0) continue;
            try (FileChannel channel = FileChannel.open(path, StandardOpenOption.READ, StandardOpenOption.WRITE)) {
                ByteBuffer one = ByteBuffer.allocate(1);
                channel.position(size - 1);
                channel.read(one);
                one.flip();
                if (one.get() == (byte) '\n') continue;
                long position = size - 1;
                while (position >= 0) {
                    one.clear();
                    channel.position(position);
                    channel.read(one);
                    one.flip();
                    if (one.get() == (byte) '\n') break;
                    position--;
                }
                long truncateAt = position < 0 ? 0 : position + 1;
                channel.truncate(truncateAt);
                channel.force(true);
                log.warn("WAL 启动恢复已截断未完成尾记录: file={}, oldBytes={}, newBytes={}",
                        path.getFileName(), size, truncateAt);
            }
        }
    }

    private long scanMaxSequence() throws IOException {
        long max = 0L;
        for (Path path : listSegments(activeDirectory)) {
            try (java.io.BufferedReader reader = Files.newBufferedReader(path, java.nio.charset.StandardCharsets.UTF_8)) {
                String line;
                while ((line = reader.readLine()) != null) {
                    if (line.trim().isEmpty()) continue;
                    JsonNode node = objectMapper.readTree(line);
                    JsonNode sequenceNode = node.get("sequence");
                    if (sequenceNode == null || !sequenceNode.isIntegralNumber()
                            || !sequenceNode.canConvertToLong() || sequenceNode.longValue() <= 0L) {
                        throw new IOException("WAL sequence 缺失或非法: " + path);
                    }
                    try {
                        SyslogWalRecordValidator.validateNode(node);
                    } catch (IllegalArgumentException invalidRecord) {
                        log.error("WAL 扫描发现损坏记录，启动后将由游标隔离: file={}, sequence={}, reason={}",
                                path, sequenceNode.longValue(), invalidRecord.getMessage());
                    }
                    max = Math.max(max, sequenceNode.longValue());
                }
            }
        }
        return max;
    }

    private Checkpoint loadCheckpoint() throws IOException {
        if (!Files.exists(checkpointFile)) return new Checkpoint();
        return objectMapper.readValue(Files.readAllBytes(checkpointFile), Checkpoint.class);
    }

    private void flushCheckpointIfDue() {
        synchronized (checkpointMonitor) {
            if (checkpointDirtyRecords == 0) return;
            long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - lastCheckpointFlushNanos);
            if (elapsedMs >= properties.getSyslogWalCheckpointFlushIntervalMs()) flushCheckpointLocked();
        }
    }

    private void flushCheckpointLocked() {
        if (checkpointDirtyRecords == 0 || pendingCheckpoint.sequence <= persistedCheckpoint.sequence) return;
        Path temporary = checkpointDirectory.resolve("dispatch-checkpoint.tmp");
        try {
            byte[] json = objectMapper.writeValueAsBytes(pendingCheckpoint);
            try (FileChannel channel = FileChannel.open(temporary, StandardOpenOption.CREATE,
                    StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.WRITE)) {
                writeFully(channel, json);
                channel.force(true);
            }
            try {
                Files.move(temporary, checkpointFile, StandardCopyOption.REPLACE_EXISTING,
                        StandardCopyOption.ATOMIC_MOVE);
            } catch (AtomicMoveNotSupportedException unsupported) {
                Files.move(temporary, checkpointFile, StandardCopyOption.REPLACE_EXISTING);
            }
            applyFilePermissions(checkpointFile);
            forceDirectoryBestEffort(checkpointDirectory);
            persistedCheckpoint = pendingCheckpoint.copy();
            checkpointDirtyRecords = 0L;
            lastCheckpointFlushNanos = System.nanoTime();
            moveCompletedSegments();
        } catch (IOException failure) {
            fatalFailure = failure;
            running = false;
            log.error("持久化 WAL ACK 游标失败，接入已停止", failure);
        }
    }

    private void moveCompletedSegments() throws IOException {
        String checkpointSegment = persistedCheckpoint.segment;
        if (checkpointSegment == null) return;
        for (Path segment : listSegments(activeDirectory)) {
            String name = segment.getFileName().toString();
            if (name.compareTo(checkpointSegment) >= 0 || segment.equals(writerPath)) continue;
            Files.move(segment, completedDirectory.resolve(name), StandardCopyOption.REPLACE_EXISTING);
        }
        cleanupCompletedSegments();
    }

    private void cleanupCompletedSegments() throws IOException {
        int retentionDays = properties.getSyslogWalRetentionDays();
        if (retentionDays <= 0) return;
        Instant cutoff = LocalDate.now().minusDays(retentionDays).atStartOfDay(ZoneId.systemDefault()).toInstant();
        for (Path path : listSegments(completedDirectory)) {
            if (Files.getLastModifiedTime(path).toInstant().isBefore(cutoff)) Files.deleteIfExists(path);
        }
    }

    private List<Path> listSegments(Path directory) throws IOException {
        if (directory == null || !Files.exists(directory)) return Collections.emptyList();
        List<Path> paths = new ArrayList<>();
        try (DirectoryStream<Path> stream = Files.newDirectoryStream(directory, SEGMENT_GLOB)) {
            for (Path path : stream) paths.add(path);
        }
        paths.sort(Comparator.comparing(path -> path.getFileName().toString()));
        return paths;
    }

    private void refreshDiskState(boolean force) {
        if (rootDirectory == null) return;
        long now = System.nanoTime();
        if (!force && TimeUnit.NANOSECONDS.toMillis(now - lastDiskCheckNanos) < 1000) return;
        lastDiskCheckNanos = now;
        try {
            FileStore store = Files.getFileStore(rootDirectory);
            long total = store.getTotalSpace();
            long usable = store.getUsableSpace();
            long usedPercent = total <= 0 ? 0 : ((total - usable) * 100L / total);
            DiskState next = usedPercent >= properties.getSyslogWalDiskCriticalWatermarkPercent()
                    || usable < properties.getSyslogWalMinFreeBytes() ? DiskState.CRITICAL
                    : usedPercent >= properties.getSyslogWalDiskHighWatermarkPercent()
                    ? DiskState.HIGH : DiskState.NORMAL;
            if (next != diskState) {
                log.warn("WAL 磁盘状态变化: previous={}, current={}, usedPercent={}%, usableBytes={}",
                        diskState, next, usedPercent, usable);
            }
            diskState = next;
        } catch (IOException failure) {
            diskState = DiskState.CRITICAL;
            fatalFailure = failure;
            log.error("无法检查 WAL 磁盘容量，按 fail-closed 处理", failure);
        }
    }

    private String sha256(byte[] value) throws Exception {
        byte[] digest = MessageDigest.getInstance("SHA-256").digest(value);
        StringBuilder hex = new StringBuilder(digest.length * 2);
        for (byte item : digest) hex.append(String.format("%02x", item & 0xff));
        return hex.toString();
    }

    private void applyFilePermissions(Path path) {
        try {
            Files.setPosixFilePermissions(path, EnumSet.of(
                    PosixFilePermission.OWNER_READ, PosixFilePermission.OWNER_WRITE));
        } catch (UnsupportedOperationException | IOException failure) {
            log.warn("当前文件系统不支持 WAL 文件 POSIX 0600 权限，请检查目录 ACL: file={}", path);
        }
    }

    private void applyDirectoryPermissions(Path path) {
        try {
            Set<PosixFilePermission> permissions = EnumSet.of(
                    PosixFilePermission.OWNER_READ,
                    PosixFilePermission.OWNER_WRITE,
                    PosixFilePermission.OWNER_EXECUTE);
            Files.setPosixFilePermissions(path, permissions);
        } catch (UnsupportedOperationException | IOException failure) {
            log.warn("当前文件系统不支持 WAL 目录 POSIX 0700 权限，请检查目录 ACL: directory={}", path);
        }
    }

    private void forceDirectoryBestEffort(Path directory) {
        try (FileChannel channel = FileChannel.open(directory, StandardOpenOption.READ)) {
            channel.force(true);
        } catch (UnsupportedOperationException | IOException failure) {
            // ATOMIC_MOVE 仍保证不会读取半个 checkpoint；Linux 生产环境应支持目录 force。
            log.warn("当前文件系统不支持 checkpoint 目录 force，请确认文件系统掉电语义: directory={}",
                    directory);
        }
    }

    private static void writeFully(FileChannel channel, byte[] bytes) throws IOException {
        ByteBuffer buffer = ByteBuffer.wrap(bytes);
        while (buffer.hasRemaining()) channel.write(buffer);
    }

    @PreDestroy
    public void shutdown() {
        running = false;
        synchronized (durableMonitor) { durableMonitor.notifyAll(); }
        if (writerThread != null) {
            try {
                writerThread.join(properties.getSyslogWalShutdownTimeoutMs());
                if (writerThread.isAlive()) {
                    log.error("WAL 写线程未在超时内退出: remaining={}", appendQueue.size());
                    writerThread.interrupt();
                }
            } catch (InterruptedException interrupted) {
                Thread.currentThread().interrupt();
            }
        }
        synchronized (checkpointMonitor) { flushCheckpointLocked(); }
    }

    public final class WalCursor implements AutoCloseable {
        private String currentSegment;
        private long position;
        private FileChannel channel;

        private WalCursor(Checkpoint checkpoint) throws IOException {
            this.currentSegment = checkpoint.segment;
            this.position = checkpoint.offset;
            if (currentSegment != null) {
                Path path = activeDirectory.resolve(currentSegment);
                if (!Files.exists(path)) {
                    throw new IOException("WAL checkpoint 指向的活动分段不存在，拒绝跳过: " + path);
                }
                channel = FileChannel.open(path, StandardOpenOption.READ);
            }
        }

        public SyslogWalRecord next(long waitMs) throws IOException, InterruptedException {
            long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(waitMs);
            while (running || getPendingRecordsApprox() > 0) {
                ensureChannel();
                if (channel == null) {
                    awaitDurable(deadline);
                    if (System.nanoTime() >= deadline) return null;
                    continue;
                }
                long start = position;
                byte[] line = readLine();
                if (line == null) {
                    if (openNextSegment()) continue;
                    awaitDurable(deadline);
                    if (System.nanoTime() >= deadline) return null;
                    continue;
                }
                SyslogWalRecord record;
                try {
                    JsonNode node = objectMapper.readTree(line);
                    try {
                        SyslogWalRecordValidator.validateNode(node);
                        record = objectMapper.treeToValue(node, SyslogWalRecord.class);
                        SyslogWalRecordValidator.validateRecord(record);
                    } catch (Exception invalidRecord) {
                        record = invalidRecordTombstone(node, invalidRecord);
                    }
                } catch (Exception failure) {
                    throw new IOException("WAL 中间记录损坏，拒绝越过: segment=" + currentSegment
                            + ", offset=" + start, failure);
                }
                if (record.getSequence() > durableSequence.get()) {
                    position = start;
                    awaitDurable(deadline);
                    if (System.nanoTime() >= deadline) return null;
                    continue;
                }
                record.setSegment(currentSegment);
                record.setStartOffset(start);
                record.setEndOffset(position);
                if (record.isQuarantinedInvalid()) {
                    quarantine(record, "INVALID_WAL_RECORD", new IllegalArgumentException(record.getValidationError()));
                }
                return record;
            }
            return null;
        }

        private SyslogWalRecord invalidRecordTombstone(JsonNode node, Exception failure) throws IOException {
            JsonNode sequenceNode = node == null ? null : node.get("sequence");
            if (sequenceNode == null || !sequenceNode.isIntegralNumber()
                    || !sequenceNode.canConvertToLong() || sequenceNode.longValue() <= 0L) {
                throw new IOException("损坏 WAL 无法取得合法 sequence，拒绝越过", failure);
            }
            SyslogWalRecord record = new SyslogWalRecord();
            record.setSequence(sequenceNode.longValue());
            record.setSchemaVersion(node.path("schema_version").asInt(-1));
            record.setEventId(node.path("event_id").asText("unknown-" + record.getSequence()));
            record.setReceiveTime(node.path("receive_time").asText(null));
            record.setQuarantinedInvalid(true);
            record.setValidationError(failure.getMessage() == null
                    ? failure.getClass().getName() : failure.getMessage());
            return record;
        }

        private void ensureChannel() throws IOException {
            if (channel != null) return;
            List<Path> segments = listSegments(activeDirectory);
            if (segments.isEmpty()) return;
            Path first = segments.get(0);
            currentSegment = first.getFileName().toString();
            position = 0L;
            channel = FileChannel.open(first, StandardOpenOption.READ);
        }

        private boolean openNextSegment() throws IOException {
            List<Path> segments = listSegments(activeDirectory);
            for (Path path : segments) {
                String name = path.getFileName().toString();
                if (currentSegment == null || name.compareTo(currentSegment) > 0) {
                    closeChannel();
                    currentSegment = name;
                    position = 0L;
                    channel = FileChannel.open(path, StandardOpenOption.READ);
                    return true;
                }
            }
            return false;
        }

        private byte[] readLine() throws IOException {
            long start = position;
            ByteArrayOutputStream output = new ByteArrayOutputStream(2048);
            ByteBuffer buffer = ByteBuffer.allocate(8192);
            while (true) {
                buffer.clear();
                channel.position(position);
                int count = channel.read(buffer);
                if (count < 0 || count == 0) {
                    position = start;
                    return null;
                }
                buffer.flip();
                int consumed = 0;
                while (buffer.hasRemaining()) {
                    byte value = buffer.get();
                    consumed++;
                    if (value == (byte) '\n') {
                        position += consumed;
                        return output.toByteArray();
                    }
                    output.write(value);
                }
                position += consumed;
            }
        }

        private void awaitDurable(long deadline) throws InterruptedException {
            long remainingNanos = deadline - System.nanoTime();
            if (remainingNanos <= 0) return;
            synchronized (durableMonitor) {
                TimeUnit.NANOSECONDS.timedWait(durableMonitor, remainingNanos);
            }
        }

        private void closeChannel() throws IOException {
            if (channel != null) channel.close();
            channel = null;
        }

        @Override public void close() throws IOException { closeChannel(); }
    }

    public static class Checkpoint {
        public long sequence;
        public String segment;
        public long offset;

        public Checkpoint() { this(0L, null, 0L); }
        public Checkpoint(long sequence, String segment, long offset) {
            this.sequence = sequence;
            this.segment = segment;
            this.offset = offset;
        }
        private Checkpoint copy() { return new Checkpoint(sequence, segment, offset); }
    }

    private enum DiskState { NORMAL, HIGH, CRITICAL }
}