--- title: "CDC Webhooks User Guide" --- # CDC Webhooks User Guide :::caution[Planned: CDC webhooks] CDC webhooks are planned for HeliosDB Full. The current release does not capture changes from the write-ahead log for delivery or call HTTP endpoints, and it does not accept the CDC stream statements on this page yet. This page describes the planned design. Today, `SHOW CDC EVENTS [LIMIT n]` returns the most recent INSERT, UPDATE and DELETE events from the server's in-memory change-event log (see the [CDC tutorial](/docs/full/tutorials/intermediate/05-cdc-events/)). ::: --- ## Table of Contents 1. [Introduction](#introduction) 2. [Getting Started](#getting-started) 3. [Creating CDC Streams](#creating-cdc-streams) 4. [Event Filtering](#event-filtering) 5. [Event Transformation](#event-transformation) 6. [Delivery Configuration](#delivery-configuration) 7. [Circuit Breaker](#circuit-breaker) 8. [Exactly-Once Delivery](#exactly-once-delivery) 9. [Security](#security) 10. [Monitoring](#monitoring) 11. [Best Practices](#best-practices) --- ## Introduction HeliosDB's planned CDC-to-Webhook integration streams database changes to external systems. In the planned design, changes are captured from the Write-Ahead Log (WAL) and delivered to HTTP endpoints with exactly-once semantics. ### Key Features - **Real-time**: Sub-50ms delivery latency (P99) - **Reliable**: Exactly-once delivery guarantee - **Scalable**: 100K+ events per second - **Resilient**: Circuit breaker protection - **Filtering**: table, operation and column filters; transformations are planned as well --- ## Getting Started ### Prerequisites - HeliosDB Full - Network access to webhook endpoints - Webhook endpoints that accept POST requests ### Quick Start ```sql -- Create a simple CDC stream CREATE CHANGE DATA CAPTURE ON orders TO KAFKA 'localhost:9092' TOPIC 'orders-events' FORMAT JSON; -- Verify the stream SHOW CDC STREAMS; ``` --- ## Creating CDC Streams ### Basic Syntax ```sql CREATE CHANGE DATA CAPTURE ON table_name TO destination_type 'address' [TOPIC 'topic_name'] [FORMAT JSON|AVRO] [AS 'stream_name']; ``` ### Kafka Destination ```sql CREATE CHANGE DATA CAPTURE ON users TO KAFKA 'kafka.example.com:9092' TOPIC 'user-events' FORMAT JSON AS 'users_stream'; ``` ### AWS Kinesis Destination ```sql CREATE CHANGE DATA CAPTURE ON transactions TO KINESIS 'transactions-stream' REGION 'us-east-1' FORMAT AVRO; ``` ### Managing Streams ```sql -- List all streams SHOW CDC STREAMS; -- View stream details SHOW CDC STREAM STATUS users_stream; -- Pause a stream ALTER CDC STREAM users_stream PAUSE; -- Resume a stream ALTER CDC STREAM users_stream RESUME; -- Drop a stream DROP CHANGE DATA CAPTURE users_stream; ``` --- ## Event Filtering ### Table Filtering ```yaml # Filter by table filters: - name: product_updates tables: - products - product_inventory ``` ### Operation Filtering ```yaml # Filter by operation filters: - name: inserts_only tables: - orders operations: - INSERT ``` ### Column Filtering ```yaml # Filter by changed columns filters: - name: price_changes tables: - products columns: products: - price - discount ``` ### Custom Predicates ```yaml # Filter with SQL predicate filters: - name: high_value_orders tables: - orders operations: - INSERT predicate: "total > 1000" ``` --- ## Event Transformation ### Event Format Default event structure: ```json { "event_id": "evt_abc123", "timestamp": "2026-01-04T12:00:00Z", "sequence": 12345, "operation": "INSERT", "database": "production", "table": "orders", "key": {"id": 123}, "value": { "id": 123, "customer_id": 456, "total": 99.99, "status": "pending" }, "old_value": null, "transaction_id": "txn_xyz789" } ``` ### Jinja2 Templates ```yaml transformations: - name: slack_format target: https://hooks.slack.com/services/... template: | { "text": "New {{ operation }} on {{ table }}", "blocks": [ { "type": "section", "text": { "type": "mrkdwn", "text": "*{{ operation }}* on `{{ table }}`\nKey: {{ key }}" } } ] } ``` ### JSONPath Extraction ```yaml transformations: - name: extract_fields target: https://analytics.io/webhooks/orders jsonpath: order_id: "$.key.id" customer: "$.value.customer_id" amount: "$.value.total" operation: "$.operation" ``` --- ## Delivery Configuration ### Retry Strategy ```yaml delivery: max_retries: 3 initial_retry_delay_ms: 100 max_retry_delay_ms: 30000 timeout_secs: 30 ``` Retry schedule with exponential backoff: - Attempt 1: Immediate - Attempt 2: 100ms + jitter - Attempt 3: 200ms + jitter - Attempt 4: 400ms + jitter ### Worker Pool ```yaml worker_pool: num_workers: 100 max_concurrent_requests: 1000 connection_pool_size: 500 ``` --- ## Circuit Breaker The circuit breaker prevents cascading failures when webhook endpoints are unhealthy. ### States | State | Description | Behavior | |-------|-------------|----------| | **Closed** | Normal operation | All requests flow through | | **Open** | Too many failures | Requests rejected immediately | | **Half-Open** | Testing recovery | Limited requests allowed | ### Configuration ```yaml circuit_breaker: failure_threshold: 5 # Open after 5 failures success_threshold: 2 # Close after 2 successes in half-open timeout_secs: 60 # Stay open for 60 seconds half_open_max_calls: 3 # Max calls in half-open state ``` --- ## Exactly-Once Delivery ### How It Works 1. Each event generates a unique idempotency key 2. Delivered events are logged with their idempotency key 3. Retries check the delivery log before sending 4. Duplicate deliveries are prevented within the delivery window ### Configuration ```yaml exactly_once: delivery_window_secs: 86400 # 24 hours ``` --- ## Security ### HMAC Signature All webhook requests include an HMAC-SHA256 signature: ```http POST /webhooks/db-changes HTTP/1.1 Host: api.example.com Content-Type: application/json X-HeliosDB-Signature: sha256=f7bc83f430538424b13298e6aa6fb143ef4d59a14946175997479dbc2d1a3cd8 X-HeliosDB-Event-ID: evt_001 X-HeliosDB-Timestamp: 2026-01-04T12:00:00Z ``` ### Verifying Signatures (Receiver Side) ```python import hmac import hashlib def verify_signature(payload, signature, secret): expected = 'sha256=' + hmac.new( secret.encode(), payload, hashlib.sha256 ).hexdigest() return hmac.compare_digest(signature, expected) ``` ### TLS Configuration ```yaml security: tls: enabled: true cert_path: /etc/heliosdb/tls/cert.pem key_path: /etc/heliosdb/tls/key.pem ip_whitelist: - 10.0.0.0/8 - 172.16.0.0/12 ``` --- ## Monitoring ### Prometheus Metrics ```promql # Event processing rate rate(heliosdb_cdc_events_processed_total[5m]) # Webhook delivery latency (P99) histogram_quantile(0.99, rate(heliosdb_webhook_delivery_duration_seconds_bucket[5m])) # Delivery success rate sum(rate(heliosdb_webhook_delivery_total{status="success"}[5m])) / sum(rate(heliosdb_webhook_delivery_total[5m])) # Circuit breaker state heliosdb_circuit_breaker_state == 2 # Open circuits # Event queue backlog heliosdb_event_queue_size{queue_type="overflow"} ``` ### Available Metrics | Metric | Type | Description | |--------|------|-------------| | `heliosdb_cdc_events_processed_total` | Counter | Events processed | | `heliosdb_cdc_events_filtered_total` | Counter | Events filtered out | | `heliosdb_webhook_delivery_duration_seconds` | Histogram | Delivery latency | | `heliosdb_webhook_delivery_total` | Counter | Delivery attempts | | `heliosdb_webhook_retry_total` | Counter | Retry attempts | | `heliosdb_circuit_breaker_state` | Gauge | Circuit state | | `heliosdb_event_queue_size` | Gauge | Queue size | ### Grafana Dashboard Import the HeliosDB CDC dashboard for real-time monitoring of: - Event throughput - Delivery latency - Success rates - Circuit breaker status - Queue backlogs --- ## Best Practices ### 1. Use Appropriate Filters ```yaml # Filter to reduce volume filters: - name: relevant_changes tables: - orders # Only tables you need operations: - INSERT - UPDATE # Skip DELETEs if not needed columns: orders: - status - total # Only columns you care about ``` ### 2. Configure Retries Appropriately ```yaml # Balance between reliability and latency delivery: max_retries: 3 # Enough for transient failures initial_retry_delay_ms: 100 # Quick first retry max_retry_delay_ms: 5000 # Don't wait too long ``` ### 3. Monitor Circuit Breakers ```promql # Alert on open circuits ALERT WebhookCircuitOpen IF heliosdb_circuit_breaker_state == 2 FOR 5m LABELS { severity = "warning" } ``` ### 4. Implement Idempotent Receivers ```python def handle_webhook(event): event_id = event['event_id'] # Check if already processed if already_processed(event_id): return {"status": "duplicate", "event_id": event_id} # Process event process_event(event) # Mark as processed mark_processed(event_id) return {"status": "success", "event_id": event_id} ``` ### 5. Set Appropriate Timeouts ```yaml delivery: timeout_secs: 30 # Match your endpoint's expected response time ``` --- ## Related Documentation - [README.md](https://www.heliosdb.com/docs/full/features/webhooks/readme/) - Feature overview - [TROUBLESHOOTING.md](https://www.heliosdb.com/docs/full/features/webhooks/troubleshooting/) - Common issues - [SQL Interface](https://www.heliosdb.com/docs/full/features/sql-interface/sql_command_extensions/) - SQL commands