/*
 * Source-neutral EPL module: DriverActivityPointEvent -> DriverActivityIntervalEvent.
 * This module is intended for Runtime Processing plans and is independent of any concrete source provider.
 */

create schema DriverActivityIntervalEvent(
    sessionId java.util.UUID,
    driverKey string,
    intervalId string,
    activityType string,
    cardSlot string,
    cardStatus string,
    drivingStatus string,
    registrationKey string,
    vehicleKey string,
    sourceKind string,
    firstSourceIntervalId string,
    lastSourceIntervalId string,
    startedAt java.time.OffsetDateTime,
    endedAt java.time.OffsetDateTime,
    startedAtEpochSecond long,
    endedAtEpochSecond long,
    durationSeconds long,
    sourceIntervalIds java.util.List,
    synthetic boolean,
    clippedToRequestedPeriod boolean,
    level string
);

create window OpenDriverActivityPoint#unique(driverKey, intervalId) as DriverActivityPointEvent;

insert into OpenDriverActivityPoint
select *
from DriverActivityPointEvent(lifecycle = 'START');

@Priority(20)
on DriverActivityPointEvent(lifecycle = 'END') as endEvent
insert into DriverActivityIntervalEvent
select
    startEvent.sessionId as sessionId,
    startEvent.driverKey as driverKey,
    startEvent.intervalId as intervalId,
    startEvent.activityType as activityType,
    startEvent.cardSlot as cardSlot,
    startEvent.cardStatus as cardStatus,
    startEvent.drivingStatus as drivingStatus,
    startEvent.registrationKey as registrationKey,
    startEvent.vehicleKey as vehicleKey,
    startEvent.sourceKind as sourceKind,
    startEvent.sourceRowId as firstSourceIntervalId,
    endEvent.sourceRowId as lastSourceIntervalId,
    startEvent.occurredAt as startedAt,
    endEvent.occurredAt as endedAt,
    startEvent.occurredAtEpochSecond as startedAtEpochSecond,
    endEvent.occurredAtEpochSecond as endedAtEpochSecond,
    endEvent.occurredAtEpochSecond - startEvent.occurredAtEpochSecond as durationSeconds,
    startEvent.sourceRowIds as sourceIntervalIds,
    startEvent.synthetic as synthetic,
    startEvent.clippedToRequestedPeriod as clippedToRequestedPeriod,
    startEvent.level as level
from OpenDriverActivityPoint as startEvent
where startEvent.driverKey = endEvent.driverKey
  and startEvent.intervalId = endEvent.intervalId
  and endEvent.occurredAtEpochSecond > startEvent.occurredAtEpochSecond;

@Priority(10)
on DriverActivityPointEvent(lifecycle = 'END') as endEvent
delete from OpenDriverActivityPoint as openEvent
where openEvent.driverKey = endEvent.driverKey
  and openEvent.intervalId = endEvent.intervalId;

@name('driverActivityIntervals')
select *
from DriverActivityIntervalEvent;
