package at.procon.eventhub.tachographfilesession.service;

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 at.procon.eventhub.tachographfilesession.model.DriverExtractionSession;
import at.procon.eventhub.tachographfilesession.model.ExtractedCardActivityInterval;
import at.procon.eventhub.tachographfilesession.model.ExtractedCardVehicleUsageInterval;
import at.procon.eventhub.tachographfilesession.model.ExtractedSupportEvent;
import at.procon.eventhub.tachographfilesession.model.ExtractionWarning;
import at.procon.eventhub.tachographfilesession.model.ResolvedActivityInterval;
import at.procon.eventhub.tachographfilesession.model.ResolvedDriverTimeline;
import at.procon.eventhub.tachographfilesession.model.ResolvedVehicleUsageInterval;
import at.procon.eventhub.tachographfilesession.model.TachographEsperActivityIntervalEvent;
import at.procon.eventhub.tachographfilesession.model.TachographEsperDrivingInterruptionIntervalEvent;
import at.procon.eventhub.tachographfilesession.model.TachographEsperPotentialHomeOvernightStayIntervalEvent;
import at.procon.eventhub.tachographfilesession.model.TachographEsperVuCardAbsentIntervalEvent;
import at.procon.eventhub.tachographfilesession.model.TachographEsperVehicleUsageIntervalEvent;
import at.procon.eventhub.tachographfilesession.model.TachographFileSession;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Component;
import org.springframework.util.StreamUtils;

@Component
public class DriverTimelineBuilder {

    private static final AtomicLong RUNTIME_COUNTER = new AtomicLong();
    private static final String ACTIVITY_INTERVAL_EVENTS_EPL =
            loadResource("esper/tachograph-activity-interval-events.epl");
    private static final String DRIVING_INTERVAL_EVENTS_EPL =
            loadResource("esper/tachograph-driving-interval-events.epl");
    private static final String DRIVING_INTERRUPTION_INTERVAL_EVENTS_EPL_TEMPLATE =
            loadResource("esper/tachograph-driving-interruption-interval-events.epl");
    private static final String DRIVING_INTERRUPTION_VEHICLE_CHANGE_INTERVAL_EVENTS_EPL =
            loadResource("esper/tachograph-driving-interruption-vehicle-change-interval-events.epl");
    private static final String DAILY_WEEKLY_REST_CANDIDATE_INTERVAL_EVENTS_EPL_TEMPLATE =
            loadResource("esper/tachograph-daily-weekly-rest-candidate-interval-events.epl");
    private static final String POTENTIAL_HOME_OVERNIGHT_STAY_INTERVAL_EVENTS_EPL_TEMPLATE =
            loadResource("esper/tachograph-potential-home-overnight-stay-interval-events.epl");
    private static final String VEHICLE_USAGE_INTERVAL_EVENTS_EPL =
            loadResource("esper/tachograph-vehicle-usage-interval-events.epl");
    private static final String VU_CARD_ABSENT_INTERVAL_EVENTS_EPL =
            loadResource("esper/tachograph-vu-card-absent-interval-events.epl");

    public ResolvedDriverTimeline build(TachographFileSession session, DriverExtractionSession driverSession) {
        String sourceKind = session.metadata().driverCardFile() ? "DRIVER_CARD" : "VEHICLE_UNIT";
        List<ResolvedVehicleUsageInterval> vehicleUsageIntervals = mergeVehicleUsageIntervals(
                session.sessionId(),
                driverSession.driverKey(),
                driverSession.cardVehicleUsageIntervals(),
                sourceKind
        );
        List<ResolvedActivityInterval> activityIntervals = resolveActivities(driverSession.cardActivityIntervals(), sourceKind);
        List<ExtractedSupportEvent> supportEvents = driverSession.supportEvents().stream()
                .sorted(Comparator.comparing(ExtractedSupportEvent::occurredAt)
                        .thenComparing(ExtractedSupportEvent::eventDomain, Comparator.nullsLast(String::compareTo))
                        .thenComparing(ExtractedSupportEvent::eventId, Comparator.nullsLast(String::compareTo)))
                .toList();
        OffsetDateTime loadedFrom = minTimestamp(vehicleUsageIntervals, activityIntervals, supportEvents);
        OffsetDateTime loadedTo = maxTimestamp(vehicleUsageIntervals, activityIntervals, supportEvents);
        List<ExtractionWarning> warnings = mergeWarnings(session.warnings(), driverSession.warnings());
        return new ResolvedDriverTimeline(
                sourceKind,
                loadedFrom,
                loadedTo,
                vehicleUsageIntervals,
                activityIntervals,
                supportEvents,
                warnings
        );
    }

    public List<TachographEsperActivityIntervalEvent> buildEsperActivityIntervalEvents(
            TachographFileSession session,
            DriverExtractionSession driverSession
    ) {
        String sourceKind = session.metadata().driverCardFile() ? "DRIVER_CARD" : "VEHICLE_UNIT";
        List<ResolvedActivityInterval> activityIntervals = resolveActivities(driverSession.cardActivityIntervals(), sourceKind);
        return buildEsperActivityIntervalEvents(session.sessionId(), driverSession.driverKey(), activityIntervals);
    }

    public List<TachographEsperActivityIntervalEvent> buildEsperActivityIntervalEvents(
            UUID sessionId,
            String driverKey,
            ResolvedDriverTimeline timeline
    ) {
        return timeline == null
                ? List.of()
                : buildEsperActivityIntervalEvents(sessionId, driverKey, timeline.activityIntervals());
    }

    public List<TachographEsperActivityIntervalEvent> buildEsperDrivingIntervalEvents(
            TachographFileSession session,
            DriverExtractionSession driverSession
    ) {
        String sourceKind = session.metadata().driverCardFile() ? "DRIVER_CARD" : "VEHICLE_UNIT";
        List<ResolvedActivityInterval> activityIntervals = resolveActivities(driverSession.cardActivityIntervals(), sourceKind);
        return buildEsperDrivingIntervalEvents(session.sessionId(), driverSession.driverKey(), activityIntervals);
    }

    public List<TachographEsperActivityIntervalEvent> buildEsperDrivingIntervalEvents(
            UUID sessionId,
            String driverKey,
            ResolvedDriverTimeline timeline
    ) {
        if (timeline == null) {
            return List.of();
        }
        return buildEsperDrivingIntervalEvents(sessionId, driverKey, timeline.activityIntervals());
    }

    public List<TachographEsperVehicleUsageIntervalEvent> buildEsperVehicleUsageIntervalEvents(
            TachographFileSession session,
            DriverExtractionSession driverSession
    ) {
        String sourceKind = session.metadata().driverCardFile() ? "DRIVER_CARD" : "VEHICLE_UNIT";
        List<ResolvedVehicleUsageInterval> vehicleUsageIntervals = mergeVehicleUsageIntervals(
                session.sessionId(),
                driverSession.driverKey(),
                driverSession.cardVehicleUsageIntervals(),
                sourceKind
        );
        return buildEsperVehicleUsageIntervalEvents(vehicleUsageIntervals);
    }

    public List<TachographEsperVehicleUsageIntervalEvent> buildEsperVehicleUsageIntervalEvents(
            ResolvedDriverTimeline timeline
    ) {
        return timeline == null ? List.of() : buildEsperVehicleUsageIntervalEvents(timeline.vehicleUsageIntervals());
    }

    public List<TachographEsperDrivingInterruptionIntervalEvent> buildEsperDrivingInterruptionIntervalEvents(
            TachographFileSession session,
            DriverExtractionSession driverSession,
            int significantDrivingMinutes
    ) {
        String sourceKind = session.metadata().driverCardFile() ? "DRIVER_CARD" : "VEHICLE_UNIT";
        List<ResolvedActivityInterval> activityIntervals = resolveActivities(driverSession.cardActivityIntervals(), sourceKind);
        return buildEsperDrivingInterruptionIntervalEvents(
                session.sessionId(),
                driverSession.driverKey(),
                activityIntervals,
                significantDrivingMinutes
        );
    }

    public List<TachographEsperDrivingInterruptionIntervalEvent> buildEsperDrivingInterruptionIntervalEvents(
            UUID sessionId,
            String driverKey,
            ResolvedDriverTimeline timeline,
            int significantDrivingMinutes
    ) {
        if (timeline == null) {
            return List.of();
        }
        return buildEsperDrivingInterruptionIntervalEvents(
                sessionId,
                driverKey,
                timeline.activityIntervals(),
                significantDrivingMinutes
        );
    }

    public List<TachographEsperDrivingInterruptionIntervalEvent> buildEsperDrivingInterruptionVehicleChangeIntervalEvents(
            List<TachographEsperDrivingInterruptionIntervalEvent> drivingInterruptionIntervals
    ) {
        if (drivingInterruptionIntervals == null || drivingInterruptionIntervals.isEmpty()) {
            return List.of();
        }
        List<TachographEsperDrivingInterruptionIntervalEvent> result = new ArrayList<>();
        executeWithRuntime(
                configuration -> configuration.getCommon().addEventType(
                        "TachographDrivingInterruptionIntervalInputEvent",
                        drivingInterruptionIntervalInputDefinition()
                ),
                DRIVING_INTERRUPTION_VEHICLE_CHANGE_INTERVAL_EVENTS_EPL,
                "drivingInterruptionVehicleChangeIntervals",
                newData -> collectDrivingInterruptionIntervalEventsFromTimestamps(newData, result),
                runtime -> {
                    for (TachographEsperDrivingInterruptionIntervalEvent interval : drivingInterruptionIntervals) {
                        runtime.getEventService().sendEventMap(
                                toDrivingInterruptionIntervalInputMap(interval),
                                "TachographDrivingInterruptionIntervalInputEvent"
                        );
                    }
                }
        );
        return result;
    }

    public List<TachographEsperPotentialHomeOvernightStayIntervalEvent> buildEsperPotentialHomeOvernightStayIntervalEvents(
            List<TachographEsperDrivingInterruptionIntervalEvent> drivingInterruptionVehicleChangeIntervals,
            List<TachographEsperVuCardAbsentIntervalEvent> vuCardAbsentIntervals
    ) {
        if (drivingInterruptionVehicleChangeIntervals == null
                || drivingInterruptionVehicleChangeIntervals.isEmpty()
                || vuCardAbsentIntervals == null
                || vuCardAbsentIntervals.isEmpty()) {
            return List.of();
        }
        List<TachographEsperPotentialHomeOvernightStayIntervalEvent> result = new ArrayList<>();
        executeWithRuntime(
                configuration -> {
                    configuration.getCommon().addEventType(
                            "TachographDrivingInterruptionVehicleChangeIntervalInputEvent",
                            drivingInterruptionIntervalInputDefinition()
                    );
                    configuration.getCommon().addEventType(
                            "TachographVuCardAbsentIntervalInputEvent",
                            vuCardAbsentIntervalInputDefinition()
                    );
                },
                POTENTIAL_HOME_OVERNIGHT_STAY_INTERVAL_EVENTS_EPL_TEMPLATE,
                "potentialHomeOvernightStayIntervals",
                newData -> collectPotentialHomeOvernightStayIntervalEvents(newData, result),
                runtime -> {
                    for (TachographEsperVuCardAbsentIntervalEvent interval : vuCardAbsentIntervals) {
                        runtime.getEventService().sendEventMap(
                                toVuCardAbsentIntervalInputMap(interval),
                                "TachographVuCardAbsentIntervalInputEvent"
                        );
                    }
                    for (TachographEsperDrivingInterruptionIntervalEvent interval : drivingInterruptionVehicleChangeIntervals) {
                        runtime.getEventService().sendEventMap(
                                toDrivingInterruptionIntervalInputMap(interval),
                                "TachographDrivingInterruptionVehicleChangeIntervalInputEvent"
                        );
                    }
                }
        );
        return result;
    }

    public List<TachographEsperDrivingInterruptionIntervalEvent> buildEsperDailyWeeklyRestCandidateIntervalEvents(
            List<TachographEsperDrivingInterruptionIntervalEvent> drivingInterruptionIntervals,
            int minimumRestPeriodMinutes
    ) {
        if (drivingInterruptionIntervals == null || drivingInterruptionIntervals.isEmpty()) {
            return List.of();
        }
        List<TachographEsperDrivingInterruptionIntervalEvent> result = new ArrayList<>();
        executeWithRuntime(
                configuration -> configuration.getCommon().addEventType(
                        "TachographDrivingInterruptionIntervalInputEvent",
                        drivingInterruptionIntervalInputDefinition()
                ),
                renderDailyWeeklyRestCandidateIntervalEventsEpl(minimumRestPeriodMinutes),
                "dailyWeeklyRestCandidateIntervals",
                newData -> collectDrivingInterruptionIntervalEventsFromTimestamps(newData, result),
                runtime -> {
                    for (TachographEsperDrivingInterruptionIntervalEvent interval : drivingInterruptionIntervals) {
                        runtime.getEventService().sendEventMap(
                                toDrivingInterruptionIntervalInputMap(interval),
                                "TachographDrivingInterruptionIntervalInputEvent"
                        );
                    }
                }
        );
        return result;
    }

    public List<TachographEsperVuCardAbsentIntervalEvent> buildEsperVuCardAbsentIntervalEvents(
            TachographFileSession session,
            DriverExtractionSession driverSession
    ) {
        String sourceKind = session.metadata().driverCardFile() ? "DRIVER_CARD" : "VEHICLE_UNIT";
        List<ResolvedVehicleUsageInterval> vehicleUsageIntervals = mergeVehicleUsageIntervals(
                session.sessionId(),
                driverSession.driverKey(),
                driverSession.cardVehicleUsageIntervals(),
                sourceKind
        );
        return buildEsperVuCardAbsentIntervalEvents(vehicleUsageIntervals);
    }

    public List<TachographEsperVuCardAbsentIntervalEvent> buildEsperVuCardAbsentIntervalEvents(
            ResolvedDriverTimeline timeline
    ) {
        return timeline == null ? List.of() : buildEsperVuCardAbsentIntervalEvents(timeline.vehicleUsageIntervals());
    }

    private List<TachographEsperActivityIntervalEvent> buildEsperActivityIntervalEvents(
            UUID sessionId,
            String driverKey,
            List<ResolvedActivityInterval> activityIntervals
    ) {
        if (activityIntervals == null || activityIntervals.isEmpty()) {
            return List.of();
        }
        List<TachographEsperActivityIntervalEvent> result = new ArrayList<>();
        executeWithRuntime(
                configuration -> configuration.getCommon().addEventType(
                        "TachographActivityIntervalInputEvent",
                        activityIntervalInputDefinition()
                ),
                ACTIVITY_INTERVAL_EVENTS_EPL,
                "activityIntervals",
                newData -> collectActivityIntervalEvents(newData, result),
                runtime -> {
                    for (ResolvedActivityInterval interval : activityIntervals) {
                        runtime.getEventService().sendEventMap(
                                toActivityIntervalInputMap(sessionId, driverKey, interval),
                                "TachographActivityIntervalInputEvent"
                        );
                    }
                }
        );
        return result;
    }

    private List<TachographEsperActivityIntervalEvent> buildEsperDrivingIntervalEvents(
            UUID sessionId,
            String driverKey,
            List<ResolvedActivityInterval> activityIntervals
    ) {
        if (activityIntervals == null || activityIntervals.isEmpty()) {
            return List.of();
        }
        List<TachographEsperActivityIntervalEvent> result = new ArrayList<>();
        executeWithRuntime(
                configuration -> configuration.getCommon().addEventType(
                        "TachographActivityIntervalInputEvent",
                        activityIntervalInputDefinition()
                ),
                DRIVING_INTERVAL_EVENTS_EPL,
                "drivingIntervals",
                newData -> collectActivityIntervalEvents(newData, result),
                runtime -> {
                    for (ResolvedActivityInterval interval : activityIntervals) {
                        runtime.getEventService().sendEventMap(
                                toActivityIntervalInputMap(sessionId, driverKey, interval),
                                "TachographActivityIntervalInputEvent"
                        );
                    }
                }
        );
        return result;
    }

    private List<TachographEsperVehicleUsageIntervalEvent> buildEsperVehicleUsageIntervalEvents(
            List<ResolvedVehicleUsageInterval> vehicleUsageIntervals
    ) {
        if (vehicleUsageIntervals == null || vehicleUsageIntervals.isEmpty()) {
            return List.of();
        }
        List<TachographEsperVehicleUsageIntervalEvent> result = new ArrayList<>();
        executeWithRuntime(
                configuration -> configuration.getCommon().addEventType(
                        "TachographVehicleUsageIntervalInputEvent",
                        vehicleUsageIntervalInputDefinition()
                ),
                VEHICLE_USAGE_INTERVAL_EVENTS_EPL,
                "vehicleUsageIntervals",
                newData -> collectVehicleUsageIntervalEvents(newData, result),
                runtime -> {
                    for (ResolvedVehicleUsageInterval interval : vehicleUsageIntervals) {
                        runtime.getEventService().sendEventMap(
                                toVehicleUsageIntervalInputMap(interval),
                                "TachographVehicleUsageIntervalInputEvent"
                        );
                    }
                }
        );
        return result;
    }

    private List<TachographEsperDrivingInterruptionIntervalEvent> buildEsperDrivingInterruptionIntervalEvents(
            UUID sessionId,
            String driverKey,
            List<ResolvedActivityInterval> activityIntervals,
            int significantDrivingMinutes
    ) {
        if (activityIntervals == null || activityIntervals.size() < 2) {
            return List.of();
        }
        List<TachographEsperDrivingInterruptionIntervalEvent> result = new ArrayList<>();
        executeWithRuntime(
                configuration -> configuration.getCommon().addEventType(
                        "TachographActivityIntervalInputEvent",
                        activityIntervalInputDefinition()
                ),
                renderDrivingInterruptionIntervalEventsEpl(significantDrivingMinutes),
                "drivingInterruptionIntervals",
                newData -> collectDrivingInterruptionIntervalEvents(newData, result),
                runtime -> {
                    for (ResolvedActivityInterval interval : activityIntervals) {
                        runtime.getEventService().sendEventMap(
                                toActivityIntervalInputMap(sessionId, driverKey, interval),
                                "TachographActivityIntervalInputEvent"
                        );
                    }
                }
        );
        return result;
    }

    private List<TachographEsperVuCardAbsentIntervalEvent> buildEsperVuCardAbsentIntervalEvents(
            List<ResolvedVehicleUsageInterval> vehicleUsageIntervals
    ) {
        if (vehicleUsageIntervals == null || vehicleUsageIntervals.size() < 2) {
            return List.of();
        }
        List<TachographEsperVuCardAbsentIntervalEvent> result = new ArrayList<>();
        executeWithRuntime(
                configuration -> configuration.getCommon().addEventType(
                        "TachographVehicleUsageIntervalInputEvent",
                        vehicleUsageIntervalInputDefinition()
                ),
                VU_CARD_ABSENT_INTERVAL_EVENTS_EPL,
                "vuCardAbsentIntervals",
                newData -> collectVuCardAbsentIntervalEvents(newData, result),
                runtime -> {
                    for (ResolvedVehicleUsageInterval interval : vehicleUsageIntervals) {
                        runtime.getEventService().sendEventMap(
                                toVehicleUsageIntervalInputMap(interval),
                                "TachographVehicleUsageIntervalInputEvent"
                        );
                    }
                }
        );
        return result;
    }

    private List<ResolvedVehicleUsageInterval> mergeVehicleUsageIntervals(
            UUID sessionId,
            String driverKey,
            List<ExtractedCardVehicleUsageInterval> rawIntervals,
            String sourceKind
    ) {
        if (rawIntervals == null || rawIntervals.isEmpty()) {
            return List.of();
        }
        List<ResolvedVehicleUsageInterval> sorted = rawIntervals.stream()
                .filter(interval -> interval.from() != null && (interval.to() == null || interval.to().isAfter(interval.from())))
                .map(interval -> ResolvedVehicleUsageInterval.resolved(
                        sessionId,
                        driverKey,
                        interval.intervalId(),
                        interval.from(),
                        interval.to(),
                        interval.odometerBeginKm(),
                        interval.odometerEndKm(),
                        interval.registrationKey(),
                        interval.vehicleKey(),
                        sourceKind,
                        List.of(interval.intervalId())
                ))
                .sorted(Comparator.comparing(ResolvedVehicleUsageInterval::from)
                        .thenComparing(ResolvedVehicleUsageInterval::to, Comparator.nullsLast(Comparator.naturalOrder())))
                .toList();
        if (sorted.isEmpty()) {
            return List.of();
        }

        List<ResolvedVehicleUsageInterval> result = new ArrayList<>();
        ResolvedVehicleUsageInterval current = sorted.get(0);
        List<String> currentSources = new ArrayList<>(current.sourceIntervalIds());
        for (int i = 1; i < sorted.size(); i++) {
            ResolvedVehicleUsageInterval next = sorted.get(i);
            if (canMerge(current, next)) {
                currentSources.addAll(next.sourceIntervalIds());
                current = ResolvedVehicleUsageInterval.resolved(
                        current.sessionId(),
                        current.driverKey(),
                        current.intervalId() + "+" + next.intervalId(),
                        current.from(),
                        mergedTo(current.to(), next.to()),
                        current.odometerBeginKm(),
                        next.odometerEndKm() != null ? next.odometerEndKm() : current.odometerEndKm(),
                        current.registrationKey(),
                        current.vehicleKey(),
                        sourceKind,
                        currentSources
                );
            } else {
                result.add(current);
                current = next;
                currentSources = new ArrayList<>(current.sourceIntervalIds());
            }
        }
        result.add(current);
        return result;
    }

    private boolean canMerge(ResolvedVehicleUsageInterval left, ResolvedVehicleUsageInterval right) {
        return Objects.equals(left.registrationKey(), right.registrationKey())
                && Objects.equals(left.vehicleKey(), right.vehicleKey())
                && !right.from().isAfter(mergeBoundary(left.to()));
    }

    private List<ResolvedActivityInterval> resolveActivities(
            List<ExtractedCardActivityInterval> rawIntervals,
            String sourceKind
    ) {
        if (rawIntervals == null || rawIntervals.isEmpty()) {
            return List.of();
        }
        return rawIntervals.stream()
                .filter(interval -> interval.from() != null && interval.to() != null && interval.to().isAfter(interval.from()))
                .map(interval -> ResolvedActivityInterval.raw(
                        interval.intervalId(),
                        interval.from(),
                        interval.to(),
                        normalizeActivity(interval.activityType()),
                        interval.slot(),
                        interval.cardStatus(),
                        interval.drivingStatus(),
                        interval.registrationKey(),
                        interval.vehicleKey(),
                        sourceKind,
                        List.of(interval.intervalId())
                ))
                .sorted(Comparator.comparing(ResolvedActivityInterval::from)
                        .thenComparing(ResolvedActivityInterval::to)
                        .thenComparing(ResolvedActivityInterval::activityType, Comparator.nullsLast(String::compareTo)))
                .toList();
    }

    private String normalizeActivity(String activityType) {
        if (activityType == null || activityType.isBlank()) {
            return "UNKNOWN";
        }
        if ("UNKNOWN_ACTIVITY".equals(activityType)) {
            return "UNKNOWN";
        }
        return activityType;
    }

    private OffsetDateTime minTimestamp(
            List<ResolvedVehicleUsageInterval> vehicleUsageIntervals,
            List<ResolvedActivityInterval> activityIntervals,
            List<ExtractedSupportEvent> supportEvents
    ) {
        OffsetDateTime min = null;
        for (ResolvedVehicleUsageInterval interval : vehicleUsageIntervals) {
            min = min(min, interval.from());
        }
        for (ResolvedActivityInterval interval : activityIntervals) {
            min = min(min, interval.from());
        }
        for (ExtractedSupportEvent supportEvent : supportEvents) {
            min = min(min, supportEvent.occurredAt());
        }
        return min;
    }

    private OffsetDateTime maxTimestamp(
            List<ResolvedVehicleUsageInterval> vehicleUsageIntervals,
            List<ResolvedActivityInterval> activityIntervals,
            List<ExtractedSupportEvent> supportEvents
    ) {
        OffsetDateTime max = null;
        for (ResolvedVehicleUsageInterval interval : vehicleUsageIntervals) {
            max = max(max, interval.to());
        }
        for (ResolvedActivityInterval interval : activityIntervals) {
            max = max(max, interval.to());
        }
        for (ExtractedSupportEvent supportEvent : supportEvents) {
            max = max(max, supportEvent.occurredAt());
        }
        return max;
    }

    private List<ExtractionWarning> mergeWarnings(List<ExtractionWarning> sessionWarnings, List<ExtractionWarning> driverWarnings) {
        LinkedHashSet<ExtractionWarning> merged = new LinkedHashSet<>();
        if (sessionWarnings != null) {
            merged.addAll(sessionWarnings);
        }
        if (driverWarnings != null) {
            merged.addAll(driverWarnings);
        }
        return List.copyOf(merged);
    }

    private OffsetDateTime min(OffsetDateTime left, OffsetDateTime right) {
        if (left == null) {
            return right;
        }
        if (right == null) {
            return left;
        }
        return left.isBefore(right) ? left : right;
    }

    private OffsetDateTime max(OffsetDateTime left, OffsetDateTime right) {
        if (left == null) {
            return right;
        }
        if (right == null) {
            return left;
        }
        return left.isAfter(right) ? left : right;
    }

    private OffsetDateTime mergedTo(OffsetDateTime left, OffsetDateTime right) {
        if (left == null || right == null) {
            return null;
        }
        return max(left, right);
    }

    private OffsetDateTime mergeBoundary(OffsetDateTime endInclusive) {
        return endInclusive == null ? OffsetDateTime.MAX : endInclusive.plusSeconds(1);
    }

    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-tachograph-projection-" + 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 tachograph projection EPL for statement '" + statementName + "'",
                    e
            );
        } finally {
            if (runtime != null) {
                runtime.destroy();
            }
        }
    }

    private Map<String, Object> activityIntervalInputDefinition() {
        Map<String, Object> definition = new LinkedHashMap<>();
        definition.put("sessionId", UUID.class);
        definition.put("driverKey", String.class);
        definition.put("intervalId", String.class);
        definition.put("activityType", String.class);
        definition.put("cardSlot", String.class);
        definition.put("cardStatus", String.class);
        definition.put("drivingStatus", String.class);
        definition.put("registrationKey", String.class);
        definition.put("vehicleKey", String.class);
        definition.put("sourceKind", String.class);
        definition.put("firstSourceIntervalId", String.class);
        definition.put("lastSourceIntervalId", String.class);
        definition.put("startedAt", OffsetDateTime.class);
        definition.put("endedAt", OffsetDateTime.class);
        definition.put("startedAtEpochSecond", long.class);
        definition.put("endedAtEpochSecond", long.class);
        definition.put("durationSeconds", long.class);
        definition.put("sourceIntervalIds", java.util.List.class);
        definition.put("synthetic", boolean.class);
        definition.put("clippedToRequestedPeriod", boolean.class);
        definition.put("level", String.class);
        return definition;
    }

    private Map<String, Object> vehicleUsageIntervalInputDefinition() {
        Map<String, Object> definition = new LinkedHashMap<>();
        definition.put("sessionId", UUID.class);
        definition.put("driverKey", String.class);
        definition.put("intervalId", String.class);
        definition.put("firstSourceIntervalId", String.class);
        definition.put("lastSourceIntervalId", String.class);
        definition.put("startedAt", OffsetDateTime.class);
        definition.put("endedAt", OffsetDateTime.class);
        definition.put("startedAtEpochSecond", long.class);
        definition.put("endedAtEpochSecond", Long.class);
        definition.put("durationSeconds", long.class);
        definition.put("odometerBeginKm", Long.class);
        definition.put("odometerEndKm", Long.class);
        definition.put("registrationKey", String.class);
        definition.put("vehicleKey", String.class);
        definition.put("sourceKind", String.class);
        definition.put("sourceIntervalIds", java.util.List.class);
        return definition;
    }

    private Map<String, Object> drivingInterruptionIntervalInputDefinition() {
        Map<String, Object> definition = new LinkedHashMap<>();
        definition.put("sessionId", UUID.class);
        definition.put("driverKey", String.class);
        definition.put("startedAt", OffsetDateTime.class);
        definition.put("endedAt", OffsetDateTime.class);
        definition.put("startedAtEpochSecond", long.class);
        definition.put("endedAtEpochSecond", long.class);
        definition.put("durationSeconds", long.class);
        definition.put("previousDrivingSourceIntervalId", String.class);
        definition.put("nextDrivingSourceIntervalId", String.class);
        definition.put("previousRegistrationKey", String.class);
        definition.put("nextRegistrationKey", String.class);
        definition.put("previousVehicleKey", String.class);
        definition.put("nextVehicleKey", String.class);
        return definition;
    }

    private Map<String, Object> vuCardAbsentIntervalInputDefinition() {
        Map<String, Object> definition = new LinkedHashMap<>();
        definition.put("sessionId", UUID.class);
        definition.put("driverKey", String.class);
        definition.put("startedAt", OffsetDateTime.class);
        definition.put("endedAt", OffsetDateTime.class);
        definition.put("startedAtEpochSecond", long.class);
        definition.put("endedAtEpochSecond", long.class);
        definition.put("durationSeconds", long.class);
        definition.put("previousUsageIntervalId", String.class);
        definition.put("nextUsageIntervalId", String.class);
        definition.put("previousRegistrationKey", String.class);
        definition.put("nextRegistrationKey", String.class);
        definition.put("previousVehicleKey", String.class);
        definition.put("nextVehicleKey", String.class);
        return definition;
    }

    private Map<String, Object> toActivityIntervalInputMap(
            UUID sessionId,
            String driverKey,
            ResolvedActivityInterval interval
    ) {
        Map<String, Object> event = new LinkedHashMap<>();
        event.put("sessionId", sessionId);
        event.put("driverKey", driverKey);
        event.put("intervalId", interval.intervalId());
        event.put("activityType", interval.activityType());
        event.put("cardSlot", interval.slot());
        event.put("cardStatus", interval.cardStatus());
        event.put("drivingStatus", interval.drivingStatus());
        event.put("registrationKey", interval.registrationKey());
        event.put("vehicleKey", interval.vehicleKey());
        event.put("sourceKind", interval.sourceKind());
        event.put("firstSourceIntervalId", firstSourceIntervalId(interval));
        event.put("lastSourceIntervalId", lastSourceIntervalId(interval));
        event.put("startedAt", interval.from());
        event.put("endedAt", interval.to());
        event.put("startedAtEpochSecond", interval.from().toEpochSecond());
        event.put("endedAtEpochSecond", interval.to().toEpochSecond());
        event.put("durationSeconds", interval.durationSeconds());
        event.put("sourceIntervalIds", interval.sourceIntervalIds());
        event.put("synthetic", interval.synthetic());
        event.put("clippedToRequestedPeriod", interval.clippedToRequestedPeriod());
        event.put("level", interval.level());
        return event;
    }

    private String firstSourceIntervalId(ResolvedActivityInterval interval) {
        return interval.sourceIntervalIds().isEmpty() ? interval.intervalId() : interval.sourceIntervalIds().get(0);
    }

    private String lastSourceIntervalId(ResolvedActivityInterval interval) {
        return interval.sourceIntervalIds().isEmpty()
                ? interval.intervalId()
                : interval.sourceIntervalIds().get(interval.sourceIntervalIds().size() - 1);
    }

    private Map<String, Object> toVehicleUsageIntervalInputMap(ResolvedVehicleUsageInterval interval) {
        Map<String, Object> event = new LinkedHashMap<>();
        event.put("sessionId", interval.sessionId());
        event.put("driverKey", interval.driverKey());
        event.put("intervalId", interval.intervalId());
        event.put("firstSourceIntervalId", firstSourceIntervalId(interval));
        event.put("lastSourceIntervalId", lastSourceIntervalId(interval));
        event.put("startedAt", interval.from());
        event.put("endedAt", interval.to());
        event.put("startedAtEpochSecond", interval.from().toEpochSecond());
        event.put("endedAtEpochSecond", interval.to() == null ? null : interval.to().toEpochSecond());
        event.put("durationSeconds", interval.durationSeconds());
        event.put("odometerBeginKm", interval.odometerBeginKm());
        event.put("odometerEndKm", interval.odometerEndKm());
        event.put("registrationKey", interval.registrationKey());
        event.put("vehicleKey", interval.vehicleKey());
        event.put("sourceKind", interval.sourceKind());
        event.put("sourceIntervalIds", interval.sourceIntervalIds());
        return event;
    }

    private Map<String, Object> toDrivingInterruptionIntervalInputMap(
            TachographEsperDrivingInterruptionIntervalEvent interval
    ) {
        Map<String, Object> event = new LinkedHashMap<>();
        event.put("sessionId", interval.sessionId());
        event.put("driverKey", interval.driverKey());
        event.put("startedAt", interval.startedAt());
        event.put("endedAt", interval.endedAt());
        event.put("startedAtEpochSecond", interval.startedAt().toEpochSecond());
        event.put("endedAtEpochSecond", interval.endedAt().toEpochSecond());
        event.put("durationSeconds", interval.durationSeconds());
        event.put("previousDrivingSourceIntervalId", interval.previousDrivingSourceIntervalId());
        event.put("nextDrivingSourceIntervalId", interval.nextDrivingSourceIntervalId());
        event.put("previousRegistrationKey", interval.previousRegistrationKey());
        event.put("nextRegistrationKey", interval.nextRegistrationKey());
        event.put("previousVehicleKey", interval.previousVehicleKey());
        event.put("nextVehicleKey", interval.nextVehicleKey());
        return event;
    }

    private Map<String, Object> toVuCardAbsentIntervalInputMap(TachographEsperVuCardAbsentIntervalEvent interval) {
        Map<String, Object> event = new LinkedHashMap<>();
        event.put("sessionId", interval.sessionId());
        event.put("driverKey", interval.driverKey());
        event.put("startedAt", interval.startedAt());
        event.put("endedAt", interval.endedAt());
        event.put("startedAtEpochSecond", interval.startedAt().toEpochSecond());
        event.put("endedAtEpochSecond", interval.endedAt().toEpochSecond());
        event.put("durationSeconds", interval.durationSeconds());
        event.put("previousUsageIntervalId", interval.previousUsageIntervalId());
        event.put("nextUsageIntervalId", interval.nextUsageIntervalId());
        event.put("previousRegistrationKey", interval.previousRegistrationKey());
        event.put("nextRegistrationKey", interval.nextRegistrationKey());
        event.put("previousVehicleKey", interval.previousVehicleKey());
        event.put("nextVehicleKey", interval.nextVehicleKey());
        return event;
    }

    private String firstSourceIntervalId(ResolvedVehicleUsageInterval interval) {
        return interval.sourceIntervalIds().isEmpty() ? interval.intervalId() : interval.sourceIntervalIds().get(0);
    }

    private String lastSourceIntervalId(ResolvedVehicleUsageInterval interval) {
        return interval.sourceIntervalIds().isEmpty()
                ? interval.intervalId()
                : interval.sourceIntervalIds().get(interval.sourceIntervalIds().size() - 1);
    }

    private void collectActivityIntervalEvents(
            EventBean[] newData,
            List<TachographEsperActivityIntervalEvent> target
    ) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            target.add(new TachographEsperActivityIntervalEvent(
                    (UUID) event.get("sessionId"),
                    (String) event.get("driverKey"),
                    (String) event.get("intervalId"),
                    (String) event.get("activityType"),
                    (String) event.get("cardSlot"),
                    (String) event.get("cardStatus"),
                    (String) event.get("drivingStatus"),
                    (String) event.get("registrationKey"),
                    (String) event.get("vehicleKey"),
                    (String) event.get("sourceKind"),
                    (OffsetDateTime) event.get("startedAt"),
                    (OffsetDateTime) event.get("endedAt"),
                    (Long) event.get("durationSeconds"),
                    castSourceIntervalIds(event.get("sourceIntervalIds")),
                    (Boolean) event.get("synthetic"),
                    (Boolean) event.get("clippedToRequestedPeriod"),
                    (String) event.get("level")
            ));
        }
    }

    private void collectVehicleUsageIntervalEvents(
            EventBean[] newData,
            List<TachographEsperVehicleUsageIntervalEvent> target
    ) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            target.add(new TachographEsperVehicleUsageIntervalEvent(
                    (UUID) event.get("sessionId"),
                    (String) event.get("driverKey"),
                    (String) event.get("intervalId"),
                    (OffsetDateTime) event.get("startedAt"),
                    (OffsetDateTime) event.get("endedAt"),
                    (Long) event.get("durationSeconds"),
                    (Long) event.get("odometerBeginKm"),
                    (Long) event.get("odometerEndKm"),
                    (String) event.get("registrationKey"),
                    (String) event.get("vehicleKey"),
                    (String) event.get("sourceKind"),
                    castSourceIntervalIds(event.get("sourceIntervalIds"))
            ));
        }
    }

    private void collectDrivingInterruptionIntervalEvents(
            EventBean[] newData,
            List<TachographEsperDrivingInterruptionIntervalEvent> target
    ) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            long startedAtEpochSecond = (Long) event.get("startedAtEpochSecond");
            long endedAtEpochSecond = (Long) event.get("endedAtEpochSecond");
            target.add(new TachographEsperDrivingInterruptionIntervalEvent(
                    (UUID) event.get("sessionId"),
                    (String) event.get("driverKey"),
                    OffsetDateTime.ofInstant(Instant.ofEpochSecond(startedAtEpochSecond), ZoneOffset.UTC),
                    OffsetDateTime.ofInstant(Instant.ofEpochSecond(endedAtEpochSecond), ZoneOffset.UTC),
                    (Long) event.get("durationSeconds"),
                    (String) event.get("previousDrivingSourceIntervalId"),
                    (String) event.get("nextDrivingSourceIntervalId"),
                    (String) event.get("previousRegistrationKey"),
                    (String) event.get("nextRegistrationKey"),
                    (String) event.get("previousVehicleKey"),
                    (String) event.get("nextVehicleKey")
            ));
        }
    }

    private void collectDrivingInterruptionIntervalEventsFromTimestamps(
            EventBean[] newData,
            List<TachographEsperDrivingInterruptionIntervalEvent> target
    ) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            target.add(new TachographEsperDrivingInterruptionIntervalEvent(
                    (UUID) event.get("sessionId"),
                    (String) event.get("driverKey"),
                    (OffsetDateTime) event.get("startedAt"),
                    (OffsetDateTime) event.get("endedAt"),
                    (Long) event.get("durationSeconds"),
                    (String) event.get("previousDrivingSourceIntervalId"),
                    (String) event.get("nextDrivingSourceIntervalId"),
                    (String) event.get("previousRegistrationKey"),
                    (String) event.get("nextRegistrationKey"),
                    (String) event.get("previousVehicleKey"),
                    (String) event.get("nextVehicleKey")
            ));
        }
    }

    private void collectVuCardAbsentIntervalEvents(
            EventBean[] newData,
            List<TachographEsperVuCardAbsentIntervalEvent> target
    ) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            long startedAtEpochSecond = (Long) event.get("startedAtEpochSecond");
            long endedAtEpochSecond = (Long) event.get("endedAtEpochSecond");
            target.add(new TachographEsperVuCardAbsentIntervalEvent(
                    (UUID) event.get("sessionId"),
                    (String) event.get("driverKey"),
                    OffsetDateTime.ofInstant(Instant.ofEpochSecond(startedAtEpochSecond), ZoneOffset.UTC),
                    OffsetDateTime.ofInstant(Instant.ofEpochSecond(endedAtEpochSecond), ZoneOffset.UTC),
                    (Long) event.get("durationSeconds"),
                    (String) event.get("previousUsageIntervalId"),
                    (String) event.get("nextUsageIntervalId"),
                    (String) event.get("previousRegistrationKey"),
                    (String) event.get("nextRegistrationKey"),
                    (String) event.get("previousVehicleKey"),
                    (String) event.get("nextVehicleKey")
            ));
        }
    }

    private void collectPotentialHomeOvernightStayIntervalEvents(
            EventBean[] newData,
            List<TachographEsperPotentialHomeOvernightStayIntervalEvent> target
    ) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            target.add(new TachographEsperPotentialHomeOvernightStayIntervalEvent(
                    (UUID) event.get("sessionId"),
                    (String) event.get("driverKey"),
                    (OffsetDateTime) event.get("startedAt"),
                    (OffsetDateTime) event.get("endedAt"),
                    (Long) event.get("durationSeconds"),
                    (Long) event.get("cardAbsentDurationSeconds"),
                    (Double) event.get("cardAbsentCoveragePercent"),
                    (String) event.get("previousDrivingSourceIntervalId"),
                    (String) event.get("nextDrivingSourceIntervalId"),
                    (String) event.get("previousRegistrationKey"),
                    (String) event.get("nextRegistrationKey"),
                    (String) event.get("previousVehicleKey"),
                    (String) event.get("nextVehicleKey"),
                    null,
                    null,
                    null,
                    null,
                    null,
                    null,
                    null,
                    null,
                    null,
                    null,
                    null,
                    null,
                    null,
                    null,
                    null,
                    null,
                    null,
                    "UNKNOWN",
                    null,
                    null
            ));
        }
    }

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

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

    private String renderDrivingInterruptionIntervalEventsEpl(int significantDrivingMinutes) {
        long thresholdSeconds = Math.max(1, significantDrivingMinutes) * 60L;
        return DRIVING_INTERRUPTION_INTERVAL_EVENTS_EPL_TEMPLATE.replace(
                "${SIGNIFICANT_DRIVING_THRESHOLD_SECONDS}",
                Long.toString(thresholdSeconds)
        );
    }

    private String renderDailyWeeklyRestCandidateIntervalEventsEpl(int minimumRestPeriodMinutes) {
        long thresholdSeconds = Math.max(1, minimumRestPeriodMinutes) * 60L;
        return DAILY_WEEKLY_REST_CANDIDATE_INTERVAL_EVENTS_EPL_TEMPLATE.replace(
                "${MINIMUM_REST_PERIOD_THRESHOLD_SECONDS}",
                Long.toString(thresholdSeconds)
        );
    }
}
