package com.zy.core.task;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.zy.asrs.entity.DeviceDataLog;
import com.zy.asrs.service.DeviceLogReplayManifestService;
import com.zy.core.enums.SlaveType;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.BufferedWriter;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Stream;

@Slf4j
@Component
public class DeviceLogFileWriter {

    private static final long MAX_FILE_SIZE_BYTES = 1024L * 1024L;
    private static final long SHUTDOWN_FLUSH_TIMEOUT_MS = 2_000L;
    private static final String DATE_PATTERN = "yyyyMMdd";
    private static final byte[] LINE_SEPARATOR_BYTES = System.lineSeparator().getBytes(StandardCharsets.UTF_8);

    private final ConcurrentHashMap<String, AtomicReference<DeviceLogFileWriteRequest>> pendingByLane = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, AtomicBoolean> runningByLane = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, LaneWriterContext> activeContexts = new ConcurrentHashMap<>();
    private final AtomicLong submittedCount = new AtomicLong();
    private final AtomicLong overwrittenCount = new AtomicLong();
    private final AtomicLong writtenCount = new AtomicLong();
    private final AtomicLong flushCount = new AtomicLong();
    private final AtomicLong writeFailureCount = new AtomicLong();
    private final AtomicLong idleCloseCount = new AtomicLong();
    private final AtomicLong droppedLaneCount = new AtomicLong();

    private final DeviceLogReplayManifestService manifestService;

    @Value("${deviceLogStorage.loggingPath}")
    private String loggingPath;
    @Value("${deviceLogStorage.asyncPublishWorkers:6}")
    private Integer asyncPublishWorkers;
    @Value("${deviceLogStorage.directFileQueueCapacity:2048}")
    private Integer directFileQueueCapacity;
    @Value("${deviceLogStorage.writerFlushBatchSize:64}")
    private Integer writerFlushBatchSize;
    @Value("${deviceLogStorage.writerFlushIntervalMs:200}")
    private Long writerFlushIntervalMs;
    @Value("${deviceLogStorage.writerIdleCloseMs:30000}")
    private Long writerIdleCloseMs;
    @Value("${deviceLogStorage.manifestRefreshDebounceMs:1000}")
    private Long manifestRefreshDebounceMs;

    private volatile boolean shuttingDown;
    private ExecutorService writerExecutor;
    private ScheduledExecutorService maintenanceExecutor;

    public DeviceLogFileWriter(DeviceLogReplayManifestService manifestService) {
        this.manifestService = manifestService;
    }

    @PostConstruct
    public void initExecutor() {
        int workerCount = Math.max(1, asyncPublishWorkers == null ? 6 : asyncPublishWorkers);
        this.writerExecutor = Executors.newFixedThreadPool(workerCount, new DeviceLogFileWriterThreadFactory());
        this.maintenanceExecutor = Executors.newSingleThreadScheduledExecutor(
                new NamedThreadFactory("device-log-file-maintenance"));
        long maintenanceIntervalMs = resolveMaintenanceIntervalMs();
        this.maintenanceExecutor.scheduleWithFixedDelay(
                this::runMaintenanceSafely,
                maintenanceIntervalMs,
                maintenanceIntervalMs,
                TimeUnit.MILLISECONDS
        );
    }

    @PreDestroy
    public void shutdownExecutor() {
        shuttingDown = true;
        long deadlineNanos = System.nanoTime() + Duration.ofMillis(SHUTDOWN_FLUSH_TIMEOUT_MS).toNanos();

        shutdownMaintenanceExecutor();
        drainPendingDirectly(deadlineNanos);
        shutdownWriterExecutor(deadlineNanos);
        flushAndCloseActiveContexts(deadlineNanos);

        log.info("Device log file writer shutdown, submittedCount={}, overwrittenCount={}, writtenCount={}, flushCount={}, " +
                        "writeFailureCount={}, idleCloseCount={}, droppedLaneCount={}, activeLaneCount={}",
                submittedCount.get(),
                overwrittenCount.get(),
                writtenCount.get(),
                flushCount.get(),
                writeFailureCount.get(),
                idleCloseCount.get(),
                droppedLaneCount.get(),
                getActiveLaneCount());
    }

    public void publishLatest(String laneKey, DeviceDataLog deviceDataLog) {
        if (shuttingDown || laneKey == null || deviceDataLog == null) {
            return;
        }
        AtomicReference<DeviceLogFileWriteRequest> pending = pendingByLane.get(laneKey);
        if (pending == null) {
            int laneCapacity = Math.max(64, directFileQueueCapacity == null ? 2048 : directFileQueueCapacity);
            if (pendingByLane.size() >= laneCapacity) {
                droppedLaneCount.incrementAndGet();
                log.warn("Device direct-file writer lane capacity exceeded, laneKey={}, pendingLanes={}", laneKey, pendingByLane.size());
                return;
            }
            AtomicReference<DeviceLogFileWriteRequest> created = new AtomicReference<>();
            AtomicReference<DeviceLogFileWriteRequest> previous = pendingByLane.putIfAbsent(laneKey, created);
            pending = previous == null ? created : previous;
        }

        DeviceLogFileWriteRequest request = new DeviceLogFileWriteRequest();
        request.setLaneKey(laneKey);
        request.setDeviceDataLog(deviceDataLog);

        DeviceLogFileWriteRequest previousRequest = pending.getAndSet(request);
        submittedCount.incrementAndGet();
        if (previousRequest != null) {
            overwrittenCount.incrementAndGet();
        }
        scheduleLaneDrain(laneKey, pending);
    }

    public long getSubmittedCount() {
        return submittedCount.get();
    }

    public long getOverwrittenCount() {
        return overwrittenCount.get();
    }

    public long getWrittenCount() {
        return writtenCount.get();
    }

    public long getFlushCount() {
        return flushCount.get();
    }

    public long getWriteFailureCount() {
        return writeFailureCount.get();
    }

    public long getIdleCloseCount() {
        return idleCloseCount.get();
    }

    public long getDroppedLaneCount() {
        return droppedLaneCount.get();
    }

    public int getActiveLaneCount() {
        return activeContexts.size();
    }

    private void scheduleLaneDrain(String laneKey, AtomicReference<DeviceLogFileWriteRequest> pending) {
        if (shuttingDown || pending == null || writerExecutor == null) {
            return;
        }
        AtomicBoolean running = runningByLane.computeIfAbsent(laneKey, key -> new AtomicBoolean(false));
        if (!running.compareAndSet(false, true)) {
            return;
        }
        try {
            writerExecutor.execute(() -> drain(laneKey, pending, running));
        } catch (Exception e) {
            running.set(false);
            log.warn("Schedule device log writer lane drain failed, laneKey={}", laneKey, e);
        }
    }

    private void drain(String laneKey, AtomicReference<DeviceLogFileWriteRequest> pending, AtomicBoolean running) {
        try {
            while (true) {
                DeviceLogFileWriteRequest request = pending.getAndSet(null);
                if (request == null) {
                    return;
                }
                writeRequest(request);
                if (shuttingDown) {
                    return;
                }
            }
        } finally {
            running.set(false);
            AtomicReference<DeviceLogFileWriteRequest> currentPending = pendingByLane.get(laneKey);
            if (currentPending == null) {
                runningByLane.remove(laneKey);
                return;
            }
            if (currentPending.get() == null) {
                pendingByLane.remove(laneKey, currentPending);
                runningByLane.remove(laneKey, running);
                return;
            }
            scheduleLaneDrain(laneKey, currentPending);
        }
    }

    private void writeRequest(DeviceLogFileWriteRequest request) {
        DeviceDataLog logItem = request.getDeviceDataLog();
        if (logItem == null) {
            return;
        }
        try {
            String day = resolveDay(logItem);
            String filePrefix = buildFilePrefix(logItem, day);
            if (filePrefix == null) {
                return;
            }
            Path deviceDir = resolveDeviceDir(Paths.get(loggingPath).resolve(day), logItem);
            if (deviceDir == null) {
                return;
            }

            String jsonLine = buildJsonLine(logItem);
            long lineBytes = jsonLine.getBytes(StandardCharsets.UTF_8).length + LINE_SEPARATOR_BYTES.length;
            long now = System.currentTimeMillis();

            LaneWriterContext context = ensureLaneContext(request.getLaneKey(), logItem, day, deviceDir, filePrefix, now);
            appendLine(context, logItem, jsonLine, lineBytes, now);
        } catch (Exception e) {
            writeFailureCount.incrementAndGet();
            log.error("设备日志写文件失败, laneKey={}, type={}, deviceNo={}, stationId={}",
                    request.getLaneKey(),
                    logItem.getType(),
                    logItem.getDeviceNo(),
                    logItem.getStationId(),
                    e);
            recycleLaneContext(request.getLaneKey(), null);
        }
    }

    private LaneWriterContext ensureLaneContext(String laneKey,
                                                DeviceDataLog logItem,
                                                String day,
                                                Path deviceDir,
                                                String filePrefix,
                                                long now) throws IOException {
        LaneWriterContext currentContext = activeContexts.get(laneKey);
        if (currentContext != null && day.equals(currentContext.day)) {
            synchronized (currentContext) {
                if (activeContexts.get(laneKey) == currentContext
                        && day.equals(currentContext.day)
                        && currentContext.bufferedWriter != null) {
                    currentContext.lastWriteTimeMs = now;
                    return currentContext;
                }
            }
        }

        if (currentContext != null) {
            closeLaneContext(laneKey, currentContext, closeReasonFlushManifest(), now);
        }

        LaneWriterContext newContext = openLaneContext(laneKey, logItem, day, deviceDir, filePrefix, now);
        LaneWriterContext previousContext = activeContexts.put(laneKey, newContext);
        if (previousContext != null && previousContext != currentContext && previousContext != newContext) {
            closeLaneContext(laneKey, previousContext, closeReasonFlushManifest(), now);
        }
        return newContext;
    }

    private LaneWriterContext openLaneContext(String laneKey,
                                              DeviceDataLog logItem,
                                              String day,
                                              Path deviceDir,
                                              String filePrefix,
                                              long now) throws IOException {
        Files.createDirectories(deviceDir);

        LaneWriterContext context = new LaneWriterContext();
        context.laneKey = laneKey;
        context.day = day;
        context.type = logItem.getType();
        context.deviceNo = logItem.getDeviceNo();
        context.stationId = logItem.getStationId();
        context.deviceDir = deviceDir;
        context.filePrefix = filePrefix;
        context.currentFileIndex = findStartIndex(deviceDir, filePrefix);
        context.currentFile = deviceDir.resolve(filePrefix + context.currentFileIndex + ".log");
        context.bufferedWriter = openBufferedWriter(context.currentFile);
        context.currentFileBytes = Files.exists(context.currentFile) ? Files.size(context.currentFile) : 0L;
        context.pendingFlushLines = 0;
        context.lastWriteTimeMs = now;
        context.lastFlushTimeMs = now;
        context.manifestDirty = false;
        context.lastManifestScheduleTimeMs = 0L;
        return context;
    }

    private void appendLine(LaneWriterContext context,
                            DeviceDataLog logItem,
                            String jsonLine,
                            long lineBytes,
                            long now) throws IOException {
        synchronized (context) {
            rotateIfNeeded(context, lineBytes, now);

            context.bufferedWriter.write(jsonLine);
            context.bufferedWriter.newLine();
            context.currentFileBytes += lineBytes;
            context.pendingFlushLines++;
            context.lastWriteTimeMs = now;
            context.manifestDirty = true;

            if (shouldFlushAfterWrite(context, now)) {
                flushContext(context, buildManifestLogItem(context), now, false);
            }
        }
        writtenCount.incrementAndGet();
    }

    private void rotateIfNeeded(LaneWriterContext context, long lineBytes, long now) throws IOException {
        if (context.currentFileBytes + lineBytes <= MAX_FILE_SIZE_BYTES) {
            return;
        }

        flushContext(context, buildManifestLogItem(context), now, true);
        closeBufferedWriterQuietly(context);

        context.currentFileIndex++;
        context.currentFile = context.deviceDir.resolve(context.filePrefix + context.currentFileIndex + ".log");
        context.bufferedWriter = openBufferedWriter(context.currentFile);
        context.currentFileBytes = Files.exists(context.currentFile) ? Files.size(context.currentFile) : 0L;
        context.pendingFlushLines = 0;
        context.lastFlushTimeMs = now;
        context.lastWriteTimeMs = now;
        context.manifestDirty = false;
    }

    private boolean shouldFlushAfterWrite(LaneWriterContext context, long now) {
        int flushBatchSize = Math.max(1, writerFlushBatchSize == null ? 64 : writerFlushBatchSize);
        long flushIntervalMs = Math.max(1L, writerFlushIntervalMs == null ? 200L : writerFlushIntervalMs);
        return context.pendingFlushLines >= flushBatchSize || now - context.lastFlushTimeMs >= flushIntervalMs;
    }

    private void runMaintenanceSafely() {
        try {
            long now = System.currentTimeMillis();
            for (LaneWriterContext context : activeContexts.values()) {
                inspectLaneContext(context, now);
            }
        } catch (Exception e) {
            log.warn("Device log writer maintenance failed", e);
        }
    }

    private void inspectLaneContext(LaneWriterContext context, long now) {
        if (context == null) {
            return;
        }
        boolean closedByIdle = false;
        synchronized (context) {
            if (activeContexts.get(context.laneKey) != context) {
                return;
            }
            long flushIntervalMs = Math.max(1L, writerFlushIntervalMs == null ? 200L : writerFlushIntervalMs);
            long idleCloseMs = Math.max(flushIntervalMs, writerIdleCloseMs == null ? 30_000L : writerIdleCloseMs);

            if (context.pendingFlushLines > 0 && now - context.lastFlushTimeMs >= flushIntervalMs) {
                try {
                    flushContext(context, buildManifestLogItem(context), now, false);
                } catch (IOException e) {
                    closeLaneContextLocked(context.laneKey, context, closeReasonSkipManifest(), now);
                    return;
                }
            }
            if (now - context.lastWriteTimeMs < idleCloseMs) {
                return;
            }

            if (activeContexts.get(context.laneKey) != context || now - context.lastWriteTimeMs < idleCloseMs) {
                return;
            }

            closeLaneContextLocked(context.laneKey, context, closeReasonFlushManifest(), now);
            closedByIdle = true;
        }

        if (closedByIdle) {
            idleCloseCount.incrementAndGet();
            log.debug("Device log writer idle close, laneKey={}, file={}", context.laneKey, context.currentFile);
        }
    }

    private void flushContext(LaneWriterContext context,
                              DeviceDataLog manifestLogItem,
                              long now,
                              boolean forceManifestSchedule) throws IOException {
        if (context.pendingFlushLines <= 0 && !shouldScheduleManifest(context, now, forceManifestSchedule)) {
            return;
        }

        try {
            if (context.pendingFlushLines > 0) {
                context.bufferedWriter.flush();
                flushCount.incrementAndGet();
            }
            context.pendingFlushLines = 0;
            context.lastFlushTimeMs = now;

            if (shouldScheduleManifest(context, now, forceManifestSchedule) && manifestLogItem != null) {
                try {
                    manifestService.scheduleManifestRefreshForLog(context.day, manifestLogItem);
                    context.manifestDirty = false;
                    context.lastManifestScheduleTimeMs = now;
                } catch (Exception e) {
                    log.warn("调度设备日志 manifest 刷新失败, laneKey={}, day={}, type={}, deviceNo={}, stationId={}",
                            context.laneKey,
                            context.day,
                            context.type,
                            context.deviceNo,
                            context.stationId,
                            e);
                }
            }
        } catch (IOException e) {
            writeFailureCount.incrementAndGet();
            log.error("设备日志 flush 失败, laneKey={}, file={}, type={}, deviceNo={}, stationId={}",
                    context.laneKey,
                    context.currentFile,
                    context.type,
                    context.deviceNo,
                    context.stationId,
                    e);
            throw e;
        }
    }

    private boolean shouldScheduleManifest(LaneWriterContext context, long now, boolean forceManifestSchedule) {
        if (!context.manifestDirty) {
            return false;
        }
        if (forceManifestSchedule) {
            return true;
        }
        long debounceMs = Math.max(0L, manifestRefreshDebounceMs == null ? 1000L : manifestRefreshDebounceMs);
        return now - context.lastManifestScheduleTimeMs >= debounceMs;
    }

    private void closeLaneContext(String laneKey, LaneWriterContext context, CloseOptions closeOptions, long now) {
        if (context == null) {
            return;
        }
        synchronized (context) {
            closeLaneContextLocked(laneKey, context, closeOptions, now);
        }
    }

    private void closeLaneContextLocked(String laneKey, LaneWriterContext context, CloseOptions closeOptions, long now) {
        if (activeContexts.get(laneKey) != context && context.bufferedWriter == null) {
            return;
        }
        try {
            if (closeOptions.flushBeforeClose) {
                flushContext(context, buildManifestLogItem(context), now, closeOptions.forceManifestSchedule);
            }
        } catch (Exception e) {
            log.warn("Close lane context after flush failure, laneKey={}, file={}", laneKey, context.currentFile, e);
        } finally {
            closeBufferedWriterQuietly(context);
            context.bufferedWriter = null;
            activeContexts.remove(laneKey, context);
        }
    }

    private void recycleLaneContext(String laneKey, LaneWriterContext expectedContext) {
        if (laneKey == null) {
            return;
        }
        LaneWriterContext context = expectedContext;
        if (context == null) {
            context = activeContexts.get(laneKey);
        }
        if (context == null) {
            return;
        }
        closeLaneContext(laneKey, context, closeReasonSkipManifest(), System.currentTimeMillis());
    }

    private void drainPendingDirectly(long deadlineNanos) {
        for (AtomicReference<DeviceLogFileWriteRequest> pending : pendingByLane.values()) {
            if (System.nanoTime() > deadlineNanos) {
                log.warn("Stop draining pending device logs during shutdown because deadline reached");
                return;
            }
            try {
                DeviceLogFileWriteRequest request = pending == null ? null : pending.getAndSet(null);
                if (request != null) {
                    writeRequest(request);
                }
            } catch (Exception e) {
                log.warn("Drain pending device log during shutdown failed", e);
            }
        }
    }

    private void flushAndCloseActiveContexts(long deadlineNanos) {
        boolean warnedDeadlineExceeded = false;
        for (LaneWriterContext context : activeContexts.values()) {
            boolean deadlineExceeded = System.nanoTime() > deadlineNanos;
            if (deadlineExceeded && !warnedDeadlineExceeded) {
                log.warn("Stop flushing active device log writers during shutdown because deadline reached, remainingActiveLanes={}",
                        activeContexts.size());
                warnedDeadlineExceeded = true;
            }
            CloseOptions closeOptions = deadlineExceeded ? closeReasonCloseOnly() : closeReasonFlushManifest();
            closeLaneContext(context.laneKey, context, closeOptions, System.currentTimeMillis());
        }
    }

    private void shutdownMaintenanceExecutor() {
        if (maintenanceExecutor == null) {
            return;
        }
        maintenanceExecutor.shutdownNow();
    }

    private void shutdownWriterExecutor(long deadlineNanos) {
        if (writerExecutor == null) {
            return;
        }
        writerExecutor.shutdown();
        long remainingMs = Math.max(1L, Duration.ofNanos(Math.max(0L, deadlineNanos - System.nanoTime())).toMillis());
        try {
            if (!writerExecutor.awaitTermination(remainingMs, TimeUnit.MILLISECONDS)) {
                log.warn("Device log writer executor did not stop before shutdown deadline");
                writerExecutor.shutdownNow();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            writerExecutor.shutdownNow();
        }
    }

    private long resolveMaintenanceIntervalMs() {
        long flushIntervalMs = Math.max(1L, writerFlushIntervalMs == null ? 200L : writerFlushIntervalMs);
        long idleCloseMs = Math.max(flushIntervalMs, writerIdleCloseMs == null ? 30_000L : writerIdleCloseMs);
        return Math.max(200L, Math.min(flushIntervalMs, Math.min(idleCloseMs, 1000L)));
    }

    private String buildJsonLine(DeviceDataLog logItem) {
        return JSON.toJSONStringWithDateFormat(
                logItem,
                "yyyy-MM-dd HH:mm:ss.SSS",
                SerializerFeature.WriteDateUseDateFormat);
    }

    private String resolveDay(DeviceDataLog logItem) {
        Date createTime = logItem.getCreateTime() == null ? new Date() : logItem.getCreateTime();
        return new SimpleDateFormat(DATE_PATTERN).format(createTime);
    }

    private Path resolveDeviceDir(Path dayDir, DeviceDataLog logItem) {
        if (dayDir == null || logItem == null || logItem.getType() == null || logItem.getDeviceNo() == null) {
            return null;
        }
        return dayDir.resolve(logItem.getType()).resolve(String.valueOf(logItem.getDeviceNo()));
    }

    private String buildFilePrefix(DeviceDataLog logItem, String day) {
        if (logItem == null || logItem.getType() == null || logItem.getDeviceNo() == null) {
            return null;
        }
        if (String.valueOf(SlaveType.Devp).equals(logItem.getType())) {
            if (logItem.getStationId() == null) {
                log.warn("跳过缺少站点号的输送设备日志, deviceNo={}, createTime={}", logItem.getDeviceNo(), logItem.getCreateTime());
                return null;
            }
            return logItem.getType() + "_" + logItem.getDeviceNo() + "_station_" + logItem.getStationId() + "_" + day + "_";
        }
        return logItem.getType() + "_" + logItem.getDeviceNo() + "_" + day + "_";
    }

    private int findStartIndex(Path baseDir, String prefix) throws IOException {
        int maxIndex = 0;
        try (Stream<Path> stream = Files.list(baseDir)) {
            for (Path path : (Iterable<Path>) stream::iterator) {
                String fileName = path.getFileName().toString();
                if (!fileName.startsWith(prefix) || !fileName.endsWith(".log")) {
                    continue;
                }
                String suffix = fileName.substring(prefix.length(), fileName.length() - 4);
                if (suffix.isBlank()) {
                    continue;
                }
                try {
                    maxIndex = Math.max(maxIndex, Integer.parseInt(suffix));
                } catch (NumberFormatException ignored) {
                }
            }
        }

        int candidateIndex = maxIndex == 0 ? 1 : maxIndex;
        Path candidatePath = baseDir.resolve(prefix + candidateIndex + ".log");
        if (Files.exists(candidatePath) && Files.size(candidatePath) >= MAX_FILE_SIZE_BYTES) {
            return candidateIndex + 1;
        }
        return candidateIndex;
    }

    private BufferedWriter openBufferedWriter(Path filePath) throws IOException {
        if (filePath.getParent() != null) {
            Files.createDirectories(filePath.getParent());
        }
        return Files.newBufferedWriter(
                filePath,
                StandardCharsets.UTF_8,
                StandardOpenOption.CREATE,
                StandardOpenOption.APPEND
        );
    }

    private DeviceDataLog buildManifestLogItem(LaneWriterContext context) {
        if (context == null) {
            return null;
        }
        DeviceDataLog logItem = new DeviceDataLog();
        logItem.setType(context.type);
        logItem.setDeviceNo(context.deviceNo);
        logItem.setStationId(context.stationId);
        return logItem;
    }

    private void closeBufferedWriterQuietly(LaneWriterContext context) {
        if (context == null || context.bufferedWriter == null) {
            return;
        }
        try {
            context.bufferedWriter.close();
        } catch (Exception e) {
            log.warn("Close device log writer failed, laneKey={}, file={}", context.laneKey, context.currentFile, e);
        }
    }

    private CloseOptions closeReasonFlushManifest() {
        return new CloseOptions(true, true);
    }

    private CloseOptions closeReasonSkipManifest() {
        return new CloseOptions(false, true);
    }

    private CloseOptions closeReasonCloseOnly() {
        return new CloseOptions(false, false);
    }

    private static class CloseOptions {

        private final boolean forceManifestSchedule;
        private final boolean flushBeforeClose;

        private CloseOptions(boolean forceManifestSchedule, boolean flushBeforeClose) {
            this.forceManifestSchedule = forceManifestSchedule;
            this.flushBeforeClose = flushBeforeClose;
        }
    }

    private static class LaneWriterContext {

        private String laneKey;
        private String day;
        private String type;
        private Integer deviceNo;
        private Integer stationId;
        private Path deviceDir;
        private String filePrefix;
        private int currentFileIndex;
        private Path currentFile;
        private BufferedWriter bufferedWriter;
        private long currentFileBytes;
        private int pendingFlushLines;
        private long lastWriteTimeMs;
        private long lastFlushTimeMs;
        private boolean manifestDirty;
        private long lastManifestScheduleTimeMs;
    }

    private static class DeviceLogFileWriterThreadFactory implements ThreadFactory {

        private final AtomicInteger index = new AtomicInteger(1);

        @Override
        public Thread newThread(Runnable runnable) {
            Thread thread = new Thread(runnable, "device-log-file-writer-" + index.getAndIncrement());
            thread.setDaemon(true);
            return thread;
        }
    }

    private static class NamedThreadFactory implements ThreadFactory {

        private final String prefix;
        private final AtomicInteger index = new AtomicInteger(1);

        private NamedThreadFactory(String prefix) {
            this.prefix = prefix;
        }

        @Override
        public Thread newThread(Runnable runnable) {
            Thread thread = new Thread(runnable, prefix + "-" + index.getAndIncrement());
            thread.setDaemon(true);
            return thread;
        }
    }
}