/*
 * Source-neutral EPL module: DriverVehicleUsagePointEvent -> DriverVehicleUsageIntervalEvent.
 * Pairs INSERT/WITHDRAW point events for the same driver and interval id.
 */

create schema DriverVehicleUsageIntervalEvent(
    sessionId java.util.UUID,
    driverKey string,
    intervalId string,
    firstSourceIntervalId string,
    lastSourceIntervalId string,
    startedAt java.time.OffsetDateTime,
    endedAt java.time.OffsetDateTime,
    startedAtEpochSecond long,
    endedAtEpochSecond java.lang.Long,
    durationSeconds long,
    odometerBeginKm java.lang.Long,
    odometerEndKm java.lang.Long,
    registrationKey string,
    vehicleKey string,
    sourceKind string,
    sourceIntervalIds java.util.List
);

create window OpenDriverVehicleUsagePoint#unique(driverKey, intervalId) as DriverVehicleUsagePointEvent;

insert into OpenDriverVehicleUsagePoint
select *
from DriverVehicleUsagePointEvent(lifecycle = 'INSERT');

@Priority(20)
on DriverVehicleUsagePointEvent(lifecycle = 'WITHDRAW') as withdrawEvent
insert into DriverVehicleUsageIntervalEvent
select
    insertEvent.sessionId as sessionId,
    insertEvent.driverKey as driverKey,
    insertEvent.intervalId as intervalId,
    insertEvent.sourceRowId as firstSourceIntervalId,
    withdrawEvent.sourceRowId as lastSourceIntervalId,
    insertEvent.occurredAt as startedAt,
    withdrawEvent.occurredAt as endedAt,
    insertEvent.occurredAtEpochSecond as startedAtEpochSecond,
    withdrawEvent.occurredAtEpochSecond as endedAtEpochSecond,
    withdrawEvent.occurredAtEpochSecond - insertEvent.occurredAtEpochSecond as durationSeconds,
    insertEvent.odometerKm as odometerBeginKm,
    withdrawEvent.odometerKm as odometerEndKm,
    insertEvent.registrationKey as registrationKey,
    insertEvent.vehicleKey as vehicleKey,
    insertEvent.sourceKind as sourceKind,
    insertEvent.sourceRowIds as sourceIntervalIds
from OpenDriverVehicleUsagePoint as insertEvent
where insertEvent.driverKey = withdrawEvent.driverKey
  and insertEvent.intervalId = withdrawEvent.intervalId
  and withdrawEvent.occurredAtEpochSecond >= insertEvent.occurredAtEpochSecond;

@Priority(10)
on DriverVehicleUsagePointEvent(lifecycle = 'WITHDRAW') as withdrawEvent
delete from OpenDriverVehicleUsagePoint as openEvent
where openEvent.driverKey = withdrawEvent.driverKey
  and openEvent.intervalId = withdrawEvent.intervalId;

@name('driverVehicleUsageIntervals')
select *
from DriverVehicleUsageIntervalEvent;
