-- Create table for streaming
CREATE TABLE streaming_events (
  event_id VARCHAR(36),
  event_type VARCHAR(50),
  user_id INT,
  event_timestamp TIMESTAMP_NTZ,
  payload VARIANT,
  ingestion_time TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

-- Create pipe for auto-ingestion
CREATE OR REPLACE PIPE streaming_pipe
  AUTO_INGEST = TRUE
  AS
  COPY INTO streaming_events
  FROM @streaming_stage
  FILE_FORMAT = (TYPE = 'JSON' STRIP_OUTER_ARRAY = TRUE)
  ON_ERROR = 'CONTINUE';

{
  "name": "snowflake-connector",
  "config": {
    "connector.class": "com.snowflake.kafka.connector.SnowflakeSinkConnector",
    "snowflake.ingestion.method": "SNOWPIPE_STREAMING",
    "snowflake.database": "my_database",
    "snowflake.schema": "streaming_schema",
    "snowflake.table": "streaming_events",
    "snowflake.role": "streaming_role",
    "snowflake.auth.type": "SNOWFLAKE_JWT",
    "snowflake.private.key": "${SNOWFLAKE_PRIVATE_KEY}",
    "topics": "events-topic",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter"
  }
}

-- Check streaming channels
SELECT *
FROM TABLE(INFORMATION_SCHEMA.STREAMING_INGESTION_CHANNELS(
  DATABASE_NAME => 'my_database',
  SCHEMA_NAME => 'streaming_schema'
));

-- Monitor channel health
SELECT
  channel_name,
  channel_state,
  earliest_commit_ts,
  latest_commit_ts,
  rows_inserted,
  bytes_inserted
FROM TABLE(INFORMATION_SCHEMA.STREAMING_INGESTION_CHANNELS(
  DATABASE_NAME => 'my_database',
  SCHEMA_NAME => 'streaming_schema'
));

-- Idempotent upsert pattern
MERGE INTO target_table t
USING staging_table s
ON t.event_id = s.event_id
WHEN MATCHED AND s.ingestion_time > t.ingestion_time THEN
  UPDATE SET
    t.payload = s.payload,
    t.ingestion_time = s.ingestion_time
WHEN NOT MATCHED THEN
  INSERT (event_id, event_type, user_id, event_timestamp, payload, ingestion_time)
  VALUES (s.event_id, s.event_type, s.user_id, s.event_timestamp, s.payload, s.ingestion_time);

-- Real-time ingestion metrics
SELECT
  DATE_TRUNC('minute', ingestion_time) as minute,
  COUNT(*) as rows_ingested,
  COUNT(DISTINCT event_type) as event_types,
  AVG(processing_time_ms) as avg_latency_ms
FROM streaming_events
WHERE ingestion_time >= DATEADD(hour, -1, CURRENT_TIMESTAMP())
GROUP BY 1
ORDER BY 1 DESC;