Skip to content

Event Pipeline Design

Best practices for designing your event pipeline: naming conventions, attribute schemas, contact field strategies, and batching patterns.

Event Naming

Use a consistent {domain}_{action} pattern in snake_case:

Good Names

order_completed
claim_submitted
payment_failed
subscription_activated
onboarding_step_completed
password_reset_requested

Avoid

OrderCompleted # camelCase -- use snake_case
order.completed # dots -- use underscores
order-completed # hyphens -- use underscores
order # too generic -- what happened to the order?
completed # no domain context
user_event # meaningless

Naming Guidelines

  1. Domain first -- order_, claim_, payment_, subscription_
  2. Past tense actions -- _completed, _submitted, _failed, _activated
  3. Specific over generic -- claim_submitted not claim_updated
  4. Consistent vocabulary -- Pick one word and stick with it (e.g., always completed not sometimes finished)
Note

Event names are the contract between your backend and Synapse flows. Treat them like API endpoints -- stable, documented, and versioned if changed.

Attribute Design

Include What Templates Need

Before designing your event attributes, list the email templates that will use the event. Include every field those templates reference.

json
{
"event_name": "claim_submitted",
"attributes": {
"reference_number": "WIC-2026-0001",
"worker_name": "John Doe",
"incident_date": "2026-03-15T14:30:00+08:00",
"claim_type": "work_injury",
"claim_status_url": "https://app.example.com/claims/WIC-2026-0001"
}
}

Attribute Best Practices

PracticeExample
Use snake_case keysreference_number, not ReferenceNumber
Include human-readable values"worker_name": "John Doe", not just "worker_id": "w_123"
Include URLs for CTAs"claim_status_url": "https://..."
Use ISO 8601 for dates"incident_date": "2026-03-15T14:30:00+08:00"
Include currency with amounts"amount": 149.99, "currency": "USD"
Avoid PII you do not needDo not send SSN, full bank account, etc.

Nested Attributes

Synapse supports nested objects and arrays:

json
{
"attributes": {
"order_id": "ORD-001",
"items": [
{"name": "Widget Pro", "quantity": 2, "price": 49.99},
{"name": "Gadget X", "quantity": 1, "price": 50.01}
],
"shipping": {
"method": "express",
"estimated_date": "2026-04-10"
}
}
}

NLT can access these as:

  • {the Order ID from the trigger event}
  • {for item in the Items from the trigger event} ... {item.Name} ... {end for}
  • {the Shipping.Method from the trigger event}

Contact Field Strategies

Send contact data with every event. This ensures contacts are always up to date:

json
{
"event_name": "order_completed",
"external_id": "user_12345",
"contact": {
"email": "[email protected]",
"first_name": "Jane",
"last_name": "Doe",
"timezone": "Asia/Singapore"
},
"attributes": {"order_id": "ORD-001"}
}

Strategy 2: Separate Contact Management

Manage contacts via POST /v1/contacts (upsert) or PATCH /v1/contacts/{external_id} and only send events with external_id:

json
{
"event_name": "order_completed",
"external_id": "user_12345",
"attributes": {"order_id": "ORD-001"}
}
Tip

Strategy 1 is simpler and more resilient -- even if a contact was not pre-created, the event creates it. Strategy 2 gives you more control over when contact data changes and supports tag operators ($add_tags, $remove_tags).

Strategy 3: Contact Data Only on Signup

Send full contact data on the first event (signup_completed) and only external_id on subsequent events:

python
# First event: create the contact
client.events.create(
event_name="signup_completed",
external_id=user.id,
contact={
"email": user.email,
"first_name": user.first_name,
"last_name": user.last_name,
"timezone": user.timezone,
"tags": ["new-user"],
"properties": {"plan": user.plan},
},
attributes={"source": "landing_page"},
)
 
# Subsequent events: just the event data
client.events.create(
event_name="order_completed",
external_id=user.id,
attributes={"order_id": order.id, "amount": order.total},
)

Batching and Throughput

Single Event Ingestion

For real-time events (user actions, transactions), send events individually:

python
# Triggered in your request handler
client.events.create(
event_name="order_completed",
external_id=order.user_id,
attributes={...},
idempotency_key=f"order_completed_{order.id}",
)

Batch Import (Historical Events)

For migrating historical data or bulk imports, use the contacts bulk endpoint for contacts and send events in parallel:

python
import concurrent.futures
 
def send_event(event_data):
return client.events.create(**event_data)
 
events = [
{"event_name": "signup_completed", "external_id": f"user_{i}", "attributes": {}}
for i in range(1000)
]
 
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
results = list(executor.map(send_event, events))
Warning

Respect rate limits when batching. The default rate limit is 100 requests per minute per API key. Use exponential backoff when you receive 429 responses.

Event Pipeline Architecture

Your Application
│
├── User action occurs (order, claim, signup)
│
├── Persist to your database (source of truth)
│
├── Publish event to Synapse (async, fire-and-forget)
│ POST /v1/events with idempotency_key
│
└── Handle 202/409/429 responses
202: Success (event queued)
409: Duplicate (already processed, safe to ignore)
429: Rate limited (retry with backoff)

Retry Strategy

python
import time
import requests
 
def send_event_with_retry(event_data, max_retries=3):
for attempt in range(max_retries):
response = requests.post(
"https://synapse-api.pyrx.tech/v1/events",
headers=headers,
json=event_data,
)
 
if response.status_code == 202:
return response.json()
elif response.status_code == 409:
return {"status": "duplicate"} # already processed
elif response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
time.sleep(retry_after)
else:
time.sleep(2 ** attempt) # exponential backoff
 
raise Exception("Event delivery failed after retries")