Traveler Lifecycle Stage

-- ==========================================
-- 1. GOOGLE BIGQUERY SQL
-- ==========================================
WITH reporting_dates AS (
  SELECT calendar_date AS snapshot_date
  FROM `your-project.reference.calendar_date`
  WHERE calendar_date BETWEEN DATE_SUB(CURRENT_DATE(), INTERVAL 89 DAY) AND CURRENT_DATE()
),
source_rows AS (
  SELECT d.snapshot_date,
    NULLIF(TRIM(CAST(h.traveler_analytics_key AS STRING)), '') AS traveler_partition_key,
    h.stage_code, h.previous_stage_code, h.rule_version, h.effective_from_date,
    h.classifier_updated_at, h.classifier_record_id
  FROM reporting_dates d
  JOIN `your-project.warehouse.traveler_lifecycle_stage_history` h
    ON d.snapshot_date >= h.effective_from_date
   AND (h.effective_to_date IS NULL OR d.snapshot_date < h.effective_to_date)
  WHERE NULLIF(TRIM(CAST(h.traveler_analytics_key AS STRING)), '') IS NOT NULL
),
stage_candidates AS (
  SELECT source_rows.*,
    ROW_NUMBER() OVER (
      PARTITION BY snapshot_date, traveler_partition_key
      ORDER BY effective_from_date DESC, classifier_updated_at DESC, classifier_record_id DESC
    ) AS stage_rank
  FROM source_rows
),
normalized AS (
  SELECT snapshot_date,
    REGEXP_REPLACE(LOWER(TRIM(stage_code)), r'[\s-]+', '_') AS stage_code,
    REGEXP_REPLACE(LOWER(TRIM(previous_stage_code)), r'[\s-]+', '_') AS previous_stage_code,
    COALESCE(NULLIF(TRIM(rule_version), ''), 'unknown') AS rule_version
  FROM stage_candidates
  WHERE stage_rank = 1
),
classified AS (
  SELECT snapshot_date, rule_version,
    CASE
      WHEN NULLIF(stage_code, '') IS NULL THEN 'unknown'
      WHEN stage_code IN ('anonymous_visitor', 'known_visitor', 'active_planner', 'booking_started', 'booked_pre_trip', 'in_trip', 'post_trip', 'repeat_traveler', 'lapsed_traveler', 'unknown') THEN stage_code
      ELSE 'unmapped'
    END AS traveler_lifecycle_stage,
    CASE
      WHEN NULLIF(previous_stage_code, '') IS NULL THEN 'unknown'
      WHEN previous_stage_code IN ('anonymous_visitor', 'known_visitor', 'active_planner', 'booking_started', 'booked_pre_trip', 'in_trip', 'post_trip', 'repeat_traveler', 'lapsed_traveler', 'unknown') THEN previous_stage_code
      ELSE 'unmapped'
    END AS previous_traveler_lifecycle_stage
  FROM normalized
)
SELECT snapshot_date, traveler_lifecycle_stage, previous_traveler_lifecycle_stage, rule_version, COUNT(*) AS traveler_count
FROM classified
GROUP BY snapshot_date, traveler_lifecycle_stage, previous_traveler_lifecycle_stage, rule_version
ORDER BY snapshot_date, traveler_lifecycle_stage, previous_traveler_lifecycle_stage;

-- ==========================================
-- 2. ADOBE AEP QUERY SERVICE SQL
-- ==========================================
WITH reporting_dates AS (
  SELECT calendar_date AS snapshot_date
  FROM enterprise_calendar
  WHERE calendar_date BETWEEN CURRENT_DATE - INTERVAL '89' DAY AND CURRENT_DATE
),
source_rows AS (
  SELECT d.snapshot_date,
    NULLIF(TRIM(CAST(h.traveler_analytics_key AS VARCHAR)), '') AS traveler_partition_key,
    h.stage_code, h.previous_stage_code, h.rule_version, h.effective_from_date,
    h.classifier_updated_at, h.classifier_record_id
  FROM reporting_dates d
  JOIN traveler_lifecycle_stage_history h
    ON d.snapshot_date >= h.effective_from_date
   AND (h.effective_to_date IS NULL OR d.snapshot_date < h.effective_to_date)
  WHERE NULLIF(TRIM(CAST(h.traveler_analytics_key AS VARCHAR)), '') IS NOT NULL
),
stage_candidates AS (
  SELECT source_rows.*,
    ROW_NUMBER() OVER (
      PARTITION BY snapshot_date, traveler_partition_key
      ORDER BY effective_from_date DESC, classifier_updated_at DESC, classifier_record_id DESC
    ) AS stage_rank
  FROM source_rows
),
normalized AS (
  SELECT snapshot_date,
    REGEXP_REPLACE(LOWER(TRIM(stage_code)), '[ -]+', '_') AS stage_code,
    REGEXP_REPLACE(LOWER(TRIM(previous_stage_code)), '[ -]+', '_') AS previous_stage_code,
    COALESCE(NULLIF(TRIM(rule_version), ''), 'unknown') AS rule_version
  FROM stage_candidates
  WHERE stage_rank = 1
),
classified AS (
  SELECT snapshot_date, rule_version,
    CASE
      WHEN stage_code IS NULL OR stage_code = '' THEN 'unknown'
      WHEN stage_code IN ('anonymous_visitor', 'known_visitor', 'active_planner', 'booking_started', 'booked_pre_trip', 'in_trip', 'post_trip', 'repeat_traveler', 'lapsed_traveler', 'unknown') THEN stage_code
      ELSE 'unmapped'
    END AS traveler_lifecycle_stage,
    CASE
      WHEN previous_stage_code IS NULL OR previous_stage_code = '' THEN 'unknown'
      WHEN previous_stage_code IN ('anonymous_visitor', 'known_visitor', 'active_planner', 'booking_started', 'booked_pre_trip', 'in_trip', 'post_trip', 'repeat_traveler', 'lapsed_traveler', 'unknown') THEN previous_stage_code
      ELSE 'unmapped'
    END AS previous_traveler_lifecycle_stage
  FROM normalized
)
SELECT snapshot_date, traveler_lifecycle_stage, previous_traveler_lifecycle_stage, rule_version, COUNT(*) AS traveler_count
FROM classified
GROUP BY snapshot_date, traveler_lifecycle_stage, previous_traveler_lifecycle_stage, rule_version
ORDER BY snapshot_date, traveler_lifecycle_stage, previous_traveler_lifecycle_stage;