package at.procon.eventhub.esperpoc.service;

import at.procon.eventhub.esperpoc.dto.ActivityIntervalDto;
import at.procon.eventhub.esperpoc.dto.RawActivityEventDto;
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.time.Duration;
import java.time.OffsetDateTime;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import org.springframework.stereotype.Component;

@Component
public class EsperDriverActivityEngine {

    private static final AtomicLong RUNTIME_COUNTER = new AtomicLong();

    private static final String INTERVAL_EPL = """
            @name('driverCardActivityIntervals')
            select
                s.driverEntityId as driverEntityId,
                s.vehicleId as vehicleId,
                s.vehicleRegistrationId as vehicleRegistrationId,
                s.eventType as activityType,
                s.cardSlot as cardSlot,
                s.cardStatus as cardStatus,
                s.drivingStatus as drivingStatus,
                s.sourceKind as sourceKind,
                s.occurredAt as startedAt,
                e.occurredAt as endedAt,
                s.sourceRowId as sourceRowId
            from pattern [
                every s = RawDriverActivityPoint(lifecycle = 'START') ->
                      e = RawDriverActivityPoint(
                          lifecycle = 'END',
                          sourceRowId = s.sourceRowId,
                          eventType = s.eventType,
                          driverEntityId = s.driverEntityId,
                          cardSlot = s.cardSlot
                      )
            ]
            """;

    private static final String INTERVAL_STREAM_EPL = """
            @name('driverActivityIntervalStream')
            select * from DriverActivityInterval
            """;

    public List<ActivityIntervalDto> buildIntervals(List<RawActivityEventDto> rawEvents) {
        if (rawEvents == null || rawEvents.isEmpty()) {
            return List.of();
        }

        List<ActivityIntervalDto> intervals = new ArrayList<>();
        executeWithRuntime(
                configuration -> configuration.getCommon().addEventType("RawDriverActivityPoint", EsperRawDriverActivityPoint.class),
                INTERVAL_EPL,
                "driverCardActivityIntervals",
                newData -> collectIntervals(newData, intervals),
                runtime -> {
                    List<EsperRawDriverActivityPoint> points = rawEvents.stream()
                            .sorted(Comparator.comparing(RawActivityEventDto::occurredAt).thenComparing(RawActivityEventDto::lifecycle))
                            .map(this::toEsperPoint)
                            .toList();
                    for (EsperRawDriverActivityPoint point : points) {
                        runtime.getEventService().sendEventBean(point, "RawDriverActivityPoint");
                    }
                }
        );

        return intervals.stream()
                .filter(interval -> interval.endedAt().isAfter(interval.startedAt()))
                .sorted(Comparator.comparing(ActivityIntervalDto::startedAt).thenComparing(ActivityIntervalDto::endedAt))
                .toList();
    }

    public List<ActivityIntervalDto> mergeConsecutiveIdenticalActivities(
            List<ActivityIntervalDto> intervals,
            Duration mergeGapTolerance
    ) {
        List<ActivityIntervalDto> sorted = sortedPositiveIntervals(intervals);
        if (sorted.isEmpty()) {
            return List.of();
        }

        MergedActivityCollector collector = new MergedActivityCollector(mergeGapTolerance);
        executeIntervalStream(sorted, collector::accept);
        return collector.finish();
    }

    public List<EsperResolvedShiftSpan> resolveShiftSpans(
            List<ActivityIntervalDto> intervals,
            Duration shiftEndThreshold,
            Duration contiguousGapTolerance
    ) {
        List<ActivityIntervalDto> sorted = sortedPositiveIntervals(intervals);
        if (sorted.isEmpty()) {
            return List.of();
        }

        ShiftSpanCollector collector = new ShiftSpanCollector(shiftEndThreshold, contiguousGapTolerance);
        executeIntervalStream(sorted, collector::accept);
        return collector.finish();
    }

    private void executeIntervalStream(List<ActivityIntervalDto> intervals, Consumer<ActivityIntervalDto> consumer) {
        executeWithRuntime(
                configuration -> configuration.getCommon().addEventType("DriverActivityInterval", EsperActivityIntervalEvent.class),
                INTERVAL_STREAM_EPL,
                "driverActivityIntervalStream",
                newData -> collectInputIntervals(newData, consumer),
                runtime -> {
                    for (ActivityIntervalDto interval : intervals) {
                        runtime.getEventService().sendEventBean(toEsperInterval(interval), "DriverActivityInterval");
                    }
                }
        );
    }

    private void executeWithRuntime(
            Consumer<Configuration> configurationSetup,
            String epl,
            String statementName,
            Consumer<EventBean[]> listener,
            Consumer<EPRuntime> sender
    ) {
        EPRuntime runtime = null;
        try {
            Configuration configuration = new Configuration();
            configurationSetup.accept(configuration);
            String runtimeUri = "eventhub-esper-poc-" + 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);
            runtime.getDeploymentService()
                    .getStatement(deployment.getDeploymentId(), statementName)
                    .addListener((newData, oldData, statement, rt) -> listener.accept(newData));

            sender.accept(runtime);
        } catch (EPCompileException | EPDeployException e) {
            throw new IllegalStateException("Cannot compile/deploy Esper PoC EPL", e);
        } finally {
            if (runtime != null) {
                runtime.destroy();
            }
        }
    }

    private void collectIntervals(EventBean[] newData, List<ActivityIntervalDto> intervals) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            OffsetDateTime startedAt = (OffsetDateTime) event.get("startedAt");
            OffsetDateTime endedAt = (OffsetDateTime) event.get("endedAt");
            intervals.add(ActivityIntervalDto.raw(
                    (UUID) event.get("driverEntityId"),
                    (UUID) event.get("vehicleId"),
                    (UUID) event.get("vehicleRegistrationId"),
                    (String) event.get("activityType"),
                    (String) event.get("cardSlot"),
                    (String) event.get("cardStatus"),
                    (String) event.get("drivingStatus"),
                    (String) event.get("sourceKind"),
                    startedAt,
                    endedAt,
                    (String) event.get("sourceRowId")
            ));
        }
    }

    private void collectInputIntervals(EventBean[] newData, Consumer<ActivityIntervalDto> consumer) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            EsperActivityIntervalEvent interval = (EsperActivityIntervalEvent) event.getUnderlying();
            consumer.accept(new ActivityIntervalDto(
                    interval.driverEntityId(),
                    interval.vehicleId(),
                    interval.vehicleRegistrationId(),
                    interval.activityType(),
                    interval.cardSlot(),
                    interval.cardStatus(),
                    interval.drivingStatus(),
                    interval.sourceKind(),
                    interval.startedAt(),
                    interval.endedAt(),
                    interval.durationSeconds(),
                    interval.sourceRowId(),
                    interval.sourceRowIds(),
                    interval.clippedToRequestedPeriod(),
                    interval.level()
            ));
        }
    }

    private EsperRawDriverActivityPoint toEsperPoint(RawActivityEventDto event) {
        return new EsperRawDriverActivityPoint(
                event.eventId(),
                event.occurredAt(),
                event.sourceRowId(),
                event.externalSourceEventId(),
                event.driverEntityId(),
                event.vehicleId(),
                event.vehicleRegistrationId(),
                event.activityType(),
                event.lifecycle(),
                event.cardSlot(),
                event.cardStatus(),
                event.drivingStatus(),
                event.sourceKind()
        );
    }

    private EsperActivityIntervalEvent toEsperInterval(ActivityIntervalDto interval) {
        return new EsperActivityIntervalEvent(
                interval.driverEntityId(),
                interval.vehicleId(),
                interval.vehicleRegistrationId(),
                interval.activityType(),
                interval.cardSlot(),
                interval.cardStatus(),
                interval.drivingStatus(),
                interval.sourceKind(),
                interval.startedAt(),
                interval.endedAt(),
                interval.durationSeconds(),
                interval.sourceRowId(),
                interval.sourceRowIds(),
                interval.clippedToRequestedPeriod(),
                interval.level()
        );
    }

    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();
    }

    private static final class MergedActivityCollector {
        private final Duration mergeGapTolerance;
        private final List<ActivityIntervalDto> merged = new ArrayList<>();
        private ActivityIntervalDto current;
        private List<String> currentSources = List.of();

        private MergedActivityCollector(Duration mergeGapTolerance) {
            this.mergeGapTolerance = mergeGapTolerance;
        }

        private void accept(ActivityIntervalDto next) {
            if (current == null) {
                current = next;
                currentSources = new ArrayList<>(next.sourceRowIds());
                return;
            }
            if (EsperActivitySemantics.canMerge(current, next, mergeGapTolerance)) {
                currentSources.addAll(next.sourceRowIds());
                OffsetDateTime newEndedAt = current.endedAt().isAfter(next.endedAt()) ? current.endedAt() : next.endedAt();
                current = new ActivityIntervalDto(
                        current.driverEntityId(),
                        current.vehicleId() == null ? next.vehicleId() : current.vehicleId(),
                        current.vehicleRegistrationId() == null ? next.vehicleRegistrationId() : current.vehicleRegistrationId(),
                        current.activityType(),
                        current.cardSlot(),
                        current.cardStatus(),
                        current.drivingStatus(),
                        current.sourceKind(),
                        current.startedAt(),
                        newEndedAt,
                        Duration.between(current.startedAt(), newEndedAt).getSeconds(),
                        current.sourceRowId(),
                        List.copyOf(currentSources),
                        current.clippedToRequestedPeriod() || next.clippedToRequestedPeriod(),
                        "MERGED_ACTIVITY"
                );
                return;
            }
            merged.add(current.asMerged(List.copyOf(currentSources)));
            current = next;
            currentSources = new ArrayList<>(next.sourceRowIds());
        }

        private List<ActivityIntervalDto> finish() {
            if (current != null) {
                merged.add(current.asMerged(List.copyOf(currentSources)));
            }
            return List.copyOf(merged);
        }
    }

    private static final class ShiftSpanCollector {
        private final Duration shiftEndThreshold;
        private final Duration contiguousGapTolerance;
        private final List<ActivityIntervalDto> seen = new ArrayList<>();
        private final List<EsperResolvedShiftSpan> spans = new ArrayList<>();
        private OffsetDateTime shiftBegin;
        private long accumulatedRestSeconds;
        private OffsetDateTime prevActivityEnd;
        private OffsetDateTime pendingShiftEndTriggerAt;

        private ShiftSpanCollector(Duration shiftEndThreshold, Duration contiguousGapTolerance) {
            this.shiftEndThreshold = shiftEndThreshold;
            this.contiguousGapTolerance = contiguousGapTolerance;
        }

        private void accept(ActivityIntervalDto interval) {
            seen.add(interval);
            if (shiftBegin == null) {
                if (EsperActivitySemantics.isShiftActivity(interval)) {
                    shiftBegin = interval.startedAt();
                }
                prevActivityEnd = interval.endedAt();
                return;
            }
            if (!shiftBegin.isBefore(interval.startedAt())) {
                prevActivityEnd = interval.endedAt();
                return;
            }

            if (pendingShiftEndTriggerAt != null && EsperActivitySemantics.isShiftActivity(interval)) {
                appendShiftSpan(shiftBegin, interval.startedAt());
                shiftBegin = interval.startedAt();
                pendingShiftEndTriggerAt = null;
                accumulatedRestSeconds = 0;
            }

            if (EsperActivitySemantics.isRestOrUnknown(interval)) {
                long gapSeconds = prevActivityEnd == null ? 0 : Math.max(0, Duration.between(prevActivityEnd, interval.startedAt()).getSeconds());
                boolean contiguous = prevActivityEnd != null
                        && Math.abs(Duration.between(prevActivityEnd, interval.startedAt()).getSeconds()) <= contiguousGapTolerance.getSeconds();
                boolean crossedRecordDate = prevActivityEnd != null
                        && !interval.startedAt().toLocalDate().equals(prevActivityEnd.toLocalDate());
                if (contiguous) {
                    accumulatedRestSeconds += interval.durationSeconds();
                } else if (crossedRecordDate) {
                    accumulatedRestSeconds = interval.durationSeconds() + gapSeconds;
                } else {
                    accumulatedRestSeconds = interval.durationSeconds();
                }
                if (accumulatedRestSeconds >= shiftEndThreshold.getSeconds() && pendingShiftEndTriggerAt == null) {
                    pendingShiftEndTriggerAt = interval.startedAt();
                }
            } else if (EsperActivitySemantics.isShiftActivity(interval)) {
                long gapSeconds = prevActivityEnd == null ? 0 : Math.max(0, Duration.between(prevActivityEnd, interval.startedAt()).getSeconds());
                boolean crossedRecordDate = prevActivityEnd != null
                        && !interval.startedAt().toLocalDate().equals(prevActivityEnd.toLocalDate());
                if (crossedRecordDate && accumulatedRestSeconds + gapSeconds >= shiftEndThreshold.getSeconds()) {
                    appendShiftSpan(shiftBegin, interval.startedAt());
                    shiftBegin = interval.startedAt();
                }
                accumulatedRestSeconds = 0;
                pendingShiftEndTriggerAt = null;
            }

            prevActivityEnd = interval.endedAt();
        }

        private List<EsperResolvedShiftSpan> finish() {
            if (shiftBegin != null) {
                appendShiftSpan(shiftBegin, null);
            }
            return List.copyOf(spans);
        }

        private void appendShiftSpan(OffsetDateTime currentShiftBegin, OffsetDateTime nextShiftBegin) {
            if (currentShiftBegin == null) {
                return;
            }
            OffsetDateTime shiftEnd = seen.stream()
                    .filter(interval -> !interval.startedAt().isBefore(currentShiftBegin))
                    .filter(interval -> nextShiftBegin == null || interval.startedAt().isBefore(nextShiftBegin))
                    .filter(EsperActivitySemantics::isShiftActivity)
                    .map(ActivityIntervalDto::endedAt)
                    .max(OffsetDateTime::compareTo)
                    .orElse(null);
            if (shiftEnd == null || !shiftEnd.isAfter(currentShiftBegin)) {
                return;
            }
            long restSeconds = nextShiftBegin == null ? 0 : Math.max(0, Duration.between(shiftEnd, nextShiftBegin).getSeconds());
            spans.add(new EsperResolvedShiftSpan(currentShiftBegin, shiftEnd, nextShiftBegin, restSeconds));
        }
    }
}
