create variable long operatingSplitIdleMs = ${operatingSplitIdleMs};

/*
 * Full-EPL operating-period state machine.
 *
 * Input contract:
 * - Java sends already-resolved intervals, not raw START/END boundaries.
 * - DRIVER_CARD remains authoritative and VU is only used to fill uncovered gaps before events reach this EPL.
 * - Synthetic uncovered gaps arrive as activityType = 'UNKNOWN'.
 *
 * Output contract:
 * - PeriodizedActivityInterval: every input interval assigned to an operating period
 * - OperatingPeriodClosed: closed operating periods, including the final FLUSH period
 */

create schema KnownOperatingInput(
    driverId java.util.UUID,
    vehicleId java.util.UUID,
    vehicleRegistrationId java.util.UUID,
    activityType string,
    cardSlot string,
    cardStatus string,
    drivingStatus string,
    sourceKind string,
    startTs long,
    endTs long,
    durationMs long,
    sourceRowId string,
    sourceRowIds java.util.List,
    clippedToRequestedPeriod boolean,
    synthetic boolean
);

create schema UnknownOperatingInput(
    driverId java.util.UUID,
    vehicleId java.util.UUID,
    vehicleRegistrationId java.util.UUID,
    activityType string,
    cardSlot string,
    cardStatus string,
    drivingStatus string,
    sourceKind string,
    startTs long,
    endTs long,
    durationMs long,
    sourceRowId string,
    sourceRowIds java.util.List,
    clippedToRequestedPeriod boolean,
    synthetic boolean
);

create schema PeriodizedActivityInterval(
    driverId java.util.UUID,
    vehicleId java.util.UUID,
    vehicleRegistrationId java.util.UUID,
    activityType string,
    cardSlot string,
    cardStatus string,
    drivingStatus string,
    sourceKind string,
    startedAtTs long,
    endedAtTs long,
    durationMs long,
    sourceRowId string,
    sourceRowIds java.util.List,
    clippedToRequestedPeriod boolean,
    operatingPeriodNo long,
    operatingPeriodStartedAtTs long,
    newOperatingPeriod boolean,
    gapSincePreviousActivityMs java.lang.Long,
    synthetic boolean
);

create schema OperatingPeriodClosed(
    driverId java.util.UUID,
    operatingPeriodNo long,
    operatingPeriodStartedAtTs long,
    operatingPeriodEndedAtTs long,
    durationMs long,
    closedBy string
);

create window OperatingPeriodState#unique(driverId) as (
    driverId java.util.UUID,
    hasOpen boolean,
    operatingPeriodNo long,
    operatingPeriodStartedAtTs long,
    lastKnownActivityEndTs long
);

/* Split the timeline into known activities and synthetic UNKNOWN gaps. */
insert into KnownOperatingInput
select * from OperatingPeriodInputMap as i where i.activityType != 'UNKNOWN';

insert into UnknownOperatingInput
select * from OperatingPeriodInputMap as i where i.activityType = 'UNKNOWN';

/* First known activity for a driver opens operating period 1. */
@Priority(200)
on KnownOperatingInput as i
insert into PeriodizedActivityInterval
select
    i.driverId as driverId,
    i.vehicleId as vehicleId,
    i.vehicleRegistrationId as vehicleRegistrationId,
    i.activityType as activityType,
    i.cardSlot as cardSlot,
    i.cardStatus as cardStatus,
    i.drivingStatus as drivingStatus,
    i.sourceKind as sourceKind,
    i.startTs as startedAtTs,
    i.endTs as endedAtTs,
    i.durationMs as durationMs,
    i.sourceRowId as sourceRowId,
    i.sourceRowIds as sourceRowIds,
    i.clippedToRequestedPeriod as clippedToRequestedPeriod,
    1L as operatingPeriodNo,
    i.startTs as operatingPeriodStartedAtTs,
    true as newOperatingPeriod,
    cast(null, java.lang.Long) as gapSincePreviousActivityMs,
    i.synthetic as synthetic
where not exists (select * from OperatingPeriodState as s where s.driverId = i.driverId);

@Priority(190)
on KnownOperatingInput as i
insert into OperatingPeriodState
select
    i.driverId as driverId,
    true as hasOpen,
    1L as operatingPeriodNo,
    i.startTs as operatingPeriodStartedAtTs,
    i.endTs as lastKnownActivityEndTs
where not exists (select * from OperatingPeriodState as s where s.driverId = i.driverId);

/* A long forward gap between known activities closes the current period with reason IDLE_GAP. */
@Priority(180)
on KnownOperatingInput as i
insert into OperatingPeriodClosed
select
    s.driverId as driverId,
    s.operatingPeriodNo as operatingPeriodNo,
    s.operatingPeriodStartedAtTs as operatingPeriodStartedAtTs,
    s.lastKnownActivityEndTs as operatingPeriodEndedAtTs,
    s.lastKnownActivityEndTs - s.operatingPeriodStartedAtTs as durationMs,
    'IDLE_GAP' as closedBy
from OperatingPeriodState as s
where s.driverId = i.driverId
  and s.hasOpen = true
  and i.startTs - s.lastKnownActivityEndTs >= operatingSplitIdleMs;

/* After a long idle gap, the next known interval is emitted as the first interval of the next period. */
@Priority(170)
on KnownOperatingInput as i
insert into PeriodizedActivityInterval
select
    i.driverId as driverId,
    i.vehicleId as vehicleId,
    i.vehicleRegistrationId as vehicleRegistrationId,
    i.activityType as activityType,
    i.cardSlot as cardSlot,
    i.cardStatus as cardStatus,
    i.drivingStatus as drivingStatus,
    i.sourceKind as sourceKind,
    i.startTs as startedAtTs,
    i.endTs as endedAtTs,
    i.durationMs as durationMs,
    i.sourceRowId as sourceRowId,
    i.sourceRowIds as sourceRowIds,
    i.clippedToRequestedPeriod as clippedToRequestedPeriod,
    s.operatingPeriodNo + 1 as operatingPeriodNo,
    i.startTs as operatingPeriodStartedAtTs,
    true as newOperatingPeriod,
    cast(i.startTs - s.lastKnownActivityEndTs, java.lang.Long) as gapSincePreviousActivityMs,
    i.synthetic as synthetic
from OperatingPeriodState as s
where s.driverId = i.driverId
  and s.hasOpen = true
  and i.startTs - s.lastKnownActivityEndTs >= operatingSplitIdleMs;

/* Update the window state to the newly opened period after an IDLE_GAP close. */
@Priority(160)
on KnownOperatingInput as i
update OperatingPeriodState as s
set
    hasOpen = true,
    operatingPeriodNo = s.operatingPeriodNo + 1,
    operatingPeriodStartedAtTs = i.startTs,
    lastKnownActivityEndTs = i.endTs
where s.driverId = i.driverId
  and s.hasOpen = true
  and i.startTs - s.lastKnownActivityEndTs >= operatingSplitIdleMs;

/* After a long UNKNOWN gap we keep the counter but mark the period closed. The next known activity reopens
 * the next period number using the retained state row. */
@Priority(155)
on KnownOperatingInput as i
insert into PeriodizedActivityInterval
select
    i.driverId as driverId,
    i.vehicleId as vehicleId,
    i.vehicleRegistrationId as vehicleRegistrationId,
    i.activityType as activityType,
    i.cardSlot as cardSlot,
    i.cardStatus as cardStatus,
    i.drivingStatus as drivingStatus,
    i.sourceKind as sourceKind,
    i.startTs as startedAtTs,
    i.endTs as endedAtTs,
    i.durationMs as durationMs,
    i.sourceRowId as sourceRowId,
    i.sourceRowIds as sourceRowIds,
    i.clippedToRequestedPeriod as clippedToRequestedPeriod,
    s.operatingPeriodNo + 1 as operatingPeriodNo,
    i.startTs as operatingPeriodStartedAtTs,
    true as newOperatingPeriod,
    cast(null, java.lang.Long) as gapSincePreviousActivityMs,
    i.synthetic as synthetic
from OperatingPeriodState as s
where s.driverId = i.driverId
  and s.hasOpen = false;

@Priority(145)
on KnownOperatingInput as i
update OperatingPeriodState as s
set
    hasOpen = true,
    operatingPeriodNo = s.operatingPeriodNo + 1,
    operatingPeriodStartedAtTs = i.startTs,
    lastKnownActivityEndTs = i.endTs
where s.driverId = i.driverId
  and s.hasOpen = false;

/* Normal same-period continuity: the gap is forward, non-negative, and still below the split threshold. */
@Priority(150)
on KnownOperatingInput as i
insert into PeriodizedActivityInterval
select
    i.driverId as driverId,
    i.vehicleId as vehicleId,
    i.vehicleRegistrationId as vehicleRegistrationId,
    i.activityType as activityType,
    i.cardSlot as cardSlot,
    i.cardStatus as cardStatus,
    i.drivingStatus as drivingStatus,
    i.sourceKind as sourceKind,
    i.startTs as startedAtTs,
    i.endTs as endedAtTs,
    i.durationMs as durationMs,
    i.sourceRowId as sourceRowId,
    i.sourceRowIds as sourceRowIds,
    i.clippedToRequestedPeriod as clippedToRequestedPeriod,
    s.operatingPeriodNo as operatingPeriodNo,
    s.operatingPeriodStartedAtTs as operatingPeriodStartedAtTs,
    false as newOperatingPeriod,
    cast(i.startTs - s.lastKnownActivityEndTs, java.lang.Long) as gapSincePreviousActivityMs,
    i.synthetic as synthetic
from OperatingPeriodState as s
where s.driverId = i.driverId
  and s.hasOpen = true
  and i.startTs - s.lastKnownActivityEndTs >= 0
  and i.startTs - s.lastKnownActivityEndTs < operatingSplitIdleMs;

@Priority(140)
on KnownOperatingInput as i
update OperatingPeriodState as s
set
    lastKnownActivityEndTs = case
        when i.endTs > s.lastKnownActivityEndTs then i.endTs
        else s.lastKnownActivityEndTs
    end
where s.driverId = i.driverId
  and s.hasOpen = true
  and i.startTs - s.lastKnownActivityEndTs >= 0
  and i.startTs - s.lastKnownActivityEndTs < operatingSplitIdleMs;

/* A long UNKNOWN interval closes the current period, but the state row remains so the next known activity
 * can reopen with the next period number. */
@Priority(130)
on UnknownOperatingInput as i
insert into OperatingPeriodClosed
select
    s.driverId as driverId,
    s.operatingPeriodNo as operatingPeriodNo,
    s.operatingPeriodStartedAtTs as operatingPeriodStartedAtTs,
    s.lastKnownActivityEndTs as operatingPeriodEndedAtTs,
    s.lastKnownActivityEndTs - s.operatingPeriodStartedAtTs as durationMs,
    'UNKNOWN_GAP' as closedBy
from OperatingPeriodState as s
where s.driverId = i.driverId
  and s.hasOpen = true
  and i.durationMs >= operatingSplitIdleMs;

@Priority(120)
on UnknownOperatingInput as i
update OperatingPeriodState as s
set hasOpen = false
where s.driverId = i.driverId
  and s.hasOpen = true
  and i.durationMs >= operatingSplitIdleMs;

/* Short UNKNOWN stays inside the open period as explicit uncertainty, without changing period state. */
@Priority(110)
on UnknownOperatingInput as i
insert into PeriodizedActivityInterval
select
    i.driverId as driverId,
    i.vehicleId as vehicleId,
    i.vehicleRegistrationId as vehicleRegistrationId,
    i.activityType as activityType,
    i.cardSlot as cardSlot,
    i.cardStatus as cardStatus,
    i.drivingStatus as drivingStatus,
    i.sourceKind as sourceKind,
    i.startTs as startedAtTs,
    i.endTs as endedAtTs,
    i.durationMs as durationMs,
    i.sourceRowId as sourceRowId,
    i.sourceRowIds as sourceRowIds,
    i.clippedToRequestedPeriod as clippedToRequestedPeriod,
    s.operatingPeriodNo as operatingPeriodNo,
    s.operatingPeriodStartedAtTs as operatingPeriodStartedAtTs,
    false as newOperatingPeriod,
    cast(i.startTs - s.lastKnownActivityEndTs, java.lang.Long) as gapSincePreviousActivityMs,
    i.synthetic as synthetic
from OperatingPeriodState as s
where s.driverId = i.driverId
  and s.hasOpen = true
  and i.durationMs < operatingSplitIdleMs;

/* Historical evaluation ends with a flush event so the final still-open period is emitted. */
@Priority(100)
on OperatingPeriodFlushEvent as f
insert into OperatingPeriodClosed
select
    s.driverId as driverId,
    s.operatingPeriodNo as operatingPeriodNo,
    s.operatingPeriodStartedAtTs as operatingPeriodStartedAtTs,
    s.lastKnownActivityEndTs as operatingPeriodEndedAtTs,
    s.lastKnownActivityEndTs - s.operatingPeriodStartedAtTs as durationMs,
    'FLUSH' as closedBy
from OperatingPeriodState as s
where s.hasOpen = true;

@Priority(90)
on OperatingPeriodFlushEvent as f
update OperatingPeriodState as s
set hasOpen = false
where s.hasOpen = true;

/* Listener-facing output statements consumed by Java. */
@name('periodizedActivityIntervals')
select * from PeriodizedActivityInterval;

@name('operatingPeriodClosed')
select * from OperatingPeriodClosed;
