Track Usage From Your Backend

This guide shows how to instrument your backend to send usage events to Brilo. Events are the foundation of usage-based billing — everything downstream (metrics, pricing, invoices) depends on events being tracked reliably.

Where to Track

Track an event as close to the action as possible — when the billable thing actually happens, not after a batch process or at the end of the day. This ensures your draft invoices and usage dashboards reflect real-time state.

Good places to track:

Inside the API handler that serves the request
In a middleware layer that wraps billable routes
In the service method that performs the billable action
In a message queue consumer if the action is async

Python

import os
import hashlib
import requests
from datetime import datetime, timezone

BRILO_API_KEY = os.environ["BRILO_API_KEY"]
BRILO_BASE_URL = "https://api.brilo.tech"

def track_event(event_name, customer_id, properties=None, timestamp=None):
    if timestamp is None:
        timestamp = datetime.now(timezone.utc).isoformat()

    # Deterministic idempotency ID
    raw = f"{customer_id}:{event_name}:{timestamp}"
    idempotency_id = hashlib.sha256(raw.encode()).hexdigest()[:32]

    payload = {
        "event_name": event_name,
        "customer_id": customer_id,
        "idempotency_id": idempotency_id,
        "time_created": timestamp,
        "properties": properties or {}
    }

    response = requests.post(
        f"{BRILO_BASE_URL}/api/track",
        headers={"X-API-KEY": BRILO_API_KEY, "Content-Type": "application/json"},
        json=payload,
        timeout=5
    )
    response.raise_for_status()
    return response

# Usage
track_event(
    event_name="api_call_made",
    customer_id="cust_123",
    properties={"endpoint": "/v1/messages", "tokens_used": 1200}
)

Non-blocking tracking with a thread

To avoid adding latency to your API response:

import threading

def track_event_async(event_name, customer_id, properties=None):
    thread = threading.Thread(
        target=track_event,
        args=(event_name, customer_id, properties)
    )
    thread.daemon = True
    thread.start()

Node.js

const crypto = require("crypto");

const BRILO_API_KEY = process.env.BRILO_API_KEY;
const BRILO_BASE_URL = "https://api.brilo.tech";

async function trackEvent(eventName, customerId, properties = {}, timestamp = null) {
  const time = timestamp || new Date().toISOString();

  const raw = `${customerId}:${eventName}:${time}`;
  const idempotencyId = crypto.createHash("sha256").update(raw).digest("hex").slice(0, 32);

  const payload = {
    event_name: eventName,
    customer_id: customerId,
    idempotency_id: idempotencyId,
    time_created: time,
    properties
  };

  const response = await fetch(`${BRILO_BASE_URL}/api/track`, {
    method: "POST",
    headers: {
      "X-API-KEY": BRILO_API_KEY,
      "Content-Type": "application/json"
    },
    body: JSON.stringify(payload)
  });

  if (!response.ok) {
    throw new Error(`Brilo tracking failed: ${response.status}`);
  }

  return response;
}

// Usage
await trackEvent("api_call_made", "cust_123", {
  endpoint: "/v1/messages",
  tokens_used: 1200
});

Fire and forget (non-blocking)

// Don't await — let it run in the background
trackEvent("api_call_made", "cust_123", { tokens_used: 1200 })
  .catch(err => console.error("Brilo tracking error:", err));

Retry Logic

Network requests fail. Add retry with exponential backoff around event tracking to ensure events reach Brilo:

import time

def track_event_with_retry(event_name, customer_id, properties=None, max_retries=3):
    for attempt in range(max_retries):
        try:
            return track_event(event_name, customer_id, properties)
        except requests.exceptions.RequestException as e:
            if attempt == max_retries - 1:
                # Log failure and move on — don't block your main flow
                print(f"Brilo tracking failed after {max_retries} attempts: {e}")
                return None
            wait = (2 ** attempt) + (random.random() * 0.5)
            time.sleep(wait)

Middleware Pattern

For APIs where every request to certain routes is billable, use middleware to track automatically:

# Flask example
from functools import wraps

def track_usage(event_name, get_properties=None):
    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            result = f(*args, **kwargs)
            properties = get_properties(result) if get_properties else {}
            track_event_async(event_name, current_user.customer_id, properties)
            return result
        return wrapper
    return decorator

@app.route("/api/messages", methods=["POST"])
@track_usage("message_sent", get_properties=lambda r: {"characters": len(r.data)})
def send_message():
    # ... handle request

Backfilling Historical Events

If you're migrating to Brilo and need to load historical usage, send events with the original time_created timestamps:

for row in historical_usage_data:
    track_event(
        event_name=row["event_type"],
        customer_id=row["customer_id"],
        properties=row["metadata"],
        timestamp=row["created_at"].isoformat()
    )
    time.sleep(0.05)  # Rate limit yourself to avoid hitting API limits

Use deterministic idempotency IDs based on the original record so re-runs don't double-count.

Verifying Your Setup

After tracking a few events, confirm they arrived:

curl -X POST https://api.brilo.tech/api/verify-events \
  -H "X-API-KEY: your-api-key" \
  -H "Content-Type: application/json" \
  -d '{
    "idempotency_ids": ["your-idempotency-id-1", "your-idempotency-id-2"],
    "lookback_days": 1
  }'

You can also view recent events in the dashboard under Events.

Next Steps

activity

Events & Metering

Understand how events become billable metrics

check-circle

Test Your Integration

Validate everything before going live