The live stream

GET /v1/stream?lat=&lon=&radius=&type=&min_confidence=
Authorization: Bearer <key>          (or ?ticket= for a browser)
Accept: text/event-stream

This is the product. Everything else — publishing, trust, geo indexing — exists so that this can be fast and correct.

It is also the part most integrations get wrong, in ways that look like success. Read the whole page before you write a client; it is short, and each section below is a bug someone has already shipped.

curl -sN -H "Authorization: Bearer $RELAY_KEY" \
  "$RELAY/v1/stream?lat=48.8698&lon=2.3078&radius=3000"

Parameters are identical to GET /v1/events, including the 9 780 m radius cap. A bad radius or an unknown type is a 422 before the stream starts — after the first byte there is no status code left to send.

The four frames

Plain SSE. Every frame is event: <name> then data: <json>.

Literally, on the wire:

// event: snapshot
{ "events": [ /* array of event objects */ ],
  "watch": {"lat": 48.8698, "lon": 2.3078, "radius_m": 3000.0},
  "max_seconds": 3300 }
 
// event: event          <- the delta. NOTE THE WRAPPER.
{ "change": "created" | "updated" | "expired", "event": { /* one event object */ } }
 
// event: keepalive
{ "dropped": 0 }
 
// event: bye
{ "reason": "max_connection_age" | "snapshot_unavailable", "reconnect": true }

We never send an SSE id: or retry: field. Last-Event-ID is therefore always empty — see No replay log.

event is an envelope, not an event

// WRONG — silently ignores every change, forever
es.addEventListener("event", e => {
  const ev = JSON.parse(e.data);
  store.set(ev.event_id, ev);        // ev.event_id is undefined
});
 
// RIGHT
es.addEventListener("event", e => {
  const { change, event } = JSON.parse(e.data);
  apply(change, event);
});

The SSE event name is event, and its payload is {"change": …, "event": {…}}. A client that reads event_id off the delta gets undefined and writes garbage into its store — or, more commonly, drops the delta on a truthiness check and moves on.

What that failure looks like: a connection that is open, authenticated, receiving bytes, emitting no errors, and delivering nothing. The snapshot arrives and renders perfectly, so the map looks right for the first frame and then freezes. There is no error anywhere in that picture, in any log, on either side.

This is not a hypothetical. It is the first bug our own Explorer hit, on day one.

Read .event for the payload and .change for what happened to it.

The three change values

change and event.status are not the same field. Switch on change for what to do with your store, and read status if you want to tell your user why it went away. An early death by deny quorum — enough vehicles reporting the hazard gone — arrives as change: "expired" with event.status: "expired" and expires_at pulled to that moment:

{"change": "expired",
 "event": {"event_id": "7fce…", "status": "expired", "confidence": 0.0,
           "deny_count": 4, "updated_at": "2026-08-03T13:19:54.931585Z", …}}

The status vocabulary also carries invalidated — "this was never true", a verdict about a publisher that is reserved to moderation: denials clear an event, they never condemn it (see Trust). It arrives through this same frame, so treat any status on an expired delta the same way: remove. expired is the only removal signal you will ever receive, and it does not cover the common case — see the next section.

Natural expiry is silent

An event that reaches its expires_at sends no delta. None. Your client must drop events on their own expires_at.

This is a contract decision, not an omission. A TTL is knowledge you already hold — it is in every copy of the event you have — and broadcasting the death of every event would double the fan-out to tell every subscriber something it can compute locally.

So there are two ways an event leaves your store, and you must implement both:

1. A delta with change: "expired" — an early death: enough vehicles denied it, or an upstream public feed stopped listing the incident.
2. Your own clock reaching expires_at — the ordinary case, the majority of events, and the one that produces no traffic at all.

A client that only implements (1) accumulates stale hazards forever and shows a driver a 20-minute-old wrong-way alert on a road that is now clear. That is not a cosmetic bug: it is phantom braking.

Note that expires_at moves. A confirmation from another vehicle pushes it back (see Trust), and so does its publisher revising it; both arrive as an updated delta. Re-read expires_at from every version you accept; do not cache the first one you saw.

lat, lon and geometry move too. A publisher may restate where its own event reaches — the tail of a queue walks upstream, a wrong-way vehicle covers 1.5 km a minute — and that also arrives as an updated delta (see Publishing). Replace the whole event object rather than merging the fields you happen to care about: a client that re-reads only confidence and expires_at keeps drawing a 40 km jam at the 200 m it had when it was first published.

Ordering

Key by event_id. Apply a payload only when its updated_at is strictly newer than the one you hold.

Not a suggestion, and not an implementation detail of ours — it is part of the permanent contract, because of how we register subscriptions.

We subscribe first and read the snapshot second. The reverse order is what most implementations do and it is wrong: an event published between the read and the registration lands in neither, and stays invisible for the whole life of the connection — up to 55 minutes. Subscribing first cannot lose an event. It can only duplicate one.

Three consequences a correct client handles:

• a delta may arrive for an event you have never seen — it crossed min_confidence after the snapshot, or entered your area;
• a delta may repeat what the snapshot already carried;
• a delta may be older than the snapshot, because it was queued while the snapshot was being read.

updated_at moves on every write, creation included, which is what makes it usable as a version. created_at never moves and cannot serve. Never append; never assume a delta is fresher than what you hold.

One caveat: updated_at is a wall clock, so two API instances with skewed clocks could in principle order two writes to the same event wrongly. The blast radius is one stale row until your next reconnect, which is why v1 does not carry a logical counter.

Keepalives and backpressure

event: keepalive
data: {"dropped": 0}

Every 20 seconds of silence. An idle stream is byte-for-byte indistinguishable from a dead one, and proxies and mobile NATs drop quiet connections without telling either end — so without a heartbeat you would believe you are subscribed while receiving nothing. That is the worst failure a safety feed can have, because it looks exactly like "the road is clear".

dropped is your own count of deltas we discarded because you were not draining the connection fast enough. Each connection has a bounded queue (256 deltas); past that we drop the oldest rather than let one stalled client grow unbounded memory on a shared instance.

A rising dropped is actionable: you are behind, and what you hold is incomplete. Reconnect for a fresh snapshot rather than trusting your state.

Reconnecting

Every connection has a hard lifetime, announced to you in the snapshot as max_seconds (currently 3300, i.e. 55 minutes). We close first, cleanly, so you reconnect on a schedule instead of discovering a reset mid-drive.

event: bye
data: {"reason": "max_connection_age", "reconnect": true}

A bearer client gets SSE's automatic reconnect for free: the key does not expire, so EventSource (or your own loop) reopens and receives a fresh snapshot.

A browser on a ?ticket= does not. The ticket lived 300 seconds and the connection lived 3300. EventSource retries the same URL, gets a 401, and per the SSE spec a non-200 is a fatal error rather than a retryable one — it stops for good, silently. You must mint a new ticket and construct a new EventSource. Full pattern in Authentication.

Reconnection is always correct because a fresh snapshot is authoritative. There is no resume token and none is needed.

No replay log

We deliberately do not implement Last-Event-ID or a delta replay log, even though SSE supports it.

Road events are ephemeral and the live set is small, so the current state is the truth. A delta log would be a second source of that truth, and two sources eventually disagree — at which point you hold an event the state says is gone, or miss one the state says is live. Losing deltas is a recoverable state by design; that is what makes the bounded queue above safe.

A complete client

Two of them — JavaScript and Python. There is deliberately no SDK behind either: the API is REST + SSE, so a complete client fits on one page of the language you already use, and a package would be one more version to pin between you and the four frames above. Each client implements every rule on this page — the envelope, the ordering guard, the silent-expiry sweep, backpressure and reconnection. Start from one of these rather than from scratch.

JavaScript

Bearer-authenticated — Node, or any non-browser runtime.

const RELAY = process.env.RELAY;
const RELAY_KEY = process.env.RELAY_KEY;
 
const store = new Map();
 
function apply(change, event) {
  const held = store.get(event.event_id);
 
  if (change === "expired") {          // early death: cleared by denials, or feed dropped it
    store.delete(event.event_id);
    return;
  }
  // Upsert, but only if this payload is newer than what we hold. A delta can arrive
  // older than the snapshot.
  if (held && Date.parse(event.updated_at) <= Date.parse(held.updated_at)) return;
  store.set(event.event_id, event);
}
 
// Natural expiry sends NO delta. This sweep is not optional.
setInterval(() => {
  const now = Date.now();
  for (const [id, ev] of store) {
    if (Date.parse(ev.expires_at) <= now) store.delete(id);
  }
}, 5000);
 
// Handle one frame. TRUE means: drop this connection and resync.
function onFrame(name, data) {
  switch (name) {
    case "snapshot":
      store.clear();                                   // the snapshot is authoritative
      for (const ev of data.events) store.set(ev.event_id, ev);
      break;
    case "event":
      apply(data.change, data.event);                  // <- the envelope
      break;
    case "keepalive":
      if (data.dropped > 0) return true;               // we are behind; resync
      break;
    case "bye":
      return true;
    default:
      break;                                           // unknown frame name: ignore it
  }
  return false;
}

And the reader, using fetch because it can set the Authorization header:

async function connect({ lat, lon, radius }) {
  const url = `${RELAY}/v1/stream?lat=${lat}&lon=${lon}&radius=${radius}`;
  const res = await fetch(url, { headers: { Authorization: `Bearer ${RELAY_KEY}` } });
  if (!res.ok) throw new Error(`stream refused: ${res.status}`);   // 401 / 422 live here
 
  const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
  let buffer = "";
 
  for (;;) {
    const { value, done } = await reader.read();
    if (done) return;                                  // connection ended: reconnect
    buffer += value;
 
    let split;
    while ((split = buffer.indexOf("\n\n")) !== -1) {   // frames are blank-line separated
      const frame = buffer.slice(0, split);
      buffer = buffer.slice(split + 2);
 
      let name = "message", data = "";
      for (const line of frame.split("\n")) {
        if (line.startsWith("event: ")) name = line.slice(7);
        else if (line.startsWith("data: ")) data += line.slice(6);
      }
      if (data && onFrame(name, JSON.parse(data))) return;
    }
  }
}
 
// Reconnection is always correct: the next snapshot is authoritative.
for (;;) {
  try {
    await connect({ lat: 48.8698, lon: 2.3078, radius: 5000 });
  } catch (err) {
    console.error(err);                                // 401 / 422 and transport faults
  }
  await new Promise((resolve) => setTimeout(resolve, 1000));
}

For a browser, replace the reader with EventSource on a ?ticket= URL and keep onFrame exactly as it is — see Authentication.

Python

Standard library only — no requests, no SSE package, nothing to install. Python 3.11+. The reader is the part to copy exactly: it reads the response line by line, never res.read() — the whole-body read is the buffering trap described below, waiting for an end that only comes when the connection does.

import json
import os
import time
import urllib.request
from datetime import datetime, timezone
 
RELAY = os.environ["RELAY"]
RELAY_KEY = os.environ["RELAY_KEY"]
 
store: dict[str, dict] = {}
 
 
def _ts(value: str) -> datetime:
    return datetime.fromisoformat(value)
 
 
def apply(change: str, event: dict) -> None:
    if change == "expired":            # early death: cleared by denials, or feed dropped it
        store.pop(event["event_id"], None)
        return
    # Upsert, but only if this payload is newer than what we hold. A delta can arrive
    # older than the snapshot.
    held = store.get(event["event_id"])
    # `updated_at` is nullable on rows written before the rule existed. With no version
    # on either side there is nothing to compare, so take the newer payload — the guard
    # exists to drop STALE deltas, not to drop unversioned ones.
    if held and event["updated_at"] and held["updated_at"]:
        if _ts(event["updated_at"]) <= _ts(held["updated_at"]):
            return
    store[event["event_id"]] = event
 
 
def sweep() -> None:
    # Natural expiry sends NO delta. This sweep is not optional. It runs after every
    # frame, and keepalives arrive every 20 s of silence, so it is never starved.
    now = datetime.now(timezone.utc)
    for event_id in [i for i, ev in store.items() if _ts(ev["expires_at"]) <= now]:
        del store[event_id]
 
 
def on_frame(name: str, data: dict) -> bool:
    """Handle one frame. True means: drop this connection and resync."""
    if name == "snapshot":
        store.clear()                                  # the snapshot is authoritative
        for ev in data["events"]:
            store[ev["event_id"]] = ev
    elif name == "event":
        apply(data["change"], data["event"])           # <- the envelope
    elif name == "keepalive":
        if data["dropped"] > 0:
            return True                                # we are behind; resync
    elif name == "bye":
        return True
    # unknown frame name: ignore it (wire compatibility)
    sweep()
    return False

And the reader — plain urlopen, read line by line:

def connect(lat: float, lon: float, radius: float) -> None:
    url = f"{RELAY}/v1/stream?lat={lat}&lon={lon}&radius={radius}"
    req = urllib.request.Request(url, headers={"Authorization": f"Bearer {RELAY_KEY}"})
 
    with urllib.request.urlopen(req) as res:           # 401 / 422 raise HTTPError here
        name, data = "message", []
        while True:
            # readline(), NEVER read(): read() waits for the end of a body that only
            # ends when the connection does. Line-oriented reads surface each frame
            # the moment it arrives.
            raw = res.readline()
            if not raw:
                return                                 # connection ended: reconnect
            line = raw.decode("utf-8").rstrip("\n")
            if line.startswith("event: "):
                name = line[len("event: "):]
            elif line.startswith("data: "):
                data.append(line[len("data: "):])
            elif line == "":                           # blank line: the frame is complete
                if data and on_frame(name, json.loads("".join(data))):
                    return
                name, data = "message", []
 
 
while True:
    # Reconnection is always correct: the next snapshot is authoritative.
    connect(lat=48.8698, lon=2.3078, radius=5000)
    time.sleep(1)

When the stream "does not deliver"

Suspect your observer before the server. This costs one command:

curl -sN -H "Authorization: Bearer $RELAY_KEY" \
  "$RELAY/v1/stream?lat=45.76&lon=4.83&radius=5000"

If curl -sN shows the deltas, the server is fine and the bug is in your harness. Four ways to fool yourself, all of which read as "the server does not deliver":

1. The envelope. Correlating on event_id at the top level of a delta. See above — this one produces perfect silence.
2. Client-side buffering. Anything that reads the response in large units buffers. A large snapshot flushes it and arrives; small deltas sit in the buffer and never surface. The symptom is diabolical: the connection works, the first message arrives, then nothing. In Python, res.read() and json.load(res) wait for the end of a body that only ends when the connection does; a subprocess pipe buffers at the producer's end. Read line by line — readline(), as the Python client above does — and use curl -sN (the -N is what disables curl's own buffering).
3. Correlating on a field we do not publish. client_event_id is deliberately absent from published events; your internal identifiers are not public data. Correlate on event_id, which the POST response returns to you.
4. Forgetting the key. A 401 body is not text/event-stream, so a probe watching only for SSE frames sees silence and blames the network. Check the status line first.

Latency

Measured on the reference stack, 15 events: p50 53.9 ms, p95 73.0 ms — a 13× margin under the design target of p95 under one second between publication and reception. Production topology adds real Pub/Sub and network hops inside that margin. Verify freshness yourself rather than taking it on trust: on public-feed events, fetched_at and upstream_updated_at are published for exactly that — see The event object.