WITH base AS (
    SELECT
        JSON_EXTRACT(logicaltimestamp_json, '$.clockIdentifier') AS logic_clock_id,
        JSON_EXTRACT(logicaltimestamp_json, '$.timestampInNanoseconds') AS logical_timestamp_ns,
        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,
        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='ephemeralToAggregationIdentifierMap', P_GET(eventdata, anyeventtype, 'ephemeralToAggregationIdentifierMap.deviceAggregationId.value'), NULL)) AS device_aggregation_id
    FROM base
    WHERE component_name = 'DIM_CLIENT_EVENT'
    GROUP BY dfc_logic_clock_id
),

sad_asset_bringup AS (
    SELECT
        logic_clock_id AS sad_logic_clock_id,
        logical_timestamp_ns AS eventLogicalTimestampNs,
        P_JSON(eventdata, anyeventtype) AS raw_payload,
        JSON_EXTRACT(P_JSON(eventdata, anyeventtype), '$.assetBringUpState') AS assetBringUpDigest,
        JSON_EXTRACT(P_JSON(eventdata, anyeventtype), '$.assetBringUpState.assetBringUpType') AS assetBringUpType
    FROM base
    WHERE component_name = 'SAD_CLIENT_EVENT' AND event_name = 'assetBringUpState'
),

include_event_time AS (
    SELECT
        g.time_interval_since1970 + (e.eventLogicalTimestampNs - g.min_clock_ns) / 1000000000.0 AS eventTimestampSecondsSince1970,
        CAST(
            CAST(g.time_interval_since1970 * 1000 AS INTEGER)
            + CAST((e.eventLogicalTimestampNs - g.min_clock_ns) / 1000000 AS INTEGER)
            AS INTEGER
        ) AS eventTimestampInMsSince1970,
        e.sad_logic_clock_id,
        g.device_aggregation_id AS deviceAggregationId,
        g.system_build AS systemBuild,
        g.device_type AS deviceType,
        g.program_code AS programCode,
        e.assetBringUpType,
        e.assetBringUpDigest,
        e.raw_payload
    FROM sad_asset_bringup e
    JOIN global_dim_agg g
        ON e.sad_logic_clock_id = g.dfc_logic_clock_id
),

-- Group by clock_id and assetBringUpType, aggregate digests into JSON array
pivoted AS (
    SELECT
        DATE(MIN(eventTimestampSecondsSince1970), 'unixepoch') AS eventDate,
        MIN(eventTimestampInMsSince1970) AS eventTimestampInMsSince1970,
        deviceAggregationId,
        systemBuild,
        deviceType,
        programCode,
        '[' || GROUP_CONCAT(assetBringUpDigest, ',') || ']' AS digests
    FROM include_event_time
    WHERE eventTimestampInMsSince1970 IS NOT NULL
      AND eventTimestampInMsSince1970 > 0
      AND eventTimestampInMsSince1970 <= CAST(strftime('%s','now') AS INTEGER) * 1000
      AND eventTimestampSecondsSince1970 <= CAST(strftime('%s','now') AS INTEGER)
    GROUP BY sad_logic_clock_id, deviceAggregationId, systemBuild, deviceType, programCode, assetBringUpType
)

SELECT 
    eventDate,
    eventTimestampInMsSince1970,
    deviceAggregationId,
    systemBuild,
    deviceType,
    programCode,
    digests
FROM pivoted
ORDER BY eventTimestampInMsSince1970;