Enforce event contracts in code and CI

GraphJSON accepts flexible JSON, but a production analytics system still needs an executable contract at the producer boundary.

A written tracking plan explains meaning. An executable schema verifies shape, types, allowed values, version, and forbidden fields before an event enters a queue or reaches GraphJSON.

product decision
  → written tracking plan
    → executable event schema
      → producer validation
        → GraphJSON transport envelope
          → contract and reconciliation tests

Separate two contracts#

Every event passes through two distinct contracts:

The OpenAPI description validates the transport. It cannot decide whether amount is in cents, whether account_id is required, or whether an email address is allowed.

Own the application event schema in the same repository as the producer.

Design a stable envelope#

Use common fields across event types:

type EventEnvelope = {
  event: string;
  event_id: string;
  occurred_at: number;
  schema_version: number;
  source: string;
  environment: "development" | "staging" | "production";
  user_id?: string;
  anonymous_id?: string;
  account_id?: string;
  properties: Record<string, unknown>;
};

Keep transport metadata outside the event where appropriate. occurred_at belongs in the event contract for validation and should also become the GraphJSON request timestamp.

Do not put the GraphJSON API key into the event object.

Zod contract#

For a TypeScript producer:

import { z } from "zod";

const baseEvent = z.object({
  event_id: z.string().uuid(),
  occurred_at: z.number().int().nonnegative(),
  schema_version: z.literal(1),
  source: z.enum(["web", "api", "worker"]),
  environment: z.enum(["development", "staging", "production"]),
  user_id: z.string().min(1).optional(),
  anonymous_id: z.string().min(1).optional(),
  account_id: z.string().min(1).optional()
}).strict();

const checkoutCompleted = baseEvent.extend({
  event: z.literal("checkout_completed"),
  account_id: z.string().min(1),
  properties: z.object({
    order_id: z.string().min(1),
    amount_minor: z.number().int().nonnegative(),
    currency: z.string().regex(/^[a-z]{3}$/),
    payment_provider: z.enum(["stripe", "other"])
  }).strict()
});

type CheckoutCompleted = z.infer<typeof checkoutCompleted>;

.strict() rejects unexpected properties. That is useful at a public first-party collection endpoint. Inside a trusted producer, you can choose an explicit strip policy instead, but do not allow arbitrary request objects to flow through accidentally.

Validate before queueing:

function enqueueCheckoutCompleted(input: unknown) {
  const event = checkoutCompleted.parse(input);
  return analyticsOutbox.insert({
    event_id: event.event_id,
    occurred_at: event.occurred_at,
    collection: "product_events",
    payload: event
  });
}

The outbox stores only validated events. A retry should resend the same event_id, occurrence time, and payload.

JSON Schema contract#

Use JSON Schema when several languages produce the same event:

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "$id": "https://example.com/events/checkout-completed-v1.json",
  "title": "checkout_completed v1",
  "type": "object",
  "required": [
    "event",
    "event_id",
    "occurred_at",
    "schema_version",
    "account_id",
    "properties"
  ],
  "properties": {
    "event": {"const": "checkout_completed"},
    "event_id": {"type": "string", "format": "uuid"},
    "occurred_at": {"type": "integer", "minimum": 0},
    "schema_version": {"const": 1},
    "account_id": {"type": "string", "minLength": 1},
    "properties": {
      "type": "object",
      "required": ["order_id", "amount_minor", "currency"],
      "properties": {
        "order_id": {"type": "string", "minLength": 1},
        "amount_minor": {"type": "integer", "minimum": 0},
        "currency": {
          "type": "string",
          "pattern": "^[a-z]{3}$"
        }
      },
      "additionalProperties": false
    }
  },
  "additionalProperties": false
}

Check the schema itself in CI, then validate fixtures in every producer language.

Do not fetch a schema from the network on each production event. Pin an approved version with the application release.

Pydantic contract#

For Python:

from typing import Literal
from pydantic import BaseModel, ConfigDict, Field

class CheckoutProperties(BaseModel):
    model_config = ConfigDict(extra="forbid")

    order_id: str = Field(min_length=1)
    amount_minor: int = Field(ge=0)
    currency: str = Field(pattern=r"^[a-z]{3}$")

class CheckoutCompleted(BaseModel):
    model_config = ConfigDict(extra="forbid")

    event: Literal["checkout_completed"]
    event_id: str
    occurred_at: int = Field(ge=0)
    schema_version: Literal[1]
    account_id: str = Field(min_length=1)
    properties: CheckoutProperties

Validate at the point where the business operation has succeeded, not in a generic HTTP logger that cannot know whether the fact is true.

Enforce privacy rules#

Shape validation alone does not prevent unsafe data.

Add a second policy check:

const forbiddenKey = /(^|_)(password|secret|token|authorization|cookie)$/i;

function rejectForbiddenKeys(
  value: unknown,
  path: string[] = []
): void {
  if (!value || typeof value !== "object") return;

  for (const [key, nested] of Object.entries(value)) {
    const nextPath = [...path, key];
    if (forbiddenKey.test(key)) {
      throw new Error(`Forbidden analytics field: ${nextPath.join(".")}`);
    }
    rejectForbiddenKeys(nested, nextPath);
  }
}

A denylist is a backstop, not the main contract. Prefer an allowlisted schema that makes each accepted property deliberate.

Also enforce:

• maximum serialized size below the GraphJSON limit
• maximum string length
• allowed enum cardinality
• no raw request headers or bodies
• no complete URLs containing sensitive query parameters
• no free-form exception objects

Define compatibility#

Classify a schema change before release:

GraphJSON retains historical events. A schema migration does not rewrite the meaning of old rows.

Use a version window:

producer v1 + query supports v1
        ↓
producer v2 + query supports v1 and v2
        ↓
v2 adoption reaches retirement threshold
        ↓
query retires v1 compatibility deliberately

Version events deliberately#

Use one of these patterns:

Additive change#

Keep schema_version: 1 when the field is optional and existing meaning is unchanged.

Breaking shape change#

Increment schema_version and make queries handle the active window:

SELECT
  JSONExtractInt(json, 'schema_version') AS schema_version,
  count() AS events
FROM product_events
WHERE JSONExtractString(json, 'event') = 'checkout_completed'
GROUP BY schema_version
ORDER BY schema_version

Breaking semantic change#

Prefer a new event or property when the business fact changed. A version number cannot make two incompatible definitions safe to sum.

Build fixture tests#

Store:

analytics/
  schemas/
    checkout-completed-v1.schema.json
  fixtures/
    checkout-completed-v1.valid.json
    checkout-completed-v1.invalid-missing-account.json
    checkout-completed-v1.invalid-secret.json

CI should verify:

1. every valid fixture passes
2. every invalid fixture fails for the expected reason
3. the serialized event stays within the size budget
4. event time is whole Unix seconds
5. forbidden properties are rejected
6. collection routing is deterministic
7. schema and sample documentation agree

Do not write a test that asserts only “the analytics function was called.” Assert the completed business fact and exact payload.

Test the GraphJSON boundary#

Keep most tests local. Add one minimal integration contract test in a non-production workspace:

const event = checkoutCompleted.parse(validFixture);

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_TEST_API_KEY,
    collection: "contract_test_events",
    timestamp: event.occurred_at,
    json: JSON.stringify(event)
  })
});

expect(response.ok).toBe(true);

Use a stable fixture and bounded test frequency. Live contract tests should not become uncontrolled production traffic.

Handle rejected events#

Choose by importance:

Never place the rejected unsafe payload into ordinary logs. Record:

• event name
• schema version
• producer version
• safe reason code
• occurrence time
• opaque event ID

Keep any full quarantine payload in a restricted system designed for that purpose, with its own retention and access controls.

Release safely#

1. write the decision and metric definition
2. add the schema and fixtures
3. validate in the producer
4. deploy to staging
5. inspect Samples
6. run the intended query
7. deploy to a small production cohort
8. monitor schema-version adoption and rejections
9. reconcile a known business total
10. retire the old version only after the threshold is met

Ownership#

Every contract needs:

• product owner for meaning
• engineering owner for production
• data consumer owner for queries and dashboards
• privacy classification
• effective date
• deprecation policy

A schema repository without ownership becomes a directory of plausible-looking JSON that nobody can safely change.

Continue with Create a tracking plan, Design your event schema, and Monitor instrumentation health.