WITH base AS (
    SELECT
        JSON_EXTRACT(logicaltimestamp_json, '$.clockIdentifier') AS logic_clock_id
        , JSON_EXTRACT(logicaltimestamp_json, '$.timestampInNanoseconds') AS logical_timestamp_ns
        , JSON_EXTRACT(clusterrepresentativeid_json, '$.uuid') AS cluster_id
        , LOOKUP_ANY_EVENT_TYPE(anyeventtype) AS component_name
        , LOOKUP_INNER_TYPE(anyeventtype, innertype) AS event_name
        , anyeventtype
        , eventdata
    FROM SiriSELFProcessedEvent
)

, global_dim_agg AS (
    SELECT
        logic_clock_id AS dfc_logic_clock_id
        , MIN(logical_timestamp_ns) AS min_clock_ns
        , IIF(event_name='deviceFixedContext', DATETIME(P_GET(eventData, anyEventType, 'deviceFixedContext.timeIntervalSince1970'), 'unixepoch'), NULL) AS event_timestamp
        , MAX(IIF(event_name = 'ephemeralToAggregationIdentifierMap', P_GET(eventdata, anyeventtype, 'ephemeralToAggregationIdentifierMap.deviceAggregationId.value'), NULL)) AS device_aggregation_id
        , MIN(IIF(event_name='deviceFixedContext', P_GET(eventData, anyEventType, 'deviceFixedContext.timeIntervalSince1970'), NULL)) AS time_interval_since1970
        , 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
    FROM base
    WHERE component_name = 'DIM_CLIENT_EVENT'
    GROUP BY dfc_logic_clock_id
)

, sanitized_device_dimensions AS (
    SELECT
        DISTINCT 
        dfc_logic_clock_id
        , device_aggregation_id AS deviceAggregationId
        , time_interval_since1970 AS timeIntervalSince1970
        , device_type AS deviceType
        , system_build AS systemBuild
        , program_code AS programCode
        , data_sharing_opt_in_state AS dataSharingOptInState
        , siri_input_locale AS siriInputLocale
        , assistant_id AS assistantId
    FROM global_dim_agg
)

, 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
)

, check_previous_system_build AS (
    SELECT 
        system_build
        , build_change_timestamp
        , ROW_NUMBER() OVER (ORDER BY build_change_timestamp DESC) AS latest_build_change_rank
    FROM organize_system_build_records
)

, extract_previous_build AS (
    SELECT 
        system_build AS previousSystemBuild
        , 1 AS dummy_key
    FROM check_previous_system_build
    WHERE latest_build_change_rank = 2
)

, extract_latest_build_change_timestamp AS (
    SELECT 
        UNIXEPOCH(build_change_timestamp)    AS buildInstallationTimestampInSecondsSince1970
        , 1 AS dummy_key
    FROM check_previous_system_build
    WHERE latest_build_change_rank = 1
)

, filtered AS (
    SELECT
        *
        , IIF(component_name = 'SAD_CLIENT_EVENT' AND event_name = 'availableAssetDailyStatus', P_JSON(eventdata, anyeventtype), NULL) AS assetsetpayload
        , IIF(component_name = 'SAD_CLIENT_EVENT' AND event_name = 'availableAssetDailyStatus', P_GET(eventdata, anyeventtype, 'availableAssetDailyStatus.statusReason'), NULL) AS status_reason
    FROM base
    WHERE
        component_name IN ('SAD_CLIENT_EVENT')
        AND event_name IN ('availableAssetDailyStatus')
        AND eventdata IS NOT NULL
)

, include_event_timestamp AS (
    SELECT
        component_name
        , event_name
        , logic_clock_id
        , assetsetpayload
        , DATETIME(
            global_dim_agg.time_interval_since1970
            + logical_timestamp_ns / 1000000000 - global_dim_agg.min_clock_ns
            / 1000000000
            , 'unixepoch'
        ) AS event_timestamp
        , CAST(logical_timestamp_ns / 1000000.0 AS DOUBLE) AS bootToAssetDeliveryInMs
    FROM filtered
    LEFT JOIN global_dim_agg
        ON filtered.logic_clock_id = global_dim_agg.dfc_logic_clock_id
    WHERE status_reason = 'SADAVAILABLEASSETSTATUSREASON_ASSET_ARRIVED'
)

, extracted AS (
    SELECT
        DATE(event_timestamp) AS eventDate
        , logic_clock_id
        , JSON_EXTRACT(assetsetpayload, '$.availableAssetDailyStatus.assetSetStatus[0].uafAssetSets') AS assetsetstatus
        , bootToAssetDeliveryInMs
    FROM include_event_timestamp
)

, unnested AS (
    SELECT
        eventDate
        , JSON_EXTRACT(json_each.value, '$.assetSetName') AS assetSetName
        , JSON_EXTRACT(json_each.value, '$.assetSetId') AS assetSetId
        , bootToAssetDeliveryInMs
        , logic_clock_id
    FROM
        extracted, JSON_EACH(assetsetstatus)
)

, agg AS (
    SELECT
        logic_clock_id
        , eventDate
        , assetSetName
        , COALESCE(assetSetId, 'missing') AS assetSetId
        , MIN(bootToAssetDeliveryInMs) AS bootToAssetDeliveryInMs
    FROM unnested
    GROUP BY
        logic_clock_id
        , eventDate
        , assetSetName
        , assetSetId
)

, include_device_dimensions AS (
    SELECT
        sanitized_device_dimensions.deviceAggregationId
        , sanitized_device_dimensions.timeIntervalSince1970
        , sanitized_device_dimensions.deviceType
        , sanitized_device_dimensions.systemBuild
        , sanitized_device_dimensions.programCode
        , sanitized_device_dimensions.dataSharingOptInState
        , sanitized_device_dimensions.siriInputLocale
        , sanitized_device_dimensions.assistantId
        , agg.assetSetName
        , agg.assetSetId
        , agg.bootToAssetDeliveryInMs
        , agg.eventDate
        , 1 AS dummy_key
    FROM agg
    LEFT JOIN sanitized_device_dimensions
        ON agg.logic_clock_id = sanitized_device_dimensions.dfc_logic_clock_id
)

SELECT
    include_device_dimensions.eventDate
    , include_device_dimensions.deviceAggregationId
    , include_device_dimensions.systemBuild
    , include_device_dimensions.deviceType
    , include_device_dimensions.programCode
    , include_device_dimensions.siriInputLocale
    , include_device_dimensions.dataSharingOptInState
    , include_device_dimensions.assetSetName
    , include_device_dimensions.assetSetId
    , include_device_dimensions.bootToAssetDeliveryInMs
    , extract_previous_build.previousSystemBuild
    , extract_latest_build_change_timestamp.buildInstallationTimestampInSecondsSince1970
FROM include_device_dimensions
LEFT JOIN extract_previous_build
    USING (dummy_key)
LEFT JOIN extract_latest_build_change_timestamp
    USING (dummy_key)
ORDER BY assetSetName
;