WITH full_days AS (
SELECT DISTINCT json_each.value AS full_event_date
  FROM (
    SELECT json_array(%@) AS d
    )t, json_each(t.d)
),

daily_stats AS (
SELECT
    json_extract(eventMetadata_json, '$.deviceAggregationId') AS device_aggregation_id
    ,json_extract(eventMetadata_json, '$.userAggregationIdRotationTimestampInMs') AS user_aggregation_id_rotation_timestamp_ms
    ,json_extract(eventMetadata_json, '$.userAggregationIdExpirationTimestampInMs') AS user_aggregation_id_expiration_timestamp_ms
    ,date(datetime(json_extract(eventMetadata_json, '$.timeInterval.startTimestampInSecondsSince1970'), 'unixepoch')) AS event_date
    ,json_extract(eventMetadata_json, '$.timeInterval.numberOfSeconds') AS time_interval_duration_s
    ,IIF(SUM(validTurnCount)>=2,1,0) AS is_assistant_2x_day
    ,IIF(SUM(asrTurnCount)>=2,1,0) AS is_asr_2x_day
FROM SiriMetricsOnDeviceDigestUsageMetrics
WHERE productId=1 --todo: UDF to convert this back to String
GROUP BY 1,4,5
),

full_days_join AS (
SELECT
    COALESCE(daily_stats.event_date, full_days.full_event_date) AS event_date
    ,device_aggregation_id
    ,user_aggregation_id_rotation_timestamp_ms
    ,user_aggregation_id_expiration_timestamp_ms
    ,time_interval_duration_s
    ,is_assistant_2x_day
    ,is_asr_2x_day
FROM full_days
FULL JOIN daily_stats
ON full_days.full_event_date = daily_stats.event_date
),

win_value AS (
SELECT
    COALESCE(device_aggregation_id, (max(device_aggregation_id) OVER win)) AS device_aggregation_id
    ,COALESCE(user_aggregation_id_rotation_timestamp_ms, (max(user_aggregation_id_rotation_timestamp_ms) OVER win))
         AS user_aggregation_id_rotation_timestamp_ms
    ,COALESCE(user_aggregation_id_expiration_timestamp_ms, (max(user_aggregation_id_expiration_timestamp_ms) OVER win))
         AS user_aggregation_id_expiration_timestamp_ms
    ,event_date
    ,COALESCE(time_interval_duration_s, max(time_interval_duration_s) OVER win) AS time_interval_duration_s
    ,COALESCE(is_assistant_2x_day, 0) AS is_assistant_2x_day
    ,COALESCE(is_asr_2x_day, 0) AS is_asr_2x_day
    ,date(event_date, '-6 day') AS six_day_ago
    ,lag(event_date, 1) OVER win AS lag_one_date
    ,lag(event_date, 2) OVER win AS lag_two_date
    ,lag(event_date, 3) OVER win AS lag_three_date
    ,lag(event_date, 4) OVER win AS lag_four_date
    ,lag(event_date, 5) OVER win AS lag_five_date
    ,lag(event_date, 6) OVER win AS lag_six_date
    ,COALESCE(lag(is_assistant_2x_day, 1) OVER win, 0) AS lag_assistant_one_value
    ,COALESCE(lag(is_assistant_2x_day, 2) OVER win, 0) AS lag_assistant_two_value
    ,COALESCE(lag(is_assistant_2x_day, 3) OVER win, 0) AS lag_assistant_three_value
    ,COALESCE(lag(is_assistant_2x_day, 4) OVER win, 0) AS lag_assistant_four_value
    ,COALESCE(lag(is_assistant_2x_day, 5) OVER win, 0) AS lag_assistant_five_value
    ,COALESCE(lag(is_assistant_2x_day, 6) OVER win, 0) AS lag_assistant_six_value
    ,COALESCE(lag(is_asr_2x_day, 1) OVER win, 0) AS lag_asr_one_value
    ,COALESCE(lag(is_asr_2x_day, 2) OVER win, 0) AS lag_asr_two_value
    ,COALESCE(lag(is_asr_2x_day, 3) OVER win, 0) AS lag_asr_three_value
    ,COALESCE(lag(is_asr_2x_day, 4) OVER win, 0) AS lag_asr_four_value
    ,COALESCE(lag(is_asr_2x_day, 5) OVER win, 0) AS lag_asr_five_value
    ,COALESCE(lag(is_asr_2x_day, 6) OVER win, 0) AS lag_asr_six_value
FROM full_days_join
WINDOW win AS (order by event_date ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)
)

SELECT
    *
    ,IIF(daysWithTwoValidAssistantTurnsPerWeek>=3, 1, 0) AS is_assistant_2x3
FROM (
SELECT
    *
    ,IIF(
    IIF(lag_one_date<=six_day_ago,1 , 0)+
    IIF(lag_two_date<=six_day_ago,1, 0)+
    IIF(lag_three_date<=six_day_ago,1, 0)+
    IIF(lag_four_date<=six_day_ago,1, 0)+
    IIF(lag_five_date<=six_day_ago,1, 0)+
    IIF(lag_six_date<=six_day_ago,1, 0) >= 1, 0, 1
    ) AS is_incompleted_7_days
    ,IIF(lag_one_date>=six_day_ago,lag_assistant_one_value, 0)+
    IIF(lag_two_date>=six_day_ago,lag_assistant_two_value, 0)+
    IIF(lag_three_date>=six_day_ago,lag_assistant_three_value, 0)+
    IIF(lag_four_date>=six_day_ago,lag_assistant_four_value, 0)+
    IIF(lag_five_date>=six_day_ago,lag_assistant_five_value, 0)+
    IIF(lag_six_date>=six_day_ago,lag_assistant_six_value, 0)+
    is_assistant_2x_day AS daysWithTwoValidAssistantTurnsPerWeek
    ,IIF(lag_one_date>=six_day_ago,lag_asr_one_value, 0)+
    IIF(lag_two_date>=six_day_ago,lag_asr_two_value, 0)+
    IIF(lag_three_date>=six_day_ago,lag_asr_three_value, 0)
    +IIF(lag_four_date>=six_day_ago,lag_asr_four_value, 0)+
    IIF(lag_five_date>=six_day_ago,lag_asr_five_value, 0)+
    IIF(lag_six_date>=six_day_ago,lag_asr_six_value, 0)
    +is_asr_2x_day AS daysWithTwoAssistantSpeechRequestsPerWeek
FROM win_value
)