package at.procon.eventhub.esperpoc.service;

import at.procon.eventhub.config.EventHubProperties;
import at.procon.eventhub.esperpoc.dto.ActivityIntervalDto;
import at.procon.eventhub.esperpoc.dto.DriverWorkSummaryDto;
import at.procon.eventhub.esperpoc.dto.DrivingInterruptionDto;
import at.procon.eventhub.esperpoc.dto.EsperActivityMergeMode;
import at.procon.eventhub.esperpoc.dto.EsperPocRequest;
import at.procon.eventhub.esperpoc.dto.EsperPocResultDto;
import at.procon.eventhub.esperpoc.dto.EsperShiftResolutionMode;
import at.procon.eventhub.esperpoc.dto.OperatingTimePeriodDto;
import at.procon.eventhub.esperpoc.dto.RawActivityEventDto;
import at.procon.eventhub.esperpoc.dto.ResolvedWorkShiftDto;
import at.procon.eventhub.esperpoc.dto.ShiftDrivingEvaluationDto;
import at.procon.eventhub.esperpoc.persistence.EsperPocActivityRepository;
import java.time.Duration;
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.Set;
import java.util.UUID;
import org.springframework.beans.factory.annotation.Autowired;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

@Service
public class EsperPocDriverCardActivityService {

    private static final Duration SHIFT_SCAN_CONTIGUOUS_GAP = Duration.ofMinutes(1);
    private static final Logger log = LoggerFactory.getLogger(EsperPocDriverCardActivityService.class);

    private final EsperPocActivityRepository activityRepository;
    private final EsperDriverActivityEngine esperEngine;
    private final EventHubProperties properties;

    public EsperPocDriverCardActivityService(
            EsperPocActivityRepository activityRepository,
            EsperDriverActivityEngine esperEngine
    ) {
        this(activityRepository, esperEngine, null);
    }

    @Autowired
    public EsperPocDriverCardActivityService(
            EsperPocActivityRepository activityRepository,
            EsperDriverActivityEngine esperEngine,
            EventHubProperties properties
    ) {
        this.activityRepository = activityRepository;
        this.esperEngine = esperEngine;
        this.properties = properties;
    }

    public EsperPocResultDto evaluate(EsperPocRequest request) {
        long startedNanos = System.nanoTime();
        EsperActivityMergeMode activityMergeMode = resolveActivityMergeMode(request);
        EsperShiftResolutionMode shiftResolutionMode = resolveShiftResolutionMode(request);
        OffsetDateTime requestedFrom = utc(request.occurredFrom());
        OffsetDateTime requestedTo = utc(request.occurredTo());
        OffsetDateTime loadedFrom = requestedFrom.minusHours(request.guardHours());
        OffsetDateTime loadedTo = requestedTo.plusHours(request.guardHours());

        long dbStartedNanos = System.nanoTime();
        List<RawActivityEventDto> rawEvents = activityRepository.findDriverActivityEvents(
                request.tenantKey(),
                request.driverEntityId(),
                loadedFrom,
                loadedTo
        );
        long dbElapsedMs = elapsedMillis(dbStartedNanos);
        List<RawActivityEventDto> driverCardRawEvents = rawEvents.stream()
                .filter(event -> "DRIVER_CARD".equals(event.sourceKind()))
                .toList();
        List<RawActivityEventDto> vehicleUnitRawEvents = rawEvents.stream()
                .filter(event -> "VEHICLE_UNIT".equals(event.sourceKind()))
                .toList();

        long cardIntervalsStartedNanos = System.nanoTime();
        List<ActivityIntervalDto> driverCardRawIntervals = esperEngine.buildIntervals(driverCardRawEvents);
        long cardIntervalsElapsedMs = elapsedMillis(cardIntervalsStartedNanos);
        long vuIntervalsStartedNanos = System.nanoTime();
        List<ActivityIntervalDto> vehicleUnitRawIntervals = esperEngine.buildIntervals(vehicleUnitRawEvents);
        long vuIntervalsElapsedMs = elapsedMillis(vuIntervalsStartedNanos);
        long vuGapFillStartedNanos = System.nanoTime();
        List<ActivityIntervalDto> resolvedLoadedIntervals = resolveVuFillGaps(
                driverCardRawIntervals,
                vehicleUnitRawIntervals
        );
        long vuGapFillElapsedMs = elapsedMillis(vuGapFillStartedNanos);

        long mergeStartedNanos = System.nanoTime();
        List<ActivityIntervalDto> mergedLoadedActivities = mergeActivities(
                resolvedLoadedIntervals,
                Duration.ofSeconds(request.mergeGapSeconds()),
                activityMergeMode
        );
        long mergeElapsedMs = elapsedMillis(mergeStartedNanos);
        long summaryStartedNanos = System.nanoTime();
        List<ActivityIntervalDto> mergedActivities = clipToPeriod(mergedLoadedActivities, requestedFrom, requestedTo);
        DriverWorkSummaryDto summary = summarize(request, requestedFrom, requestedTo, mergedActivities);
        long summaryElapsedMs = elapsedMillis(summaryStartedNanos);
        long operatingPeriodsStartedNanos = System.nanoTime();
        List<OperatingTimePeriodDto> operatingTimePeriods = buildOperatingTimePeriods(
                request,
                requestedFrom,
                requestedTo,
                mergedLoadedActivities
        );
        long operatingPeriodsElapsedMs = elapsedMillis(operatingPeriodsStartedNanos);
        long workingShiftsStartedNanos = System.nanoTime();
        List<ResolvedWorkShiftDto> workingShifts = buildResolvedWorkShifts(
                request,
                requestedFrom,
                requestedTo,
                resolvedLoadedIntervals,
                activityMergeMode,
                shiftResolutionMode
        );
        long workingShiftsElapsedMs = elapsedMillis(workingShiftsStartedNanos);
        long totalElapsedMs = elapsedMillis(startedNanos);

        log.info("Esper PoC working-shift evaluation tenant={} driverId={} requestedFrom={} requestedTo={} loadedFrom={} loadedTo={} mergeMode={} shiftMode={} rawEvents={} cardRawEvents={} vuRawEvents={} cardIntervals={} vuIntervals={} resolvedIntervals={} mergedActivities={} operatingPeriods={} workingShifts={} timingsMs={{dbRetrieve={}, cardIntervalEsper={}, vuIntervalEsper={}, vuGapFill={}, merge={}, summaryAndDriving={}, operatingPeriods={}, workingShifts={}, total={}}}",
                request.tenantKey(),
                request.driverEntityId(),
                requestedFrom,
                requestedTo,
                loadedFrom,
                loadedTo,
                activityMergeMode,
                shiftResolutionMode,
                rawEvents.size(),
                driverCardRawEvents.size(),
                vehicleUnitRawEvents.size(),
                driverCardRawIntervals.size(),
                vehicleUnitRawIntervals.size(),
                resolvedLoadedIntervals.size(),
                mergedActivities.size(),
                operatingTimePeriods.size(),
                workingShifts.size(),
                dbElapsedMs,
                cardIntervalsElapsedMs,
                vuIntervalsElapsedMs,
                vuGapFillElapsedMs,
                mergeElapsedMs,
                summaryElapsedMs,
                operatingPeriodsElapsedMs,
                workingShiftsElapsedMs,
                totalElapsedMs);

        return new EsperPocResultDto(
                request.tenantKey(),
                request.driverEntityId(),
                requestedFrom,
                requestedTo,
                loadedFrom,
                loadedTo,
                rawEvents.size(),
                driverCardRawEvents.size(),
                vehicleUnitRawEvents.size(),
                driverCardRawIntervals.size(),
                vehicleUnitRawIntervals.size(),
                resolvedLoadedIntervals.size(),
                mergedActivities.size(),
                operatingTimePeriods.size(),
                workingShifts.size(),
                request.operatingPeriodSplitRestHours(),
                request.shiftEndMarkPeriodMinutes(),
                request.absenceBeginEndMinActivityMinutes(),
                activityMergeMode,
                shiftResolutionMode,
                rawEvents,
                resolvedLoadedIntervals,
                mergedActivities,
                operatingTimePeriods,
                workingShifts,
                summary,
                summary,
                null,
                notes(request, activityMergeMode, shiftResolutionMode)
        );
    }

    public List<ActivityIntervalDto> mergeConsecutiveIdenticalActivities(
            List<ActivityIntervalDto> intervals,
            Duration mergeGapTolerance
    ) {
        if (intervals == null || intervals.isEmpty()) {
            return List.of();
        }
        List<ActivityIntervalDto> sorted = sortedPositiveIntervals(intervals);
        List<ActivityIntervalDto> result = new ArrayList<>();
        ActivityIntervalDto current = null;
        List<String> currentSources = new ArrayList<>();
        for (ActivityIntervalDto next : sorted) {
            if (current == null) {
                current = next;
                currentSources = new ArrayList<>(next.sourceRowIds());
                continue;
            }
            if (EsperActivitySemantics.canMerge(current, next, mergeGapTolerance)) {
                currentSources.addAll(next.sourceRowIds());
                OffsetDateTime newEndedAt = max(current.endedAt(), next.endedAt());
                current = new ActivityIntervalDto(
                        current.driverEntityId(),
                        current.vehicleId() == null ? next.vehicleId() : current.vehicleId(),
                        current.vehicleRegistrationId() == null ? next.vehicleRegistrationId() : current.vehicleRegistrationId(),
                        current.activityType(),
                        current.cardSlot(),
                        current.cardStatus(),
                        current.drivingStatus(),
                        current.sourceKind(),
                        current.startedAt(),
                        newEndedAt,
                        Duration.between(current.startedAt(), newEndedAt).getSeconds(),
                        current.sourceRowId(),
                        List.copyOf(currentSources),
                        current.clippedToRequestedPeriod() || next.clippedToRequestedPeriod(),
                        "MERGED_ACTIVITY"
                );
            } else {
                result.add(current.asMerged(List.copyOf(currentSources)));
                current = next;
                currentSources = new ArrayList<>(next.sourceRowIds());
            }
        }
        if (current != null) {
            result.add(current.asMerged(List.copyOf(currentSources)));
        }
        return result;
    }

    public List<OperatingTimePeriodDto> buildOperatingTimePeriods(
            EsperPocRequest request,
            OffsetDateTime requestedFrom,
            OffsetDateTime requestedTo,
            List<ActivityIntervalDto> mergedLoadedActivities
    ) {
        Duration splitRestThreshold = Duration.ofHours(request.operatingPeriodSplitRestHours());
        List<ActivityIntervalDto> sorted = sortedPositiveIntervals(mergedLoadedActivities);
        List<ActivityIntervalDto> longRests = sorted.stream()
                .filter(interval -> isOperatingPeriodSplitRest(interval, splitRestThreshold))
                .toList();

        List<OperatingTimePeriodDto> result = new ArrayList<>();
        OffsetDateTime currentSpanStart = requestedFrom;
        ActivityIntervalDto previousLongRest = null;
        int sequenceNumber = 1;

        for (ActivityIntervalDto longRest : longRests) {
            if (!longRest.endedAt().isAfter(requestedFrom)) {
                previousLongRest = longRest;
                continue;
            }
            if (!longRest.startedAt().isBefore(requestedTo)) {
                break;
            }

            OffsetDateTime restStart = max(longRest.startedAt(), requestedFrom);
            OffsetDateTime restEnd = min(longRest.endedAt(), requestedTo);
            if (!restEnd.isAfter(restStart)) {
                continue;
            }
            if (restStart.isAfter(currentSpanStart)) {
                OperatingTimePeriodDto period = buildOperatingPeriod(
                        sequenceNumber,
                        request,
                        currentSpanStart,
                        restStart,
                        previousLongRest,
                        longRest,
                        sorted
                );
                if (period != null) {
                    result.add(period);
                    sequenceNumber++;
                }
            }
            if (restEnd.isAfter(currentSpanStart)) {
                currentSpanStart = restEnd;
            }
            previousLongRest = longRest;
        }

        if (requestedTo.isAfter(currentSpanStart)) {
            OperatingTimePeriodDto period = buildOperatingPeriod(
                    sequenceNumber,
                    request,
                    currentSpanStart,
                    requestedTo,
                    previousLongRest,
                    null,
                    sorted
            );
            if (period != null) {
                result.add(period);
            }
        }
        return result;
    }

    public List<ResolvedWorkShiftDto> buildResolvedWorkShifts(
            EsperPocRequest request,
            OffsetDateTime requestedFrom,
            OffsetDateTime requestedTo,
            List<ActivityIntervalDto> resolvedLoadedIntervals
    ) {
        return buildResolvedWorkShifts(
                request,
                requestedFrom,
                requestedTo,
                resolvedLoadedIntervals,
                resolveActivityMergeMode(request),
                resolveShiftResolutionMode(request)
        );
    }

    public List<ResolvedWorkShiftDto> buildResolvedWorkShifts(
            EsperPocRequest request,
            OffsetDateTime requestedFrom,
            OffsetDateTime requestedTo,
            List<ActivityIntervalDto> resolvedLoadedIntervals,
            EsperActivityMergeMode activityMergeMode,
            EsperShiftResolutionMode shiftResolutionMode
    ) {
        long startedNanos = System.nanoTime();
        List<ActivityIntervalDto> sorted = sortedPositiveIntervals(resolvedLoadedIntervals);
        if (sorted.isEmpty()) {
            log.info("Esper PoC work-shift resolution requestedFrom={} requestedTo={} shiftMode={} sourceIntervals=0 resolvedShifts=0 timingsMs={{resolveSpans=0, enrichShifts=0, total={}}}",
                    requestedFrom,
                    requestedTo,
                    shiftResolutionMode,
                    elapsedMillis(startedNanos));
            return List.of();
        }

        long resolveSpansStartedNanos = System.nanoTime();
        List<EsperResolvedShiftSpan> shiftSpans = resolveShiftSpans(sorted, request, shiftResolutionMode);
        long resolveSpansElapsedMs = elapsedMillis(resolveSpansStartedNanos);
        if (shiftSpans.isEmpty()) {
            log.info("Esper PoC work-shift resolution requestedFrom={} requestedTo={} shiftMode={} sourceIntervals={} resolvedShiftSpans=0 resolvedShifts=0 timingsMs={{resolveSpans={}, enrichShifts=0, total={}}}",
                    requestedFrom,
                    requestedTo,
                    shiftResolutionMode,
                    sorted.size(),
                    resolveSpansElapsedMs,
                    elapsedMillis(startedNanos));
            return List.of();
        }

        ZoneOffset evaluationOffset = request.occurredFrom().getOffset();
        ShiftDeviationStats deviationStats = calculateShiftDeviationStats(shiftSpans, evaluationOffset);
        List<ResolvedWorkShiftDto> result = new ArrayList<>();
        int sequenceNumber = 1;
        long enrichShiftsStartedNanos = System.nanoTime();

        for (int index = 0; index < shiftSpans.size(); index++) {
            EsperResolvedShiftSpan span = shiftSpans.get(index);
            if (!span.endedAt().isAfter(requestedFrom) || !span.startedAt().isBefore(requestedTo)) {
                continue;
            }

            List<ActivityIntervalDto> rawShiftActivities = clipToPeriod(sorted, span.startedAt(), span.endedAt());
            if (rawShiftActivities.isEmpty()) {
                continue;
            }
            List<ActivityIntervalDto> mergedShiftActivities = mergeActivities(
                    rawShiftActivities,
                    Duration.ofSeconds(request.mergeGapSeconds()),
                    activityMergeMode
            );
            DriverWorkSummaryDto summary = summarize(request, span.startedAt(), span.endedAt(), mergedShiftActivities);
            ShiftDrivingEvaluationDto drivingEvaluation = evaluateSignificantDriving(
                    mergedShiftActivities,
                    request.significantDrivingMinutes()
            );
            ShiftDeviationSnapshot deviation = deviationStats.snapshot(span, evaluationOffset);

            result.add(new ResolvedWorkShiftDto(
                    sequenceNumber++,
                    span.startedAt(),
                    span.endedAt(),
                    span.nextShiftStartedAt(),
                    span.durationSeconds(),
                    span.dailyRestingTimeSeconds(),
                    firstDrivingStart(rawShiftActivities),
                    lastDrivingEnd(rawShiftActivities),
                    firstDrivingStart(rawShiftActivities, request.absenceBeginEndMinActivityMinutes()),
                    lastDrivingEnd(rawShiftActivities, request.absenceBeginEndMinActivityMinutes()),
                    distinctVehicleIds(rawShiftActivities),
                    distinctVehicleRegistrationIds(rawShiftActivities),
                    determineUsedDataKind(rawShiftActivities),
                    determineShiftKind(span, evaluationOffset),
                    deviation.beginInRange(),
                    deviation.endInRange(),
                    summary.drivingSeconds() == 0,
                    index > 0 && shiftSpans.get(index - 1).dailyRestingTimeSeconds() > Duration.ofHours(24).getSeconds(),
                    span.durationSeconds() > Duration.ofHours(24).getSeconds(),
                    span.dailyRestingTimeSeconds() >= Duration.ofHours(9).getSeconds()
                            && span.dailyRestingTimeSeconds() <= Duration.ofHours(11).getSeconds(),
                    span.dailyRestingTimeSeconds() >= Duration.ofHours(7).getSeconds()
                            && span.dailyRestingTimeSeconds() <= Duration.ofHours(9).getSeconds(),
                    rawShiftActivities.size(),
                    countActivities(rawShiftActivities, "DRIVE"),
                    countActivities(rawShiftActivities, "AVAILABILITY"),
                    countActivities(rawShiftActivities, "WORK"),
                    countActivities(rawShiftActivities, "BREAK_REST"),
                    countUnknownActivities(rawShiftActivities),
                    countWorkingActivities(rawShiftActivities),
                    summary,
                    drivingEvaluation,
                    rawShiftActivities
            ));
        }
        long enrichShiftsElapsedMs = elapsedMillis(enrichShiftsStartedNanos);
        log.info("Esper PoC work-shift resolution requestedFrom={} requestedTo={} shiftMode={} sourceIntervals={} resolvedShiftSpans={} resolvedShifts={} timingsMs={{resolveSpans={}, enrichShifts={}, total={}}}",
                requestedFrom,
                requestedTo,
                shiftResolutionMode,
                sorted.size(),
                shiftSpans.size(),
                result.size(),
                resolveSpansElapsedMs,
                enrichShiftsElapsedMs,
                elapsedMillis(startedNanos));
        return result;
    }

    List<ActivityIntervalDto> resolveVuFillGaps(
            List<ActivityIntervalDto> driverCardRawIntervals,
            List<ActivityIntervalDto> vehicleUnitRawIntervals
    ) {
        List<ActivityIntervalDto> driverCard = sortedPositiveIntervals(driverCardRawIntervals);
        List<ActivityIntervalDto> vehicleUnit = sortedPositiveIntervals(vehicleUnitRawIntervals);
        if (driverCard.isEmpty()) {
            return vehicleUnit;
        }
        if (vehicleUnit.isEmpty()) {
            return driverCard;
        }

        List<ActivityIntervalDto> resolved = new ArrayList<>(driverCard);
        for (ActivityIntervalDto vuInterval : vehicleUnit) {
            resolved.addAll(subtractCoverage(vuInterval, driverCard));
        }
        return resolved.stream()
                .sorted(Comparator.comparing(ActivityIntervalDto::startedAt)
                        .thenComparing(ActivityIntervalDto::endedAt)
                        .thenComparing(ActivityIntervalDto::sourceKind, Comparator.nullsLast(String::compareTo)))
                .toList();
    }

    private List<ActivityIntervalDto> mergeActivities(
            List<ActivityIntervalDto> intervals,
            Duration mergeGapTolerance,
            EsperActivityMergeMode mode
    ) {
        if (mode == EsperActivityMergeMode.ESPER) {
            return requireEsperEngine().mergeConsecutiveIdenticalActivities(intervals, mergeGapTolerance);
        }
        return mergeConsecutiveIdenticalActivities(intervals, mergeGapTolerance);
    }

    private List<ActivityIntervalDto> subtractCoverage(
            ActivityIntervalDto candidate,
            List<ActivityIntervalDto> coverage
    ) {
        List<ActivityIntervalDto> result = new ArrayList<>();
        OffsetDateTime cursor = candidate.startedAt();
        for (ActivityIntervalDto covered : coverage) {
            if (!covered.endedAt().isAfter(cursor)) {
                continue;
            }
            if (!covered.startedAt().isBefore(candidate.endedAt())) {
                break;
            }
            OffsetDateTime overlapStart = max(cursor, covered.startedAt());
            if (overlapStart.isAfter(cursor)) {
                result.add(candidate.withTime(cursor, overlapStart, candidate.clippedToRequestedPeriod()));
            }
            if (covered.endedAt().isAfter(cursor)) {
                cursor = max(cursor, covered.endedAt());
            }
            if (!candidate.endedAt().isAfter(cursor)) {
                break;
            }
        }
        if (candidate.endedAt().isAfter(cursor)) {
            result.add(candidate.withTime(cursor, candidate.endedAt(), candidate.clippedToRequestedPeriod()));
        }
        return result.stream()
                .filter(interval -> interval.endedAt().isAfter(interval.startedAt()))
                .toList();
    }

    private List<EsperResolvedShiftSpan> resolveShiftSpans(
            List<ActivityIntervalDto> sorted,
            EsperPocRequest request,
            EsperShiftResolutionMode mode
    ) {
        if (mode == EsperShiftResolutionMode.ESPER) {
            return requireEsperEngine().resolveShiftSpans(
                    sorted,
                    Duration.ofMinutes(request.shiftEndMarkPeriodMinutes()),
                    SHIFT_SCAN_CONTIGUOUS_GAP
            );
        }

        ActivityIntervalDto firstShiftActivity = sorted.stream()
                .filter(EsperActivitySemantics::isShiftActivity)
                .findFirst()
                .orElse(null);
        if (firstShiftActivity == null) {
            return List.of();
        }

        List<EsperResolvedShiftSpan> result = new ArrayList<>();
        OffsetDateTime shiftBegin = firstShiftActivity.startedAt();
        long accumulatedRestSeconds = 0;
        OffsetDateTime prevActivityEnd = null;
        long shiftEndThresholdSeconds = Duration.ofMinutes(request.shiftEndMarkPeriodMinutes()).getSeconds();

        for (ActivityIntervalDto interval : sorted) {
            if (!shiftBegin.isBefore(interval.startedAt())) {
                prevActivityEnd = interval.endedAt();
                continue;
            }

            if (EsperActivitySemantics.isRestOrUnknown(interval)) {
                if (prevActivityEnd != null
                        && Math.abs(Duration.between(prevActivityEnd, interval.startedAt()).getSeconds()) <= SHIFT_SCAN_CONTIGUOUS_GAP.getSeconds()) {
                    accumulatedRestSeconds += interval.durationSeconds();
                } else if (prevActivityEnd != null
                        && !Objects.equals(EsperActivitySemantics.recordDate(interval), prevActivityEnd.toLocalDate())) {
                    accumulatedRestSeconds = interval.durationSeconds() + Math.max(0, Duration.between(prevActivityEnd, interval.startedAt()).getSeconds());
                } else {
                    accumulatedRestSeconds = interval.durationSeconds();
                }

                if (accumulatedRestSeconds >= shiftEndThresholdSeconds) {
                    OffsetDateTime nextShiftBegin = findNextActiveStartAfter(sorted, interval.startedAt());
                    appendShiftSpan(result, sorted, shiftBegin, nextShiftBegin);
                    shiftBegin = nextShiftBegin;
                    accumulatedRestSeconds = 0;
                }
            } else if (EsperActivitySemantics.isShiftActivity(interval)) {
                if (prevActivityEnd != null
                        && !Objects.equals(EsperActivitySemantics.recordDate(interval), prevActivityEnd.toLocalDate())
                        && accumulatedRestSeconds + Math.max(0, Duration.between(prevActivityEnd, interval.startedAt()).getSeconds()) >= shiftEndThresholdSeconds) {
                    OffsetDateTime nextShiftBegin = interval.startedAt();
                    appendShiftSpan(result, sorted, shiftBegin, nextShiftBegin);
                    shiftBegin = nextShiftBegin;
                }
                accumulatedRestSeconds = 0;
            }

            prevActivityEnd = interval.endedAt();
        }

        if (shiftBegin != null) {
            appendShiftSpan(result, sorted, shiftBegin, null);
        }
        return result;
    }

    private void appendShiftSpan(
            List<EsperResolvedShiftSpan> target,
            List<ActivityIntervalDto> sorted,
            OffsetDateTime shiftBegin,
            OffsetDateTime nextShiftBegin
    ) {
        if (shiftBegin == null) {
            return;
        }
        OffsetDateTime shiftEnd = findLastActiveEnd(sorted, shiftBegin, nextShiftBegin);
        if (shiftEnd == null || !shiftEnd.isAfter(shiftBegin)) {
            return;
        }
        long restSeconds = nextShiftBegin == null ? 0 : Math.max(0, Duration.between(shiftEnd, nextShiftBegin).getSeconds());
        target.add(new EsperResolvedShiftSpan(shiftBegin, shiftEnd, nextShiftBegin, restSeconds));
    }

    private OffsetDateTime findNextActiveStartAfter(List<ActivityIntervalDto> sorted, OffsetDateTime activityTime) {
        return sorted.stream()
                .filter(interval -> interval.startedAt().isAfter(activityTime))
                .filter(EsperActivitySemantics::isShiftActivity)
                .map(ActivityIntervalDto::startedAt)
                .findFirst()
                .orElse(null);
    }

    private OffsetDateTime findLastActiveEnd(
            List<ActivityIntervalDto> sorted,
            OffsetDateTime shiftBegin,
            OffsetDateTime nextShiftBegin
    ) {
        return sorted.stream()
                .filter(interval -> !interval.startedAt().isBefore(shiftBegin))
                .filter(interval -> nextShiftBegin == null || interval.startedAt().isBefore(nextShiftBegin))
                .filter(EsperActivitySemantics::isShiftActivity)
                .map(ActivityIntervalDto::endedAt)
                .max(OffsetDateTime::compareTo)
                .orElse(null);
    }

    private ShiftDeviationStats calculateShiftDeviationStats(List<EsperResolvedShiftSpan> shiftSpans, ZoneOffset evaluationOffset) {
        Map<String, List<EsperResolvedShiftSpan>> grouped = new LinkedHashMap<>();
        for (EsperResolvedShiftSpan span : shiftSpans) {
            grouped.computeIfAbsent(determineShiftKind(span, evaluationOffset), ignored -> new ArrayList<>()).add(span);
        }

        Map<String, DeviationSnapshot> stats = new LinkedHashMap<>();
        for (Map.Entry<String, List<EsperResolvedShiftSpan>> entry : grouped.entrySet()) {
            List<EsperResolvedShiftSpan> spans = entry.getValue();
            if (spans.size() < 4) {
                stats.put(entry.getKey(), new DeviationSnapshot(Double.NaN, Double.NaN, Double.NaN, Double.NaN, false));
                continue;
            }
            double beginMean = spans.stream().mapToInt(span -> minutesOfDay(span.startedAt(), evaluationOffset)).average().orElse(Double.NaN);
            double endMean = spans.stream().mapToInt(span -> minutesOfDay(span.endedAt(), evaluationOffset)).average().orElse(Double.NaN);
            double beginDeviation = standardDeviation(spans.stream().mapToInt(span -> minutesOfDay(span.startedAt(), evaluationOffset)).toArray(), beginMean);
            double endDeviation = standardDeviation(spans.stream().mapToInt(span -> minutesOfDay(span.endedAt(), evaluationOffset)).toArray(), endMean);
            stats.put(entry.getKey(), new DeviationSnapshot(beginMean, endMean, beginDeviation, endDeviation, true));
        }
        return new ShiftDeviationStats(stats);
    }

    private double standardDeviation(int[] values, double mean) {
        double squaredSum = 0;
        for (int value : values) {
            double delta = value - mean;
            squaredSum += delta * delta;
        }
        return Math.sqrt(squaredSum / values.length);
    }

    private int minutesOfDay(OffsetDateTime value, ZoneOffset evaluationOffset) {
        OffsetDateTime local = value.withOffsetSameInstant(evaluationOffset);
        return local.getHour() * 60 + local.getMinute();
    }

    private String determineShiftKind(EsperResolvedShiftSpan span, ZoneOffset evaluationOffset) {
        OffsetDateTime localStart = span.startedAt().withOffsetSameInstant(evaluationOffset);
        OffsetDateTime localEnd = span.endedAt().withOffsetSameInstant(evaluationOffset);
        if (localStart.toLocalDate().equals(localEnd.toLocalDate())) {
            int startSecond = localStart.getHour() * 3600 + localStart.getMinute() * 60 + localStart.getSecond();
            int endSecond = localEnd.getHour() * 3600 + localEnd.getMinute() * 60 + localEnd.getSecond();
            return startSecond < 5 * 3600 || endSecond > 20 * 3600 ? "SPECIAL_HOURS" : "DAY";
        }
        return "OVERNIGHT";
    }

    private String determineUsedDataKind(List<ActivityIntervalDto> activities) {
        boolean hasCard = activities.stream().anyMatch(activity -> "DRIVER_CARD".equals(activity.sourceKind()));
        boolean hasVu = activities.stream().anyMatch(activity -> "VEHICLE_UNIT".equals(activity.sourceKind()));
        if (hasCard && hasVu) {
            return "BOTH";
        }
        if (hasVu) {
            return "VEHICLE_UNIT";
        }
        return "DRIVER_CARD";
    }

    private boolean isOperatingPeriodSplitRest(ActivityIntervalDto interval, Duration splitRestThreshold) {
        return "BREAK_REST".equals(interval.activityType())
                && interval.durationSeconds() > splitRestThreshold.getSeconds();
    }

    private List<ActivityIntervalDto> clipToPeriod(
            List<ActivityIntervalDto> intervals,
            OffsetDateTime periodFrom,
            OffsetDateTime periodTo
    ) {
        return intervals.stream()
                .map(interval -> {
                    OffsetDateTime start = max(interval.startedAt(), periodFrom);
                    OffsetDateTime end = min(interval.endedAt(), periodTo);
                    if (!end.isAfter(start)) {
                        return null;
                    }
                    boolean clipped = !start.equals(interval.startedAt()) || !end.equals(interval.endedAt());
                    return interval.withTime(start, end, clipped);
                })
                .filter(Objects::nonNull)
                .sorted(Comparator.comparing(ActivityIntervalDto::startedAt).thenComparing(ActivityIntervalDto::endedAt))
                .toList();
    }

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

    private DriverWorkSummaryDto summarize(
            EsperPocRequest request,
            OffsetDateTime periodFrom,
            OffsetDateTime periodTo,
            List<ActivityIntervalDto> activities
    ) {
        Map<String, Long> secondsByActivity = new LinkedHashMap<>();
        for (ActivityIntervalDto activity : activities) {
            secondsByActivity.merge(activity.activityType(), activity.durationSeconds(), Long::sum);
        }
        long drivingSeconds = secondsByActivity.getOrDefault("DRIVE", 0L);
        long workSeconds = secondsByActivity.getOrDefault("WORK", 0L);
        long availabilitySeconds = secondsByActivity.getOrDefault("AVAILABILITY", 0L);
        long breakRestSeconds = secondsByActivity.getOrDefault("BREAK_REST", 0L);
        long workingSeconds = drivingSeconds + workSeconds;
        long operationSeconds = drivingSeconds + workSeconds + availabilitySeconds;
        return new DriverWorkSummaryDto(
                request.driverEntityId(),
                periodFrom,
                periodTo,
                drivingSeconds,
                workSeconds,
                availabilitySeconds,
                breakRestSeconds,
                workingSeconds,
                operationSeconds,
                secondsByActivity
        );
    }

    private ShiftDrivingEvaluationDto evaluateSignificantDriving(
            List<ActivityIntervalDto> activities,
            int significantDrivingMinutes
    ) {
        long significantSeconds = Duration.ofMinutes(significantDrivingMinutes).getSeconds();
        List<ActivityIntervalDto> significantDrivingPeriods = activities.stream()
                .filter(activity -> "DRIVE".equals(activity.activityType()))
                .filter(activity -> activity.durationSeconds() > significantSeconds)
                .sorted(Comparator.comparing(ActivityIntervalDto::startedAt))
                .toList();
        if (significantDrivingPeriods.isEmpty()) {
            return new ShiftDrivingEvaluationDto(
                    significantDrivingMinutes,
                    null,
                    null,
                    null,
                    null,
                    List.of()
            );
        }
        List<DrivingInterruptionDto> interruptions = new ArrayList<>();
        for (int i = 1; i < significantDrivingPeriods.size(); i++) {
            ActivityIntervalDto previous = significantDrivingPeriods.get(i - 1);
            ActivityIntervalDto next = significantDrivingPeriods.get(i);
            if (next.startedAt().isAfter(previous.endedAt())) {
                interruptions.add(new DrivingInterruptionDto(
                        previous.endedAt(),
                        next.startedAt(),
                        Duration.between(previous.endedAt(), next.startedAt()).getSeconds(),
                        previous.sourceRowId(),
                        next.sourceRowId()
                ));
            }
        }
        ActivityIntervalDto first = significantDrivingPeriods.get(0);
        ActivityIntervalDto last = significantDrivingPeriods.get(significantDrivingPeriods.size() - 1);
        return new ShiftDrivingEvaluationDto(
                significantDrivingMinutes,
                first.startedAt(),
                last.endedAt(),
                first,
                last,
                interruptions
        );
    }

    private OperatingTimePeriodDto buildOperatingPeriod(
            int sequenceNumber,
            EsperPocRequest request,
            OffsetDateTime spanFrom,
            OffsetDateTime spanTo,
            ActivityIntervalDto splitStartedAfterLongRest,
            ActivityIntervalDto splitEndedByLongRest,
            List<ActivityIntervalDto> allActivities
    ) {
        List<ActivityIntervalDto> activities = allActivities.stream()
                .filter(activity -> activity.startedAt().isBefore(spanTo) && activity.endedAt().isAfter(spanFrom))
                .map(activity -> {
                    OffsetDateTime start = max(activity.startedAt(), spanFrom);
                    OffsetDateTime end = min(activity.endedAt(), spanTo);
                    if (!end.isAfter(start)) {
                        return null;
                    }
                    boolean clipped = !start.equals(activity.startedAt()) || !end.equals(activity.endedAt());
                    return activity.withTime(start, end, clipped);
                })
                .filter(Objects::nonNull)
                .filter(activity -> activity.endedAt().isAfter(activity.startedAt()))
                .sorted(Comparator.comparing(ActivityIntervalDto::startedAt).thenComparing(ActivityIntervalDto::endedAt))
                .toList();

        if (activities.isEmpty()) {
            return null;
        }

        OffsetDateTime startedAt = activities.get(0).startedAt();
        OffsetDateTime endedAt = activities.get(activities.size() - 1).endedAt();
        DriverWorkSummaryDto summary = summarize(request, startedAt, endedAt, activities);
        ShiftDrivingEvaluationDto drivingEvaluation = evaluateSignificantDriving(
                mergeActivities(
                        activities,
                        Duration.ofSeconds(request.mergeGapSeconds()),
                        resolveActivityMergeMode(request)
                ),
                request.significantDrivingMinutes()
        );
        return new OperatingTimePeriodDto(
                sequenceNumber,
                startedAt,
                endedAt,
                Duration.between(startedAt, endedAt).getSeconds(),
                splitStartedAfterLongRest,
                splitEndedByLongRest,
                activities,
                summary,
                drivingEvaluation
        );
    }

    private OffsetDateTime firstDrivingStart(List<ActivityIntervalDto> activities) {
        return activities.stream()
                .filter(this::isDrivingActivity)
                .map(ActivityIntervalDto::startedAt)
                .min(OffsetDateTime::compareTo)
                .orElse(null);
    }

    private OffsetDateTime lastDrivingEnd(List<ActivityIntervalDto> activities) {
        return activities.stream()
                .filter(this::isDrivingActivity)
                .map(ActivityIntervalDto::endedAt)
                .max(OffsetDateTime::compareTo)
                .orElse(null);
    }

    private OffsetDateTime firstDrivingStart(List<ActivityIntervalDto> activities, int minMinutes) {
        long threshold = Duration.ofMinutes(minMinutes).getSeconds();
        return activities.stream()
                .filter(this::isDrivingActivity)
                .filter(activity -> activity.durationSeconds() >= threshold)
                .map(ActivityIntervalDto::startedAt)
                .min(OffsetDateTime::compareTo)
                .orElse(null);
    }

    private OffsetDateTime lastDrivingEnd(List<ActivityIntervalDto> activities, int minMinutes) {
        long threshold = Duration.ofMinutes(minMinutes).getSeconds();
        return activities.stream()
                .filter(this::isDrivingActivity)
                .filter(activity -> activity.durationSeconds() >= threshold)
                .map(ActivityIntervalDto::endedAt)
                .max(OffsetDateTime::compareTo)
                .orElse(null);
    }

    private boolean isDrivingActivity(ActivityIntervalDto activity) {
        return EsperActivitySemantics.isKnownActivity(activity) && "DRIVE".equals(activity.activityType());
    }

    private int countActivities(List<ActivityIntervalDto> activities, String activityType) {
        return (int) activities.stream()
                .filter(EsperActivitySemantics::isKnownActivity)
                .filter(activity -> activityType.equals(activity.activityType()))
                .count();
    }

    private int countUnknownActivities(List<ActivityIntervalDto> activities) {
        return (int) activities.stream()
                .filter(EsperActivitySemantics::isUnknownActivity)
                .count();
    }

    private int countWorkingActivities(List<ActivityIntervalDto> activities) {
        return (int) activities.stream()
                .filter(EsperActivitySemantics::isShiftActivity)
                .count();
    }

    private List<UUID> distinctVehicleIds(List<ActivityIntervalDto> activities) {
        Set<UUID> values = new LinkedHashSet<>();
        for (ActivityIntervalDto activity : activities) {
            if (activity.vehicleId() != null) {
                values.add(activity.vehicleId());
            }
        }
        return List.copyOf(values);
    }

    private List<UUID> distinctVehicleRegistrationIds(List<ActivityIntervalDto> activities) {
        Set<UUID> values = new LinkedHashSet<>();
        for (ActivityIntervalDto activity : activities) {
            if (activity.vehicleRegistrationId() != null) {
                values.add(activity.vehicleRegistrationId());
            }
        }
        return List.copyOf(values);
    }

    private List<String> notes(
            EsperPocRequest request,
            EsperActivityMergeMode activityMergeMode,
            EsperShiftResolutionMode shiftResolutionMode
    ) {
        return List.of(
                "PoC reads tachograph DRIVER_CARD/CARD_ACTIVITY events and VEHICLE_UNIT/VU_ACTIVITY events from eventhub.event.",
                "Driver-card intervals remain authoritative. VEHICLE_UNIT activity fills only uncovered time gaps in the driver-card timeline.",
                "Configured activity merge mode: " + activityMergeMode + ". Configured shift resolution mode: " + shiftResolutionMode + ".",
                "Resolved work shifts follow the stored-procedure split rule: BREAK_REST or unknown coverage accumulating to "
                        + request.shiftEndMarkPeriodMinutes() + " minutes ends a shift.",
                "Driving-time interruption evaluation is reported per operating time period and per working shift, not as a single evaluation for the complete requested period.",
                "Activities are merged in the guard window first and then clipped to the requested period, so BREAK_REST across the period boundary can still split operating periods correctly.",
                "Operating time periods are split by BREAK_REST activities longer than " + request.operatingPeriodSplitRestHours() + " hours.",
                "Working seconds = DRIVE + WORK. Operation seconds = DRIVE + WORK + AVAILABILITY. BREAK_REST is reported separately."
        );
    }

    private EsperActivityMergeMode resolveActivityMergeMode(EsperPocRequest request) {
        if (request.activityMergeMode() != null) {
            return request.activityMergeMode();
        }
        return properties == null ? EsperActivityMergeMode.JAVA : properties.getEsperPoc().getActivityMergeMode();
    }

    private EsperShiftResolutionMode resolveShiftResolutionMode(EsperPocRequest request) {
        if (request.shiftResolutionMode() != null) {
            return request.shiftResolutionMode();
        }
        return properties == null ? EsperShiftResolutionMode.JAVA : properties.getEsperPoc().getShiftResolutionMode();
    }

    private EsperDriverActivityEngine requireEsperEngine() {
        if (esperEngine == null) {
            throw new IllegalStateException("Esper engine is required for ESPER execution modes");
        }
        return esperEngine;
    }

    private OffsetDateTime utc(OffsetDateTime value) {
        return value.withOffsetSameInstant(ZoneOffset.UTC);
    }

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

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

    private long elapsedMillis(long startedNanos) {
        return Duration.ofNanos(System.nanoTime() - startedNanos).toMillis();
    }

    private record ShiftDeviationStats(Map<String, DeviationSnapshot> stats) {
        ShiftDeviationSnapshot snapshot(EsperResolvedShiftSpan span, ZoneOffset evaluationOffset) {
            DeviationSnapshot snapshot = stats.get(determineShiftKindStatic(span, evaluationOffset));
            if (snapshot == null || !snapshot.computed()) {
                return new ShiftDeviationSnapshot(true, true);
            }
            int beginMinutes = minutesOfDayStatic(span.startedAt(), evaluationOffset);
            int endMinutes = minutesOfDayStatic(span.endedAt(), evaluationOffset);
            boolean beginInRange = beginMinutes >= snapshot.beginMean() - snapshot.beginDeviation()
                    && beginMinutes <= snapshot.beginMean() + snapshot.beginDeviation();
            boolean endInRange = endMinutes >= snapshot.endMean() - snapshot.endDeviation()
                    && endMinutes <= snapshot.endMean() + snapshot.endDeviation();
            return new ShiftDeviationSnapshot(beginInRange, endInRange);
        }

        private static String determineShiftKindStatic(EsperResolvedShiftSpan span, ZoneOffset evaluationOffset) {
            OffsetDateTime localStart = span.startedAt().withOffsetSameInstant(evaluationOffset);
            OffsetDateTime localEnd = span.endedAt().withOffsetSameInstant(evaluationOffset);
            if (localStart.toLocalDate().equals(localEnd.toLocalDate())) {
                int startSecond = localStart.getHour() * 3600 + localStart.getMinute() * 60 + localStart.getSecond();
                int endSecond = localEnd.getHour() * 3600 + localEnd.getMinute() * 60 + localEnd.getSecond();
                return startSecond < 5 * 3600 || endSecond > 20 * 3600 ? "SPECIAL_HOURS" : "DAY";
            }
            return "OVERNIGHT";
        }

        private static int minutesOfDayStatic(OffsetDateTime value, ZoneOffset evaluationOffset) {
            OffsetDateTime local = value.withOffsetSameInstant(evaluationOffset);
            return local.getHour() * 60 + local.getMinute();
        }
    }

    private record DeviationSnapshot(
            double beginMean,
            double endMean,
            double beginDeviation,
            double endDeviation,
            boolean computed
    ) {
    }

    private record ShiftDeviationSnapshot(boolean beginInRange, boolean endInRange) {
    }
}
