​

How Webhooks Work in Nekt

​

Configuring Webhook as a Source

​

1. Configure webhook settings

• Webhook API Key: Enable this option if you want to require an API key for incoming requests.
  • API Key Header: The HTTP header name where the API key will be sent (e.g., x-api-key, Authorization).
  • API Key Value: The expected value of the API key. Requests without this key (or with an incorrect key) will be rejected.
• Use payload schema template: (Default: false) If enabled, each message is projected into typed columns (with dynamic type inference) instead of wrapping the whole body into a single payload column. The following options apply only when this is enabled:
  • Payload records JSONPath: (Default: $) JSONPath evaluated against each message body to extract record(s). The default $ emits the whole body as one record. Use $.records[*] to fan a single envelope into one row per item, or $.data to unwrap a sub-object.
  • Primary keys: The primary keys that should be used to uniquely identify each message.
  • Fixed schema (JSON file): Optional. Upload a JSON file with a Singer schema (e.g., {"type": "object", "properties": {"id": {"type": ["string", "null"]}}}) to pin the column types instead of inferring them from the messages. This is recommended whenever the producer isn’t strict about types: inference only looks at a sample, so a field that arrives as "123" in one message and 123 in the next can change type between runs and break the destination. The _nekt_* metadata columns are added automatically. When set, schema inference is skipped entirely.
  • Number of messages to infer the schema: (Default: 10) Dictates how many messages to sample when inferring column types. Sampling more messages makes the inferred types more accurate: a field seen with conflicting scalar types across the sample becomes a string instead of changing type between runs. This setting is ignored when a fixed schema is uploaded.
• Delete messages: (Default: false) Mark as true if you want messages to be deleted (SQS) or acknowledged (Pub/Sub) after processing.

​

2. Get your Webhook URL

​

3. Configure your external system

1. Use the POST method
2. Send data as JSON in the request body
3. Include the API key header (if you configured authentication)
4. Set Content-Type: application/json

curl -X POST "https://your-webhook-url.nekt.ai/webhook/abc123" \
  -H "Content-Type: application/json" \
  -H "x-api-key: your-api-key" \
  -d '{
    "event_type": "order_created",
    "order_id": "12345",
    "customer_email": "customer@example.com",
    "total": 99.99,
    "items": [
      {"sku": "PROD-001", "quantity": 2}
    ]
  }'

​

4. Configure data streams

• Layer: Choose the layer in your catalog where the webhook data will be stored.
• Folder: Optionally create a folder to organize your webhook data.
• Table name: Name for the table that will contain the webhook events.

​

5. Configure data source

• Every 15 minutes: Near real-time processing for time-sensitive data
• Every hour: Balanced approach for most use cases
• Daily: Cost-effective for non-urgent data

​

6. Check your new source

​

Streams and Fields

​

Use Cases for Data Analysis

​

1. Parse JSON Payload

SELECT
  message_id,
  webhook_received_at,
  JSON_EXTRACT_SCALAR(payload, '$.event_type') AS event_type,
  JSON_EXTRACT_SCALAR(payload, '$.order_id') AS order_id,
  JSON_EXTRACT_SCALAR(payload, '$.customer_email') AS customer_email,
  CAST(JSON_EXTRACT_SCALAR(payload, '$.total') AS DOUBLE) AS total_amount
FROM
  nekt_raw.webhooks
WHERE
  webhook_received_at >= CURRENT_DATE - INTERVAL '7' DAY
ORDER BY
  webhook_received_at DESC

SELECT
  message_id,
  webhook_received_at,
  JSON_VALUE(payload, '$.event_type') AS event_type,
  JSON_VALUE(payload, '$.order_id') AS order_id,
  JSON_VALUE(payload, '$.customer_email') AS customer_email,
  CAST(JSON_VALUE(payload, '$.total') AS FLOAT64) AS total_amount
FROM
  `nekt_raw.webhooks`
WHERE
  DATE(webhook_received_at) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)
ORDER BY
  webhook_received_at DESC

​

2. Event Type Distribution

SELECT
  JSON_EXTRACT_SCALAR(payload, '$.event_type') AS event_type,
  COUNT(*) AS event_count,
  MIN(webhook_received_at) AS first_received,
  MAX(webhook_received_at) AS last_received
FROM
  nekt_raw.webhooks
WHERE
  webhook_received_at >= CURRENT_DATE - INTERVAL '30' DAY
GROUP BY
  JSON_EXTRACT_SCALAR(payload, '$.event_type')
ORDER BY
  event_count DESC

SELECT
  JSON_VALUE(payload, '$.event_type') AS event_type,
  COUNT(*) AS event_count,
  MIN(webhook_received_at) AS first_received,
  MAX(webhook_received_at) AS last_received
FROM
  `nekt_raw.webhooks`
WHERE
  DATE(webhook_received_at) >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
GROUP BY
  event_type
ORDER BY
  event_count DESC

​

3. Create a Materialized View

-- Create a structured view of your webhook data
SELECT
  message_id,
  webhook_received_at,
  JSON_EXTRACT_SCALAR(payload, '$.event_type') AS event_type,
  JSON_EXTRACT_SCALAR(payload, '$.order_id') AS order_id,
  JSON_EXTRACT_SCALAR(payload, '$.customer_email') AS customer_email,
  JSON_EXTRACT_SCALAR(payload, '$.customer_name') AS customer_name,
  CAST(JSON_EXTRACT_SCALAR(payload, '$.total') AS DOUBLE) AS total_amount,
  JSON_EXTRACT_SCALAR(payload, '$.currency') AS currency,
  JSON_EXTRACT_SCALAR(payload, '$.status') AS status,
  JSON_EXTRACT(payload, '$.items') AS items_json
FROM
  nekt_raw.webhooks

-- Create a structured view of your webhook data
SELECT
  message_id,
  webhook_received_at,
  JSON_VALUE(payload, '$.event_type') AS event_type,
  JSON_VALUE(payload, '$.order_id') AS order_id,
  JSON_VALUE(payload, '$.customer_email') AS customer_email,
  JSON_VALUE(payload, '$.customer_name') AS customer_name,
  CAST(JSON_VALUE(payload, '$.total') AS FLOAT64) AS total_amount,
  JSON_VALUE(payload, '$.currency') AS currency,
  JSON_VALUE(payload, '$.status') AS status,
  JSON_QUERY(payload, '$.items') AS items_json
FROM
  `nekt_raw.webhooks`

​

Implementation Notes

​

Data Format

• If using the default raw payload, all webhook data is stored as a JSON string in the payload field. Use your data warehouse’s JSON functions to extract specific fields, or consider creating transformations to parse frequently accessed data into structured tables.
• If using payload schema extraction, JSON fields are mapped directly to table columns. A field seen with conflicting scalar types during inference (e.g., an integer in one message and a string in the next) will be automatically cast to a string type to prevent data coercion failures.

​

Message Handling

• Each webhook request creates one record in the stream.
• Messages are processed incrementally on each extraction run.
• For raw payloads, the message_id field uniquely identifies each webhook event, and webhook_received_at reflects when Nekt received the webhook (not when the event occurred in the source system).
• For parsed payloads, the metadata columns _nekt_message_id and _nekt_webhook_received_at serve these same purposes.
• When performing schema inference, the connector temporarily peeks at messages in the queue without consuming them, keeping them available to be processed normally during the actual sync.

​

Best Practices

1. Provide a Fixed Schema: If your external system tends to send mixed types for the same field, upload a fixed schema JSON file to enforce strict structure and avoid flapping schema definitions between runs.
2. Include timestamps in your payload: Add an event_timestamp or similar field in your webhook payload to track when the event actually occurred in the source system.
3. Use consistent payload structures: Maintain a consistent JSON structure across all webhook events to simplify data parsing.
4. Include event types: Add an event_type field to categorize different types of events.
5. Add idempotency keys: Include a unique identifier in your payload to detect and handle duplicate webhook deliveries.
6. Test with sample data: Send test webhooks to verify your integration before enabling production webhooks.

​

Troubleshooting

​

Skills for agents