Server-Side Events

POST /v1/events

HMAC-SHA256(secret_key, timestamp + "." + request_body)

{
  "events": [
    {
      "type": "track",
      "event": "order_completed",
      "user_id": "user_12345",
      "properties": {
        "order_id": "ORD-98765",
        "revenue": 109.97,
        "currency": "GBP",
        "products": [
          { "product_id": "SKU-1234", "name": "Wireless Headphones", "price": 79.99, "quantity": 1 },
          { "product_id": "SKU-5678", "name": "USB-C Cable", "price": 14.99, "quantity": 2 }
        ]
      },
      "context": {
        "ip": "203.0.113.42",
        "user_agent": "Mozilla/5.0 ..."
      },
      "timestamp": "2026-02-25T14:30:00.000Z"
    }
  ]
}

{
  "rejected": [
    { "message_id": "evt_002", "reason": "invalid" }
  ]
}

SECRET="your_pipeline_secret"
BODY='{"events":[{"type":"track","event":"order_completed","user_id":"user_12345","properties":{"order_id":"ORD-98765","revenue":109.97,"currency":"GBP"}}]}'
TIMESTAMP=$(date +%s000)
SIGNATURE=$(printf '%s.%s' "$TIMESTAMP" "$BODY" | openssl dgst -sha256 -hmac "$SECRET" | awk '{print $2}')
 
curl -X POST https://collect.example.com/v1/events \
  -H "Content-Type: application/json" \
  -H "X-Pipeline-Key: your_pipeline_key" \
  -H "X-Signature: sha256=$SIGNATURE" \
  -H "X-Timestamp: $TIMESTAMP" \
  -d "$BODY"

import crypto from 'node:crypto'
 
const SECRET = process.env.DATAFLY_PIPELINE_SECRET
const ENDPOINT = 'https://collect.example.com/v1/events'
 
async function send(events) {
  // Serialise once and sign exactly what is sent.
  const body = JSON.stringify({ events })
  const timestamp = Date.now().toString()
  const signature = crypto
    .createHmac('sha256', SECRET)
    .update(`${timestamp}.${body}`)
    .digest('hex')
 
  const res = await fetch(ENDPOINT, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'X-Pipeline-Key': process.env.DATAFLY_PIPELINE_KEY,
      'X-Signature': `sha256=${signature}`,
      'X-Timestamp': timestamp,
    },
    body,
  })
 
  if (!res.ok) {
    throw new Error(`Datafly ingest failed: ${res.status} ${await res.text()}`)
  }
}
 
await send([
  {
    type: 'track',
    event: 'order_completed',
    user_id: 'user_12345',
    properties: { order_id: 'ORD-98765', revenue: 109.97, currency: 'GBP' },
  },
])

import hashlib
import hmac
import json
import os
import time
 
import requests
 
SECRET = os.environ["DATAFLY_PIPELINE_SECRET"]
ENDPOINT = "https://collect.example.com/v1/events"
 
 
def send(events):
    # Serialise once and sign exactly what is sent.
    body = json.dumps({"events": events}, separators=(",", ":"))
    timestamp = str(int(time.time() * 1000))
    signature = hmac.new(
        SECRET.encode(),
        f"{timestamp}.{body}".encode(),
        hashlib.sha256,
    ).hexdigest()
 
    response = requests.post(
        ENDPOINT,
        data=body,
        headers={
            "Content-Type": "application/json",
            "X-Pipeline-Key": os.environ["DATAFLY_PIPELINE_KEY"],
            "X-Signature": f"sha256={signature}",
            "X-Timestamp": timestamp,
        },
        timeout=10,
    )
    response.raise_for_status()
 
 
send([
    {
        "type": "track",
        "event": "order_completed",
        "user_id": "user_12345",
        "properties": {"order_id": "ORD-98765", "revenue": 109.97, "currency": "GBP"},
    }
])

package main
 
import (
	"bytes"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"time"
)
 
type Event struct {
	Type       string                 `json:"type"`
	Event      string                 `json:"event,omitempty"`
	UserID     string                 `json:"user_id,omitempty"`
	Properties map[string]interface{} `json:"properties,omitempty"`
}
 
func send(events []Event) error {
	// Marshal once and sign exactly what is sent.
	body, err := json.Marshal(map[string][]Event{"events": events})
	if err != nil {
		return err
	}
 
	timestamp := fmt.Sprintf("%d", time.Now().UnixMilli())
	mac := hmac.New(sha256.New, []byte(os.Getenv("DATAFLY_PIPELINE_SECRET")))
	mac.Write([]byte(timestamp + "." + string(body)))
	signature := hex.EncodeToString(mac.Sum(nil))
 
	req, err := http.NewRequest("POST", "https://collect.example.com/v1/events", bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("X-Pipeline-Key", os.Getenv("DATAFLY_PIPELINE_KEY"))
	req.Header.Set("X-Signature", "sha256="+signature)
	req.Header.Set("X-Timestamp", timestamp)
 
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
 
	// 204: all accepted. 200: accepted except the events listed in "rejected".
	if resp.StatusCode < 200 || resp.StatusCode > 299 {
		return fmt.Errorf("datafly ingest failed: %s", resp.Status)
	}
	return nil
}