Server runtime recipes

GraphJSON uses a small HTTP API, so you do not need an SDK. The production rule is the same in every language: call from trusted server code, serialize the event once, send a whole Unix-second timestamp, check the HTTP status, and keep analytics failure handling explicit.

For important events, these snippets should run inside a durable job worker. Read Reliable event delivery before treating an inline request as a production pipeline.

For an end-to-end implementation with application boundaries, durable delivery, and tests, use the detailed reference for Next.js and Node, FastAPI and Django, Ruby on Rails, or Go.

Shared configuration#

Store these values in your deployment secret manager:

GRAPHJSON_API_KEY=...
GRAPHJSON_COLLECTION=product_events

Never use a public environment-variable prefix such as NEXT_PUBLIC_ for the API key.

All examples send:

{
  "event": "report_exported",
  "event_id": "evt_01J...",
  "user_id": "usr_42",
  "account_id": "acct_7",
  "format": "csv"
}

Create event_id when the business action occurs, before any retry. Keep it unchanged across delivery attempts.

Node.js and TypeScript#

Node 18 and later include fetch:

type AnalyticsEvent = {
  event: string;
  event_id: string;
  user_id?: string;
  account_id?: string;
  [key: string]: unknown;
};

export async function track(
  event: AnalyticsEvent,
  occurredAt = new Date()
) {
  const response = await fetch("https://api.graphjson.com/api/log", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      api_key: process.env.GRAPHJSON_API_KEY,
      collection:
        process.env.GRAPHJSON_COLLECTION || "product_events",
      timestamp: Math.floor(occurredAt.getTime() / 1000),
      json: JSON.stringify(event),
    }),
    signal: AbortSignal.timeout(8_000),
  });

  const body = await response.text();
  if (!response.ok) {
    throw new Error(`GraphJSON ${response.status}: ${body}`);
  }
}

Call track from a background job for durable delivery. If you use it inline for best-effort events, catch failures at the analytics boundary and report them without failing the completed user action.

Python#

This example uses httpx:

import json
import os
import time

import httpx

ENDPOINT = "https://api.graphjson.com/api/log"


def track(event: dict, occurred_at: float | None = None) -> None:
    response = httpx.post(
        ENDPOINT,
        json={
            "api_key": os.environ["GRAPHJSON_API_KEY"],
            "collection": os.getenv(
                "GRAPHJSON_COLLECTION", "product_events"
            ),
            "timestamp": int(occurred_at or time.time()),
            "json": json.dumps(event, separators=(",", ":")),
        },
        timeout=8.0,
    )
    response.raise_for_status()

In Django, FastAPI, or Flask, enqueue the event after the application transaction commits. Let Celery, RQ, Dramatiq, or your existing worker system call this function with a bounded retry policy.

Go#

Use one shared http.Client; do not create a new transport for every event:

package analytics

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"time"
)

var client = &http.Client{Timeout: 8 * time.Second}

func Track(ctx context.Context, event map[string]any, occurredAt time.Time) error {
	eventJSON, err := json.Marshal(event)
	if err != nil {
		return err
	}

	body, err := json.Marshal(map[string]any{
		"api_key":   os.Getenv("GRAPHJSON_API_KEY"),
		"collection": "product_events",
		"timestamp": occurredAt.Unix(),
		"json":      string(eventJSON),
	})
	if err != nil {
		return err
	}

	req, err := http.NewRequestWithContext(
		ctx,
		http.MethodPost,
		"https://api.graphjson.com/api/log",
		bytes.NewReader(body),
	)
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")

	response, err := client.Do(req)
	if err != nil {
		return err
	}
	defer response.Body.Close()

	if response.StatusCode < 200 || response.StatusCode >= 300 {
		return fmt.Errorf("graphjson returned %s", response.Status)
	}
	return nil
}

Pass the request context only when canceling delivery with the request is intentional. A durable worker should use its own job-scoped context so a disconnected browser does not cancel analytics delivery.

Ruby on Rails#

Put delivery in an ApplicationJob:

require "net/http"
require "time"

class DeliverAnalyticsEventJob < ApplicationJob
  queue_as :analytics

  def perform(event, occurred_at)
    uri = URI("https://api.graphjson.com/api/log")
    request = Net::HTTP::Post.new(
      uri,
      "Content-Type" => "application/json"
    )
    request.body = {
      api_key: ENV.fetch("GRAPHJSON_API_KEY"),
      collection: ENV.fetch(
        "GRAPHJSON_COLLECTION",
        "product_events"
      ),
      timestamp: Time.iso8601(occurred_at).to_i,
      json: JSON.generate(event)
    }.to_json

    response = Net::HTTP.start(
      uri.hostname,
      uri.port,
      use_ssl: true,
      open_timeout: 3,
      read_timeout: 8
    ) { |http| http.request(request) }

    return if response.is_a?(Net::HTTPSuccess)

    raise "GraphJSON returned #{response.code}"
  end
end

Configure job retries for network errors, 429, and server failures. A validation 400 should move to review rather than loop unchanged.

Enqueue after commit so a rolled-back record does not produce a successful analytics event:

DeliverAnalyticsEventJob.perform_later(event, Time.current.iso8601)

PHP and Laravel#

Laravel’s HTTP client provides timeouts and status handling:

use Illuminate\Support\Facades\Http;

function trackEvent(array $event, ?DateTimeInterface $occurredAt = null): void
{
    $response = Http::asJson()
        ->connectTimeout(3)
        ->timeout(8)
        ->post('https://api.graphjson.com/api/log', [
            'api_key' => config('services.graphjson.key'),
            'collection' => config(
                'services.graphjson.collection',
                'product_events'
            ),
            'timestamp' => ($occurredAt ?? now())->getTimestamp(),
            'json' => json_encode(
                $event,
                JSON_THROW_ON_ERROR
            ),
        ]);

    $response->throw();
}

Wrap this function in a queued job and customize retry decisions by status. Do not call it from a Blade template or expose the configured key to JavaScript.

Serverless and edge runtimes#

Short-lived runtimes need special care:

• an unawaited promise may be frozen or terminated
• an in-memory retry queue disappears with the instance
• retry storms can multiply concurrency
• edge runtimes may not support every Node package

For low-value events, await a direct fetch within the function’s execution budget. For important events, publish a small event envelope to a durable platform queue and let a separate consumer call GraphJSON.

Cloudflare Workers, Deno, Bun, Vercel functions, and modern AWS Lambda runtimes can all use the same fetch shape as the Node example. Store the key in the platform’s server-side secret system, set an explicit timeout, and never use a client-visible environment binding.

Send bulk batches from any runtime#

When a queue consumer has several events for one collection, call /api/bulk-log:

{
  "api_key": "your_api_key",
  "collection": "product_events",
  "timestamps": [1785081600, 1785081612],
  "jsons": [
    "{\"event\":\"report_exported\",\"event_id\":\"evt_1\"}",
    "{\"event\":\"report_exported\",\"event_id\":\"evt_2\"}"
  ]
}

Limit each request to 50 events. Keep source order and timestamp paired by array index, and retain the batch until a 2xx response.

Runtime checklist#

• The API key exists only in trusted server configuration.
• The collection is explicit for the deployment environment.
• Event JSON is serialized exactly once into the json string.
• Timestamps are whole Unix seconds.
• HTTP timeouts and non-2xx responses are handled.
• Important events use a durable queue or outbox.
• Retries keep the same event_id.
• Logs redact secrets and sensitive payload fields.

For complete endpoint fields and limits, see Log an event and Bulk log events.