snowflake-webhooks-events

Snowflake Webhooks & Events

Overview

Snowflake uses alerts, email notifications, external functions, and notification integrations for event-driven patterns (not traditional webhooks).

Prerequisites

• ACCOUNTADMIN or role with CREATE ALERT privilege
• Email notification integration configured
• For external functions: API Gateway (AWS/GCP/Azure) configured
• For S3/GCS event notifications: Snowpipe configured

Instructions

Step 1: Snowflake Alerts (Built-in Event System)

-- Alert when daily revenue drops below threshold
CREATE OR REPLACE ALERT revenue_drop_alert
  WAREHOUSE = ANALYTICS_WH
  SCHEDULE = '60 MINUTE'
  IF (EXISTS (
    SELECT 1
    FROM daily_order_metrics
    WHERE metric_date = CURRENT_DATE()
      AND total_revenue < (
        SELECT AVG(total_revenue) * 0.5
        FROM daily_order_metrics
        WHERE metric_date BETWEEN DATEADD(days, -30, CURRENT_DATE())
          AND DATEADD(days, -1, CURRENT_DATE())
      )
  ))
  THEN
    CALL SYSTEM$SEND_EMAIL(
      'revenue_notifications',
      '[email protected]',
      'Revenue Alert: Below 50% of 30-day average',
      'Daily revenue has dropped significantly. Check dashboard.'
    );

ALTER ALERT revenue_drop_alert RESUME;

-- Alert when warehouse credits exceed daily budget
CREATE OR REPLACE ALERT credit_usage_alert
  WAREHOUSE = ANALYTICS_WH
  SCHEDULE = '30 MINUTE'
  IF (EXISTS (
    SELECT 1
    FROM SNOWFLAKE.ACCOUNT_USAGE.WAREHOUSE_METERING_HISTORY
    WHERE start_time >= CURRENT_DATE()
    GROUP BY ALL
    HAVING SUM(credits_used) > 100  -- Daily budget: 100 credits
  ))
  THEN
    CALL SYSTEM$SEND_EMAIL(
      'ops_notifications',
      '[email protected]',
      'Snowflake Credit Alert',
      'Daily credit usage has exceeded 100 credits.'
    );

-- Monitor alert history
SELECT *
FROM TABLE(INFORMATION_SCHEMA.ALERT_HISTORY(
  SCHEDULED_TIME_RANGE_START => DATEADD(hours, -24, CURRENT_TIMESTAMP())
))
ORDER BY scheduled_time DESC;

Step 2: Email Notification Integration

-- Set up email notification integration
CREATE OR REPLACE NOTIFICATION INTEGRATION email_notifications
  TYPE = EMAIL
  ENABLED = TRUE
  ALLOWED_RECIPIENTS = (
    '[email protected]',
    '[email protected]',
    '[email protected]'
  );

-- Send email from stored procedure
CREATE OR REPLACE PROCEDURE send_data_quality_report()
  RETURNS VARCHAR
  LANGUAGE SQL
AS
$$
BEGIN
  LET row_count INTEGER;
  SELECT COUNT(*) INTO :row_count
  FROM orders WHERE order_date = CURRENT_DATE() AND amount IS NULL;

  IF (row_count > 0) THEN
    CALL SYSTEM$SEND_EMAIL(
      'email_notifications',
      '[email protected]',
      'Data Quality Issue: NULL amounts detected',
      row_count || ' orders have NULL amounts today.'
    );
    RETURN 'Alert sent for ' || row_count || ' records';
  END IF;
  RETURN 'No issues found';
END;
$$;

Step 3: External Functions (Call External APIs)

-- Create API integration for external function
CREATE OR REPLACE API INTEGRATION my_api_integration
  API_PROVIDER = aws_api_gateway
  API_AWS_ROLE_ARN = 'arn:aws:iam::123456789:role/snowflake-api-role'
  ENABLED = TRUE
  API_ALLOWED_PREFIXES = ('https://api.execute-api.us-east-1.amazonaws.com/');

-- Create external function that calls your API
CREATE OR REPLACE EXTERNAL FUNCTION notify_slack(message VARCHAR)
  RETURNS VARIANT
  API_INTEGRATION = my_api_integration
  AS 'https://api.execute-api.us-east-1.amazonaws.com/prod/slack-notify';

-- Use in a task to notify on data changes
CREATE OR REPLACE TASK notify_new_high_value_orders
  WAREHOUSE = ANALYTICS_WH
  SCHEDULE = '15 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('orders_stream')
AS
  SELECT notify_slack(
    'New high-value order: $' || amount || ' from customer ' || customer_id
  )
  FROM orders_stream
  WHERE METADATA$ACTION = 'INSERT' AND amount >= 10000;

Step 4: Cloud Event Notifications (S3/GCS/Azure)

-- S3 event notification for auto-ingest (Snowpipe)
-- This triggers when new files land in S3
CREATE OR REPLACE PIPE auto_ingest_pipe
  AUTO_INGEST = TRUE
  AS
  COPY INTO raw_events
    FROM @my_s3_stage/events/
    FILE_FORMAT = my_json_format;

-- Get the SQS queue ARN to configure S3 event notifications
SHOW PIPES LIKE 'auto_ingest_pipe';
-- Copy notification_channel value → S3 bucket event configuration

-- GCS pub/sub notification
CREATE OR REPLACE NOTIFICATION INTEGRATION gcs_notification
  TYPE = QUEUE
  NOTIFICATION_PROVIDER = GCP_PUBSUB
  ENABLED = TRUE
  GCP_PUBSUB_SUBSCRIPTION_NAME = 'projects/my-project/subscriptions/snowflake-sub';

Step 5: Application-Side Event Processing

// src/snowflake/event-processor.ts
// Poll for changes using streams (app-side consumption)

interface OrderEvent {
  ORDER_ID: number;
  CUSTOMER_ID: number;
  AMOUNT: number;
  METADATA$ACTION: 'INSERT' | 'DELETE';
}

async function processOrderEvents(conn: snowflake.Connection) {
  // Check if stream has data
  const hasData = await query<{ HAS_DATA: boolean }>(conn,
    "SELECT SYSTEM$STREAM_HAS_DATA('orders_stream') AS HAS_DATA"
  );
  if (!hasData.rows[0]?.HAS_DATA) return;

  // Consume stream within a transaction
  await query(conn, 'BEGIN');
  try {
    const events = await query<OrderEvent>(conn,
      'SELECT * FROM orders_stream'
    );

    for (const event of events.rows) {
      if (event.METADATA$ACTION === 'INSERT' && event.AMOUNT >= 10000) {
        await notifySlack(`High-value order: $${event.AMOUNT}`);
      }
    }

    // Advance the stream offset by writing to target
    await query(conn, `
      INSERT INTO processed_orders
      SELECT order_id, customer_id, amount, CURRENT_TIMESTAMP()
      FROM orders_stream WHERE METADATA$ACTION = 'INSERT'
    `);

    await query(conn, 'COMMIT');
  } catch (err) {
    await query(conn, 'ROLLBACK');
    throw err;
  }
}

Error Handling

Issue	Cause	Solution
Alert not firing	Alert suspended	ALTER ALERT x RESUME
Email not delivered	Recipient not in allowlist	Add to ALLOWED_RECIPIENTS
External function timeout	API too slow	Increase timeout, check API health
Snowpipe not triggering	S3 event config missing	Configure SQS notification from SHOW PIPES
Stream data loss	Stream stale	Recreate stream, increase retention

Resources

• Snowflake Alerts
• External Functions
• Snowpipe Auto-Ingest

Next Steps

For performance optimization, see snowflake-performance-tuning.