fleet-platform/db/migrations/20260601000019_fn_vehicle_trips.sql

-- migrate:up
--
-- Trip detection on demand from state.position_history.
--
-- Algorithm (state machine, single forward pass over the day's positions):
--
--   reporting_time := first occurred_at where acc_state = 1
--
--   A trip starts at:
--     - an ACC_ON transition (or the first row of the day if it's already ACC_ON)
--   A trip ends at:
--     - ACC_OFF + stationary (speed < 5 km/h) for >= 5 min  →  end_reason='work_stop'
--     - a fix-reporting silence of >= 5 min                 →  end_reason='nofix_stop'
--     - a fix gap > 30 min                                  →  end_reason='long_gap'
--     - the end of the day's data                           →  end_reason='day_end'
--
-- The nofix_stop rule matches legacy dispatcher semantics: when a polled
-- device goes silent for >=5 min mid-trip we assume the engine is off
-- (validated against legacy 638J full-day: 15 trips legacy = 15 trips here).
-- Pure stop-and-go traffic patterns still consolidate cleanly because each
-- stop produces explicit slow/stationary fixes, not silences.
--
--   Within a trip, an ACC_ON + stationary stretch of >= 5 min is recorded
--   as an idling segment (no trip split — engine still running, treated
--   as a customer-stop with engine on).
--
--   Distance accumulates per moving step (curr.speed_kmh >= 5 km/h);
--   GPS jitter at standstill is excluded.
--
-- Day totals (driving_min / idling_min / stopped_min / unknown_min) are
-- bucketed by the classification of the *previous* fix at each step.
--
-- Falls back to movement-only segmentation (has_acc_data=false in the
-- response) when every position for the day has acc_state IS NULL.

-- Drop any prior signature; the function identity is (name + arg types),
-- so a CREATE OR REPLACE with a different arg type would create a sibling.
DROP FUNCTION IF EXISTS serve.fn_vehicle_trips(integer, date);
DROP FUNCTION IF EXISTS serve.fn_vehicle_trips(bigint, date);

CREATE OR REPLACE FUNCTION serve.fn_vehicle_trips(
    p_vehicle_id bigint,
    p_date_eat   date
) RETURNS jsonb
LANGUAGE plpgsql STABLE
AS $fn$
DECLARE
    EAT_OFFSET    interval := interval '3 hours';
    STOP_THRESH   interval := interval '5 minutes';
    NOFIX_THRESH  interval := interval '5 minutes';
    GAP_THRESH    interval := interval '30 minutes';
    STAT_KMH      numeric  := 5;

    day_start_utc timestamptz := (p_date_eat::timestamp - EAT_OFFSET) AT TIME ZONE 'UTC';
    day_end_utc   timestamptz := day_start_utc + interval '24 hours';

    rec record;
    v_plate text;
    v_imei  text;

    reporting_time timestamptz;
    has_acc bool := false;

    -- trip-in-progress state
    in_trip          bool        := false;
    trip_started_at  timestamptz;
    trip_started_geom geometry;
    trip_path        geometry[]  := ARRAY[]::geometry[];
    trip_path_times  timestamptz[] := ARRAY[]::timestamptz[];
    trip_path_speeds numeric[]   := ARRAY[]::numeric[];
    trip_distance_m  numeric     := 0;
    trip_idling_sec  numeric     := 0;
    trip_stops       jsonb       := '[]'::jsonb;

    -- mid-trip ACC_OFF run (might become a work stop)
    off_run_start    timestamptz;
    off_run_geom     geometry;
    -- mid-trip idle (ACC_ON + stationary) run
    idle_run_start   timestamptz;
    idle_run_geom    geometry;

    prev_at          timestamptz;
    prev_geom        geometry;
    prev_state       text;        -- 'moving' | 'idling' | 'stopped' | 'unknown'

    trips_out        jsonb := '[]'::jsonb;
    n_trips          int   := 0;

    total_distance_m  numeric := 0;
    total_driving_sec numeric := 0;
    total_idling_sec  numeric := 0;
    total_stopped_sec numeric := 0;
    total_unknown_sec numeric := 0;
    longest_gap_sec   numeric := 0;
    fix_count         int     := 0;

    pos_state        text;
    step_sec         numeric;
    step_m           numeric;

    -- closure helper inline (PL/pgSQL has no nested funcs, so we inline)
BEGIN
    SELECT v.plate INTO v_plate FROM domain.vehicles v WHERE v.vehicle_id = p_vehicle_id;
    IF v_plate IS NULL THEN
        RETURN jsonb_build_object(
            'error', 'vehicle not found',
            'vehicle_id', p_vehicle_id,
            'date', to_char(p_date_eat, 'YYYY-MM-DD')
        );
    END IF;

    SELECT d.imei INTO v_imei
      FROM domain.devices d
     WHERE d.vehicle_id = p_vehicle_id
     ORDER BY CASE d.device_type WHEN 'tracker' THEN 0 ELSE 1 END, d.imei
     LIMIT 1;

    FOR rec IN
        SELECT occurred_at,
               geom,
               COALESCE(speed_kmh, 0)::numeric AS speed_kmh,
               acc_state
          FROM state.position_history
         WHERE vehicle_id = p_vehicle_id
           AND occurred_at >= day_start_utc
           AND occurred_at <  day_end_utc
         ORDER BY occurred_at
    LOOP
        fix_count := fix_count + 1;

        IF rec.acc_state IS NOT NULL THEN
            has_acc := true;
        END IF;

        IF rec.acc_state = 1 AND reporting_time IS NULL THEN
            reporting_time := rec.occurred_at;
        END IF;

        -- nofix_stop: if mid-trip and the prior fix was >= NOFIX_THRESH ago
        -- but < GAP_THRESH, treat the silence as engine-off and close the
        -- trip at prev_at before processing this row.
        IF in_trip
           AND prev_at IS NOT NULL
           AND rec.occurred_at - prev_at >= NOFIX_THRESH
           AND rec.occurred_at - prev_at <= GAP_THRESH
        THEN
            trips_out := trips_out || jsonb_build_object(
                'trip_id', n_trips + 1,
                'started_at', trip_started_at,
                'ended_at',   prev_at,
                'duration_min', round(EXTRACT(EPOCH FROM (prev_at - trip_started_at))/60.0, 1),
                'distance_km', round((trip_distance_m / 1000.0)::numeric, 2),
                'idling_min',  round((trip_idling_sec / 60.0)::numeric, 1),
                'end_reason',  'nofix_stop',
                'stops',       trip_stops,
                'path',        CASE WHEN array_length(trip_path,1) >= 2
                                    THEN ST_AsGeoJSON(ST_MakeLine(trip_path))::jsonb
                                    ELSE NULL END
            );
            n_trips := n_trips + 1;
            in_trip := false;
            trip_path := ARRAY[]::geometry[];
            trip_path_times := ARRAY[]::timestamptz[];
            trip_path_speeds := ARRAY[]::numeric[];
            trip_distance_m := 0;
            trip_idling_sec := 0;
            trip_stops := '[]'::jsonb;
            off_run_start := NULL;
            idle_run_start := NULL;
        END IF;

        -- Classify this position
        IF prev_at IS NOT NULL AND rec.occurred_at - prev_at > GAP_THRESH THEN
            pos_state := 'unknown';
        ELSIF rec.speed_kmh >= STAT_KMH THEN
            pos_state := 'moving';
        ELSIF rec.acc_state = 1 THEN
            pos_state := 'idling';
        ELSIF rec.acc_state = 0 THEN
            pos_state := 'stopped';
        ELSE
            -- no acc data: fall back to speed-only
            pos_state := CASE WHEN rec.speed_kmh >= STAT_KMH THEN 'moving' ELSE 'stopped' END;
        END IF;

        -- Bucket the time *since* prev_at into totals
        IF prev_at IS NOT NULL THEN
            step_sec := EXTRACT(EPOCH FROM (rec.occurred_at - prev_at));
            longest_gap_sec := GREATEST(longest_gap_sec, step_sec);
            CASE prev_state
                WHEN 'moving'  THEN total_driving_sec := total_driving_sec + step_sec;
                WHEN 'idling'  THEN total_idling_sec  := total_idling_sec  + step_sec;
                WHEN 'stopped' THEN total_stopped_sec := total_stopped_sec + step_sec;
                ELSE                total_unknown_sec := total_unknown_sec + step_sec;
            END CASE;

            IF in_trip AND prev_state = 'moving' AND pos_state IN ('moving','idling') THEN
                step_m := ST_Distance(prev_geom::geography, rec.geom::geography);
                trip_distance_m := trip_distance_m + step_m;
                total_distance_m := total_distance_m + step_m;
            ELSIF prev_state = 'moving' THEN
                step_m := ST_Distance(prev_geom::geography, rec.geom::geography);
                total_distance_m := total_distance_m + step_m;
            END IF;
        END IF;

        ----------------------------------------------------------------------
        -- State machine
        ----------------------------------------------------------------------

        IF pos_state = 'unknown' THEN
            -- Gap longer than GAP_THRESH; close any open trip.
            IF in_trip THEN
                trips_out := trips_out || jsonb_build_object(
                    'trip_id', n_trips + 1,
                    'started_at', trip_started_at,
                    'ended_at',   prev_at,
                    'duration_min', round(EXTRACT(EPOCH FROM (prev_at - trip_started_at))/60.0, 1),
                    'distance_km', round((trip_distance_m / 1000.0)::numeric, 2),
                    'idling_min',  round((trip_idling_sec / 60.0)::numeric, 1),
                    'end_reason',  'long_gap',
                    'stops',       trip_stops,
                    'path',        CASE WHEN array_length(trip_path,1) >= 2
                                        THEN ST_AsGeoJSON(ST_MakeLine(trip_path))::jsonb
                                        ELSE NULL END
                );
                n_trips := n_trips + 1;
                in_trip := false;
                trip_path := ARRAY[]::geometry[];
                trip_path_times := ARRAY[]::timestamptz[];
                trip_path_speeds := ARRAY[]::numeric[];
                trip_distance_m := 0;
                trip_idling_sec := 0;
                trip_stops := '[]'::jsonb;
                off_run_start := NULL;
                idle_run_start := NULL;
            END IF;

        ELSIF NOT in_trip THEN
            -- Outside any trip. Start one when we see motion or ACC_ON.
            IF pos_state IN ('moving','idling') OR rec.acc_state = 1 THEN
                in_trip := true;
                trip_started_at := rec.occurred_at;
                trip_started_geom := rec.geom;
                trip_path := ARRAY[rec.geom];
                trip_path_times := ARRAY[rec.occurred_at];
                trip_path_speeds := ARRAY[rec.speed_kmh];
                trip_distance_m := 0;
                trip_idling_sec := 0;
                trip_stops := '[]'::jsonb;
                off_run_start := NULL;
                idle_run_start := CASE WHEN pos_state = 'idling' THEN rec.occurred_at ELSE NULL END;
            END IF;

        ELSE
            -- In trip — append point, then handle stop/idle runs.
            trip_path := trip_path || rec.geom;
            trip_path_times := trip_path_times || rec.occurred_at;
            trip_path_speeds := trip_path_speeds || rec.speed_kmh;

            IF pos_state = 'stopped' THEN
                IF off_run_start IS NULL THEN
                    off_run_start := rec.occurred_at;
                    off_run_geom  := rec.geom;
                END IF;
                IF idle_run_start IS NOT NULL THEN
                    idle_run_start := NULL;
                END IF;

                IF rec.occurred_at - off_run_start >= STOP_THRESH THEN
                    -- Work stop confirmed: close the trip at off_run_start.
                    trips_out := trips_out || jsonb_build_object(
                        'trip_id', n_trips + 1,
                        'started_at', trip_started_at,
                        'ended_at',   off_run_start,
                        'duration_min', round(EXTRACT(EPOCH FROM (off_run_start - trip_started_at))/60.0, 1),
                        'distance_km', round((trip_distance_m / 1000.0)::numeric, 2),
                        'idling_min',  round((trip_idling_sec / 60.0)::numeric, 1),
                        'end_reason',  'work_stop',
                        'stops',       trip_stops,
                        'path',        CASE WHEN array_length(trip_path,1) >= 2
                                            THEN ST_AsGeoJSON(ST_MakeLine(trip_path))::jsonb
                                            ELSE NULL END
                    );
                    n_trips := n_trips + 1;
                    in_trip := false;
                    trip_path := ARRAY[]::geometry[];
                    trip_path_times := ARRAY[]::timestamptz[];
                    trip_path_speeds := ARRAY[]::numeric[];
                    trip_distance_m := 0;
                    trip_idling_sec := 0;
                    trip_stops := '[]'::jsonb;
                    off_run_start := NULL;
                END IF;

            ELSIF pos_state = 'idling' THEN
                IF off_run_start IS NOT NULL THEN
                    -- ACC came back on before STOP_THRESH; abandon the off-run.
                    off_run_start := NULL;
                END IF;
                IF idle_run_start IS NULL THEN
                    idle_run_start := rec.occurred_at;
                    idle_run_geom  := rec.geom;
                ELSIF rec.occurred_at - idle_run_start >= STOP_THRESH THEN
                    -- Already recorded as idling within trip; nothing to flush
                    -- until run ends.
                    NULL;
                END IF;

            ELSE  -- moving
                IF off_run_start IS NOT NULL THEN
                    off_run_start := NULL;
                END IF;
                IF idle_run_start IS NOT NULL THEN
                    -- Idle run ended; if it was long enough, record it.
                    IF rec.occurred_at - idle_run_start >= STOP_THRESH THEN
                        trip_idling_sec := trip_idling_sec
                            + EXTRACT(EPOCH FROM (rec.occurred_at - idle_run_start));
                        trip_stops := trip_stops || jsonb_build_object(
                            'at', idle_run_start,
                            'duration_min', round(EXTRACT(EPOCH FROM (rec.occurred_at - idle_run_start))/60.0, 1),
                            'kind', 'idling',
                            'lng', ST_X(idle_run_geom),
                            'lat', ST_Y(idle_run_geom)
                        );
                    END IF;
                    idle_run_start := NULL;
                END IF;
            END IF;
        END IF;

        prev_at    := rec.occurred_at;
        prev_geom  := rec.geom;
        prev_state := pos_state;
    END LOOP;

    -- Close any trip still open at end of loop.
    IF in_trip THEN
        -- If an unflushed idle run is open, account for it.
        IF idle_run_start IS NOT NULL AND prev_at - idle_run_start >= STOP_THRESH THEN
            trip_idling_sec := trip_idling_sec
                + EXTRACT(EPOCH FROM (prev_at - idle_run_start));
            trip_stops := trip_stops || jsonb_build_object(
                'at', idle_run_start,
                'duration_min', round(EXTRACT(EPOCH FROM (prev_at - idle_run_start))/60.0, 1),
                'kind', 'idling',
                'lng', ST_X(idle_run_geom),
                'lat', ST_Y(idle_run_geom)
            );
        END IF;
        trips_out := trips_out || jsonb_build_object(
            'trip_id', n_trips + 1,
            'started_at', trip_started_at,
            'ended_at',   prev_at,
            'duration_min', round(EXTRACT(EPOCH FROM (prev_at - trip_started_at))/60.0, 1),
            'distance_km', round((trip_distance_m / 1000.0)::numeric, 2),
            'idling_min',  round((trip_idling_sec / 60.0)::numeric, 1),
            'end_reason',  'day_end',
            'stops',       trip_stops,
            'path',        CASE WHEN array_length(trip_path,1) >= 2
                                THEN ST_AsGeoJSON(ST_MakeLine(trip_path))::jsonb
                                ELSE NULL END
        );
        n_trips := n_trips + 1;
    END IF;

    RETURN jsonb_build_object(
        'vehicle_id', p_vehicle_id,
        'plate', v_plate,
        'imei', v_imei,
        'date', to_char(p_date_eat, 'YYYY-MM-DD'),
        'reporting_time', reporting_time,
        'totals', jsonb_build_object(
            'distance_km', round((total_distance_m / 1000.0)::numeric, 2),
            'driving_min', round((total_driving_sec / 60.0)::numeric, 1),
            'idling_min',  round((total_idling_sec  / 60.0)::numeric, 1),
            'stopped_min', round((total_stopped_sec / 60.0)::numeric, 1),
            'unknown_min', round((total_unknown_sec / 60.0)::numeric, 1),
            'trip_count',  n_trips
        ),
        'data_quality', jsonb_build_object(
            'fix_count', fix_count,
            'has_acc_data', has_acc,
            'longest_gap_sec', round(longest_gap_sec::numeric, 0)
        ),
        'trips', trips_out
    );
END
$fn$;

-- migrate:down

DROP FUNCTION IF EXISTS serve.fn_vehicle_trips(bigint, date);
DROP FUNCTION IF EXISTS serve.fn_vehicle_trips(integer, date);