package com.ruoyi.shadow.kernel;

import com.ruoyi.shadow.facts.ShadowFactBatch;
import com.ruoyi.shadow.facts.ShadowFactReadException;
import com.ruoyi.shadow.facts.ShadowQueueFact;
import com.ruoyi.shadow.facts.ShadowReadCursor;
import com.ruoyi.shadow.facts.SqlServerShadowFactReader;
import com.ruoyi.shadow.startup.ShadowStartupReport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;

import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.Objects;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;

public final class Ft546ShadowScheduler implements AutoCloseable {
    private static final Logger LOGGER =
            LoggerFactory.getLogger(Ft546ShadowScheduler.class);
    private static final Duration CYCLE_DEADLINE = Duration.ofSeconds(10);
    private static final Duration SOURCE_READ_DEADLINE = Duration.ofSeconds(4);
    private static final Duration CRASH_BACKOFF = Duration.ofSeconds(1);
    private static final Duration COMMIT_RECOVERY_WINDOW = Duration.ofSeconds(30);
    private static final String FINISH_QUEUE_VIEW =
            "shadow_ro.v_wcs_srgv_queue_finish";

    private final SqlServerShadowFactReader factReader;
    private final Ft546ShadowCycle shadowCycle;
    private final ShadowStartupReport statusReport;
    private final Clock clock;
    private final Duration cycleDeadline;
    private final Duration sourceReadDeadline;
    private final ExecutorService cycleExecutor;
    private final Consumer<Map<String, Object>> runtimeWriter;
    private final Object runtimeLock = new Object();
    private final AtomicBoolean closed = new AtomicBoolean();
    private volatile Instant lastAttemptAt;
    private final AtomicBoolean inFlight = new AtomicBoolean();
    private final AtomicBoolean permanentlyStopped = new AtomicBoolean();
    private volatile CycleStage activeStage = CycleStage.IDLE;
    private volatile ShadowReadCursor cursor = ShadowReadCursor.initial();
    private volatile Instant nextAttemptAt = Instant.EPOCH;
    private volatile Thread cycleWorker;
    private long lastTimingLogNanos;
    private long lastStoppedWarningNanos;
    private boolean runtimeInitialized;
    private volatile PendingCommit pendingCommit;

    public Ft546ShadowScheduler(SqlServerShadowFactReader factReader,
                                Ft546ShadowCycle shadowCycle,
                                ShadowStartupReport statusReport,
                                Clock clock) {
        this(factReader, shadowCycle, statusReport, clock, CYCLE_DEADLINE);
    }

    public Ft546ShadowScheduler(SqlServerShadowFactReader factReader,
                                Ft546ShadowCycle shadowCycle,
                                ShadowStartupReport statusReport,
                                Clock clock,
                                Duration cycleDeadline) {
        this(factReader, shadowCycle, statusReport, clock, cycleDeadline,
                shorter(cycleDeadline, SOURCE_READ_DEADLINE));
    }

    Ft546ShadowScheduler(SqlServerShadowFactReader factReader,
                         Ft546ShadowCycle shadowCycle,
                         ShadowStartupReport statusReport,
                         Clock clock,
                         Duration cycleDeadline,
                         Duration sourceReadDeadline) {
        this(factReader, shadowCycle, statusReport, clock, cycleDeadline, sourceReadDeadline, state -> { });
    }

    public Ft546ShadowScheduler(SqlServerShadowFactReader factReader,
                                Ft546ShadowCycle shadowCycle, ShadowStartupReport statusReport,
                                Clock clock, Duration cycleDeadline,
                                Consumer<Map<String, Object>> runtimeWriter) {
        this(factReader, shadowCycle, statusReport, clock, cycleDeadline,
                shorter(cycleDeadline, SOURCE_READ_DEADLINE), runtimeWriter);
    }

    private Ft546ShadowScheduler(SqlServerShadowFactReader factReader,
                                 Ft546ShadowCycle shadowCycle, ShadowStartupReport statusReport,
                                 Clock clock, Duration cycleDeadline, Duration sourceReadDeadline,
                                 Consumer<Map<String, Object>> runtimeWriter) {
        this.factReader = Objects.requireNonNull(factReader, "factReader");
        this.shadowCycle = Objects.requireNonNull(shadowCycle, "shadowCycle");
        this.statusReport = Objects.requireNonNull(statusReport,
                "statusReport");
        this.clock = Objects.requireNonNull(clock, "clock");
        this.cycleDeadline = requirePositive(cycleDeadline);
        this.sourceReadDeadline = requirePositive(sourceReadDeadline);
        this.runtimeWriter = Objects.requireNonNull(runtimeWriter, "runtimeWriter");
        this.cycleExecutor = Executors.newSingleThreadExecutor(
                shadowCycleThreadFactory());
    }

    @Scheduled(
            initialDelayString = "${shadow.poll.initial-delay-ms:5000}",
            fixedDelayString = "${shadow.poll.fixed-delay-ms:5000}")
    public void runScheduledCycle() {
        if (permanentlyStopped.get() || clock.instant().isBefore(nextAttemptAt)) {
            warnIfStopped();
            publishRuntimeState();
        } else {
            runOnce();
        }
    }

    public boolean runOnce() {
        long startedNanos = System.nanoTime();
        Instant now = clock.instant();
        if (closed.get() || permanentlyStopped.get() || now.isBefore(nextAttemptAt)
                || !inFlight.compareAndSet(false, true)) {
            return false;
        }
        if (pendingCommit != null) {
            try {
                return reconcilePendingCommit();
            } finally {
                try { publishRuntimeState(); } finally { inFlight.set(false); }
            }
        }
        ShadowCycleDiagnostics diagnostics = new ShadowCycleDiagnostics();
        lastAttemptAt = now;
        if (!runtimeInitialized) {
            diagnostics.enter("RUNTIME_PUBLICATION");
            if (!publishRuntimeState()) {
                inFlight.set(false);
                return false;
            }
            runtimeInitialized = true;
        }
        cycleWorker = null;
        String outcome = "FAILED";
        Future<CycleAttempt> future = null;
        try {
            if (permanentlyStopped.get()) return false;
            Instant overallDeadline = now.plus(cycleDeadline);
            long remainingNanos = Math.min(cycleDeadline.toNanos() - (System.nanoTime() - startedNanos),
                    Duration.between(clock.instant(), overallDeadline).toNanos());
            if (remainingNanos <= 0L) throw new TimeoutException("runtime publication exhausted cycle budget");
            diagnostics.enter("QUEUED");
            future = cycleExecutor.submit(() -> executeCycle(overallDeadline, diagnostics));
            CycleAttempt attempt = future.get(remainingNanos, TimeUnit.NANOSECONDS);
            outcome = attempt.result.getStatusCode();
            statusReport.recordCycleDiagnostics(diagnostics.finish(outcome, null));
            if ("DEGRADED".equals(attempt.result.getStatusCode())) {
                permanentlyStopped.set(true);
                statusReport.markPermanentlyDegraded(
                        attempt.result.getReasonCode(), clock.instant());
                LOGGER.error("FT546 shadow collection stopped: {}",
                        attempt.result.getReasonCode());
            } else {
                cursor = advanceFinishCursor(cursor, attempt.batch);
            }
            nextAttemptAt = Instant.EPOCH;
            return true;
        } catch (TimeoutException timeout) {
            CycleStage timedOutStage = activeStage;
            Map<String, Object> snapshot = diagnostics.snapshot("TIMEOUT", cycleWorker);
            String phase = String.valueOf(snapshot.get("phase"));
            if (future != null && ("TRANSACTION_COMMIT".equals(phase) || "RECORDING".equals(phase))) {
                // Retain the future. cancel(true) loses the terminal result and can itself
                // block inside JDBC cancellation. No new batch starts while this is pending.
                pendingCommit = new PendingCommit(future, diagnostics, clock.instant());
                statusReport.recordCycleDiagnostics(snapshot);
                statusReport.awaitCommitCompletion(clock.instant());
                statusReport.recordCollectionFailure("SHADOW_COMMIT_AWAITING_COMPLETION", phase,
                        timeout, clock.instant(), null);
                LOGGER.warn("FT546 shadow commit exceeded cycle budget; awaiting worker completion; diagnostics={}", snapshot);
                return false;
            }
            statusReport.recordCycleDiagnostics(diagnostics.cancel("TIMEOUT", cycleWorker));
            if (future != null) future.cancel(true);
            return stopPermanently("SHADOW_CYCLE_TIMEOUT", timeout,
                    timedOutStage);
        } catch (InterruptedException interrupted) {
            CycleStage interruptedStage = activeStage;
            statusReport.recordCycleDiagnostics(diagnostics.cancel("INTERRUPTED", cycleWorker));
            if (future != null) future.cancel(true);
            Thread.currentThread().interrupt();
            return stopPermanently("SHADOW_CYCLE_INTERRUPTED", interrupted,
                    interruptedStage);
        } catch (ExecutionException wrapped) {
            statusReport.recordCycleDiagnostics(diagnostics.finish("FAILED", null));
            return handleFailure(wrapped.getCause());
        } catch (java.util.concurrent.RejectedExecutionException shutdown) {
            return stopPermanently("SHADOW_EXECUTOR_STOPPED", shutdown, activeStage);
        } finally {
            if (pendingCommit == null) {
                Map<String, Object> completed = diagnostics.finish(outcome, null);
                statusReport.recordCycleDiagnostics(completed);
                long logTime = System.nanoTime();
                if (!permanentlyStopped.get() && (lastTimingLogNanos == 0L
                        || logTime - lastTimingLogNanos >= TimeUnit.MINUTES.toNanos(1))) {
                    LOGGER.info("FT546 shadow cycle diagnostics: {}", completed);
                    lastTimingLogNanos = logTime;
                }
                activeStage = CycleStage.IDLE;
            }
            try { publishRuntimeState(); } finally { inFlight.set(false); }
        }
    }

    private boolean reconcilePendingCommit() {
        PendingCommit pending = pendingCommit;
        if (!pending.future.isDone()) {
            if (pending.expired(clock.instant())) {
                statusReport.recordCycleDiagnostics(pending.diagnostics.cancel("COMMIT_UNCONFIRMED", cycleWorker));
                return stopPermanently("SHADOW_COMMIT_COMPLETION_UNCONFIRMED",
                        new TimeoutException("worker did not finish within commit recovery window"), CycleStage.SHADOW_CYCLE);
            }
            statusReport.recordCycleDiagnostics(pending.diagnostics.snapshot("AWAITING_COMMIT", cycleWorker));
            return false;
        }
        try {
            CycleAttempt attempt = pending.future.get();
            pendingCommit = null;
            activeStage = CycleStage.IDLE;
            if (closed.get() || permanentlyStopped.get()) return false;
            if ("DEGRADED".equals(attempt.result.getStatusCode())) {
                permanentlyStopped.set(true);
                statusReport.markPermanentlyDegraded(attempt.result.getReasonCode(), clock.instant());
                statusReport.recordCycleDiagnostics(pending.diagnostics.finish("COMMIT_FAILED", null));
                LOGGER.error("FT546 shadow late commit did not complete successfully; collection stopped: {}",
                        attempt.result.getReasonCode());
                return false;
            }
            // The future has completed normally, including JDBC cleanup and after-commit
            // callbacks. Do not repeat the transaction or compare across this delayed gap.
            shadowCycle.invalidatePendingComparisons();
            cursor = advanceFinishCursor(cursor, attempt.batch);
            Long lag = attempt.batch.getDatabaseCapturedAt() == null ? null : Long.valueOf(Math.max(0L,
                    Duration.between(attempt.batch.getDatabaseCapturedAt(), attempt.completedAt).toMillis()));
            statusReport.confirmCommitCompletion(attempt.completedAt, lag);
            statusReport.recordCycleDiagnostics(pending.diagnostics.finish("COMMIT_CONFIRMED", null));
            nextAttemptAt = Instant.EPOCH;
            LOGGER.info("FT546 shadow late commit confirmed; collection can resume from the next batch; completedAt={}",
                    attempt.completedAt);
            return true;
        } catch (InterruptedException interrupted) {
            Thread.currentThread().interrupt();
            return stopPermanently("SHADOW_CYCLE_INTERRUPTED", interrupted, CycleStage.SHADOW_CYCLE);
        } catch (ExecutionException | java.util.concurrent.CancellationException failure) {
            statusReport.recordCycleDiagnostics(pending.diagnostics.finish("COMMIT_FAILED", null));
            return stopPermanently("SHADOW_COMMIT_COMPLETION_FAILED", failure, CycleStage.SHADOW_CYCLE);
        }
    }

    private void warnIfStopped() {
        if (!permanentlyStopped.get() || closed.get()) return;
        long now = System.nanoTime();
        if (lastStoppedWarningNanos == 0L || now - lastStoppedWarningNanos >= TimeUnit.MINUTES.toNanos(1)) {
            LOGGER.warn("FT546 shadow collection remains stopped; reason={}; lastCompletedCycle={}; heartbeat is not collection",
                    statusReport.getReasonCode(), statusReport.getLastCompletedCycle());
            lastStoppedWarningNanos = now;
        }
    }

    public boolean isPermanentlyStopped() {
        return permanentlyStopped.get();
    }

    public Duration getRemainingBackoff() {
        Duration remaining = Duration.between(clock.instant(), nextAttemptAt);
        return remaining.isNegative() ? Duration.ZERO : remaining;
    }

    @Override
    public void close() {
        if (!closed.compareAndSet(false, true)) return;
        permanentlyStopped.set(true);
        statusReport.markStopped(clock.instant());
        PendingCommit pending = pendingCommit;
        if (pending != null) {
            pending.diagnostics.cancel("STOPPED", cycleWorker);
            cycleExecutor.shutdown();
        } else {
            cycleExecutor.shutdownNow();
        }
        publishRuntimeState();
    }

    private boolean publishRuntimeState() {
        synchronized (runtimeLock) {
            try {
                runtimeWriter.accept(statusReport.snapshotRuntimeState(
                        clock.instant(), lastAttemptAt, permanentlyStopped.get()));
                return true;
            } catch (RuntimeException failure) {
                permanentlyStopped.set(true);
                statusReport.markPermanentlyDegraded("EVIDENCE_RUNTIME_WRITE_FAILED", clock.instant());
                LOGGER.error("FT546 shadow runtime state unavailable; collection stopped", failure);
                return false;
            }
        }
    }

    private CycleAttempt executeCycle(Instant overallDeadline, ShadowCycleDiagnostics diagnostics) {
        cycleWorker = Thread.currentThread();
        activeStage = CycleStage.SOURCE_READ;
        diagnostics.enter("SOURCE_READ");
        Instant sourceDeadline = clock.instant().plus(sourceReadDeadline);
        if (sourceDeadline.isAfter(overallDeadline)) sourceDeadline = overallDeadline;
        ShadowFactBatch batch = factReader.read(cursor, sourceDeadline);
        activeStage = CycleStage.SHADOW_CYCLE;
        ShadowCycleResult result = shadowCycle.run(batch, clock.instant(), diagnostics);
        return new CycleAttempt(batch, result, clock.instant());
    }

    private boolean handleFailure(Throwable failure) {
        shadowCycle.invalidatePendingComparisons();
        if (failure instanceof ShadowFactReadException) {
            ShadowFactReadException readFailure =
                    (ShadowFactReadException) failure;
            if ("SOURCE_TRANSACTION_FAILED".equals(readFailure.getReasonCode())
                    || "SOURCE_CYCLE_ALREADY_IN_FLIGHT".equals(readFailure.getReasonCode())) {
                return stopPermanently(readFailure.getReasonCode(), readFailure, CycleStage.SOURCE_READ);
            }
            // ExecutionException is delivered only after the reader and its transaction have unwound.
            // A watchdog timeout still stops permanently because cancellation does not prove termination.
            String timeoutReason = sourceTimeoutReason(readFailure);
            String reason = timeoutReason == null ? readFailure.getReasonCode() : timeoutReason;
            Instant failedAt = clock.instant();
            Duration backoff = readFailure.getRetryAfter().isZero() ? CRASH_BACKOFF : readFailure.getRetryAfter();
            statusReport.markDegraded(reason, failedAt);
            nextAttemptAt = failedAt.plus(backoff);
            statusReport.recordCollectionFailure(reason, "SOURCE_READ", readFailure, failedAt, nextAttemptAt);
            LOGGER.warn("FT546 shadow source read failed: {}; retry after {} ms",
                    reason, backoff.toMillis());
            if (readFailure.getCause() != null) {
                LOGGER.warn("FT546 shadow source read root cause",
                        readFailure.getCause());
            }
            return false;
        }
        if (failure instanceof Error) {
            throw (Error) failure;
        }
        RuntimeException runtimeFailure = failure instanceof RuntimeException
                ? (RuntimeException) failure
                : new IllegalStateException("shadow cycle failed", failure);
        Instant failedAt = clock.instant();
        statusReport.markDegraded("SHADOW_CYCLE_FAILED", failedAt);
        nextAttemptAt = failedAt.plus(CRASH_BACKOFF);
        statusReport.recordCollectionFailure("SHADOW_CYCLE_FAILED", activeStage.name(), failure, failedAt, nextAttemptAt);
        LOGGER.error("FT546 shadow cycle failed", runtimeFailure);
        return false;
    }

    private static String sourceTimeoutReason(
            ShadowFactReadException failure) {
        String reasonCode = failure.getReasonCode();
        if ("SOURCE_QUERY_TIMEOUT".equals(reasonCode)
                || "SOURCE_BATCH_DEADLINE_EXCEEDED".equals(reasonCode)) {
            return reasonCode;
        }
        Throwable cause = failure.getCause();
        while (cause != null) {
            String type = cause.getClass().getSimpleName()
                    .toUpperCase(java.util.Locale.ROOT);
            if (type.contains("TIMEOUT")) {
                return "SOURCE_QUERY_TIMEOUT";
            }
            cause = cause.getCause();
        }
        return null;
    }

    private boolean stopPermanently(String reasonCode, Exception failure,
                                    CycleStage stage) {
        shadowCycle.invalidatePendingComparisons();
        permanentlyStopped.set(true);
        nextAttemptAt = Instant.EPOCH;
        statusReport.markPermanentlyDegraded(reasonCode, clock.instant());
        statusReport.recordCollectionFailure(reasonCode, stage.name(), failure, clock.instant(), null);
        LOGGER.error("FT546 shadow collection stopped: {}; stage={}; diagnostics={}",
                reasonCode, stage, statusReport.getCycleDiagnostics(), failure);
        return false;
    }

    private static Duration shorter(Duration left, Duration right) {
        Duration first = requirePositive(left);
        Duration second = requirePositive(right);
        return first.compareTo(second) <= 0 ? first : second;
    }

    private static Duration requirePositive(Duration value) {
        Duration duration = Objects.requireNonNull(value, "cycleDeadline");
        if (duration.isZero() || duration.isNegative()
                || duration.toMillis() == 0L) {
            throw new IllegalArgumentException(
                    "cycleDeadline must be at least one millisecond");
        }
        return duration;
    }

    private static ThreadFactory shadowCycleThreadFactory() {
        return runnable -> {
            Thread thread = new Thread(runnable, "ft546-shadow-cycle");
            thread.setDaemon(true);
            return thread;
        };
    }

    private enum CycleStage {
        IDLE,
        SOURCE_READ,
        SHADOW_CYCLE
    }

    private static ShadowReadCursor advanceFinishCursor(
            ShadowReadCursor current, ShadowFactBatch batch) {
        Instant latestDate = current.getLastCreateDate();
        String latestGuid = current.getLastGuid();
        for (ShadowQueueFact fact : batch.getSourceQueues()) {
            if (!FINISH_QUEUE_VIEW.equals(fact.getTableName())
                    || fact.getCreatedAt() == null) {
                continue;
            }
            int dateOrder = fact.getCreatedAt().compareTo(latestDate);
            if (dateOrder > 0 || (dateOrder == 0
                    && fact.getGuid().compareTo(latestGuid) > 0)) {
                latestDate = fact.getCreatedAt();
                latestGuid = fact.getGuid();
            }
        }
        return new ShadowReadCursor(latestDate, latestGuid);
    }

    private static final class CycleAttempt {
        private final ShadowFactBatch batch;
        private final ShadowCycleResult result;
        private final Instant completedAt;

        private CycleAttempt(ShadowFactBatch batch, ShadowCycleResult result, Instant completedAt) {
            this.batch = batch;
            this.result = result;
            this.completedAt = completedAt;
        }
    }

    private static final class PendingCommit {
        private final Future<CycleAttempt> future;
        private final ShadowCycleDiagnostics diagnostics;
        private final Instant deadline;
        private final long startedNanos = System.nanoTime();

        private PendingCommit(Future<CycleAttempt> future, ShadowCycleDiagnostics diagnostics, Instant startedAt) {
            this.future = future;
            this.diagnostics = diagnostics;
            this.deadline = startedAt.plus(COMMIT_RECOVERY_WINDOW);
        }

        private boolean expired(Instant now) {
            return !now.isBefore(deadline) || System.nanoTime() - startedNanos >= COMMIT_RECOVERY_WINDOW.toNanos();
        }
    }
}