​

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:
  • Save payload as JSON string (Advanced): (Default: false) If enabled, each message is stored whole as a JSON string in a single payload column instead of being projected into typed columns. Use this when the producer sends a different shape per event type, or keeps adding fields inside nested objects: those changes cannot be applied to an existing nested column and break the sync. Takes precedence over the fixed schema and disables schema inference.
  • 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. Note that if Save payload as JSON string is enabled, keys referencing fields inside the payload are ignored; use _nekt_message_id to keep rows deduplicated.
  • 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 between the existing layers on your catalog. This is where you will find your new extracted tables as the extraction runs successfully.
• Folder: a folder can be created inside the selected layer to group all tables being created from this new data source.
• Table name: we suggest a name, but feel free to customize it. You have the option to add a prefix to all tables at once and make this process faster!
• Sync Type: you can choose between INCREMENTAL and FULL_TABLE. For webhooks, the sync type is always INCREMENTAL since webhooks are event-driven and each event is a new record.

​

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

• Configure Delta Log Retention and determine for how long we should store old states of this table as it gets updated. Read more about this resource here.

​

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. Alternatively, enable Save payload as JSON string to bypass column projection and store each extracted record safely as a full JSON string in a single payload column.

​

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.

​

Cloud Provider Infrastructure

​

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. Use JSON String Fallback: If the producer sends a completely different shape per event type or continually adds fields inside nested objects, enable Save payload as JSON string to prevent nested column errors and schema breakage.
3. 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.
4. Use consistent payload structures: Maintain a consistent JSON structure across all webhook events to simplify data parsing.
5. Include event types: Add an event_type field to categorize different types of events.
6. Add idempotency keys: Include a unique identifier in your payload to detect and handle duplicate webhook deliveries.
7. Test with sample data: Send test webhooks to verify your integration before enabling production webhooks.

​

Troubleshooting

​

Skills for agents