package at.procon.eventhub.esperpoc.service;

import at.procon.eventhub.esperpoc.dto.ActivityIntervalDto;
import at.procon.eventhub.esperpoc.dto.EsperOperatingPeriodEngineMode;
import at.procon.eventhub.esperpoc.dto.OperatingPeriodActivityIntervalDto;
import com.espertech.esper.common.client.EPCompiled;
import com.espertech.esper.common.client.EventBean;
import com.espertech.esper.common.client.configuration.Configuration;
import com.espertech.esper.compiler.client.CompilerArguments;
import com.espertech.esper.compiler.client.EPCompileException;
import com.espertech.esper.compiler.client.EPCompilerProvider;
import com.espertech.esper.runtime.client.EPDeployException;
import com.espertech.esper.runtime.client.EPDeployment;
import com.espertech.esper.runtime.client.EPRuntime;
import com.espertech.esper.runtime.client.EPRuntimeProvider;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.Duration;
import java.time.OffsetDateTime;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Component;
import org.springframework.util.StreamUtils;

@Component
public class EsperOperatingPeriodEngine {

    private static final AtomicLong RUNTIME_COUNTER = new AtomicLong();
    // Minimal stream-only mode: Esper preserves event ordering, Java owns the period state machine.
    private static final String INPUT_STREAM_EPL = """
            @name('operatingPeriodIntervalStream')
            select * from OperatingPeriodIntervalInputEvent
            """;
    // Full-EPL mode: Esper owns the operating-period state machine and emits periodized intervals/closures.
    private static final String FULL_EPL_TEMPLATE = loadResource("esper/operating-period-state-machine.epl");

    public EsperOperatingPeriodEvaluation evaluate(
            List<ActivityIntervalDto> intervals,
            Duration operatingSplitIdleThreshold,
            EsperOperatingPeriodEngineMode mode
    ) {
        List<ActivityIntervalDto> sorted = sortedPositiveIntervals(intervals);
        if (sorted.isEmpty()) {
            return new EsperOperatingPeriodEvaluation(List.of(), List.of());
        }
        if (mode == EsperOperatingPeriodEngineMode.FULL_EPL) {
            return evaluateFullEpl(sorted, operatingSplitIdleThreshold);
        }
        return evaluateStreamCollector(sorted, operatingSplitIdleThreshold);
    }

    public EsperOperatingPeriodEvaluation evaluate(
            List<ActivityIntervalDto> intervals,
            Duration operatingSplitIdleThreshold
    ) {
        return evaluate(intervals, operatingSplitIdleThreshold, EsperOperatingPeriodEngineMode.STREAM_COLLECTOR);
    }

    private EsperOperatingPeriodEvaluation evaluateStreamCollector(
            List<ActivityIntervalDto> sorted,
            Duration operatingSplitIdleThreshold
    ) {
        // In stream-collector mode Esper only serializes the stream; period transitions are evaluated in Java.
        PeriodizationCollector collector = new PeriodizationCollector(operatingSplitIdleThreshold);
        executeWithRuntime(
                configuration -> configuration.getCommon().addEventType("OperatingPeriodIntervalInputEvent", EsperOperatingPeriodIntervalInputEvent.class),
                INPUT_STREAM_EPL,
                List.of("operatingPeriodIntervalStream"),
                (statementName, newData) -> collectInputIntervals(newData, collector),
                runtime -> {
                    for (ActivityIntervalDto interval : sorted) {
                        runtime.getEventService().sendEventBean(toInputEvent(interval), "OperatingPeriodIntervalInputEvent");
                    }
                }
        );
        return collector.finish();
    }

    private EsperOperatingPeriodEvaluation evaluateFullEpl(
            List<ActivityIntervalDto> sorted,
            Duration operatingSplitIdleThreshold
    ) {
        // The full-EPL script is parameterized per request so the idle-split threshold matches the request/config.
        String epl = FULL_EPL_TEMPLATE.replace("${operatingSplitIdleMs}", Long.toString(operatingSplitIdleThreshold.toMillis()));
        List<OperatingPeriodActivityIntervalDto> periodizedIntervals = new ArrayList<>();
        List<EsperClosedOperatingPeriod> closedPeriods = new ArrayList<>();
        executeWithRuntime(
                configuration -> {
                    // Full-EPL mode uses explicit map schemas so EPL can own the whole state machine without
                    // relying on Java bean-property resolution during compilation.
                    Map<String, Object> inputDefinition = new LinkedHashMap<>();
                    inputDefinition.put("driverId", java.util.UUID.class);
                    inputDefinition.put("vehicleId", java.util.UUID.class);
                    inputDefinition.put("vehicleRegistrationId", java.util.UUID.class);
                    inputDefinition.put("activityType", String.class);
                    inputDefinition.put("cardSlot", String.class);
                    inputDefinition.put("cardStatus", String.class);
                    inputDefinition.put("drivingStatus", String.class);
                    inputDefinition.put("sourceKind", String.class);
                    inputDefinition.put("startTs", long.class);
                    inputDefinition.put("endTs", long.class);
                    inputDefinition.put("durationMs", long.class);
                    inputDefinition.put("sourceRowId", String.class);
                    inputDefinition.put("sourceRowIds", java.util.List.class);
                    inputDefinition.put("clippedToRequestedPeriod", boolean.class);
                    inputDefinition.put("synthetic", boolean.class);
                    configuration.getCommon().addEventType("OperatingPeriodInputMap", inputDefinition);
                    configuration.getCommon().addEventType("OperatingPeriodFlushEvent", Map.of("reason", String.class));
                },
                epl,
                List.of("periodizedActivityIntervals", "operatingPeriodClosed"),
                (statementName, newData) -> {
                    if ("periodizedActivityIntervals".equals(statementName)) {
                        collectPeriodizedOutputs(newData, periodizedIntervals);
                    } else if ("operatingPeriodClosed".equals(statementName)) {
                        collectClosedOutputs(newData, closedPeriods);
                    }
                },
                runtime -> {
                    // Historical evaluation sends the complete interval timeline first and then a single flush event
                    // so the EPL state machine can emit the final still-open operating period.
                    for (ActivityIntervalDto interval : sorted) {
                        runtime.getEventService().sendEventMap(toInputMap(interval), "OperatingPeriodInputMap");
                    }
                    runtime.getEventService().sendEventMap(Map.of("reason", "HISTORICAL_EVALUATION"), "OperatingPeriodFlushEvent");
                }
        );
        return new EsperOperatingPeriodEvaluation(
                periodizedIntervals.stream()
                        .sorted(Comparator.comparing(OperatingPeriodActivityIntervalDto::startedAt)
                                .thenComparing(OperatingPeriodActivityIntervalDto::endedAt)
                                .thenComparing(OperatingPeriodActivityIntervalDto::activityType, Comparator.nullsLast(String::compareTo)))
                        .toList(),
                closedPeriods.stream()
                        .sorted(Comparator.comparing(EsperClosedOperatingPeriod::startedAt))
                        .toList()
        );
    }

    private void executeWithRuntime(
            java.util.function.Consumer<Configuration> configurationSetup,
            String epl,
            List<String> statementNames,
            StatementListener listener,
            java.util.function.Consumer<EPRuntime> sender
    ) {
        EPRuntime runtime = null;
        try {
            Configuration configuration = new Configuration();
            configurationSetup.accept(configuration);
            String runtimeUri = "eventhub-esper-operating-period-" + RUNTIME_COUNTER.incrementAndGet();
            runtime = EPRuntimeProvider.getRuntime(runtimeUri, configuration);

            CompilerArguments arguments = new CompilerArguments(configuration);
            EPCompiled compiled = EPCompilerProvider.getCompiler().compile(epl, arguments);
            EPDeployment deployment = runtime.getDeploymentService().deploy(compiled);
            // Multiple statements may emit outputs from a single deployment; we dispatch by statement name.
            for (String statementName : statementNames) {
                runtime.getDeploymentService()
                        .getStatement(deployment.getDeploymentId(), statementName)
                        .addListener((newData, oldData, statement, rt) -> listener.accept(statementName, newData));
            }
            sender.accept(runtime);
        } catch (EPCompileException | EPDeployException e) {
            throw new IllegalStateException("Cannot compile/deploy Esper operating-period EPL", e);
        } finally {
            if (runtime != null) {
                runtime.destroy();
            }
        }
    }

    private void collectInputIntervals(EventBean[] newData, PeriodizationCollector collector) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            // Stream-collector mode receives the ordered interval stream back from Esper and applies the
            // deterministic Java state machine to it.
            collector.accept((EsperOperatingPeriodIntervalInputEvent) event.getUnderlying());
        }
    }

    private void collectPeriodizedOutputs(EventBean[] newData, List<OperatingPeriodActivityIntervalDto> target) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            target.add(new OperatingPeriodActivityIntervalDto(
                    (java.util.UUID) event.get("driverId"),
                    (java.util.UUID) event.get("vehicleId"),
                    (java.util.UUID) event.get("vehicleRegistrationId"),
                    (String) event.get("activityType"),
                    (String) event.get("cardSlot"),
                    (String) event.get("cardStatus"),
                    (String) event.get("drivingStatus"),
                    (String) event.get("sourceKind"),
                    OffsetDateTime.ofInstant(Instant.ofEpochMilli((Long) event.get("startedAtTs")), java.time.ZoneOffset.UTC),
                    OffsetDateTime.ofInstant(Instant.ofEpochMilli((Long) event.get("endedAtTs")), java.time.ZoneOffset.UTC),
                    ((Long) event.get("durationMs")) / 1000L,
                    (String) event.get("sourceRowId"),
                    castSourceRowIds(event.get("sourceRowIds")),
                    (Boolean) event.get("clippedToRequestedPeriod"),
                    "PERIODIZED_ACTIVITY",
                    (Long) event.get("operatingPeriodNo"),
                    OffsetDateTime.ofInstant(Instant.ofEpochMilli((Long) event.get("operatingPeriodStartedAtTs")), java.time.ZoneOffset.UTC),
                    (Boolean) event.get("newOperatingPeriod"),
                    nullableMillisToSeconds((Long) event.get("gapSincePreviousActivityMs")),
                    (Boolean) event.get("synthetic")
            ));
        }
    }

    private void collectClosedOutputs(EventBean[] newData, List<EsperClosedOperatingPeriod> target) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            target.add(new EsperClosedOperatingPeriod(
                    (Long) event.get("operatingPeriodNo"),
                    OffsetDateTime.ofInstant(Instant.ofEpochMilli((Long) event.get("operatingPeriodStartedAtTs")), java.time.ZoneOffset.UTC),
                    OffsetDateTime.ofInstant(Instant.ofEpochMilli((Long) event.get("operatingPeriodEndedAtTs")), java.time.ZoneOffset.UTC),
                    ((Long) event.get("durationMs")) / 1000L,
                    (String) event.get("closedBy")
            ));
        }
    }

    @SuppressWarnings("unchecked")
    private List<String> castSourceRowIds(Object value) {
        return value == null ? List.of() : List.copyOf((List<String>) value);
    }

    private Long nullableMillisToSeconds(Long value) {
        return value == null ? null : value / 1000L;
    }

    private EsperOperatingPeriodIntervalInputEvent toInputEvent(ActivityIntervalDto interval) {
        return new EsperOperatingPeriodIntervalInputEvent(
                interval.driverEntityId(),
                interval.vehicleId(),
                interval.vehicleRegistrationId(),
                interval.activityType(),
                interval.cardSlot(),
                interval.cardStatus(),
                interval.drivingStatus(),
                interval.sourceKind(),
                interval.startedAt().toInstant().toEpochMilli(),
                interval.endedAt().toInstant().toEpochMilli(),
                interval.durationSeconds() * 1000L,
                interval.sourceRowId(),
                interval.sourceRowIds(),
                interval.clippedToRequestedPeriod(),
                "UNKNOWN_GAP".equals(interval.level())
        );
    }

    private Map<String, Object> toInputMap(ActivityIntervalDto interval) {
        Map<String, Object> map = new LinkedHashMap<>();
        map.put("driverId", interval.driverEntityId());
        map.put("vehicleId", interval.vehicleId());
        map.put("vehicleRegistrationId", interval.vehicleRegistrationId());
        map.put("activityType", interval.activityType());
        map.put("cardSlot", interval.cardSlot());
        map.put("cardStatus", interval.cardStatus());
        map.put("drivingStatus", interval.drivingStatus());
        map.put("sourceKind", interval.sourceKind());
        map.put("startTs", interval.startedAt().toInstant().toEpochMilli());
        map.put("endTs", interval.endedAt().toInstant().toEpochMilli());
        map.put("durationMs", interval.durationSeconds() * 1000L);
        map.put("sourceRowId", interval.sourceRowId());
        map.put("sourceRowIds", interval.sourceRowIds());
        map.put("clippedToRequestedPeriod", interval.clippedToRequestedPeriod());
        map.put("synthetic", "UNKNOWN_GAP".equals(interval.level()));
        return map;
    }

    private List<ActivityIntervalDto> sortedPositiveIntervals(List<ActivityIntervalDto> intervals) {
        if (intervals == null || intervals.isEmpty()) {
            return List.of();
        }
        return intervals.stream()
                .filter(interval -> interval.endedAt().isAfter(interval.startedAt()))
                .sorted(Comparator.comparing(ActivityIntervalDto::startedAt)
                        .thenComparing(ActivityIntervalDto::endedAt)
                        .thenComparing(ActivityIntervalDto::activityType, Comparator.nullsLast(String::compareTo)))
                .toList();
    }

    public record EsperOperatingPeriodEvaluation(
            List<OperatingPeriodActivityIntervalDto> periodizedIntervals,
            List<EsperClosedOperatingPeriod> closedPeriods
    ) {
    }

    private interface StatementListener {
        void accept(String statementName, EventBean[] newData);
    }

    private static String loadResource(String path) {
        try {
            return StreamUtils.copyToString(new ClassPathResource(path).getInputStream(), StandardCharsets.UTF_8);
        } catch (IOException e) {
            throw new IllegalStateException("Cannot load Esper resource " + path, e);
        }
    }

    private static final class PeriodizationCollector {
        private final Duration operatingSplitIdleThreshold;
        private final List<OperatingPeriodActivityIntervalDto> periodizedIntervals = new ArrayList<>();
        private final List<EsperClosedOperatingPeriod> closedPeriods = new ArrayList<>();
        private boolean hasOpenPeriod;
        private long operatingPeriodNo;
        private OffsetDateTime operatingPeriodStartedAt;
        private OffsetDateTime lastKnownActivityEndAt;

        private PeriodizationCollector(Duration operatingSplitIdleThreshold) {
            this.operatingSplitIdleThreshold = operatingSplitIdleThreshold;
        }

        private void accept(EsperOperatingPeriodIntervalInputEvent interval) {
            ActivityIntervalDto dto = new ActivityIntervalDto(
                    interval.driverId(),
                    interval.vehicleId(),
                    interval.vehicleRegistrationId(),
                    interval.activityType(),
                    interval.cardSlot(),
                    interval.cardStatus(),
                    interval.drivingStatus(),
                    interval.sourceKind(),
                    OffsetDateTime.ofInstant(java.time.Instant.ofEpochMilli(interval.startTs()), java.time.ZoneOffset.UTC),
                    OffsetDateTime.ofInstant(java.time.Instant.ofEpochMilli(interval.endTs()), java.time.ZoneOffset.UTC),
                    interval.durationMs() / 1000L,
                    interval.sourceRowId(),
                    interval.sourceRowIds(),
                    interval.clippedToRequestedPeriod(),
                    interval.synthetic() ? "UNKNOWN_GAP" : "RAW_INTERVAL"
            );

            if ("UNKNOWN".equals(dto.activityType())) {
                if (!hasOpenPeriod) {
                    // Unknown time before the first known activity does not belong to any operating period.
                    return;
                }
                if (dto.durationSeconds() >= operatingSplitIdleThreshold.getSeconds()) {
                    // Long UNKNOWN behaves like a closing gap: close the current period and wait for the next
                    // known activity to reopen a new period number.
                    closeCurrent("UNKNOWN_GAP");
                    return;
                }
                // Short UNKNOWN stays inside the current period as explicit uncertainty.
                periodizedIntervals.add(OperatingPeriodActivityIntervalDto.periodized(
                        dto,
                        operatingPeriodNo,
                        operatingPeriodStartedAt,
                        false,
                        Math.max(0, Duration.between(lastKnownActivityEndAt, dto.startedAt()).getSeconds())
                ));
                return;
            }

            if (!hasOpenPeriod) {
                // First known activity, or first activity after a long closing gap, opens a new operating period.
                operatingPeriodNo = operatingPeriodNo < 1 ? 1 : operatingPeriodNo + 1;
                hasOpenPeriod = true;
                operatingPeriodStartedAt = dto.startedAt();
                lastKnownActivityEndAt = dto.endedAt();
                periodizedIntervals.add(OperatingPeriodActivityIntervalDto.periodized(
                        dto,
                        operatingPeriodNo,
                        operatingPeriodStartedAt,
                        true,
                        null
                ));
                return;
            }

            long gapSeconds = Math.max(0, Duration.between(lastKnownActivityEndAt, dto.startedAt()).getSeconds());
            if (gapSeconds >= operatingSplitIdleThreshold.getSeconds()) {
                // Long idle time between known activities closes the current period and starts the next one.
                closeCurrent("IDLE_GAP");
                operatingPeriodNo++;
                hasOpenPeriod = true;
                operatingPeriodStartedAt = dto.startedAt();
                lastKnownActivityEndAt = dto.endedAt();
                periodizedIntervals.add(OperatingPeriodActivityIntervalDto.periodized(
                        dto,
                        operatingPeriodNo,
                        operatingPeriodStartedAt,
                        true,
                        gapSeconds
                ));
                return;
            }

            // Normal forward continuity inside the same period.
            periodizedIntervals.add(OperatingPeriodActivityIntervalDto.periodized(
                    dto,
                    operatingPeriodNo,
                    operatingPeriodStartedAt,
                    false,
                    gapSeconds
            ));
            if (dto.endedAt().isAfter(lastKnownActivityEndAt)) {
                lastKnownActivityEndAt = dto.endedAt();
            }
        }

        private EsperOperatingPeriodEvaluation finish() {
            if (hasOpenPeriod) {
                // Historical evaluation has no future event to close the final period, so emit it explicitly.
                closeCurrent("FLUSH");
            }
            return new EsperOperatingPeriodEvaluation(
                    periodizedIntervals.stream()
                            .sorted(Comparator.comparing(OperatingPeriodActivityIntervalDto::startedAt)
                                    .thenComparing(OperatingPeriodActivityIntervalDto::endedAt)
                                    .thenComparing(OperatingPeriodActivityIntervalDto::activityType, Comparator.nullsLast(String::compareTo)))
                            .toList(),
                    closedPeriods.stream()
                            .sorted(Comparator.comparing(EsperClosedOperatingPeriod::startedAt))
                            .toList()
            );
        }

        private void closeCurrent(String closedBy) {
            if (!hasOpenPeriod || operatingPeriodStartedAt == null || lastKnownActivityEndAt == null) {
                hasOpenPeriod = false;
                return;
            }
            // A closed period always ends at the last known non-rest activity end, never at the synthetic UNKNOWN.
            closedPeriods.add(new EsperClosedOperatingPeriod(
                    operatingPeriodNo,
                    operatingPeriodStartedAt,
                    lastKnownActivityEndAt,
                    Duration.between(operatingPeriodStartedAt, lastKnownActivityEndAt).getSeconds(),
                    closedBy
            ));
            hasOpenPeriod = false;
        }
    }
}
