-- UAF status reason codes:
--   1 = _Scheduled (daily status snapshot)
--   2 = _Subscription_Change
--   3 = _Asset_Arrived
--   4 = _Download_Error
--   5 = _AssetSet_Download_Requested
--   6 = _AssetSet_Altered
--   7 = _AssetSet_Eliminated

WITH uaf_events_raw AS (
    SELECT DISTINCT
        eventTimestamp AS uaf_ts_sec,
        DATE(eventTimestamp, 'unixepoch') AS event_date,
        DATETIME(eventTimestamp, 'unixepoch') AS event_datetime,
        CASE CAST(JSON_EXTRACT(availableAssetDailyStatus_json, '$.statusReason') AS INTEGER)
            WHEN 1 THEN '_Scheduled'
            WHEN 2 THEN '_Subscription_Change'
            WHEN 3 THEN '_Asset_Arrived'
            WHEN 4 THEN '_Download_Error'
            WHEN 5 THEN '_AssetSet_Download_Requested'
            WHEN 6 THEN '_AssetSet_Altered'
            WHEN 7 THEN '_AssetSet_Eliminated'
            ELSE 'SADAVAILABLEASSETSTATUSREASON_UNKNOWN'
        END AS status_reason,
        availableAssetDailyStatus_json
    FROM "AssetDelivery.UAF.DailyStatus"
),
uaf_events_ranked AS (
    SELECT
        *,
        ROW_NUMBER() OVER (
            PARTITION BY status_reason, event_date, availableAssetDailyStatus_json
            ORDER BY uaf_ts_sec ASC
        ) AS row_num
    FROM uaf_events_raw
    WHERE
        availableAssetDailyStatus_json IS NOT NULL
        AND uaf_ts_sec IS NOT NULL
        AND status_reason IN ('_Scheduled', '_Download_Error', '_Asset_Arrived')
),

uaf_events_deduped AS (
    SELECT
        uaf_ts_sec,
        event_datetime,
        status_reason,
        availableAssetDailyStatus_json AS payload_json
    FROM uaf_events_ranked
    WHERE
        -- Take only the earliest _Scheduled event per day when the payload content is the same.
        (status_reason = '_Scheduled' AND row_num = 1)
        OR status_reason IN ('_Download_Error', '_Asset_Arrived')
),
-- SELF logical clock rows for DFC (Device Fixed Context)
base AS (
    SELECT
        JSON_EXTRACT(logicalTimestamp_json, '$.clockIdentifier') AS logic_clock_id,
        JSON_EXTRACT(logicalTimestamp_json, '$.timestampInNanoseconds') AS logical_timestamp_ns,
        JSON_EXTRACT(COALESCE(clusterRepresentativeId_json, componentId_json), '$.uuid') AS cluster_id,
        lookup_any_event_type(anyEventType) AS component_name,
        lookup_inner_type(anyEventType, innerType) AS event_name,
        anyEventType,
        eventData,
        componentId_json
    FROM SiriSELFProcessedEvent
),

-- Collapse SELF dims per logical clock
global_dim_agg AS (
    SELECT
        logic_clock_id AS dfc_logic_clock_id,
        MIN(logical_timestamp_ns) AS min_clock_ns,
        MIN(
            IIF(
                event_name = 'deviceFixedContext',
                P_GET(eventData, anyEventType, 'deviceFixedContext.timeIntervalSince1970'),
                NULL
            )
        ) AS event_timestamp,
        MAX(IIF(event_name = 'deviceFixedContext', P_GET(eventData, anyEventType, 'deviceFixedContext.deviceType'), NULL)) AS device_type,
        MAX(IIF(event_name = 'deviceFixedContext', P_GET(eventData, anyEventType, 'deviceFixedContext.systemBuild'), NULL)) AS system_build,
        MAX(IIF(event_name = 'deviceFixedContext', P_GET(eventData, anyEventType, 'deviceFixedContext.programCode'), NULL)) AS program_code,
        MAX(IIF(event_name = 'deviceFixedContext', P_GET(eventData, anyEventType, 'deviceFixedContext.dataSharingOptInState'), NULL)) AS data_sharing_opt_in_state,
        MAX(IIF(event_name = 'deviceFixedContext', P_GET(eventData, anyEventType, 'deviceFixedContext.siriInputLocale'), NULL)) AS siri_input_locale,
        MAX(IIF(event_name = 'siriAccountInformation', P_GET(eventData, anyEventType, 'siriAccountInformation.siriDeviceId'), NULL)) AS assistant_id,
        MAX(IIF(event_name = 'ephemeralToAggregationIdentifierMap', P_GET(eventData, anyEventType, 'ephemeralToAggregationIdentifierMap.deviceAggregationId.value'), NULL)) AS device_aggregation_id,
        MAX(IIF(event_name = 'ephemeralToAggregationIdentifierMap', P_GET(eventData, anyEventType, 'ephemeralToAggregationIdentifierMap.userAggregationIdRotationTimestampMs'), NULL)) AS user_aggregation_id_rotation_timestamp_ms,
        MAX(IIF(event_name = 'ephemeralToAggregationIdentifierMap', P_GET(eventData, anyEventType, 'ephemeralToAggregationIdentifierMap.userAggregationIdExpirationTimestampMs'), NULL)) AS user_aggregation_id_expiration_timestamp_ms
    FROM base
    WHERE component_name = 'DIM_CLIENT_EVENT'
    GROUP BY dfc_logic_clock_id
),
-- Boilerplate for tracking device build history
organize_system_build_records AS (
    SELECT
        system_build,
        MIN(event_timestamp) AS build_change_timestamp
    FROM global_dim_agg
    WHERE system_build IS NOT NULL
    GROUP BY system_build
),

extract_previous_build AS (
    SELECT
        system_build,
        build_change_timestamp,
        LAG(system_build) OVER (ORDER BY build_change_timestamp) AS previousSystemBuild
    FROM organize_system_build_records
),

device_dimensions AS (
    SELECT
        agg.dfc_logic_clock_id,
        agg.event_timestamp,
        agg.device_type,
        agg.system_build,
        agg.program_code,
        agg.siri_input_locale,
        agg.device_aggregation_id,
        agg.data_sharing_opt_in_state,
        agg.user_aggregation_id_rotation_timestamp_ms,
        agg.user_aggregation_id_expiration_timestamp_ms,
        prev.previousSystemBuild,
        prev.build_change_timestamp AS buildInstallationTimestampInSecondsSince1970
    FROM global_dim_agg agg
    LEFT JOIN extract_previous_build prev
        ON agg.system_build = prev.system_build
),

uaf_time_bounds AS (
    SELECT
        MIN(uaf_ts_sec) AS min_uaf_ts,
        MAX(uaf_ts_sec) AS max_uaf_ts
    FROM uaf_events_deduped
),

latest_device_dimensions AS (
    SELECT
        MAX(event_timestamp) AS max_event_timestamp
    FROM device_dimensions
    WHERE
        (SELECT min_uaf_ts FROM uaf_time_bounds) IS NOT NULL
        AND event_timestamp < (SELECT min_uaf_ts FROM uaf_time_bounds)
),
-- Seed the device dimensions table with the latest device dimensions
seeded_device_dimensions AS (
    SELECT * FROM device_dimensions
    UNION ALL
    SELECT *
    FROM device_dimensions
    WHERE
        (SELECT max_event_timestamp FROM latest_device_dimensions) IS NOT NULL
        AND event_timestamp = (SELECT max_event_timestamp FROM latest_device_dimensions)
),

self_time_bounds AS (
    SELECT
        MIN(event_timestamp) AS min_ts,
        MAX(event_timestamp) AS max_ts
    FROM seeded_device_dimensions
),

-- Restrict UAF events to the SELF stream window
ordered_uaf AS (
    SELECT
        src.uaf_ts_sec AS uaf_event_timestamp,
        src.event_datetime,
        src.payload_json AS asset_payload_json,
        src.status_reason,
        ROW_NUMBER() OVER (ORDER BY src.uaf_ts_sec) AS uaf_row_id
    FROM uaf_events_deduped src, self_time_bounds bounds
    WHERE bounds.min_ts IS NOT NULL
        AND bounds.max_ts IS NOT NULL
        AND src.uaf_ts_sec >= bounds.min_ts
        AND src.uaf_ts_sec <= bounds.max_ts
),

-- Get the single latest device dimension snapshot (fallback when no temporal match)
latest_single_dim AS (
    SELECT *
    FROM seeded_device_dimensions
    ORDER BY event_timestamp DESC NULLS LAST
    LIMIT 1
),

-- Match each UAF row with its most recent SELF dims (the latest dimension event at or before the UAF event)
-- If no seeded dims exist, the unconditional LEFT JOIN below behaves like a cross join so every
-- UAF row still sees the single fallback snapshot.
uaf_with_dimensions AS (
    SELECT
        uaf.uaf_row_id,
        uaf.uaf_event_timestamp,
        uaf.event_datetime,
        uaf.asset_payload_json,
        uaf.status_reason,
        COALESCE(dims.device_aggregation_id, fallback.device_aggregation_id) AS device_aggregation_id,
        COALESCE(dims.event_timestamp, fallback.event_timestamp) AS event_timestamp,
        COALESCE(dims.device_type, fallback.device_type) AS device_type,
        COALESCE(dims.system_build, fallback.system_build) AS system_build,
        COALESCE(dims.program_code, fallback.program_code) AS program_code,
        COALESCE(dims.siri_input_locale, fallback.siri_input_locale) AS siri_input_locale,
        COALESCE(dims.data_sharing_opt_in_state, fallback.data_sharing_opt_in_state) AS data_sharing_opt_in_state,
        COALESCE(dims.user_aggregation_id_rotation_timestamp_ms, fallback.user_aggregation_id_rotation_timestamp_ms) AS user_aggregation_id_rotation_timestamp_ms,
        COALESCE(dims.user_aggregation_id_expiration_timestamp_ms, fallback.user_aggregation_id_expiration_timestamp_ms) AS user_aggregation_id_expiration_timestamp_ms,
        COALESCE(dims.previousSystemBuild, fallback.previousSystemBuild) AS previousSystemBuild,
        COALESCE(dims.buildInstallationTimestampInSecondsSince1970, fallback.buildInstallationTimestampInSecondsSince1970) AS buildInstallationTimestampInSecondsSince1970,
        ROW_NUMBER() OVER (
            PARTITION BY uaf.uaf_row_id
            ORDER BY dims.event_timestamp DESC NULLS LAST
        ) AS dim_rank
    FROM
        ordered_uaf uaf
        LEFT JOIN seeded_device_dimensions dims
            ON dims.event_timestamp <= uaf.uaf_event_timestamp
        LEFT JOIN latest_single_dim fallback ON 1=1
),

combined_uaf_dimensions AS (
    SELECT *
    FROM uaf_with_dimensions
    WHERE dim_rank = 1
),

resolved_asset_penetration_events AS (
    SELECT
        uaf_row_id,
        uaf_event_timestamp,
        event_datetime,
        asset_payload_json,
        status_reason,
        device_aggregation_id,
        event_timestamp,
        device_type,
        system_build,
        program_code,
        siri_input_locale,
        data_sharing_opt_in_state,
        user_aggregation_id_rotation_timestamp_ms,
        user_aggregation_id_expiration_timestamp_ms,
        previousSystemBuild,
        buildInstallationTimestampInSecondsSince1970
    FROM
        combined_uaf_dimensions
)
SELECT
    device_aggregation_id,
    uaf_event_timestamp AS event_timestamp,
    event_datetime,
    device_type,
    system_build,
    program_code,
    siri_input_locale,
    data_sharing_opt_in_state,
    previousSystemBuild,
    buildInstallationTimestampInSecondsSince1970,
    user_aggregation_id_rotation_timestamp_ms,
    user_aggregation_id_expiration_timestamp_ms,
    status_reason,
    asset_payload_json AS assetSetPayload
FROM
    resolved_asset_penetration_events
ORDER BY
    event_timestamp;