create schema SignificantDrivingInterval(
    sessionId java.util.UUID,
    driverKey string,
    firstSourceIntervalId string,
    lastSourceIntervalId string,
    startedAtEpochSecond long,
    endedAtEpochSecond long,
    durationSeconds long,
    registrationKey string,
    vehicleKey string
);

create schema DrivingInterruptionInterval(
    sessionId java.util.UUID,
    driverKey string,
    startedAtEpochSecond long,
    endedAtEpochSecond long,
    durationSeconds long,
    previousDrivingSourceIntervalId string,
    nextDrivingSourceIntervalId string,
    previousRegistrationKey string,
    nextRegistrationKey string,
    previousVehicleKey string,
    nextVehicleKey string
);

create schema DailyWeeklyRestCandidateInterval(
    sessionId java.util.UUID,
    driverKey string,
    startedAtEpochSecond long,
    endedAtEpochSecond long,
    durationSeconds long,
    previousDrivingSourceIntervalId string,
    nextDrivingSourceIntervalId string,
    previousRegistrationKey string,
    nextRegistrationKey string,
    previousVehicleKey string,
    nextVehicleKey string
);

create schema DrivingInterruptionVehicleChangeInterval(
    sessionId java.util.UUID,
    driverKey string,
    startedAtEpochSecond long,
    endedAtEpochSecond long,
    durationSeconds long,
    previousDrivingSourceIntervalId string,
    nextDrivingSourceIntervalId string,
    previousRegistrationKey string,
    nextRegistrationKey string,
    previousVehicleKey string,
    nextVehicleKey string
);

create schema DrivingInterruptionVehicleNotChangedInterval(
    sessionId java.util.UUID,
    driverKey string,
    startedAtEpochSecond long,
    endedAtEpochSecond long,
    durationSeconds long,
    previousDrivingSourceIntervalId string,
    nextDrivingSourceIntervalId string,
    previousRegistrationKey string,
    nextRegistrationKey string,
    previousVehicleKey string,
    nextVehicleKey string
);

create schema DailyWeeklyRestCandidateCoverageUnknownResolvedInterval(
    sessionId java.util.UUID,
    driverKey string,
    startedAtEpochSecond long,
    endedAtEpochSecond long,
    durationSeconds long,
    unknownDurationSeconds long,
    previousDrivingSourceIntervalId string,
    nextDrivingSourceIntervalId string,
    previousRegistrationKey string,
    nextRegistrationKey string,
    previousVehicleKey string,
    nextVehicleKey string
);

create schema DailyWeeklyRestCandidateCoverageCardResolvedInterval(
    sessionId java.util.UUID,
    driverKey string,
    startedAtEpochSecond long,
    endedAtEpochSecond long,
    durationSeconds long,
    cardPresentDurationSeconds long,
    unknownDurationSeconds long,
    previousDrivingSourceIntervalId string,
    nextDrivingSourceIntervalId string,
    previousRegistrationKey string,
    nextRegistrationKey string,
    previousVehicleKey string,
    nextVehicleKey string
);

create schema DailyWeeklyRestCandidateCoverageInterval(
    sessionId java.util.UUID,
    driverKey string,
    startedAtEpochSecond long,
    endedAtEpochSecond long,
    durationSeconds long,
    cardPresentDurationSeconds long,
    cardPresentCoveragePercent double,
    unknownDurationSeconds long,
    unknownCoveragePercent double,
    previousDrivingSourceIntervalId string,
    nextDrivingSourceIntervalId string,
    previousRegistrationKey string,
    nextRegistrationKey string,
    previousVehicleKey string,
    nextVehicleKey string
);

create context PerDriver partition by driverKey from TachographVehicleUsageIntervalInputEvent;

create schema VuCardAbsentInterval(
    sessionId java.util.UUID,
    driverKey string,
    startedAtEpochSecond long,
    endedAtEpochSecond long,
    durationSeconds long,
    previousUsageIntervalId string,
    nextUsageIntervalId string,
    previousRegistrationKey string,
    nextRegistrationKey string,
    previousVehicleKey string,
    nextVehicleKey string
);

create schema PotentialHomeOvernightStayInterval(
    sessionId java.util.UUID,
    driverKey string,
    startedAtEpochSecond long,
    endedAtEpochSecond long,
    durationSeconds long,
    unknownDurationSeconds long,
    unknownCoveragePercent double,
    previousDrivingSourceIntervalId string,
    nextDrivingSourceIntervalId string,
    previousRegistrationKey string,
    nextRegistrationKey string,
    previousVehicleKey string,
    nextVehicleKey string
);

create schema PotentialInVehicleOvernightStayInterval(
    sessionId java.util.UUID,
    driverKey string,
    startedAtEpochSecond long,
    endedAtEpochSecond long,
    durationSeconds long,
    cardPresentDurationSeconds long,
    cardPresentCoveragePercent double,
    previousDrivingSourceIntervalId string,
    nextDrivingSourceIntervalId string,
    previousRegistrationKey string,
    nextRegistrationKey string,
    previousVehicleKey string,
    nextVehicleKey string
);

create schema UnclassifiedDailyWeeklyRestCandidateCoverageInterval(
    sessionId java.util.UUID,
    driverKey string,
    startedAtEpochSecond long,
    endedAtEpochSecond long,
    durationSeconds long,
    cardPresentDurationSeconds long,
    cardPresentCoveragePercent double,
    unknownDurationSeconds long,
    unknownCoveragePercent double,
    previousDrivingSourceIntervalId string,
    nextDrivingSourceIntervalId string,
    previousRegistrationKey string,
    nextRegistrationKey string,
    previousVehicleKey string,
    nextVehicleKey string
);

insert into SignificantDrivingInterval
select
    sessionId,
    driverKey,
    firstSourceIntervalId,
    lastSourceIntervalId,
    startedAtEpochSecond,
    endedAtEpochSecond,
    durationSeconds,
    registrationKey,
    vehicleKey
from TachographActivityIntervalInputEvent(activityType = 'DRIVE', durationSeconds > ${SIGNIFICANT_DRIVING_THRESHOLD_SECONDS});

create window PreviousSignificantDrivingInterval#unique(driverKey) as SignificantDrivingInterval;

on SignificantDrivingInterval as next
insert into DrivingInterruptionInterval
select
    priorInterval.sessionId as sessionId,
    priorInterval.driverKey as driverKey,
    priorInterval.endedAtEpochSecond as startedAtEpochSecond,
    next.startedAtEpochSecond as endedAtEpochSecond,
    next.startedAtEpochSecond - priorInterval.endedAtEpochSecond as durationSeconds,
    priorInterval.lastSourceIntervalId as previousDrivingSourceIntervalId,
    next.firstSourceIntervalId as nextDrivingSourceIntervalId,
    priorInterval.registrationKey as previousRegistrationKey,
    next.registrationKey as nextRegistrationKey,
    priorInterval.vehicleKey as previousVehicleKey,
    next.vehicleKey as nextVehicleKey
from PreviousSignificantDrivingInterval as priorInterval
where priorInterval.driverKey = next.driverKey
  and next.startedAtEpochSecond > priorInterval.endedAtEpochSecond;

@Priority(20)
on SignificantDrivingInterval
delete from PreviousSignificantDrivingInterval;

@Priority(10)
on SignificantDrivingInterval as current
insert into PreviousSignificantDrivingInterval
select *;

insert into DailyWeeklyRestCandidateInterval
select *
from DrivingInterruptionInterval(durationSeconds > ${MINIMUM_REST_PERIOD_THRESHOLD_SECONDS});

insert into DrivingInterruptionVehicleChangeInterval
select *
from DailyWeeklyRestCandidateInterval(
    previousRegistrationKey is not null,
    nextRegistrationKey is not null,
    previousRegistrationKey != nextRegistrationKey
);

insert into DrivingInterruptionVehicleNotChangedInterval
select *
from DailyWeeklyRestCandidateInterval(
    previousRegistrationKey is not null,
    nextRegistrationKey is not null,
    previousRegistrationKey = nextRegistrationKey
);

insert into DailyWeeklyRestCandidateCoverageUnknownResolvedInterval
select
    c.sessionId as sessionId,
    c.driverKey as driverKey,
    c.startedAtEpochSecond as startedAtEpochSecond,
    c.endedAtEpochSecond as endedAtEpochSecond,
    c.durationSeconds as durationSeconds,
    sum(
        case
            when u.startedAtEpochSecond <= c.startedAtEpochSecond and u.endedAtEpochSecond >= c.endedAtEpochSecond
                then c.durationSeconds
            when u.startedAtEpochSecond <= c.startedAtEpochSecond
                then u.endedAtEpochSecond - c.startedAtEpochSecond
            when u.endedAtEpochSecond >= c.endedAtEpochSecond
                then c.endedAtEpochSecond - u.startedAtEpochSecond
            else u.endedAtEpochSecond - u.startedAtEpochSecond
        end
    ) as unknownDurationSeconds,
    c.previousDrivingSourceIntervalId as previousDrivingSourceIntervalId,
    c.nextDrivingSourceIntervalId as nextDrivingSourceIntervalId,
    c.previousRegistrationKey as previousRegistrationKey,
    c.nextRegistrationKey as nextRegistrationKey,
    c.previousVehicleKey as previousVehicleKey,
    c.nextVehicleKey as nextVehicleKey
from DailyWeeklyRestCandidateInterval as c unidirectional,
     VuCardAbsentInterval#keepall as u
where u.driverKey = c.driverKey
  and u.startedAtEpochSecond < c.endedAtEpochSecond
  and u.endedAtEpochSecond > c.startedAtEpochSecond
group by
    c.sessionId,
    c.driverKey,
    c.startedAtEpochSecond,
    c.endedAtEpochSecond,
    c.durationSeconds,
    c.previousDrivingSourceIntervalId,
    c.nextDrivingSourceIntervalId,
    c.previousRegistrationKey,
    c.nextRegistrationKey,
    c.previousVehicleKey,
    c.nextVehicleKey;

insert into DailyWeeklyRestCandidateCoverageUnknownResolvedInterval
select
    c.sessionId as sessionId,
    c.driverKey as driverKey,
    c.startedAtEpochSecond as startedAtEpochSecond,
    c.endedAtEpochSecond as endedAtEpochSecond,
    c.durationSeconds as durationSeconds,
    0L as unknownDurationSeconds,
    c.previousDrivingSourceIntervalId as previousDrivingSourceIntervalId,
    c.nextDrivingSourceIntervalId as nextDrivingSourceIntervalId,
    c.previousRegistrationKey as previousRegistrationKey,
    c.nextRegistrationKey as nextRegistrationKey,
    c.previousVehicleKey as previousVehicleKey,
    c.nextVehicleKey as nextVehicleKey
from DailyWeeklyRestCandidateInterval as c
where not exists (
    select * from VuCardAbsentInterval#keepall as u
    where u.driverKey = c.driverKey
      and u.startedAtEpochSecond < c.endedAtEpochSecond
      and u.endedAtEpochSecond > c.startedAtEpochSecond
);

insert into DailyWeeklyRestCandidateCoverageCardResolvedInterval
select
    c.sessionId as sessionId,
    c.driverKey as driverKey,
    c.startedAtEpochSecond as startedAtEpochSecond,
    c.endedAtEpochSecond as endedAtEpochSecond,
    c.durationSeconds as durationSeconds,
    sum(
        case
            when v.startedAtEpochSecond <= c.startedAtEpochSecond and (v.endedAtEpochSecond is null or v.endedAtEpochSecond >= c.endedAtEpochSecond)
                then c.durationSeconds
            when v.startedAtEpochSecond <= c.startedAtEpochSecond
                then v.endedAtEpochSecond - c.startedAtEpochSecond
            when v.endedAtEpochSecond is null or v.endedAtEpochSecond >= c.endedAtEpochSecond
                then c.endedAtEpochSecond - v.startedAtEpochSecond
            else v.endedAtEpochSecond - v.startedAtEpochSecond
        end
    ) as cardPresentDurationSeconds,
    c.unknownDurationSeconds as unknownDurationSeconds,
    c.previousDrivingSourceIntervalId as previousDrivingSourceIntervalId,
    c.nextDrivingSourceIntervalId as nextDrivingSourceIntervalId,
    c.previousRegistrationKey as previousRegistrationKey,
    c.nextRegistrationKey as nextRegistrationKey,
    c.previousVehicleKey as previousVehicleKey,
    c.nextVehicleKey as nextVehicleKey
from DailyWeeklyRestCandidateCoverageUnknownResolvedInterval as c unidirectional,
     TachographVehicleUsageIntervalInputEvent#keepall as v
where v.driverKey = c.driverKey
  and v.startedAtEpochSecond < c.endedAtEpochSecond
  and (v.endedAtEpochSecond is null or v.endedAtEpochSecond > c.startedAtEpochSecond)
group by
    c.sessionId,
    c.driverKey,
    c.startedAtEpochSecond,
    c.endedAtEpochSecond,
    c.durationSeconds,
    c.unknownDurationSeconds,
    c.previousDrivingSourceIntervalId,
    c.nextDrivingSourceIntervalId,
    c.previousRegistrationKey,
    c.nextRegistrationKey,
    c.previousVehicleKey,
    c.nextVehicleKey;

insert into DailyWeeklyRestCandidateCoverageCardResolvedInterval
select
    c.sessionId as sessionId,
    c.driverKey as driverKey,
    c.startedAtEpochSecond as startedAtEpochSecond,
    c.endedAtEpochSecond as endedAtEpochSecond,
    c.durationSeconds as durationSeconds,
    0L as cardPresentDurationSeconds,
    c.unknownDurationSeconds as unknownDurationSeconds,
    c.previousDrivingSourceIntervalId as previousDrivingSourceIntervalId,
    c.nextDrivingSourceIntervalId as nextDrivingSourceIntervalId,
    c.previousRegistrationKey as previousRegistrationKey,
    c.nextRegistrationKey as nextRegistrationKey,
    c.previousVehicleKey as previousVehicleKey,
    c.nextVehicleKey as nextVehicleKey
from DailyWeeklyRestCandidateCoverageUnknownResolvedInterval as c
where not exists (
    select * from TachographVehicleUsageIntervalInputEvent#keepall as v
    where v.driverKey = c.driverKey
      and v.startedAtEpochSecond < c.endedAtEpochSecond
      and (v.endedAtEpochSecond is null or v.endedAtEpochSecond > c.startedAtEpochSecond)
);

insert into DailyWeeklyRestCandidateCoverageInterval
select
    sessionId,
    driverKey,
    startedAtEpochSecond,
    endedAtEpochSecond,
    durationSeconds,
    cardPresentDurationSeconds,
    (cardPresentDurationSeconds * 100.0d) / durationSeconds as cardPresentCoveragePercent,
    unknownDurationSeconds,
    (unknownDurationSeconds * 100.0d) / durationSeconds as unknownCoveragePercent,
    previousDrivingSourceIntervalId,
    nextDrivingSourceIntervalId,
    previousRegistrationKey,
    nextRegistrationKey,
    previousVehicleKey,
    nextVehicleKey
from DailyWeeklyRestCandidateCoverageCardResolvedInterval;

context PerDriver
create window PreviousVehicleUsageInterval#lastevent as TachographVehicleUsageIntervalInputEvent;

@Priority(30)
context PerDriver
on TachographVehicleUsageIntervalInputEvent as next
insert into VuCardAbsentInterval
select
    priorInterval.sessionId as sessionId,
    priorInterval.driverKey as driverKey,
    priorInterval.endedAtEpochSecond + 1L as startedAtEpochSecond,
    next.startedAtEpochSecond as endedAtEpochSecond,
    next.startedAtEpochSecond - (priorInterval.endedAtEpochSecond + 1L) as durationSeconds,
    priorInterval.lastSourceIntervalId as previousUsageIntervalId,
    next.firstSourceIntervalId as nextUsageIntervalId,
    priorInterval.registrationKey as previousRegistrationKey,
    next.registrationKey as nextRegistrationKey,
    priorInterval.vehicleKey as previousVehicleKey,
    next.vehicleKey as nextVehicleKey
from PreviousVehicleUsageInterval as priorInterval
where priorInterval.endedAt is not null
  and next.startedAt is not null
  and next.startedAtEpochSecond > priorInterval.endedAtEpochSecond + 1L;

@Priority(20)
context PerDriver
on TachographVehicleUsageIntervalInputEvent
delete from PreviousVehicleUsageInterval;

@Priority(10)
context PerDriver
on TachographVehicleUsageIntervalInputEvent as current
insert into PreviousVehicleUsageInterval
select *;

insert into PotentialHomeOvernightStayInterval
select
    c.sessionId as sessionId,
    c.driverKey as driverKey,
    c.startedAtEpochSecond as startedAtEpochSecond,
    c.endedAtEpochSecond as endedAtEpochSecond,
    c.durationSeconds as durationSeconds,
    c.unknownDurationSeconds as unknownDurationSeconds,
    c.unknownCoveragePercent as unknownCoveragePercent,
    c.previousDrivingSourceIntervalId as previousDrivingSourceIntervalId,
    c.nextDrivingSourceIntervalId as nextDrivingSourceIntervalId,
    c.previousRegistrationKey as previousRegistrationKey,
    c.nextRegistrationKey as nextRegistrationKey,
    c.previousVehicleKey as previousVehicleKey,
    c.nextVehicleKey as nextVehicleKey
from DailyWeeklyRestCandidateCoverageInterval as c
where c.previousRegistrationKey is not null
  and c.nextRegistrationKey is not null
  and c.previousRegistrationKey != c.nextRegistrationKey
  and c.unknownDurationSeconds * 100L >= c.durationSeconds * 95L;

insert into PotentialInVehicleOvernightStayInterval
select
    c.sessionId as sessionId,
    c.driverKey as driverKey,
    c.startedAtEpochSecond as startedAtEpochSecond,
    c.endedAtEpochSecond as endedAtEpochSecond,
    c.durationSeconds as durationSeconds,
    c.cardPresentDurationSeconds as cardPresentDurationSeconds,
    c.cardPresentCoveragePercent as cardPresentCoveragePercent,
    c.previousDrivingSourceIntervalId as previousDrivingSourceIntervalId,
    c.nextDrivingSourceIntervalId as nextDrivingSourceIntervalId,
    c.previousRegistrationKey as previousRegistrationKey,
    c.nextRegistrationKey as nextRegistrationKey,
    c.previousVehicleKey as previousVehicleKey,
    c.nextVehicleKey as nextVehicleKey
from DailyWeeklyRestCandidateCoverageInterval as c
where c.previousRegistrationKey is not null
  and c.nextRegistrationKey is not null
  and c.previousRegistrationKey = c.nextRegistrationKey
  and c.cardPresentDurationSeconds >= c.durationSeconds;

insert into UnclassifiedDailyWeeklyRestCandidateCoverageInterval
select
    c.sessionId as sessionId,
    c.driverKey as driverKey,
    c.startedAtEpochSecond as startedAtEpochSecond,
    c.endedAtEpochSecond as endedAtEpochSecond,
    c.durationSeconds as durationSeconds,
    c.cardPresentDurationSeconds as cardPresentDurationSeconds,
    c.cardPresentCoveragePercent as cardPresentCoveragePercent,
    c.unknownDurationSeconds as unknownDurationSeconds,
    c.unknownCoveragePercent as unknownCoveragePercent,
    c.previousDrivingSourceIntervalId as previousDrivingSourceIntervalId,
    c.nextDrivingSourceIntervalId as nextDrivingSourceIntervalId,
    c.previousRegistrationKey as previousRegistrationKey,
    c.nextRegistrationKey as nextRegistrationKey,
    c.previousVehicleKey as previousVehicleKey,
    c.nextVehicleKey as nextVehicleKey
from DailyWeeklyRestCandidateCoverageInterval as c
where not (
        c.previousRegistrationKey is not null
    and c.nextRegistrationKey is not null
    and c.previousRegistrationKey != c.nextRegistrationKey
    and c.unknownDurationSeconds * 100L >= c.durationSeconds * 95L
)
  and not (
        c.previousRegistrationKey is not null
    and c.nextRegistrationKey is not null
    and c.previousRegistrationKey = c.nextRegistrationKey
    and c.cardPresentDurationSeconds >= c.durationSeconds
);

@name('drivingInterruptionIntervals')
select * from DrivingInterruptionInterval;

@name('dailyWeeklyRestCandidateIntervals')
select * from DailyWeeklyRestCandidateInterval;

@name('dailyWeeklyRestCandidateCoverageIntervals')
select * from DailyWeeklyRestCandidateCoverageInterval;

@name('drivingInterruptionVehicleChangeIntervals')
select * from DrivingInterruptionVehicleChangeInterval;

@name('vuCardAbsentIntervals')
select * from VuCardAbsentInterval;

@name('potentialHomeOvernightStayIntervals')
select * from PotentialHomeOvernightStayInterval;

@name('potentialInVehicleOvernightStayIntervals')
select * from PotentialInVehicleOvernightStayInterval;

@name('unclassifiedDailyWeeklyRestCandidateCoverageIntervals')
select * from UnclassifiedDailyWeeklyRestCandidateCoverageInterval;