WITH event_enhanced AS (
  SELECT
    *
  FROM TEST.PUBLIC_amplitude.amplitude__event_enhanced
), date_spine AS (
  SELECT
    spine.*
  FROM TEST.PUBLIC_int_amplitude.int_amplitude__date_spine AS spine
), agg_event_data AS (
  SELECT
    event_day,
    event_type,
    COUNT(DISTINCT unique_event_id) AS number_events,
    COUNT(DISTINCT unique_session_id) AS number_sessions,
    COUNT(DISTINCT amplitude_user_id) AS number_users,
    COUNT(
      DISTINCT (
        CASE
          WHEN CAST(DATE_TRUNC('DAY', user_creation_time) AS DATE) = event_day
          THEN amplitude_user_id
        END
      )
    ) AS number_new_users
  FROM event_enhanced
  GROUP BY
    1,
    2
), spine_joined AS (
  SELECT
    date_spine.event_day,
    date_spine.event_type,
    agg_event_data.number_events,
    agg_event_data.number_sessions,
    agg_event_data.number_users,
    agg_event_data.number_new_users
  FROM date_spine
  LEFT JOIN agg_event_data
    ON date_spine.event_day = agg_event_data.event_day
    AND date_spine.event_type = agg_event_data.event_type
), final AS (
  SELECT
    event_day,
    event_type,
    COALESCE(number_events, 0) AS number_events,
    COALESCE(number_sessions, 0) AS number_sessions,
    COALESCE(number_users, 0) AS number_users,
    COALESCE(number_new_users, 0) AS number_new_users,
    MD5(
      CAST(COALESCE(CAST(event_day AS TEXT), '_dbt_utils_surrogate_key_null_') || '-' || COALESCE(CAST(event_type AS TEXT), '_dbt_utils_surrogate_key_null_') AS TEXT)
    ) AS daily_unique_key
  FROM spine_joined
)
SELECT
  *
FROM final

WITH event_data_raw AS (
  SELECT
    events.*
  FROM TEST.PUBLIC__source_amplitude.stg_amplitude__event AS events
), event_data /* deduplicate */ AS (
  SELECT
    *
  FROM (
    SELECT
      *,
      CASE
        WHEN NOT _insert_id IS NULL
        THEN ROW_NUMBER() OVER (PARTITION BY _insert_id ORDER BY client_upload_time DESC)
        ELSE ROW_NUMBER() OVER (PARTITION BY event_id, device_id, client_event_time, amplitude_user_id ORDER BY client_upload_time DESC)
      END AS nth_event_record
    FROM event_data_raw
  ) AS duplicates
  WHERE
    nth_event_record = 1
), event_type AS (
  SELECT
    *
  FROM TEST.PUBLIC__source_amplitude.stg_amplitude__event_type
), session_data AS (
  SELECT
    *
  FROM TEST.PUBLIC_amplitude.amplitude__sessions
), event_enhanced AS (
  SELECT
    event_data.unique_event_id,
    event_data.unique_session_id,
    CAST(event_data.amplitude_user_id AS TEXT) AS amplitude_user_id,
    event_data.event_id,
    event_data.event_type,
    event_data.event_time,
    event_data.event_day,
    event_type.event_type_id,
    event_type.event_type_name,
    event_data.session_id,
    ROW_NUMBER() OVER (PARTITION BY session_id ORDER BY event_time ASC) AS session_event_number,
    ROW_NUMBER() OVER (PARTITION BY amplitude_user_id ORDER BY event_time ASC) AS user_event_number,
    event_data.group_types,
    CAST(event_data.user_id AS TEXT) AS user_id,
    event_data.user_creation_time,
    event_data.amplitude_id,
    event_data.app,
    event_data.project_name,
    event_data.version_name,
    event_data.client_event_time,
    event_data.client_upload_time,
    event_data.server_received_time,
    event_data.server_upload_time,
    event_data.city,
    event_data.country,
    event_data.region,
    event_data.data,
    event_data.location_lat,
    event_data.location_lng,
    event_data.device_brand,
    event_data.device_carrier,
    event_data.device_family,
    event_data.device_id,
    event_data.device_manufacturer,
    event_data.device_model,
    event_data.device_type,
    event_data.ip_address,
    event_data.os_name,
    event_data.os_version,
    event_data.platform,
    event_data.language,
    event_data.dma,
    event_data.schema,
    event_data.start_version,
    event_type.totals,
    event_type.value,
    session_data.events_per_session,
    session_data.session_started_at,
    session_data.session_ended_at,
    session_data.user_session_number,
    session_data.session_started_at_day,
    session_data.session_ended_at_day,
    session_data.session_length_in_minutes,
    session_data.is_first_user_session,
    session_data.minutes_in_between_sessions
  FROM event_data
  LEFT JOIN event_type
    ON event_data.event_type_id = event_type.event_type_id
  LEFT JOIN session_data
    ON event_data.unique_session_id = session_data.unique_session_id
), final AS (
  SELECT
    *,
    MD5(
      CAST(COALESCE(CAST(unique_event_id AS TEXT), '_dbt_utils_surrogate_key_null_') || '-' || COALESCE(CAST(event_day AS TEXT), '_dbt_utils_surrogate_key_null_') AS TEXT)
    ) AS unique_key
  FROM event_enhanced
)
SELECT
  *
FROM final