package at.procon.eventhub.processing.driverworkingtime.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.processing.driverworkingtime.model.DriverWorkingTimeActivityInterval;
import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeDerivedProjectionBundle;
import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeDrivingInterruptionInterval;
import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeGeoEvidenceEvent;
import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimePotentialInVehicleTripInterval;
import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeProcessingInput;
import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeRestCoverageInterval;
import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeVehicleUsageInterval;
import at.procon.eventhub.processing.driverworkingtime.model.DriverWorkingTimeVuCardAbsentInterval;
import at.procon.eventhub.processing.eventprocessing.support.RuntimeSupportEvidenceEvent;
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.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Service;
import org.springframework.util.StreamUtils;

@Service
public class DriverWorkingTimeReusableProjectionBuilder {

    private static final AtomicLong RUNTIME_COUNTER = new AtomicLong();
    private static final String DRIVING_DERIVED_PROJECTION_BUNDLE_EPL_TEMPLATE =
            loadResource("esper/driver-working-time-derived-projections.epl");

    private final EventHubProperties properties;

    @Autowired
    public DriverWorkingTimeReusableProjectionBuilder(
            EventHubProperties properties
    ) {
        this.properties = properties;
    }

    public DriverWorkingTimeDerivedProjectionBundle buildDerivedProjectionBundle(
            DriverWorkingTimeProcessingInput input
    ) {
        if (input == null) {
            return emptyDerivedProjectionBundle();
        }
        return buildDerivedProjectionBundle(
                buildActivityIntervalInputEvents(input.activityIntervals()),
                buildVehicleUsageIntervalInputEventsCommon(input.vehicleUsageIntervals()),
                buildSupportGeoInputEventsCommon(input.sessionId(), input.supportEvidenceEvents()),
                input.significantDrivingMinutes(),
                input.minimumRestPeriodMinutes()
        );
    }

    private DriverWorkingTimeDerivedProjectionBundle buildDerivedProjectionBundle(
            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 emptyDerivedProjectionBundle();
        }

        List<DriverWorkingTimeDrivingInterruptionInterval> drivingInterruptionIntervals = new ArrayList<>();
        List<DriverWorkingTimeDrivingInterruptionInterval> dailyWeeklyRestCandidateIntervals = new ArrayList<>();
        List<DriverWorkingTimeRestCoverageInterval> dailyWeeklyRestCandidateCoverageIntervals = new ArrayList<>();
        List<DriverWorkingTimeRestCoverageInterval> unclassifiedDailyWeeklyRestCandidateCoverageIntervals = new ArrayList<>();
        List<DriverWorkingTimeDrivingInterruptionInterval> drivingInterruptionVehicleChangeIntervals = new ArrayList<>();
        List<DriverWorkingTimeVuCardAbsentInterval> vuCardAbsentIntervals = new ArrayList<>();
        List<DriverWorkingTimeRestCoverageInterval> potentialHomeOvernightStayIntervals = new ArrayList<>();
        List<DriverWorkingTimeRestCoverageInterval> potentialInVehicleOvernightStayIntervals = new ArrayList<>();
        List<DriverWorkingTimePotentialInVehicleTripInterval> 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 -> collectDrivingInterruptionIntervalModels(newData, drivingInterruptionIntervals),
                        "dailyWeeklyRestCandidateIntervals",
                        newData -> collectDrivingInterruptionIntervalModels(newData, dailyWeeklyRestCandidateIntervals),
                        "dailyWeeklyRestCandidateCoverageIntervals",
                        newData -> collectDailyWeeklyRestCandidateCoverageIntervalModels(newData, dailyWeeklyRestCandidateCoverageIntervals),
                        "unclassifiedDailyWeeklyRestCandidateCoverageIntervals",
                        newData -> collectDailyWeeklyRestCandidateCoverageIntervalModels(newData, unclassifiedDailyWeeklyRestCandidateCoverageIntervals),
                        "drivingInterruptionVehicleChangeIntervals",
                        newData -> collectDrivingInterruptionIntervalModels(newData, drivingInterruptionVehicleChangeIntervals),
                        "vuCardAbsentIntervals",
                        newData -> collectVuCardAbsentIntervalModels(newData, vuCardAbsentIntervals),
                        "potentialHomeOvernightStayIntervals",
                        newData -> collectPotentialHomeOvernightStayIntervalModels(newData, potentialHomeOvernightStayIntervals),
                        "potentialInVehicleOvernightStayIntervals",
                        newData -> collectPotentialInVehicleOvernightStayIntervalModels(newData, potentialInVehicleOvernightStayIntervals),
                        "potentialInVehicleTripIntervals",
                        newData -> collectPotentialInVehicleTripIntervalModels(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"
                            );
                        }
                    }
                }
        );

        List<DriverWorkingTimeRestCoverageInterval> sortedDailyWeeklyRestCandidateCoverageIntervals =
                sortDailyWeeklyRestCandidateCoverageIntervalsCommon(dailyWeeklyRestCandidateCoverageIntervals);
        List<DriverWorkingTimeRestCoverageInterval> sortedUnclassifiedDailyWeeklyRestCandidateCoverageIntervals =
                sortDailyWeeklyRestCandidateCoverageIntervalsCommon(unclassifiedDailyWeeklyRestCandidateCoverageIntervals);
        List<DriverWorkingTimeRestCoverageInterval> sortedPotentialHomeOvernightStayIntervals =
                sortPotentialHomeOvernightStayIntervalsCommon(potentialHomeOvernightStayIntervals);
        List<DriverWorkingTimeRestCoverageInterval> sortedPotentialInVehicleOvernightStayIntervals =
                sortPotentialInVehicleOvernightStayIntervalsCommon(potentialInVehicleOvernightStayIntervals);

        return new DriverWorkingTimeDerivedProjectionBundle(
                sortDrivingInterruptionIntervalsCommon(drivingInterruptionIntervals),
                sortDrivingInterruptionIntervalsCommon(dailyWeeklyRestCandidateIntervals),
                sortedDailyWeeklyRestCandidateCoverageIntervals,
                sortedUnclassifiedDailyWeeklyRestCandidateCoverageIntervals,
                sortDrivingInterruptionIntervalsCommon(drivingInterruptionVehicleChangeIntervals),
                sortVuCardAbsentIntervalsCommon(vuCardAbsentIntervals),
                sortedPotentialHomeOvernightStayIntervals,
                sortedPotentialInVehicleOvernightStayIntervals,
                sortPotentialInVehicleTripIntervalsCommon(potentialInVehicleTripIntervals)
        );
    }

    private List<Map<String, Object>> buildActivityIntervalInputEvents(
            List<DriverWorkingTimeActivityInterval> activityIntervals
    ) {
        return safeList(activityIntervals).stream()
                .map(this::toActivityIntervalInputMap)
                .filter(Objects::nonNull)
                .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>> buildVehicleUsageIntervalInputEventsCommon(
            List<DriverWorkingTimeVehicleUsageInterval> vehicleUsageIntervals
    ) {
        return safeList(vehicleUsageIntervals).stream()
                .map(this::toVehicleUsageIntervalInputMap)
                .filter(Objects::nonNull)
                .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>> buildSupportGeoInputEventsCommon(
            UUID sessionId,
            List<RuntimeSupportEvidenceEvent> 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 DriverWorkingTimeDerivedProjectionBundle emptyDerivedProjectionBundle() {
        return new DriverWorkingTimeDerivedProjectionBundle(
                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(
            DriverWorkingTimeActivityInterval interval
    ) {
        if (interval == null || interval.startedAt() == null || interval.endedAt() == null) {
            return null;
        }
        Map<String, Object> event = new LinkedHashMap<>();
        event.put("sessionId", interval.sessionId());
        event.put("driverKey", interval.driverKey());
        event.put("intervalId", interval.intervalId());
        event.put("activityType", interval.activityType());
        event.put("cardSlot", interval.cardSlot());
        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", interval.firstSourceIntervalId());
        event.put("lastSourceIntervalId", interval.lastSourceIntervalId());
        event.put("startedAt", interval.startedAt());
        event.put("endedAt", interval.endedAt());
        event.put("startedAtEpochSecond", interval.startedAtEpochSecond());
        event.put("endedAtEpochSecond", interval.endedAtEpochSecond());
        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(DriverWorkingTimeVehicleUsageInterval interval) {
        if (interval == null || interval.startedAt() == null) {
            return null;
        }
        Map<String, Object> event = new LinkedHashMap<>();
        event.put("sessionId", interval.sessionId());
        event.put("driverKey", interval.driverKey());
        event.put("intervalId", interval.intervalId());
        event.put("firstSourceIntervalId", interval.firstSourceIntervalId());
        event.put("lastSourceIntervalId", interval.lastSourceIntervalId());
        event.put("startedAt", interval.startedAt());
        event.put("endedAt", interval.endedAt());
        event.put("startedAtEpochSecond", interval.startedAtEpochSecond());
        event.put("endedAtEpochSecond", interval.endedAtEpochSecond());
        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,
            RuntimeSupportEvidenceEvent 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 == null ? new UUID(0L, 0L) : 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 void collectDrivingInterruptionIntervalModels(
            EventBean[] newData,
            List<DriverWorkingTimeDrivingInterruptionInterval> target
    ) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            long startedAtEpochSecond = (Long) event.get("startedAtEpochSecond");
            long endedAtEpochSecond = (Long) event.get("endedAtEpochSecond");
            target.add(new DriverWorkingTimeDrivingInterruptionInterval(
                    (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 collectVuCardAbsentIntervalModels(
            EventBean[] newData,
            List<DriverWorkingTimeVuCardAbsentInterval> target
    ) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            long startedAtEpochSecond = (Long) event.get("startedAtEpochSecond");
            long endedAtEpochSecond = (Long) event.get("endedAtEpochSecond");
            target.add(new DriverWorkingTimeVuCardAbsentInterval(
                    (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 collectDailyWeeklyRestCandidateCoverageIntervalModels(
            EventBean[] newData,
            List<DriverWorkingTimeRestCoverageInterval> target
    ) {
        collectRestCoverageModels(newData, target);
    }

    private void collectPotentialHomeOvernightStayIntervalModels(
            EventBean[] newData,
            List<DriverWorkingTimeRestCoverageInterval> target
    ) {
        collectRestCoverageModels(newData, target);
    }

    private void collectPotentialInVehicleOvernightStayIntervalModels(
            EventBean[] newData,
            List<DriverWorkingTimeRestCoverageInterval> target
    ) {
        collectRestCoverageModels(newData, target);
    }

    private void collectRestCoverageModels(
            EventBean[] newData,
            List<DriverWorkingTimeRestCoverageInterval> target
    ) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            long startedAtEpochSecond = (Long) event.get("startedAtEpochSecond");
            long endedAtEpochSecond = (Long) event.get("endedAtEpochSecond");
            DriverWorkingTimeGeoEvidenceEvent beginGeoEvent = geoEvidenceModel(
                    (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"))
            );
            DriverWorkingTimeGeoEvidenceEvent endGeoEvent = geoEvidenceModel(
                    (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 DriverWorkingTimeRestCoverageInterval(
                    (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 collectPotentialInVehicleTripIntervalModels(
            EventBean[] newData,
            List<DriverWorkingTimePotentialInVehicleTripInterval> target
    ) {
        if (newData == null) {
            return;
        }
        for (EventBean event : newData) {
            long startedAtEpochSecond = (Long) event.get("startedAtEpochSecond");
            long endedAtEpochSecond = (Long) event.get("endedAtEpochSecond");
            target.add(new DriverWorkingTimePotentialInVehicleTripInterval(
                    (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<DriverWorkingTimeDrivingInterruptionInterval> sortDrivingInterruptionIntervalsCommon(
            List<DriverWorkingTimeDrivingInterruptionInterval> intervals
    ) {
        return intervals.stream()
                .sorted(Comparator.comparing(DriverWorkingTimeDrivingInterruptionInterval::startedAt)
                        .thenComparing(DriverWorkingTimeDrivingInterruptionInterval::endedAt))
                .toList();
    }

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

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

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

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

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

    private List<DriverWorkingTimeRestCoverageInterval> deduplicateRestCoverageIntervalsCommon(
            List<DriverWorkingTimeRestCoverageInterval> intervals
    ) {
        Map<String, DriverWorkingTimeRestCoverageInterval> deduplicated = new LinkedHashMap<>();
        for (DriverWorkingTimeRestCoverageInterval 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 DriverWorkingTimeGeoEvidenceEvent geoEvidenceModel(
            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 DriverWorkingTimeGeoEvidenceEvent(
                eventId,
                eventDomain,
                occurredAt,
                latitude,
                longitude,
                distanceSeconds,
                odometerKm
        );
    }

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