Dual-Constraint Sessionization: 30-min Inactivity AND 4-Hour Absolute Hard Cap

WITH RECURSIVE numbered_events AS
  (SELECT user_id,
          event_time,
          row_number() OVER (PARTITION BY user_id
                             ORDER BY event_time ASC) AS rn
   FROM clickstream_events),
               sessionized_stream AS
  (SELECT user_id,
          event_time,
          rn,
          event_time AS session_start_time,
          event_time AS prev_event_time,
          1::BIGINT AS session_id
   FROM numbered_events
   WHERE rn = 1
   UNION ALL SELECT curr.user_id,
                    curr.event_time,
                    curr.rn,
                    CASE
                        WHEN curr.event_time - prev.prev_event_time > INTERVAL '30 minutes'
                             OR curr.event_time - prev.session_start_time > INTERVAL '4 hours' THEN curr.event_time
                        ELSE prev.session_start_time
                    END AS session_start_time,
                    curr.event_time AS prev_event_time,
                    CASE
                        WHEN curr.event_time - prev.prev_event_time > INTERVAL '30 minutes'
                             OR curr.event_time - prev.session_start_time > INTERVAL '4 hours' THEN prev.session_id + 1
                        ELSE prev.session_id
                    END AS session_id
   FROM numbered_events curr
   JOIN sessionized_stream prev ON curr.user_id = prev.user_id
   AND curr.rn = prev.rn + 1)
SELECT user_id,
       event_time,
       session_id
FROM sessionized_stream
ORDER BY user_id ASC,
         event_time ASC;