-- Tumbling Window: Count events every 5 minutes
SELECT
    System.Timestamp() AS window_end,
    COUNT(*) AS event_count,
    AVG(temperature) AS avg_temperature,
    MAX(temperature) AS max_temperature,
    MIN(temperature) AS min_temperature
INTO [cosmos-output]
FROM [eventhub-input] PARTITION BY device_id
GROUP BY
    TumblingWindow(minute, 5),
    device_id;

-- Hopping Window: Rolling 10-minute average
SELECT
    System.Timestamp() AS window_end,
    device_id,
    AVG(temperature) AS rolling_avg_temp
INTO [powerbi-output]
FROM [eventhub-input]
GROUP BY
    HoppingWindow(minute, 10, 5),  -- 10-min window, 5-min hop
    device_id;

-- Sliding Window: Detect anomalies
SELECT
    System.Timestamp() AS window_end,
    device_id,
    AVG(temperature) AS avg_temp,
    STDEV(temperature) AS std_temp,
    CASE
        WHEN AVG(temperature) > (SELECT AVG(temperature) + 3 * STDEV(temperature)
                                  FROM [eventhub-input]
                                  WHERE device_id = Input.device_id
                                  TIMESTAMP BY event_time
                                  AND DATEDIFF(minute, System.Timestamp(), event_time) BETWEEN -60 AND 0)
        THEN 'ANOMALY'
        ELSE 'NORMAL'
    END AS status
INTO [alert-output]
FROM [eventhub-input] PARTITION BY device_id
GROUP BY
    TumblingWindow(minute, 1),
    device_id;

-- Temporal Join: Enrich with reference data
SELECT
    e.device_id,
    e.temperature,
    e.event_time,
    r.device_location,
    r.device_type,
    r.threshold
INTO [enriched-output]
FROM [eventhub-input] e
JOIN [reference-input] r
    ON e.device_id = r.device_id
    AND DATEDIFF(minute, e, r) BETWEEN -5 AND 5;

-- Session Window: User activity
SELECT
    device_id,
    MIN(event_time) AS session_start,
    MAX(event_time) AS session_end,
    COUNT(*) AS events_in_session
INTO [session-output]
FROM [eventhub-input]
GROUP BY
    SessionWindow(minute, 10),  -- 10-min timeout
    device_id;

-- Built-in anomaly detection
SELECT
    device_id,
    System.Timestamp() AS window_end,
    AVG(temperature) AS avg_temp,
    AnomalyDetection_SpikeAndDip(
        temperature, 95, 100,
        DATEDIFF(minute, System.Timestamp(), event_time)
    ) AS spike_result
INTO [anomaly-output]
FROM [eventhub-input]
GROUP BY
    TumblingWindow(minute, 5),
    device_id;

-- CNTK-based anomaly detection
SELECT
    device_id,
    System.Timestamp() AS window_end,
    AnomalyDetection_ChangePoint(
        temperature, 95, 200,
        DATEDIFF(minute, System.Timestamp(), event_time)
    ) AS change_point_result
INTO [changepoint-output]
FROM [eventhub-input]
GROUP BY
    TumblingWindow(minute, 10),
    device_id;