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.config.EventHubProperties;
import at.procon.eventhub.tachographfilesession.model.DriverExtractionSession;
import at.procon.eventhub.tachographfilesession.model.ExtractedSupportEvent;
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.TachographEsperDailyWeeklyRestCandidateCoverageIntervalEvent;
import at.procon.eventhub.tachographfilesession.model.TachographEsperDrivingDerivedProjectionBundle;
import at.procon.eventhub.tachographfilesession.model.TachographEsperDrivingInterruptionIntervalEvent;
import at.procon.eventhub.tachographfilesession.model.TachographEsperGeoEvidenceEvent;
import at.procon.eventhub.tachographfilesession.model.TachographEsperPotentialHomeOvernightStayIntervalEvent;
import at.procon.eventhub.tachographfilesession.model.TachographEsperPotentialInVehicleOvernightStayIntervalEvent;
import at.procon.eventhub.tachographfilesession.model.TachographEsperPotentialInVehicleTripIntervalEvent;
import at.procon.eventhub.tachographfilesession.model.TachographEsperVuCardAbsentIntervalEvent;
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.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 DriverTimelineReusableProjectionBuilder {

    private static final AtomicLong RUNTIME_COUNTER = new AtomicLong();
    private static final String DRIVING_DERIVED_PROJECTION_BUNDLE_EPL_TEMPLATE =
            loadResource("esper/tachograph-driving-derived-projection-bundle.epl");

    private final DriverTimelineBuilder driverTimelineBuilder;
    private final EventHubProperties properties;

    public DriverTimelineReusableProjectionBuilder(
            DriverTimelineBuilder driverTimelineBuilder,
            EventHubProperties properties
    ) {
        this.driverTimelineBuilder = driverTimelineBuilder;
        this.properties = properties;
    }

    public TachographEsperDrivingDerivedProjectionBundle buildEsperDrivingDerivedProjectionBundle(
            TachographFileSession session,
            DriverExtractionSession driverSession,
            int significantDrivingMinutes,
            int minimumRestPeriodMinutes
    ) {
        ResolvedDriverTimeline timeline = driverTimelineBuilder.build(session, driverSession);
        return buildEsperDrivingDerivedProjectionBundle(
                session.sessionId(),
                driverSession.driverKey(),
                timeline,
                significantDrivingMinutes,
                minimumRestPeriodMinutes
        );
    }

    public TachographEsperDrivingDerivedProjectionBundle buildEsperDrivingDerivedProjectionBundle(
            TachographFileSession session,
            int significantDrivingMinutes,
            int minimumRestPeriodMinutes
    ) {
        if (session == null || session.driversByKey() == null || session.driversByKey().isEmpty()) {
            return emptyBundle();
        }

        List<Map<String, Object>> activityInputEvents = new ArrayList<>();
        List<Map<String, Object>> vehicleUsageInputEvents = new ArrayList<>();
        List<Map<String, Object>> supportGeoInputEvents = new ArrayList<>();

        for (DriverExtractionSession driverSession : session.driversByKey().values()) {
            if (driverSession == null || driverSession.driverKey() == null) {
                continue;
            }
            ResolvedDriverTimeline timeline = driverTimelineBuilder.build(session, driverSession);
            if (timeline == null) {
                continue;
            }
            for (ResolvedActivityInterval interval : safeList(timeline.activityIntervals())) {
                activityInputEvents.add(toActivityIntervalInputMap(
                        session.sessionId(),
                        driverSession.driverKey(),
                        interval
                ));
            }
            for (ResolvedVehicleUsageInterval interval : safeList(timeline.vehicleUsageIntervals())) {
                vehicleUsageInputEvents.add(toVehicleUsageIntervalInputMap(interval));
            }
            for (ExtractedSupportEvent supportEvent : safeList(timeline.supportEvents())) {
                Map<String, Object> supportGeoEvidence = toSupportGeoEvidenceInputMap(session.sessionId(), supportEvent);
                if (supportGeoEvidence != null) {
                    supportGeoInputEvents.add(supportGeoEvidence);
                }
            }
        }

        return buildEsperDrivingDerivedProjectionBundle(
                activityInputEvents,
                vehicleUsageInputEvents,
                supportGeoInputEvents,
                significantDrivingMinutes,
                minimumRestPeriodMinutes
        );
    }

    public TachographEsperDrivingDerivedProjectionBundle buildEsperDrivingDerivedProjectionBundle(
            UUID sessionId,
            String driverKey,
            ResolvedDriverTimeline timeline,
            int significantDrivingMinutes,
            int minimumRestPeriodMinutes
    ) {
        if (timeline == null) {
            return emptyBundle();
        }
        return buildEsperDrivingDerivedProjectionBundle(
                buildActivityIntervalInputEvents(sessionId, driverKey, timeline.activityIntervals()),
                buildVehicleUsageIntervalInputEvents(timeline.vehicleUsageIntervals()),
                buildSupportGeoInputEvents(sessionId, timeline.supportEvents()),
                significantDrivingMinutes,
                minimumRestPeriodMinutes
        );
    }

    private TachographEsperDrivingDerivedProjectionBundle buildEsperDrivingDerivedProjectionBundle(
            List<Map<String, Object>> activityInputEvents,
            List<Map<String, Object>> vehicleUsageInputEvents,
            List<Map<String, Object>> supportGeoInputEvents,
            int significantDrivingMinutes,
            int minimumRestPeriodMinutes
    ) {
        if ((activityInputEvents == null || activityInputEvents.isEmpty())
                && (vehicleUsageInputEvents == null || vehicleUsageInputEvents.isEmpty())) {
            return emptyBundle();
        }

        List<TachographEsperDrivingInterruptionIntervalEvent> drivingInterruptionIntervals = new ArrayList<>();
        List<TachographEsperDrivingInterruptionIntervalEvent> dailyWeeklyRestCandidateIntervals = new ArrayList<>();
        List<TachographEsperDailyWeeklyRestCandidateCoverageIntervalEvent> dailyWeeklyRestCandidateCoverageIntervals = new ArrayList<>();
        List<TachographEsperDailyWeeklyRestCandidateCoverageIntervalEvent> unclassifiedDailyWeeklyRestCandidateCoverageIntervals = new ArrayList<>();
        List<TachographEsperDrivingInterruptionIntervalEvent> drivingInterruptionVehicleChangeIntervals = new ArrayList<>();
        List<TachographEsperVuCardAbsentIntervalEvent> vuCardAbsentIntervals = new ArrayList<>();
        List<TachographEsperPotentialHomeOvernightStayIntervalEvent> potentialHomeOvernightStayIntervals = new ArrayList<>();
        List<TachographEsperPotentialInVehicleOvernightStayIntervalEvent> potentialInVehicleOvernightStayIntervals = new ArrayList<>();
        List<TachographEsperPotentialInVehicleTripIntervalEvent> potentialInVehicleTripIntervals = new ArrayList<>();

        executeWithRuntime(
                configuration -> {
                    configuration.getCommon().addEventType(
                            "TachographActivityIntervalInputEvent",
                            activityIntervalInputDefinition()
                    );
                    configuration.getCommon().addEventType(
                            "TachographVehicleUsageIntervalInputEvent",
                            vehicleUsageIntervalInputDefinition()
                    );
                    configuration.getCommon().addEventType(
                            "TachographSupportGeoEvidenceInputEvent",
                            supportGeoEvidenceInputDefinition()
                    );
                },
                renderDrivingDerivedProjectionBundleEpl(significantDrivingMinutes, minimumRestPeriodMinutes),
                Map.of(
                        "drivingInterruptionIntervals", newData -> collectDrivingInterruptionIntervalEvents(newData, drivingInterruptionIntervals),
                        "dailyWeeklyRestCandidateIntervals", newData -> collectDrivingInterruptionIntervalEvents(newData, dailyWeeklyRestCandidateIntervals),
                        "dailyWeeklyRestCandidateCoverageIntervals", newData -> collectDailyWeeklyRestCandidateCoverageIntervalEvents(newData, dailyWeeklyRestCandidateCoverageIntervals),
                        "unclassifiedDailyWeeklyRestCandidateCoverageIntervals", newData -> collectDailyWeeklyRestCandidateCoverageIntervalEvents(newData, unclassifiedDailyWeeklyRestCandidateCoverageIntervals),
                        "drivingInterruptionVehicleChangeIntervals", newData -> collectDrivingInterruptionIntervalEvents(newData, drivingInterruptionVehicleChangeIntervals),
                        "vuCardAbsentIntervals", newData -> collectVuCardAbsentIntervalEvents(newData, vuCardAbsentIntervals),
                        "potentialHomeOvernightStayIntervals", newData -> collectPotentialHomeOvernightStayIntervalEvents(newData, potentialHomeOvernightStayIntervals),
                        "potentialInVehicleOvernightStayIntervals", newData -> collectPotentialInVehicleOvernightStayIntervalEvents(newData, potentialInVehicleOvernightStayIntervals),
                        "potentialInVehicleTripIntervals", newData -> collectPotentialInVehicleTripIntervalEvents(newData, potentialInVehicleTripIntervals)
                ),
                runtime -> {
                    if (supportGeoInputEvents != null) {
                        for (Map<String, Object> supportGeoEvidence : supportGeoInputEvents) {
                            runtime.getEventService().sendEventMap(
                                    supportGeoEvidence,
                                    "TachographSupportGeoEvidenceInputEvent"
                            );
                        }
                    }
                    if (vehicleUsageInputEvents != null) {
                        for (Map<String, Object> interval : vehicleUsageInputEvents) {
                            runtime.getEventService().sendEventMap(
                                    interval,
                                    "TachographVehicleUsageIntervalInputEvent"
                            );
                        }
                    }
                    if (activityInputEvents != null) {
                        for (Map<String, Object> interval : activityInputEvents) {
                            runtime.getEventService().sendEventMap(
                                    interval,
                                    "TachographActivityIntervalInputEvent"
                            );
                        }
                    }
                }
        );

        return new TachographEsperDrivingDerivedProjectionBundle(
                sortDrivingInterruptionIntervals(drivingInterruptionIntervals),
                sortDrivingInterruptionIntervals(dailyWeeklyRestCandidateIntervals),
                sortDailyWeeklyRestCandidateCoverageIntervals(dailyWeeklyRestCandidateCoverageIntervals),
                sortDailyWeeklyRestCandidateCoverageIntervals(unclassifiedDailyWeeklyRestCandidateCoverageIntervals),
                sortDrivingInterruptionIntervals(drivingInterruptionVehicleChangeIntervals),
                sortVuCardAbsentIntervals(vuCardAbsentIntervals),
                sortPotentialHomeOvernightStayIntervals(potentialHomeOvernightStayIntervals),
                sortPotentialInVehicleOvernightStayIntervals(potentialInVehicleOvernightStayIntervals),
                sortPotentialInVehicleTripIntervals(potentialInVehicleTripIntervals)
        );
    }

    private List<Map<String, Object>> buildActivityIntervalInputEvents(
            UUID sessionId,
            String driverKey,
            List<ResolvedActivityInterval> activityIntervals
    ) {
        return safeList(activityIntervals).stream()
                .map(interval -> toActivityIntervalInputMap(sessionId, driverKey, interval))
                .sorted(Comparator
                        .comparing((Map<String, Object> event) -> (Long) event.get("startedAtEpochSecond"))
                        .thenComparing(event -> Objects.toString(event.get("driverKey"), ""))
                        .thenComparing(event -> Objects.toString(event.get("intervalId"), "")))
                .toList();
    }

    private List<Map<String, Object>> buildVehicleUsageIntervalInputEvents(
            List<ResolvedVehicleUsageInterval> vehicleUsageIntervals
    ) {
        return safeList(vehicleUsageIntervals).stream()
                .map(this::toVehicleUsageIntervalInputMap)
                .sorted(Comparator
                        .comparing((Map<String, Object> event) -> (Long) event.get("startedAtEpochSecond"))
                        .thenComparing(event -> Objects.toString(event.get("driverKey"), ""))
                        .thenComparing(event -> Objects.toString(event.get("intervalId"), "")))
                .toList();
    }

    private List<Map<String, Object>> buildSupportGeoInputEvents(
            UUID sessionId,
            List<ExtractedSupportEvent> supportEvents
    ) {
        return safeList(supportEvents).stream()
                .map(event -> toSupportGeoEvidenceInputMap(sessionId, event))
                .filter(Objects::nonNull)
                .sorted(Comparator
                        .comparing((Map<String, Object> event) -> (Long) event.get("occurredAtEpochSecond"))
                        .thenComparing(event -> Objects.toString(event.get("driverKey"), ""))
                        .thenComparing(event -> Objects.toString(event.get("eventId"), "")))
                .toList();
    }

    private <T> List<T> safeList(List<T> values) {
        return values == null ? List.of() : values;
    }

    private TachographEsperDrivingDerivedProjectionBundle emptyBundle() {
        return new TachographEsperDrivingDerivedProjectionBundle(
                List.of(),
                List.of(),
                List.of(),
                List.of(),
                List.of(),
                List.of(),
                List.of(),
                List.of(),
                List.of()
        );
    }

    private void executeWithRuntime(
            Consumer<Configuration> configurationSetup,
            String epl,
            Map<String, Consumer<EventBean[]>> listeners,
            Consumer<EPRuntime> sender
    ) {
        EPRuntime runtime = null;
        try {
            Configuration configuration = new Configuration();
            configurationSetup.accept(configuration);
            String runtimeUri = "eventhub-tachograph-reusable-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);
            for (Map.Entry<String, Consumer<EventBean[]>> entry : listeners.entrySet()) {
                runtime.getDeploymentService()
                        .getStatement(deployment.getDeploymentId(), entry.getKey())
                        .addListener((newData, oldData, statement, rt) -> entry.getValue().accept(newData));
            }

            sender.accept(runtime);
        } catch (EPCompileException | EPDeployException e) {
            throw new IllegalStateException("Cannot compile/deploy reusable tachograph projection EPL bundle", 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> supportGeoEvidenceInputDefinition() {
        Map<String, Object> definition = new LinkedHashMap<>();
        definition.put("sessionId", UUID.class);
        definition.put("driverKey", String.class);
        definition.put("eventId", String.class);
        definition.put("eventDomain", String.class);
        definition.put("occurredAt", OffsetDateTime.class);
        definition.put("occurredAtEpochSecond", long.class);
        definition.put("registrationKey", String.class);
        definition.put("vehicleKey", String.class);
        definition.put("latitude", Double.class);
        definition.put("longitude", Double.class);
        definition.put("odometerKm", Long.class);
        definition.put("priority", int.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 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> toSupportGeoEvidenceInputMap(
            UUID sessionId,
            ExtractedSupportEvent supportEvent
    ) {
        if (supportEvent == null
                || supportEvent.driverKey() == null
                || supportEvent.occurredAt() == null
                || supportEvent.latitude() == null
                || supportEvent.longitude() == null) {
            return null;
        }
        int priority = supportGeoPriority(supportEvent.eventDomain());
        if (priority <= 0) {
            return null;
        }
        Map<String, Object> event = new LinkedHashMap<>();
        event.put("sessionId", sessionId);
        event.put("driverKey", supportEvent.driverKey());
        event.put("eventId", supportEvent.eventId());
        event.put("eventDomain", supportEvent.eventDomain());
        event.put("occurredAt", supportEvent.occurredAt());
        event.put("occurredAtEpochSecond", supportEvent.occurredAt().toEpochSecond());
        event.put("registrationKey", supportEvent.registrationKey());
        event.put("vehicleKey", supportEvent.vehicleKey());
        event.put("latitude", supportEvent.latitude().doubleValue());
        event.put("longitude", supportEvent.longitude().doubleValue());
        event.put("odometerKm", supportEvent.odometerKm());
        event.put("priority", priority);
        return event;
    }

    private int supportGeoPriority(String eventDomain) {
        if (eventDomain == null || eventDomain.isBlank()) {
            return 0;
        }
        return switch (eventDomain.trim().toUpperCase()) {
            case "POSITION" -> 500;
            case "PLACE" -> 400;
            case "BORDER_CROSSING" -> 300;
            case "LOAD_UNLOAD" -> 250;
            default -> 0;
        };
    }

    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 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 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 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 collectDailyWeeklyRestCandidateCoverageIntervalEvents(
            EventBean[] newData,
            List<TachographEsperDailyWeeklyRestCandidateCoverageIntervalEvent> target
    ) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            long startedAtEpochSecond = (Long) event.get("startedAtEpochSecond");
            long endedAtEpochSecond = (Long) event.get("endedAtEpochSecond");
            TachographEsperGeoEvidenceEvent beginGeoEvent = geoEvidenceEvent(
                    (String) event.get("beginGeoEventId"),
                    (String) event.get("beginGeoEventDomain"),
                    offsetDateTime(event.get("beginGeoOccurredAtEpochSecond")),
                    (Double) event.get("beginLatitude"),
                    (Double) event.get("beginLongitude"),
                    longOrNull(event.get("beginGeoDistanceSeconds")),
                    longOrNull(event.get("beginGeoOdometerKm"))
            );
            TachographEsperGeoEvidenceEvent endGeoEvent = geoEvidenceEvent(
                    (String) event.get("endGeoEventId"),
                    (String) event.get("endGeoEventDomain"),
                    offsetDateTime(event.get("endGeoOccurredAtEpochSecond")),
                    (Double) event.get("endLatitude"),
                    (Double) event.get("endLongitude"),
                    longOrNull(event.get("endGeoDistanceSeconds")),
                    longOrNull(event.get("endGeoOdometerKm"))
            );
            target.add(new TachographEsperDailyWeeklyRestCandidateCoverageIntervalEvent(
                    (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"),
                    (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"),
                    longOrNull(event.get("beginBoundaryOdometerKm")),
                    longOrNull(event.get("endBoundaryOdometerKm")),
                    (String) event.get("beginGeoEventId"),
                    (String) event.get("beginGeoEventDomain"),
                    offsetDateTime(event.get("beginGeoOccurredAtEpochSecond")),
                    (Double) event.get("beginLatitude"),
                    (Double) event.get("beginLongitude"),
                    longOrNull(event.get("beginGeoDistanceSeconds")),
                    longOrNull(event.get("beginGeoOdometerKm")),
                    (String) event.get("endGeoEventId"),
                    (String) event.get("endGeoEventDomain"),
                    offsetDateTime(event.get("endGeoOccurredAtEpochSecond")),
                    (Double) event.get("endLatitude"),
                    (Double) event.get("endLongitude"),
                    longOrNull(event.get("endGeoDistanceSeconds")),
                    longOrNull(event.get("endGeoOdometerKm")),
                    longOrNull(event.get("geoEvidenceMovementMeters")),
                    (String) event.get("geoEvidenceMovementCategory"),
                    beginGeoEvent,
                    endGeoEvent
            ));
        }
    }

    private void collectPotentialHomeOvernightStayIntervalEvents(
            EventBean[] newData,
            List<TachographEsperPotentialHomeOvernightStayIntervalEvent> target
    ) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            long startedAtEpochSecond = (Long) event.get("startedAtEpochSecond");
            long endedAtEpochSecond = (Long) event.get("endedAtEpochSecond");
            TachographEsperGeoEvidenceEvent beginGeoEvent = geoEvidenceEvent(
                    (String) event.get("beginGeoEventId"),
                    (String) event.get("beginGeoEventDomain"),
                    offsetDateTime(event.get("beginGeoOccurredAtEpochSecond")),
                    (Double) event.get("beginLatitude"),
                    (Double) event.get("beginLongitude"),
                    longOrNull(event.get("beginGeoDistanceSeconds")),
                    longOrNull(event.get("beginGeoOdometerKm"))
            );
            TachographEsperGeoEvidenceEvent endGeoEvent = geoEvidenceEvent(
                    (String) event.get("endGeoEventId"),
                    (String) event.get("endGeoEventDomain"),
                    offsetDateTime(event.get("endGeoOccurredAtEpochSecond")),
                    (Double) event.get("endLatitude"),
                    (Double) event.get("endLongitude"),
                    longOrNull(event.get("endGeoDistanceSeconds")),
                    longOrNull(event.get("endGeoOdometerKm"))
            );
            target.add(new TachographEsperPotentialHomeOvernightStayIntervalEvent(
                    (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"),
                    (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"),
                    longOrNull(event.get("beginBoundaryOdometerKm")),
                    longOrNull(event.get("endBoundaryOdometerKm")),
                    (String) event.get("beginGeoEventId"),
                    (String) event.get("beginGeoEventDomain"),
                    offsetDateTime(event.get("beginGeoOccurredAtEpochSecond")),
                    (Double) event.get("beginLatitude"),
                    (Double) event.get("beginLongitude"),
                    longOrNull(event.get("beginGeoDistanceSeconds")),
                    longOrNull(event.get("beginGeoOdometerKm")),
                    (String) event.get("endGeoEventId"),
                    (String) event.get("endGeoEventDomain"),
                    offsetDateTime(event.get("endGeoOccurredAtEpochSecond")),
                    (Double) event.get("endLatitude"),
                    (Double) event.get("endLongitude"),
                    longOrNull(event.get("endGeoDistanceSeconds")),
                    longOrNull(event.get("endGeoOdometerKm")),
                    longOrNull(event.get("geoEvidenceMovementMeters")),
                    (String) event.get("geoEvidenceMovementCategory"),
                    beginGeoEvent,
                    endGeoEvent
            ));
        }
    }

    private void collectPotentialInVehicleOvernightStayIntervalEvents(
            EventBean[] newData,
            List<TachographEsperPotentialInVehicleOvernightStayIntervalEvent> target
    ) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            long startedAtEpochSecond = (Long) event.get("startedAtEpochSecond");
            long endedAtEpochSecond = (Long) event.get("endedAtEpochSecond");
            TachographEsperGeoEvidenceEvent beginGeoEvent = geoEvidenceEvent(
                    (String) event.get("beginGeoEventId"),
                    (String) event.get("beginGeoEventDomain"),
                    offsetDateTime(event.get("beginGeoOccurredAtEpochSecond")),
                    (Double) event.get("beginLatitude"),
                    (Double) event.get("beginLongitude"),
                    longOrNull(event.get("beginGeoDistanceSeconds")),
                    longOrNull(event.get("beginGeoOdometerKm"))
            );
            TachographEsperGeoEvidenceEvent endGeoEvent = geoEvidenceEvent(
                    (String) event.get("endGeoEventId"),
                    (String) event.get("endGeoEventDomain"),
                    offsetDateTime(event.get("endGeoOccurredAtEpochSecond")),
                    (Double) event.get("endLatitude"),
                    (Double) event.get("endLongitude"),
                    longOrNull(event.get("endGeoDistanceSeconds")),
                    longOrNull(event.get("endGeoOdometerKm"))
            );
            target.add(new TachographEsperPotentialInVehicleOvernightStayIntervalEvent(
                    (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"),
                    (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"),
                    longOrNull(event.get("beginBoundaryOdometerKm")),
                    longOrNull(event.get("endBoundaryOdometerKm")),
                    (String) event.get("beginGeoEventId"),
                    (String) event.get("beginGeoEventDomain"),
                    offsetDateTime(event.get("beginGeoOccurredAtEpochSecond")),
                    (Double) event.get("beginLatitude"),
                    (Double) event.get("beginLongitude"),
                    longOrNull(event.get("beginGeoDistanceSeconds")),
                    longOrNull(event.get("beginGeoOdometerKm")),
                    (String) event.get("endGeoEventId"),
                    (String) event.get("endGeoEventDomain"),
                    offsetDateTime(event.get("endGeoOccurredAtEpochSecond")),
                    (Double) event.get("endLatitude"),
                    (Double) event.get("endLongitude"),
                    longOrNull(event.get("endGeoDistanceSeconds")),
                    longOrNull(event.get("endGeoOdometerKm")),
                    longOrNull(event.get("geoEvidenceMovementMeters")),
                    (String) event.get("geoEvidenceMovementCategory"),
                    beginGeoEvent,
                    endGeoEvent
            ));
        }
    }

    private void collectPotentialInVehicleTripIntervalEvents(
            EventBean[] newData,
            List<TachographEsperPotentialInVehicleTripIntervalEvent> target
    ) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            long startedAtEpochSecond = (Long) event.get("startedAtEpochSecond");
            long endedAtEpochSecond = (Long) event.get("endedAtEpochSecond");
            target.add(new TachographEsperPotentialInVehicleTripIntervalEvent(
                    (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("registrationKey"),
                    (String) event.get("vehicleKey"),
                    (Integer) event.get("containedPotentialInVehicleOvernightStayIntervalCount"),
                    (Long) event.get("containedPotentialInVehicleOvernightStayDurationSeconds"),
                    (Long) event.get("containedCardAbsentDurationSeconds"),
                    OffsetDateTime.ofInstant(Instant.ofEpochSecond((Long) event.get("firstPotentialInVehicleOvernightStayStartedAtEpochSecond")), ZoneOffset.UTC),
                    OffsetDateTime.ofInstant(Instant.ofEpochSecond((Long) event.get("lastPotentialInVehicleOvernightStayEndedAtEpochSecond")), ZoneOffset.UTC),
                    (String) event.get("firstPreviousDrivingSourceIntervalId"),
                    (String) event.get("lastNextDrivingSourceIntervalId"),
                    List.of()
            ));
        }
    }

    private List<TachographEsperDrivingInterruptionIntervalEvent> sortDrivingInterruptionIntervals(
            List<TachographEsperDrivingInterruptionIntervalEvent> intervals
    ) {
        return intervals.stream()
                .sorted(Comparator.comparing(TachographEsperDrivingInterruptionIntervalEvent::startedAt)
                        .thenComparing(TachographEsperDrivingInterruptionIntervalEvent::endedAt))
                .toList();
    }

    private List<TachographEsperVuCardAbsentIntervalEvent> sortVuCardAbsentIntervals(
            List<TachographEsperVuCardAbsentIntervalEvent> intervals
    ) {
        return intervals.stream()
                .sorted(Comparator.comparing(TachographEsperVuCardAbsentIntervalEvent::startedAt)
                        .thenComparing(TachographEsperVuCardAbsentIntervalEvent::endedAt))
                .toList();
    }

    private List<TachographEsperDailyWeeklyRestCandidateCoverageIntervalEvent> sortDailyWeeklyRestCandidateCoverageIntervals(
            List<TachographEsperDailyWeeklyRestCandidateCoverageIntervalEvent> intervals
    ) {
        return deduplicateRestCoverageIntervals(intervals).stream()
                .sorted(Comparator.comparing(TachographEsperDailyWeeklyRestCandidateCoverageIntervalEvent::startedAt)
                        .thenComparing(TachographEsperDailyWeeklyRestCandidateCoverageIntervalEvent::endedAt))
                .toList();
    }

    private List<TachographEsperPotentialHomeOvernightStayIntervalEvent> sortPotentialHomeOvernightStayIntervals(
            List<TachographEsperPotentialHomeOvernightStayIntervalEvent> intervals
    ) {
        return deduplicatePotentialHomeOvernightStayIntervals(intervals).stream()
                .sorted(Comparator.comparing(TachographEsperPotentialHomeOvernightStayIntervalEvent::startedAt)
                        .thenComparing(TachographEsperPotentialHomeOvernightStayIntervalEvent::endedAt))
                .toList();
    }

    private List<TachographEsperPotentialInVehicleOvernightStayIntervalEvent> sortPotentialInVehicleOvernightStayIntervals(
            List<TachographEsperPotentialInVehicleOvernightStayIntervalEvent> intervals
    ) {
        return deduplicatePotentialInVehicleOvernightStayIntervals(intervals).stream()
                .sorted(Comparator.comparing(TachographEsperPotentialInVehicleOvernightStayIntervalEvent::startedAt)
                        .thenComparing(TachographEsperPotentialInVehicleOvernightStayIntervalEvent::endedAt))
                .toList();
    }

    private List<TachographEsperDailyWeeklyRestCandidateCoverageIntervalEvent> deduplicateRestCoverageIntervals(
            List<TachographEsperDailyWeeklyRestCandidateCoverageIntervalEvent> intervals
    ) {
        Map<String, TachographEsperDailyWeeklyRestCandidateCoverageIntervalEvent> deduplicated = new LinkedHashMap<>();
        for (TachographEsperDailyWeeklyRestCandidateCoverageIntervalEvent interval : intervals) {
            deduplicated.put(restCoverageIntervalKey(
                    interval.driverKey(),
                    interval.startedAt(),
                    interval.endedAt(),
                    interval.previousDrivingSourceIntervalId(),
                    interval.nextDrivingSourceIntervalId()
            ), interval);
        }
        return new ArrayList<>(deduplicated.values());
    }

    private List<TachographEsperPotentialHomeOvernightStayIntervalEvent> deduplicatePotentialHomeOvernightStayIntervals(
            List<TachographEsperPotentialHomeOvernightStayIntervalEvent> intervals
    ) {
        Map<String, TachographEsperPotentialHomeOvernightStayIntervalEvent> deduplicated = new LinkedHashMap<>();
        for (TachographEsperPotentialHomeOvernightStayIntervalEvent interval : intervals) {
            deduplicated.put(restCoverageIntervalKey(
                    interval.driverKey(),
                    interval.startedAt(),
                    interval.endedAt(),
                    interval.previousDrivingSourceIntervalId(),
                    interval.nextDrivingSourceIntervalId()
            ), interval);
        }
        return new ArrayList<>(deduplicated.values());
    }

    private List<TachographEsperPotentialInVehicleOvernightStayIntervalEvent> deduplicatePotentialInVehicleOvernightStayIntervals(
            List<TachographEsperPotentialInVehicleOvernightStayIntervalEvent> intervals
    ) {
        Map<String, TachographEsperPotentialInVehicleOvernightStayIntervalEvent> deduplicated = new LinkedHashMap<>();
        for (TachographEsperPotentialInVehicleOvernightStayIntervalEvent interval : intervals) {
            deduplicated.put(restCoverageIntervalKey(
                    interval.driverKey(),
                    interval.startedAt(),
                    interval.endedAt(),
                    interval.previousDrivingSourceIntervalId(),
                    interval.nextDrivingSourceIntervalId()
            ), interval);
        }
        return new ArrayList<>(deduplicated.values());
    }

    private String restCoverageIntervalKey(
            String driverKey,
            OffsetDateTime startedAt,
            OffsetDateTime endedAt,
            String previousDrivingSourceIntervalId,
            String nextDrivingSourceIntervalId
    ) {
        return String.join("|",
                driverKey == null ? "" : driverKey,
                startedAt == null ? "" : startedAt.toString(),
                endedAt == null ? "" : endedAt.toString(),
                previousDrivingSourceIntervalId == null ? "" : previousDrivingSourceIntervalId,
                nextDrivingSourceIntervalId == null ? "" : nextDrivingSourceIntervalId
        );
    }

    private List<TachographEsperPotentialInVehicleTripIntervalEvent> sortPotentialInVehicleTripIntervals(
            List<TachographEsperPotentialInVehicleTripIntervalEvent> intervals
    ) {
        return intervals.stream()
                .sorted(Comparator.comparing(TachographEsperPotentialInVehicleTripIntervalEvent::startedAt)
                        .thenComparing(TachographEsperPotentialInVehicleTripIntervalEvent::endedAt))
                .toList();
    }

    private String renderDrivingDerivedProjectionBundleEpl(int significantDrivingMinutes, int minimumRestPeriodMinutes) {
        return DRIVING_DERIVED_PROJECTION_BUNDLE_EPL_TEMPLATE
                .replace(
                        "${SIGNIFICANT_DRIVING_THRESHOLD_SECONDS}",
                        Long.toString(Math.max(1, significantDrivingMinutes) * 60L)
                )
                .replace(
                        "${MINIMUM_REST_PERIOD_THRESHOLD_SECONDS}",
                        Long.toString(Math.max(1, minimumRestPeriodMinutes) * 60L)
                )
                .replace(
                        "${REST_GEO_LOOKBACK_SECONDS}",
                        Long.toString(Math.max(1, properties.getTachographFileSession().getProcessing().getRestCandidateGeoLookbackMinutes()) * 60L)
                )
                .replace(
                        "${REST_GEO_LOOKAHEAD_SECONDS}",
                        Long.toString(Math.max(1, properties.getTachographFileSession().getProcessing().getRestCandidateGeoLookaheadMinutes()) * 60L)
                )
                .replace(
                        "${REST_GEO_STATIONARY_MAX_METERS}",
                        Integer.toString(Math.max(0, properties.getTachographFileSession().getProcessing().getRestCandidateGeoStationaryMaxMeters()))
                )
                .replace(
                        "${REST_GEO_MINOR_MOVEMENT_MAX_METERS}",
                        Integer.toString(Math.max(
                                properties.getTachographFileSession().getProcessing().getRestCandidateGeoStationaryMaxMeters(),
                                properties.getTachographFileSession().getProcessing().getRestCandidateGeoMinorMovementMaxMeters()
                        ))
                );
    }

    private OffsetDateTime offsetDateTime(Object epochSecond) {
        if (!(epochSecond instanceof Long value)) {
            return null;
        }
        return OffsetDateTime.ofInstant(Instant.ofEpochSecond(value), ZoneOffset.UTC);
    }

    private Long longOrNull(Object value) {
        return value instanceof Long number ? number : null;
    }

    private TachographEsperGeoEvidenceEvent geoEvidenceEvent(
            String eventId,
            String eventDomain,
            OffsetDateTime occurredAt,
            Double latitude,
            Double longitude,
            Long distanceSeconds,
            Long odometerKm
    ) {
        if (eventId == null
                && eventDomain == null
                && occurredAt == null
                && latitude == null
                && longitude == null
                && distanceSeconds == null
                && odometerKm == null) {
            return null;
        }
        return new TachographEsperGeoEvidenceEvent(
                eventId,
                eventDomain,
                occurredAt,
                latitude,
                longitude,
                distanceSeconds,
                odometerKm
        );
    }

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