[Examples & benchmarks](https://cdpfleet.com/docs/cases) / #89 · 2026-11-10 · Example · Medium · for [monitoring](https://cdpfleet.com/monitoring), [web scraping](https://cdpfleet.com/web-scraping)

# Watch a page live: its XHR calls, DOM changes and WebSocket messages over SSE and a signed webhook

Keep a Chrome of yours on a page that updates itself (the ISS position, polled about once a second), and get every matching XHR response and every change to one element as it happens: on the job's SSE stream within a second, in signed \`job.event\` webhook batches, and stored with the job. Stop it, compare the three, read what it cost — and capture a second page's WebSocket messages both ways. In five languages.

[Google Chrome](https://cdpfleet.com/docs/browsers/chrome)

Run on production on 2026-11-10: ✓ Node.js ✓ Python ✓ Java ✓ C# ✓ Go

## The problem

Prices, scores, order books, queues, dashboards: the data on many pages arrives after the page has loaded, through the page's own XHR, fetch or WebSocket calls, and changes every second. Re-scraping such a page on a schedule misses most of it and costs a page load each time; reverse-engineering its API breaks when the site changes its tokens or headers. A [watch job](https://cdpfleet.com/docs/jobs#watch) does neither: it keeps one real browser on the page and reports what the page itself receives — the JSON responses, the WebSocket messages, the changes to an element — as they happen. This case runs one for real on a public page that updates itself without a reload, checks the three ways the events reach you — live on SSE, as signed webhooks, and stored with the job — and then watches a second page that gets its data over a WebSocket.

## What we used, and why

| What | Why |
| --- | --- |
| `POST /v1/functions/watch` | `{url, browser: "chrome", duration: 300, filter: {url: ["*/w/ajax/realtime*"], selector: ".span3 table.table-striped", types: ["xhr", "fetch", "dom"]}}` → `202 {job_id}`. `duration` is only a safety net; the program stops the job itself. No `reload_every`: the page updates itself. |
| `filter` | `url`: globs on the request (or WebSocket) URL — only the page's position calls, not its maps, fonts and analytics. `selector`: the element whose changes count (debounced 0.5 s), reported as its `text` and `html`. `types`: which kinds of events to send — `xhr`, `fetch`, `dom`, and `ws` for WebSocket messages (Chromium-based browsers only). |
| `deliver: {webhook: {url, events: ["event", "done"]}}` | A `job.event` POST with a batch of events at most every 2 s, and `job.done` at the end, signed with the account's secret; without `"event"` in `events` a watch sends only `job.done`. The receiver in the programs is cdpfleet.com's own test inbox, which keeps the raw body and headers an hour and reads them back with GET; use your own public https endpoint. |
| `GET /v1/jobs/{id}/events` | Server-sent events: `event` messages with `{events: [...]}` the moment the service has them, plus `progress`, `part` and `done`. Read with a plain HTTP client in every language — no SSE library. |
| `DELETE /v1/jobs/{id}` | Stops the watch 40 s after it was started: `stopping`, then `stopped`, and the SSE stream ends with `done`. |
| `GET /v1/jobs/{id}/results` | The same events, stored with the job (kept 14 days). |
| `GET /v1/jobs/{id}` and `GET /v1/me/sessions` | A watch runs in your own browser session: `own_session: true`, `billed_seconds` in thread-seconds, and the session (engine, weight) in the account's session history. |
| A second watch: `{url: "https://mempool.space/", filter: {url: ["*/api/v1/ws*"], types: ["ws"]}}` | mempool.space pushes Bitcoin mempool statistics to its page over one WebSocket. No `browser`: a watch starts Chrome by default. Stopped after 45 s; its stored events are read with `GET /v1/jobs/{id}/results`. |

## How it works

1.  Read the account's webhook secret and start the watch with a webhook to a fresh inbox URL.
2.  Open the job's SSE stream at once and read it until `done`, counting events by type and timing each one from its `at` to its arrival; 40 s after the start, `DELETE` the job.
3.  Read the job as it ended and its stored events; check they are exactly the events that arrived live.
4.  Read the webhook deliveries: verify every signature on the raw body (and that a changed body fails), count the events in the `job.event` batches, the gaps between batches, and the `job.done`.
5.  Parse the events: the ISS position from each XHR response (how far it moved, how fast), and the table's text from a DOM event.
6.  Start the WebSocket watch, stop it after 45 s, and read its messages: the ones the page sent, the ones it received, and what they carry.
7.  Read the bill: `billed_seconds` against the job's running time, and the sessions behind both watches.

## The code

The same program in five languages (also [on GitHub](https://github.com/cdpfleet/cdpfleet/tree/main/cases/89-watch-a-page-live), with the raw output). Set these environment variables first:

-   `CDPFLEET_API_KEY` — your API key (dashboard → API keys)

 

```node
// Node.js 18+, no packages. env: CDPFLEET_API_KEY (a paid plan: jobs aren't on Community)
import crypto from 'node:crypto';

const KEY = process.env.CDPFLEET_API_KEY;
const API = 'https://cdpfleet.com';
// A public page that changes on its own: it asks its server for the ISS position about once a second
// (XHR to /w/ajax/realtime) and rewrites a small table with it.
const PAGE = 'https://wheretheiss.at/';
const WATCH = {
  url: PAGE,
  browser: 'chrome',
  duration: 300, // a safety net: the program stops the job itself
  filter: { url: ['*/w/ajax/realtime*'], selector: '.span3 table.table-striped', types: ['xhr', 'fetch', 'dom'] },
};
const STOP_AFTER_S = 40; // DELETE the job this long after starting it
// A second, short watch for WebSocket messages: mempool.space pushes Bitcoin mempool stats to its page over one.
// No browser given: watch defaults to Chrome (WebSocket capture needs a Chromium-based browser).
const WS_WATCH = { url: 'https://mempool.space/', duration: 120, filter: { url: ['*/api/v1/ws*'], types: ['ws'] } };
const WS_STOP_AFTER_S = 45;
const ENDED = ['done', 'failed', 'stopped', 'interrupted'];
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const secs = (ms) => Math.round(ms / 100) / 10;
const round = (x, d = 1) => Math.round(x * 10 ** d) / 10 ** d;

async function api(method, path, body) {
  const res = await fetch(`${API}${path}`, {
    method,
    headers: { 'x-api-key': KEY, ...(body ? { 'content-type': 'application/json' } : {}) },
    body: body ? JSON.stringify(body) : undefined,
  });
  const text = await res.text();
  let json = null;
  try { json = JSON.parse(text); } catch { /* not JSON */ }
  return { status: res.status, json, text };
}

// x-cdpfleet-signature: t=<unix>,v1=<hex HMAC-SHA256(secret, "<t>.<raw body>")>, checked on the raw bytes.
function verify(secret, header, rawBody) {
  const { t, v1 } = Object.fromEntries((header || '').split(',').map((kv) => kv.split('=')));
  if (!t || !v1) return false;
  const expected = crypto.createHmac('sha256', secret).update(`${t}.${rawBody}`).digest();
  const given = Buffer.from(v1, 'hex');
  return given.length === expected.length && crypto.timingSafeEqual(given, expected);
}

// Great-circle distance in km between two {lat, lon}.
function km(a, b) {
  const rad = (d) => (d * Math.PI) / 180;
  const h = Math.sin(rad(b.lat - a.lat) / 2) ** 2 + Math.cos(rad(a.lat)) * Math.cos(rad(b.lat)) * Math.sin(rad(b.lon - a.lon) / 2) ** 2;
  return 2 * 6371 * Math.asin(Math.sqrt(h));
}

// The webhook receiver here is cdpfleet.com's own test inbox: it keeps deliveries an hour (raw body and
// headers) and reads them back with GET. Use your own public https endpoint.
const hook = `${API}/e2e/hook/${crypto.randomBytes(18).toString('base64url')}`;
const secret = (await api('GET', '/v1/me/webhook-secret')).json.secret;

// 1. Start the watch: XHR/fetch calls matching the glob, and changes under the selector, with a webhook.
const t0 = Date.now();
const start = await api('POST', '/v1/functions/watch', { ...WATCH, deliver: { webhook: { url: hook, events: ['event', 'done'] } } });
if (start.status !== 202) throw new Error(`watch ${start.status} ${start.text}`);
const id = start.json.job_id;

// 2. Follow it live on the SSE stream; stop it with DELETE after STOP_AFTER_S seconds.
let stop = null;
const stopTimer = setTimeout(async () => {
  const t = Date.now();
  const r = await api('DELETE', `/v1/jobs/${id}`);
  stop = { at: t, status: r.status, reply: r.json?.status ?? null };
}, STOP_AFTER_S * 1000);

const live = []; // every event as it arrived: {ev, received}
const seen = { progress: 0, part: 0, event: 0, done: 0 };
let doneData = null;
const sse = await fetch(`${API}/v1/jobs/${id}/events`, { headers: { 'x-api-key': KEY, accept: 'text/event-stream' }, signal: AbortSignal.timeout(240000) });
const decoder = new TextDecoder();
let buf = '';
read: for await (const chunk of sse.body) {
  buf += decoder.decode(chunk, { stream: true });
  for (let i; (i = buf.indexOf('\n\n')) >= 0;) {
    const block = buf.slice(0, i);
    buf = buf.slice(i + 2);
    const name = /^event: ?(.*)$/m.exec(block)?.[1];
    const data = block.split('\n').filter((l) => l.startsWith('data:')).map((l) => l.slice(5).trim()).join('\n');
    if (!name) continue; // a keep-alive comment
    seen[name] = (seen[name] || 0) + 1;
    if (name === 'event') for (const ev of JSON.parse(data).events) live.push({ ev, received: Date.now() });
    if (name === 'done') { doneData = JSON.parse(data); break read; }
  }
}
clearTimeout(stopTimer);
const tEnd = Date.now();

// 3. The job as it ended, and its events as stored.
let job = (await api('GET', `/v1/jobs/${id}`)).json;
for (let i = 0; i < 20 && !ENDED.includes(job.status); i++) { await sleep(2000); job = (await api('GET', `/v1/jobs/${id}`)).json; }
const stored = [];
for (let cursor = 0; cursor !== null;) {
  const r = (await api('GET', `/v1/jobs/${id}/results?cursor=${cursor}`)).json;
  stored.push(...r.results);
  if (r.next_cursor === cursor) break;
  cursor = r.next_cursor;
}

// 4. The webhook deliveries: job.event batches (at most one per 2 s) and job.done. Wait for job.done.
let items = [];
for (const end = Date.now() + 90000; Date.now() < end; await sleep(2000)) {
  items = (await (await fetch(hook)).json()).items;
  if (items.some((d) => d.event === 'job.done')) break;
}
const hooks = items.map((d) => ({ ...d, json: JSON.parse(d.body) }));
const batches = hooks.filter((d) => d.event === 'job.event');
const doneHook = hooks.find((d) => d.event === 'job.done');
const hookEvents = batches.flatMap((d) => d.json.data?.events || []);
const sent = batches.map((d) => Date.parse(d.json.created_at)).sort((a, b) => a - b);
const gaps = sent.slice(1).map((t, i) => t - sent[i]);
// How long after its newest event each batch was signed (the signature's t, whole seconds).
const delays = batches.map((d) => Number(/t=(\d+)/.exec(d.signature || '')?.[1]) - Math.max(...d.json.data.events.map((e) => Date.parse(e.at))) / 1000).sort((a, b) => a - b);
const key = (e) => `${e.type} ${e.at} ${e.url || e.selector}`;
const storedKeys = new Set(stored.map(key));
const hookKeys = new Set(hookEvents.map(key));

// 5. What the events say: the ISS position from each XHR response, the table text from each DOM change.
// An XHR still in flight when the page reloads is reported without a response body.
const positions = stored.filter((e) => e.type === 'xhr' && e.response_body).map((e) => {
  const d = JSON.parse(e.response_body).data;
  return { at: Date.parse(e.at), lat: d.lat, lon: d.lon, time: d.time };
});
const first = positions[0];
const last = positions.at(-1);
const lags = live.map((x) => (x.received - Date.parse(x.ev.at)) / 1000).sort((a, b) => a - b);
const byType = (list) => Object.fromEntries(['xhr', 'fetch', 'ws', 'dom'].map((t) => [t, list.filter((e) => e.type === t).length]).filter(([, n]) => n));
const firstOf = (t) => { const x = live.find((e) => e.ev.type === t); return x ? secs(x.received - t0) : null; };
const dom = stored.filter((e) => e.type === 'dom');

// 6. WebSocket messages: start the second watch, stop it after WS_STOP_AFTER_S, read what it stored.
const tWs = Date.now();
const wsStart = await api('POST', '/v1/functions/watch', WS_WATCH);
if (wsStart.status !== 202) throw new Error(`ws watch ${wsStart.status} ${wsStart.text}`);
const wsId = wsStart.json.job_id;
await sleep(WS_STOP_AFTER_S * 1000 - (Date.now() - tWs));
await api('DELETE', `/v1/jobs/${wsId}`);
let wsJob = (await api('GET', `/v1/jobs/${wsId}`)).json;
for (let i = 0; i < 20 && !ENDED.includes(wsJob.status); i++) { await sleep(2000); wsJob = (await api('GET', `/v1/jobs/${wsId}`)).json; }
const wsEvents = [];
for (let cursor = 0; cursor !== null;) {
  const r = (await api('GET', `/v1/jobs/${wsId}/results?cursor=${cursor}`)).json;
  wsEvents.push(...r.results);
  if (r.next_cursor === cursor) break;
  cursor = r.next_cursor;
}
// Each message is the frame's text: here JSON, cut at 64 KB (a cut one no longer parses).
const parse = (text) => { try { return JSON.parse(text); } catch { return null; } };
const wsIn = wsEvents.filter((e) => e.type === 'ws' && e.direction === 'in');
const wsOut = wsEvents.filter((e) => e.type === 'ws' && e.direction === 'out');
const inKeys = {};
for (const e of wsIn) { const k = Object.keys(parse(e.data) || {})[0] ?? '(cut at 64 KB)'; inKeys[k] = (inKeys[k] || 0) + 1; }
const mempool = wsIn.map((e) => parse(e.data)?.mempoolInfo).filter(Boolean);

// 7. The bill: a watch runs in your own browser session, one thread for a headless Chrome.
const sessions = (await api('GET', `/v1/me/sessions?from=${encodeURIComponent(new Date(t0 - 60000).toISOString())}&limit=200`)).json?.sessions || [];
const session = sessions.find((s) => s.session_id === job.session_id || s.id === job.session_id);
const wsSession = sessions.find((s) => s.session_id === wsJob.session_id || s.id === wsJob.session_id);

console.log(JSON.stringify({
  language: 'node',
  page: PAGE,
  watch: { ...WATCH, stop_after_s: STOP_AFTER_S, start_status: start.status },
  sse: {
    events: live.length, by_type: byType(live.map((x) => x.ev)), messages: seen,
    first_event_seconds: live.length ? secs(live[0].received - t0) : null,
    first_xhr_seconds: firstOf('xhr'), first_dom_seconds: firstOf('dom'),
    lag_seconds: lags.length ? { min: round(lags[0], 2), median: round(lags[Math.floor(lags.length / 2)], 2), max: round(lags.at(-1), 2) } : null,
    done_status: doneData?.status ?? null, stream_seconds: secs(tEnd - t0),
  },
  stop: stop && { after_seconds: secs(stop.at - t0), status: stop.status, reply: stop.reply, seconds_to_ended: secs(Date.parse(job.ended_at) - stop.at) },
  job: {
    status: job.status, own_session: job.own_session, concurrency: job.concurrency, runs: job.runs, progress: job.progress,
    billed_seconds: job.billed_seconds, run_seconds: secs(Date.parse(job.ended_at) - Date.parse(job.started_at)),
    results: job.results ? { parts: job.results.parts, bytes: job.results.bytes } : null, error: job.error,
  },
  session: session ? { engine: session.engine, weight: session.weight, seconds: secs(Date.parse(session.ended_at) - Date.parse(session.started_at)) } : null,
  stored: {
    events: stored.length, by_type: byType(stored), same_as_live: stored.length === live.length && live.every((x) => storedKeys.has(key(x.ev))),
    xhr_urls_match_glob: stored.filter((e) => e.type === 'xhr').every((e) => e.url.includes('/w/ajax/realtime') && e.status === 200 && e.method === 'GET'),
    xhr_without_body: stored.filter((e) => e.type === 'xhr' && !e.response_body).length,
    positions: positions.length, distinct_positions: new Set(positions.map((p) => `${p.lat},${p.lon}`)).size,
    first: first && { lat: round(first.lat, 3), lon: round(first.lon, 3), time: first.time },
    last: last && { lat: round(last.lat, 3), lon: round(last.lon, 3), time: last.time },
    ground_km: first && round(km(first, last)), ground_km_per_s: first && last.at > first.at ? round(km(first, last) / ((last.at - first.at) / 1000), 2) : null,
    dom_events: dom.length, dom_text_sample: dom[0]?.text?.replace(/\s+/g, ' ').trim() ?? null,
    dom_html_bytes: dom[0] ? Buffer.byteLength(dom[0].html) : null,
  },
  webhook: {
    deliveries: hooks.length, inbox_full: items.length >= 20, by_event: Object.fromEntries([...new Set(hooks.map((d) => d.event))].map((e) => [e, hooks.filter((d) => d.event === e).length])),
    signatures_valid: hooks.filter((d) => verify(secret, d.signature, d.body)).length,
    tampered_body_rejected: hooks.length > 0 && !verify(secret, hooks[0].signature, `${hooks[0].body} `),
    delivery_ids_unique_uuids: new Set(hooks.map((d) => d.delivery)).size === hooks.length && hooks.every((d) => /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/.test(d.delivery || '')),
    job_id_matches: hooks.every((d) => d.json.job_id === id),
    events: hookEvents.length, events_by_type: byType(hookEvents), events_all_stored: hookEvents.every((e) => storedKeys.has(key(e))),
    stored_not_in_webhooks: stored.filter((e) => !hookKeys.has(key(e))).length,
    not_sent_are_the_last: stored.filter((e) => !hookKeys.has(key(e))).every((e) => hookEvents.every((h) => Date.parse(h.at) <= Date.parse(e.at))),
    signed_after_newest_event_seconds: delays.length ? { min: round(delays[0]), max: round(delays.at(-1)) } : null,
    min_gap_seconds: gaps.length ? secs(Math.min(...gaps)) : null,
    min_gap_before_stop_seconds: gaps.length > 1 ? secs(Math.min(...gaps.slice(0, -1))) : null, max_events_per_batch: Math.max(0, ...batches.map((d) => d.json.data?.events?.length || 0)),
    done: doneHook ? { status: doneHook.json.data?.status, progress: doneHook.json.data?.progress, error: doneHook.json.data?.error ?? null } : null,
  },
  ws: {
    page: WS_WATCH.url, filter: WS_WATCH.filter, start_status: wsStart.status, status: wsJob.status, session_engine: wsSession?.engine ?? null,
    events: wsEvents.length, in: wsIn.length, out: wsOut.length, urls: [...new Set(wsEvents.map((e) => e.url))],
    first_in_seconds: wsIn.length ? secs(Date.parse(wsIn[0].at) - tWs) : null,
    sent: wsOut.map((e) => e.data), received: inKeys,
    mempool_txs: mempool.length ? { first: mempool[0].size, last: mempool.at(-1).size, updates: mempool.length } : null,
    billed_seconds: wsJob.billed_seconds,
  },
  measured_at: new Date().toISOString().slice(0, 10),
}, null, 2));
```

```python
# pip install requests
# env: CDPFLEET_API_KEY (a paid plan: jobs aren't on Community)
import hashlib
import hmac
import json
import math
import os
import re
import secrets
import threading
import time
from datetime import datetime, timezone

import requests

KEY = os.environ["CDPFLEET_API_KEY"]
API = "https://cdpfleet.com"
# A public page that changes on its own: it asks its server for the ISS position about once a second
# (XHR to /w/ajax/realtime) and rewrites a small table with it.
PAGE = "https://wheretheiss.at/"
WATCH = {
    "url": PAGE,
    "browser": "chrome",
    "duration": 300,  # a safety net: the program stops the job itself
    "filter": {"url": ["*/w/ajax/realtime*"], "selector": ".span3 table.table-striped", "types": ["xhr", "fetch", "dom"]},
}
STOP_AFTER_S = 40  # DELETE the job this long after starting it
# A second, short watch for WebSocket messages: mempool.space pushes Bitcoin mempool stats to its page over one.
# No browser given: watch defaults to Chrome (WebSocket capture needs a Chromium-based browser).
WS_WATCH = {"url": "https://mempool.space/", "duration": 120, "filter": {"url": ["*/api/v1/ws*"], "types": ["ws"]}}
WS_STOP_AFTER_S = 45
ENDED = ["done", "failed", "stopped", "interrupted"]
HEADERS = {"x-api-key": KEY}
UUID = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")

def api(method, path, body=None):
    return requests.request(method, API + path, headers=HEADERS, json=body, timeout=90)

def ms(iso):  # an API timestamp as epoch milliseconds
    return datetime.fromisoformat(iso.replace("Z", "+00:00")).timestamp() * 1000

def now_ms():
    return time.time() * 1000

def secs(m):
    return round(m / 1000, 1)

# x-cdpfleet-signature: t=<unix>,v1=<hex HMAC-SHA256(secret, "<t>.<raw body>")>, checked on the raw bytes.
def verify(secret, header, raw_body):
    parts = dict(kv.split("=", 1) for kv in (header or "").split(",") if "=" in kv)
    if "t" not in parts or "v1" not in parts:
        return False
    expected = hmac.new(secret.encode(), f"{parts['t']}.{raw_body}".encode(), hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, parts["v1"])

def km(a, b):  # great-circle distance in km
    r = math.radians
    h = math.sin(r(b["lat"] - a["lat"]) / 2) ** 2 + math.cos(r(a["lat"])) * math.cos(r(b["lat"])) * math.sin(r(b["lon"] - a["lon"]) / 2) ** 2
    return 2 * 6371 * math.asin(math.sqrt(h))

def by_type(events):
    return {t: n for t in ["xhr", "fetch", "ws", "dom"] if (n := sum(1 for e in events if e["type"] == t))}

def key(e):
    return f"{e['type']} {e['at']} {e.get('url') or e.get('selector')}"

# The webhook receiver here is cdpfleet.com's own test inbox: it keeps deliveries an hour (raw body and
# headers) and reads them back with GET. Use your own public https endpoint.
hook = f"{API}/e2e/hook/{secrets.token_urlsafe(18)}"
secret = api("GET", "/v1/me/webhook-secret").json()["secret"]

# 1. Start the watch: XHR/fetch calls matching the glob, and changes under the selector, with a webhook.
t0 = now_ms()
start = api("POST", "/v1/functions/watch", {**WATCH, "deliver": {"webhook": {"url": hook, "events": ["event", "done"]}}})
if start.status_code != 202:
    raise SystemExit(f"watch {start.status_code} {start.text}")
job_id = start.json()["job_id"]

# 2. Follow it live on the SSE stream; stop it with DELETE after STOP_AFTER_S seconds.
stop = {}

def do_stop():
    t = now_ms()
    r = api("DELETE", f"/v1/jobs/{job_id}")
    stop.update(at=t, status=r.status_code, reply=r.json().get("status"))

timer = threading.Timer(STOP_AFTER_S, do_stop)
timer.start()

live = []  # every event as it arrived: (event, received)
seen = {"progress": 0, "part": 0, "event": 0, "done": 0}
done_data = None
with requests.get(f"{API}/v1/jobs/{job_id}/events", headers={**HEADERS, "accept": "text/event-stream"}, stream=True, timeout=(30, 240)) as sse:
    name, data = None, []
    for line in sse.iter_lines(decode_unicode=True):
        if line:
            if line.startswith("event:"):
                name = line[6:].strip()
            elif line.startswith("data:"):
                data.append(line[5:].strip())
            continue
        if name:  # a blank line ends a message; a keep-alive comment has no event name
            seen[name] = seen.get(name, 0) + 1
            payload = json.loads("\n".join(data)) if data else {}
            if name == "event":
                live += [(ev, now_ms()) for ev in payload["events"]]
            if name == "done":
                done_data = payload
                break
        name, data = None, []
timer.cancel()
t_end = now_ms()

# 3. The job as it ended, and its events as stored.
job = api("GET", f"/v1/jobs/{job_id}").json()
for _ in range(20):
    if job["status"] in ENDED:
        break
    time.sleep(2)
    job = api("GET", f"/v1/jobs/{job_id}").json()
stored, cursor = [], 0
while cursor is not None:
    r = api("GET", f"/v1/jobs/{job_id}/results?cursor={cursor}").json()
    stored += r["results"]
    if r["next_cursor"] == cursor:
        break
    cursor = r["next_cursor"]

# 4. The webhook deliveries: job.event batches (at most one per 2 s) and job.done. Wait for job.done.
items = []
deadline = time.time() + 90
while time.time() < deadline:
    items = requests.get(hook, timeout=30).json()["items"]
    if any(d["event"] == "job.done" for d in items):
        break
    time.sleep(2)
hooks = [{**d, "json": json.loads(d["body"])} for d in items]
batches = [d for d in hooks if d["event"] == "job.event"]
done_hook = next((d for d in hooks if d["event"] == "job.done"), None)
hook_events = [e for d in batches for e in d["json"]["data"]["events"]]
sent = sorted(ms(d["json"]["created_at"]) for d in batches)
gaps = [b - a for a, b in zip(sent, sent[1:])]
# How long after its newest event each batch was signed (the signature's t, whole seconds).
delays = sorted(int(re.search(r"t=(\d+)", d["signature"]).group(1)) - max(ms(e["at"]) for e in d["json"]["data"]["events"]) / 1000 for d in batches)
stored_keys = {key(e) for e in stored}
hook_keys = {key(e) for e in hook_events}

# 5. What the events say: the ISS position from each XHR response, the table text from each DOM change.
positions = []
for e in stored:
    if e["type"] == "xhr" and e.get("response_body"):  # an XHR still in flight when the page reloads has no body
        d = json.loads(e["response_body"])["data"]
        positions.append({"at": ms(e["at"]), "lat": d["lat"], "lon": d["lon"], "time": d["time"]})
first = positions[0] if positions else None
last = positions[-1] if positions else None
lags = sorted((rec - ms(ev["at"])) / 1000 for ev, rec in live)
dom = [e for e in stored if e["type"] == "dom"]

def first_of(t):
    return next((secs(rec - t0) for ev, rec in live if ev["type"] == t), None)

# 6. WebSocket messages: start the second watch, stop it after WS_STOP_AFTER_S, read what it stored.
t_ws = now_ms()
ws_start = api("POST", "/v1/functions/watch", WS_WATCH)
if ws_start.status_code != 202:
    raise RuntimeError(f"ws watch {ws_start.status_code} {ws_start.text}")
ws_id = ws_start.json()["job_id"]
time.sleep(max(0.0, WS_STOP_AFTER_S - (now_ms() - t_ws) / 1000))
api("DELETE", f"/v1/jobs/{ws_id}")
ws_job = api("GET", f"/v1/jobs/{ws_id}").json()
for _ in range(20):
    if ws_job["status"] in ENDED:
        break
    time.sleep(2)
    ws_job = api("GET", f"/v1/jobs/{ws_id}").json()
ws_events, cursor = [], 0
while cursor is not None:
    r = api("GET", f"/v1/jobs/{ws_id}/results?cursor={cursor}").json()
    ws_events += r["results"]
    if r["next_cursor"] == cursor:
        break
    cursor = r["next_cursor"]

def parse(text):  # each message is the frame's text: here JSON, cut at 64 KB (a cut one no longer parses)
    try:
        return json.loads(text)
    except ValueError:
        return None

ws_in = [e for e in ws_events if e["type"] == "ws" and e["direction"] == "in"]
ws_out = [e for e in ws_events if e["type"] == "ws" and e["direction"] == "out"]
in_keys = {}
for e in ws_in:
    msg = parse(e["data"])
    k = next(iter(msg), "(cut at 64 KB)") if isinstance(msg, dict) else "(cut at 64 KB)"
    in_keys[k] = in_keys.get(k, 0) + 1
mempool = [m["mempoolInfo"] for m in map(lambda e: parse(e["data"]), ws_in) if isinstance(m, dict) and m.get("mempoolInfo")]

# 7. The bill: a watch runs in your own browser session, one thread for a headless Chrome.
since = datetime.fromtimestamp(t0 / 1000 - 60, timezone.utc).isoformat().replace("+00:00", "Z")
sessions = requests.get(f"{API}/v1/me/sessions", headers=HEADERS, params={"from": since, "limit": 200}, timeout=60).json().get("sessions", [])
session = next((s for s in sessions if job.get("session_id") in (s.get("session_id"), s.get("id"))), None)
ws_session = next((s for s in sessions if ws_job.get("session_id") in (s.get("session_id"), s.get("id"))), None)

print(json.dumps({
    "language": "python",
    "page": PAGE,
    "watch": {**WATCH, "stop_after_s": STOP_AFTER_S, "start_status": start.status_code},
    "sse": {
        "events": len(live), "by_type": by_type([ev for ev, _ in live]), "messages": seen,
        "first_event_seconds": secs(live[0][1] - t0) if live else None,
        "first_xhr_seconds": first_of("xhr"), "first_dom_seconds": first_of("dom"),
        "lag_seconds": {"min": round(lags[0], 2), "median": round(lags[len(lags) // 2], 2), "max": round(lags[-1], 2)} if lags else None,
        "done_status": (done_data or {}).get("status"), "stream_seconds": secs(t_end - t0),
    },
    "stop": {"after_seconds": secs(stop["at"] - t0), "status": stop["status"], "reply": stop["reply"], "seconds_to_ended": secs(ms(job["ended_at"]) - stop["at"])} if stop else None,
    "job": {
        "status": job["status"], "own_session": job.get("own_session"), "concurrency": job.get("concurrency"), "runs": job.get("runs"),
        "progress": job.get("progress"), "billed_seconds": job.get("billed_seconds"),
        "run_seconds": secs(ms(job["ended_at"]) - ms(job["started_at"])),
        "results": {"parts": job["results"]["parts"], "bytes": job["results"]["bytes"]} if job.get("results") else None, "error": job.get("error"),
    },
    "session": {"engine": session["engine"], "weight": session["weight"], "seconds": secs(ms(session["ended_at"]) - ms(session["started_at"]))} if session else None,
    "stored": {
        "events": len(stored), "by_type": by_type(stored),
        "same_as_live": len(stored) == len(live) and all(key(ev) in stored_keys for ev, _ in live),
        "xhr_urls_match_glob": all("/w/ajax/realtime" in e["url"] and e["status"] == 200 and e["method"] == "GET" for e in stored if e["type"] == "xhr"),
        "xhr_without_body": sum(1 for e in stored if e["type"] == "xhr" and not e.get("response_body")),
        "positions": len(positions), "distinct_positions": len({(p["lat"], p["lon"]) for p in positions}),
        "first": {"lat": round(first["lat"], 3), "lon": round(first["lon"], 3), "time": first["time"]} if first else None,
        "last": {"lat": round(last["lat"], 3), "lon": round(last["lon"], 3), "time": last["time"]} if last else None,
        "ground_km": round(km(first, last), 1) if first else None,
        "ground_km_per_s": round(km(first, last) / ((last["at"] - first["at"]) / 1000), 2) if first and last["at"] > first["at"] else None,
        "dom_events": len(dom), "dom_text_sample": " ".join(dom[0]["text"].split()) if dom else None,
        "dom_html_bytes": len(dom[0]["html"].encode()) if dom else None,
    },
    "webhook": {
        "deliveries": len(hooks), "inbox_full": len(items) >= 20,
        "by_event": {e: sum(1 for d in hooks if d["event"] == e) for e in dict.fromkeys(d["event"] for d in hooks)},
        "signatures_valid": sum(1 for d in hooks if verify(secret, d["signature"], d["body"])),
        "tampered_body_rejected": bool(hooks) and not verify(secret, hooks[0]["signature"], hooks[0]["body"] + " "),
        "delivery_ids_unique_uuids": len({d["delivery"] for d in hooks}) == len(hooks) and all(UUID.match(d["delivery"] or "") for d in hooks),
        "job_id_matches": all(d["json"].get("job_id") == job_id for d in hooks),
        "events": len(hook_events), "events_by_type": by_type(hook_events), "events_all_stored": all(key(e) in stored_keys for e in hook_events),
        "stored_not_in_webhooks": sum(1 for e in stored if key(e) not in hook_keys),
        "not_sent_are_the_last": all(ms(e["at"]) >= max((ms(h["at"]) for h in hook_events), default=0) for e in stored if key(e) not in hook_keys),
        "signed_after_newest_event_seconds": {"min": round(delays[0], 1), "max": round(delays[-1], 1)} if delays else None,
        "min_gap_seconds": secs(min(gaps)) if gaps else None,
        "min_gap_before_stop_seconds": secs(min(gaps[:-1])) if len(gaps) > 1 else None,
        "max_events_per_batch": max([len(d["json"]["data"]["events"]) for d in batches], default=0),
        "done": {"status": done_hook["json"]["data"].get("status"), "progress": done_hook["json"]["data"].get("progress"), "error": done_hook["json"]["data"].get("error")} if done_hook else None,
    },
    "ws": {
        "page": WS_WATCH["url"], "filter": WS_WATCH["filter"], "start_status": ws_start.status_code, "status": ws_job["status"],
        "session_engine": ws_session["engine"] if ws_session else None,
        "events": len(ws_events), "in": len(ws_in), "out": len(ws_out), "urls": list(dict.fromkeys(e["url"] for e in ws_events)),
        "first_in_seconds": secs(ms(ws_in[0]["at"]) - t_ws) if ws_in else None,
        "sent": [e["data"] for e in ws_out], "received": in_keys,
        "mempool_txs": {"first": mempool[0]["size"], "last": mempool[-1]["size"], "updates": len(mempool)} if mempool else None,
        "billed_seconds": ws_job.get("billed_seconds"),
    },
    "measured_at": datetime.now(timezone.utc).strftime("%Y-%m-%d"),
}, indent=2, ensure_ascii=False))
```

```java
// Java 21. Maven: com.google.code.gson:gson:2.11.0
// env: CDPFLEET_API_KEY (a paid plan: jobs aren't on Community)
import com.google.gson.*;
import java.io.*;
import java.net.URI;
import java.net.URLEncoder;
import java.net.http.*;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.SecureRandom;
import java.time.*;
import java.util.*;
import java.util.regex.*;
import java.util.stream.*;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;

public class Main {
  static final String KEY = System.getenv("CDPFLEET_API_KEY");
  static final String API = "https://cdpfleet.com";
  // A public page that changes on its own: it asks its server for the ISS position about once a second
  // (XHR to /w/ajax/realtime) and rewrites a small table with it.
  static final String PAGE = "https://wheretheiss.at/";
  static final int STOP_AFTER_S = 40; // DELETE the job this long after starting it
  // A second, short watch for WebSocket messages: mempool.space pushes Bitcoin mempool stats to its page over one.
  // No browser given: watch defaults to Chrome (WebSocket capture needs a Chromium-based browser).
  static final String WS_WATCH = """
      {"url": "https://mempool.space/", "duration": 120, "filter": {"url": ["*/api/v1/ws*"], "types": ["ws"]}}""";
  static final int WS_STOP_AFTER_S = 45;
  static final List<String> ENDED = List.of("done", "failed", "stopped", "interrupted");
  static final List<String> TYPES = List.of("xhr", "fetch", "ws", "dom");
  static final Pattern UUID_RE = Pattern.compile("^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$");
  static final HttpClient HTTP = HttpClient.newBuilder().followRedirects(HttpClient.Redirect.NORMAL).build();
  static final Gson GSON = new GsonBuilder().serializeNulls().setPrettyPrinting().disableHtmlEscaping().create();

  record Res(int status, JsonObject json, String text) {}
  record Live(JsonObject ev, long received) {}

  static Res send(String method, String url, JsonObject body, boolean withKey) throws Exception {
    HttpRequest.Builder b = HttpRequest.newBuilder(URI.create(url)).timeout(Duration.ofSeconds(120));
    if (withKey) b.header("x-api-key", KEY);
    if (body != null) b.header("content-type", "application/json").method(method, HttpRequest.BodyPublishers.ofString(body.toString()));
    else b.method(method, HttpRequest.BodyPublishers.noBody());
    HttpResponse<byte[]> r = HTTP.send(b.build(), HttpResponse.BodyHandlers.ofByteArray());
    String text = new String(r.body(), StandardCharsets.UTF_8);
    JsonObject json;
    try { json = JsonParser.parseString(text).getAsJsonObject(); } catch (Exception e) { json = new JsonObject(); }
    return new Res(r.statusCode(), json, text);
  }
  static Res api(String method, String path, JsonObject body) throws Exception { return send(method, API + path, body, true); }
  static String str(JsonObject o, String k) { return o != null && o.has(k) && !o.get(k).isJsonNull() ? o.get(k).getAsString() : null; }
  static JsonObject obj(JsonObject o, String k) { return o != null && o.has(k) && o.get(k).isJsonObject() ? o.getAsJsonObject(k) : null; }
  static double secs(double ms) { return Math.round(ms / 100) / 10.0; }
  static double round(double x, int d) { double p = Math.pow(10, d); return Math.round(x * p) / p; }
  static long ms(String iso) { return Instant.parse(iso).toEpochMilli(); } // an API timestamp
  static String key(JsonObject e) { return str(e, "type") + " " + str(e, "at") + " " + (e.has("url") ? str(e, "url") : str(e, "selector")); }

  // x-cdpfleet-signature: t=<unix>,v1=<hex HMAC-SHA256(secret, "<t>.<raw body>")>, checked on the raw bytes.
  static boolean verify(String secret, String header, String rawBody) throws Exception {
    Map<String, String> parts = new HashMap<>();
    for (String kv : (header == null ? "" : header).split(",")) { int i = kv.indexOf('='); if (i > 0) parts.put(kv.substring(0, i), kv.substring(i + 1)); }
    if (!parts.containsKey("t") || !parts.containsKey("v1")) return false;
    Mac mac = Mac.getInstance("HmacSHA256");
    mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
    byte[] expected = mac.doFinal((parts.get("t") + "." + rawBody).getBytes(StandardCharsets.UTF_8));
    byte[] given;
    try { given = HexFormat.of().parseHex(parts.get("v1")); } catch (Exception e) { return false; }
    return MessageDigest.isEqual(expected, given);
  }

  // Great-circle distance in km.
  static double km(double lat1, double lon1, double lat2, double lon2) {
    double h = Math.pow(Math.sin(Math.toRadians(lat2 - lat1) / 2), 2)
        + Math.cos(Math.toRadians(lat1)) * Math.cos(Math.toRadians(lat2)) * Math.pow(Math.sin(Math.toRadians(lon2 - lon1) / 2), 2);
    return 2 * 6371 * Math.asin(Math.sqrt(h));
  }

  static JsonObject byType(List<JsonObject> events) {
    JsonObject o = new JsonObject();
    for (String t : TYPES) { long n = events.stream().filter(e -> t.equals(str(e, "type"))).count(); if (n > 0) o.addProperty(t, n); }
    return o;
  }

  public static void main(String[] args) throws Exception {
    // The webhook receiver here is cdpfleet.com's own test inbox: it keeps deliveries an hour (raw body and
    // headers) and reads them back with GET. Use your own public https endpoint.
    byte[] tok = new byte[18];
    new SecureRandom().nextBytes(tok);
    String hook = API + "/e2e/hook/" + Base64.getUrlEncoder().withoutPadding().encodeToString(tok);
    String secret = str(api("GET", "/v1/me/webhook-secret", null).json(), "secret");

    // 1. Start the watch: XHR/fetch calls matching the glob, and changes under the selector, with a webhook.
    JsonObject watch = JsonParser.parseString("""
        {"url": "%s", "browser": "chrome", "duration": 300,
         "filter": {"url": ["*/w/ajax/realtime*"], "selector": ".span3 table.table-striped", "types": ["xhr", "fetch", "dom"]}}"""
        .formatted(PAGE)).getAsJsonObject(); // duration 300: a safety net, the program stops the job itself
    JsonObject body = watch.deepCopy();
    body.add("deliver", JsonParser.parseString("{\"webhook\": {\"url\": \"" + hook + "\", \"events\": [\"event\", \"done\"]}}"));
    long t0 = System.currentTimeMillis();
    Res start = api("POST", "/v1/functions/watch", body);
    if (start.status() != 202) throw new RuntimeException("watch " + start.status() + " " + start.text());
    String id = str(start.json(), "job_id");

    // 2. Follow it live on the SSE stream; stop it with DELETE after STOP_AFTER_S seconds.
    JsonObject stop = new JsonObject();
    Thread stopper = Thread.ofVirtual().start(() -> {
      try {
        Thread.sleep(STOP_AFTER_S * 1000L);
        long t = System.currentTimeMillis();
        Res r = api("DELETE", "/v1/jobs/" + id, null);
        synchronized (stop) { stop.addProperty("at", t); stop.addProperty("status", r.status()); stop.addProperty("reply", str(r.json(), "status")); }
      } catch (InterruptedException e) { /* the job ended first */ } catch (Exception e) { throw new RuntimeException(e); }
    });

    List<Live> live = new ArrayList<>();
    Map<String, Integer> seen = new LinkedHashMap<>();
    for (String n : List.of("progress", "part", "event", "done")) seen.put(n, 0);
    JsonObject doneData = null;
    HttpRequest sseReq = HttpRequest.newBuilder(URI.create(API + "/v1/jobs/" + id + "/events"))
        .header("x-api-key", KEY).header("accept", "text/event-stream").GET().build();
    HttpResponse<InputStream> sse = HTTP.send(sseReq, HttpResponse.BodyHandlers.ofInputStream());
    try (BufferedReader reader = new BufferedReader(new InputStreamReader(sse.body(), StandardCharsets.UTF_8))) {
      String name = null;
      StringBuilder data = new StringBuilder();
      for (String line; (line = reader.readLine()) != null;) {
        if (!line.isEmpty()) {
          if (line.startsWith("event:")) name = line.substring(6).trim();
          else if (line.startsWith("data:")) data.append(data.length() > 0 ? "\n" : "").append(line.substring(5).trim());
          continue;
        }
        if (name != null) { // a blank line ends a message; a keep-alive comment has no event name
          seen.merge(name, 1, Integer::sum);
          JsonObject payload = data.length() > 0 ? JsonParser.parseString(data.toString()).getAsJsonObject() : new JsonObject();
          if (name.equals("event")) for (JsonElement ev : payload.getAsJsonArray("events")) live.add(new Live(ev.getAsJsonObject(), System.currentTimeMillis()));
          if (name.equals("done")) { doneData = payload; break; }
        }
        name = null;
        data.setLength(0);
      }
    }
    stopper.interrupt();
    stopper.join();
    long tEnd = System.currentTimeMillis();

    // 3. The job as it ended, and its events as stored.
    JsonObject job = api("GET", "/v1/jobs/" + id, null).json();
    for (int i = 0; i < 20 && !ENDED.contains(str(job, "status")); i++) { Thread.sleep(2000); job = api("GET", "/v1/jobs/" + id, null).json(); }
    List<JsonObject> stored = new ArrayList<>();
    for (long cursor = 0; ; ) {
      JsonObject r = api("GET", "/v1/jobs/" + id + "/results?cursor=" + cursor, null).json();
      r.getAsJsonArray("results").forEach(x -> stored.add(x.getAsJsonObject()));
      if (r.get("next_cursor").isJsonNull() || r.get("next_cursor").getAsLong() == cursor) break;
      cursor = r.get("next_cursor").getAsLong();
    }

    // 4. The webhook deliveries: job.event batches (at most one per 2 s) and job.done. Wait for job.done.
    List<JsonObject> items = new ArrayList<>();
    for (long end = System.currentTimeMillis() + 90000; System.currentTimeMillis() < end; Thread.sleep(2000)) {
      items.clear();
      send("GET", hook, null, false).json().getAsJsonArray("items").forEach(x -> items.add(x.getAsJsonObject()));
      if (items.stream().anyMatch(d -> "job.done".equals(str(d, "event")))) break;
    }
    List<JsonObject> batches = items.stream().filter(d -> "job.event".equals(str(d, "event"))).toList();
    JsonObject doneHook = items.stream().filter(d -> "job.done".equals(str(d, "event"))).findFirst().orElse(null);
    List<JsonObject> hookEvents = new ArrayList<>();
    List<Long> sent = new ArrayList<>();
    List<Double> delays = new ArrayList<>();
    int maxPerBatch = 0;
    for (JsonObject d : batches) {
      JsonObject b = JsonParser.parseString(str(d, "body")).getAsJsonObject();
      JsonArray evs = b.getAsJsonObject("data").getAsJsonArray("events");
      evs.forEach(x -> hookEvents.add(x.getAsJsonObject()));
      maxPerBatch = Math.max(maxPerBatch, evs.size());
      sent.add(ms(str(b, "created_at")));
      // How long after its newest event each batch was signed (the signature's t, whole seconds).
      Matcher m = Pattern.compile("t=(\\d+)").matcher(str(d, "signature"));
      long newest = 0;
      for (JsonElement x : evs) newest = Math.max(newest, ms(str(x.getAsJsonObject(), "at")));
      if (m.find()) delays.add(Long.parseLong(m.group(1)) - newest / 1000.0);
    }
    Collections.sort(sent);
    Collections.sort(delays);
    long minGap = Long.MAX_VALUE, minGapBeforeStop = Long.MAX_VALUE;
    for (int i = 1; i < sent.size(); i++) {
      minGap = Math.min(minGap, sent.get(i) - sent.get(i - 1));
      if (i < sent.size() - 1) minGapBeforeStop = Math.min(minGapBeforeStop, sent.get(i) - sent.get(i - 1));
    }
    Set<String> storedKeys = stored.stream().map(Main::key).collect(Collectors.toSet());
    Set<String> hookKeys = hookEvents.stream().map(Main::key).collect(Collectors.toSet());

    // 5. What the events say: the ISS position from each XHR response, the table text from each DOM change.
    List<double[]> positions = new ArrayList<>(); // at, lat, lon
    List<String> times = new ArrayList<>();
    for (JsonObject e : stored) {
      if (!"xhr".equals(str(e, "type")) || str(e, "response_body") == null) continue; // an XHR still in flight when the page reloads has no body
      JsonObject d = JsonParser.parseString(str(e, "response_body")).getAsJsonObject().getAsJsonObject("data");
      positions.add(new double[] { ms(str(e, "at")), d.get("lat").getAsDouble(), d.get("lon").getAsDouble() });
      times.add(str(d, "time"));
    }
    List<Double> lags = live.stream().map(x -> (x.received() - ms(str(x.ev(), "at"))) / 1000.0).sorted().toList();
    List<JsonObject> dom = stored.stream().filter(e -> "dom".equals(str(e, "type"))).toList();

    // 6. WebSocket messages: start the second watch, stop it after WS_STOP_AFTER_S, read what it stored.
    long tWs = System.currentTimeMillis();
    Res wsStart = api("POST", "/v1/functions/watch", JsonParser.parseString(WS_WATCH).getAsJsonObject());
    if (wsStart.status() != 202) throw new RuntimeException("ws watch " + wsStart.status() + " " + wsStart.text());
    String wsId = str(wsStart.json(), "job_id");
    Thread.sleep(Math.max(0, WS_STOP_AFTER_S * 1000L - (System.currentTimeMillis() - tWs)));
    api("DELETE", "/v1/jobs/" + wsId, null);
    JsonObject wsJob = api("GET", "/v1/jobs/" + wsId, null).json();
    for (int i = 0; i < 20 && !ENDED.contains(str(wsJob, "status")); i++) { Thread.sleep(2000); wsJob = api("GET", "/v1/jobs/" + wsId, null).json(); }
    List<JsonObject> wsEvents = new ArrayList<>();
    for (long cursor = 0; ; ) {
      JsonObject r = api("GET", "/v1/jobs/" + wsId + "/results?cursor=" + cursor, null).json();
      r.getAsJsonArray("results").forEach(x -> wsEvents.add(x.getAsJsonObject()));
      if (r.get("next_cursor").isJsonNull() || r.get("next_cursor").getAsLong() == cursor) break;
      cursor = r.get("next_cursor").getAsLong();
    }
    // Each message is the frame's text: here JSON, cut at 64 KB (a cut one no longer parses).
    List<JsonObject> wsIn = wsEvents.stream().filter(e -> "ws".equals(str(e, "type")) && "in".equals(str(e, "direction"))).toList();
    List<JsonObject> wsOut = wsEvents.stream().filter(e -> "ws".equals(str(e, "type")) && "out".equals(str(e, "direction"))).toList();
    Map<String, Integer> inKeys = new LinkedHashMap<>();
    List<JsonElement> mempool = new ArrayList<>(); // mempoolInfo.size of each update
    for (JsonObject e : wsIn) {
      JsonObject msg = null;
      try { msg = JsonParser.parseString(str(e, "data")).getAsJsonObject(); } catch (Exception ex) { /* cut at 64 KB */ }
      String k = msg != null && !msg.keySet().isEmpty() ? msg.keySet().iterator().next() : "(cut at 64 KB)";
      inKeys.merge(k, 1, Integer::sum);
      if (msg != null && obj(msg, "mempoolInfo") != null) mempool.add(obj(msg, "mempoolInfo").get("size"));
    }

    // 7. The bill: a watch runs in your own browser session, one thread for a headless Chrome.
    String since = Instant.ofEpochMilli(t0 - 60000).toString();
    JsonArray sessions = api("GET", "/v1/me/sessions?from=" + URLEncoder.encode(since, StandardCharsets.UTF_8) + "&limit=200", null).json().getAsJsonArray("sessions");
    String sessionId = str(job, "session_id"), wsSessionId = str(wsJob, "session_id");
    JsonObject session = null, wsSession = null;
    for (JsonElement s : sessions) {
      JsonObject so = s.getAsJsonObject();
      if (sessionId != null && (sessionId.equals(str(so, "session_id")) || sessionId.equals(str(so, "id")))) session = so;
      if (wsSessionId != null && (wsSessionId.equals(str(so, "session_id")) || wsSessionId.equals(str(so, "id")))) wsSession = so;
    }

    JsonObject out = new JsonObject();
    out.addProperty("language", "java");
    out.addProperty("page", PAGE);
    JsonObject w = watch.deepCopy();
    w.addProperty("stop_after_s", STOP_AFTER_S);
    w.addProperty("start_status", start.status());
    out.add("watch", w);

    JsonObject s = new JsonObject();
    s.addProperty("events", live.size());
    s.add("by_type", byType(live.stream().map(Live::ev).toList()));
    s.add("messages", GSON.toJsonTree(seen));
    s.addProperty("first_event_seconds", live.isEmpty() ? null : secs(live.get(0).received() - t0));
    for (String t : List.of("xhr", "dom")) {
      Live f = live.stream().filter(x -> t.equals(str(x.ev(), "type"))).findFirst().orElse(null);
      s.addProperty("first_" + t + "_seconds", f == null ? null : secs(f.received() - t0));
    }
    if (lags.isEmpty()) s.add("lag_seconds", JsonNull.INSTANCE);
    else {
      JsonObject l = new JsonObject();
      l.addProperty("min", round(lags.get(0), 2));
      l.addProperty("median", round(lags.get(lags.size() / 2), 2));
      l.addProperty("max", round(lags.get(lags.size() - 1), 2));
      s.add("lag_seconds", l);
    }
    s.addProperty("done_status", str(doneData, "status"));
    s.addProperty("stream_seconds", secs(tEnd - t0));
    out.add("sse", s);

    if (stop.has("at")) {
      JsonObject st = new JsonObject();
      long at = stop.get("at").getAsLong();
      st.addProperty("after_seconds", secs(at - t0));
      st.add("status", stop.get("status"));
      st.add("reply", stop.get("reply"));
      st.addProperty("seconds_to_ended", secs(ms(str(job, "ended_at")) - at));
      out.add("stop", st);
    } else out.add("stop", JsonNull.INSTANCE);

    JsonObject j = new JsonObject();
    for (String k : List.of("status", "own_session", "concurrency", "runs", "progress", "billed_seconds")) j.add(k, job.has(k) ? job.get(k) : JsonNull.INSTANCE);
    j.addProperty("run_seconds", secs(ms(str(job, "ended_at")) - ms(str(job, "started_at"))));
    JsonObject res = obj(job, "results");
    if (res != null) { JsonObject r = new JsonObject(); r.add("parts", res.get("parts")); r.add("bytes", res.get("bytes")); j.add("results", r); }
    else j.add("results", JsonNull.INSTANCE);
    j.add("error", job.has("error") ? job.get("error") : JsonNull.INSTANCE);
    out.add("job", j);

    if (session != null) {
      JsonObject se = new JsonObject();
      se.add("engine", session.get("engine"));
      se.add("weight", session.get("weight"));
      se.addProperty("seconds", secs(ms(str(session, "ended_at")) - ms(str(session, "started_at"))));
      out.add("session", se);
    } else out.add("session", JsonNull.INSTANCE);

    JsonObject so = new JsonObject();
    so.addProperty("events", stored.size());
    so.add("by_type", byType(stored));
    so.addProperty("same_as_live", stored.size() == live.size() && live.stream().allMatch(x -> storedKeys.contains(key(x.ev()))));
    so.addProperty("xhr_urls_match_glob", stored.stream().filter(e -> "xhr".equals(str(e, "type")))
        .allMatch(e -> str(e, "url").contains("/w/ajax/realtime") && e.get("status").getAsInt() == 200 && "GET".equals(str(e, "method"))));
    so.addProperty("xhr_without_body", stored.stream().filter(e -> "xhr".equals(str(e, "type")) && str(e, "response_body") == null).count());
    so.addProperty("positions", positions.size());
    so.addProperty("distinct_positions", positions.stream().map(p -> p[1] + "," + p[2]).distinct().count());
    if (!positions.isEmpty()) {
      double[] f = positions.get(0);
      double[] l = positions.get(positions.size() - 1);
      for (Object[] fl : new Object[][] { { "first", f, times.get(0) }, { "last", l, times.get(times.size() - 1) } }) {
        double[] p = (double[]) fl[1];
        JsonObject o = new JsonObject();
        o.addProperty("lat", round(p[1], 3));
        o.addProperty("lon", round(p[2], 3));
        o.addProperty("time", (String) fl[2]);
        so.add((String) fl[0], o);
      }
      double dist = km(f[1], f[2], l[1], l[2]);
      so.addProperty("ground_km", round(dist, 1));
      so.addProperty("ground_km_per_s", l[0] > f[0] ? round(dist / ((l[0] - f[0]) / 1000), 2) : null);
    } else { so.add("first", JsonNull.INSTANCE); so.add("last", JsonNull.INSTANCE); so.add("ground_km", JsonNull.INSTANCE); so.add("ground_km_per_s", JsonNull.INSTANCE); }
    so.addProperty("dom_events", dom.size());
    so.addProperty("dom_text_sample", dom.isEmpty() ? null : str(dom.get(0), "text").replaceAll("\\s+", " ").trim());
    so.addProperty("dom_html_bytes", dom.isEmpty() ? null : str(dom.get(0), "html").getBytes(StandardCharsets.UTF_8).length);
    out.add("stored", so);

    JsonObject wh = new JsonObject();
    wh.addProperty("deliveries", items.size());
    wh.addProperty("inbox_full", items.size() >= 20);
    JsonObject byEvent = new JsonObject();
    for (JsonObject d : items) byEvent.addProperty(str(d, "event"), (byEvent.has(str(d, "event")) ? byEvent.get(str(d, "event")).getAsInt() : 0) + 1);
    wh.add("by_event", byEvent);
    int valid = 0;
    for (JsonObject d : items) if (verify(secret, str(d, "signature"), str(d, "body"))) valid++;
    wh.addProperty("signatures_valid", valid);
    wh.addProperty("tampered_body_rejected", !items.isEmpty() && !verify(secret, str(items.get(0), "signature"), str(items.get(0), "body") + " "));
    wh.addProperty("delivery_ids_unique_uuids", items.stream().map(d -> str(d, "delivery")).distinct().count() == items.size()
        && items.stream().allMatch(d -> str(d, "delivery") != null && UUID_RE.matcher(str(d, "delivery")).matches()));
    wh.addProperty("job_id_matches", items.stream().allMatch(d -> id.equals(str(JsonParser.parseString(str(d, "body")).getAsJsonObject(), "job_id"))));
    wh.addProperty("events", hookEvents.size());
    wh.add("events_by_type", byType(hookEvents));
    wh.addProperty("events_all_stored", hookEvents.stream().allMatch(e -> storedKeys.contains(key(e))));
    wh.addProperty("stored_not_in_webhooks", stored.stream().filter(e -> !hookKeys.contains(key(e))).count());
    long newestSent = hookEvents.stream().mapToLong(e -> ms(str(e, "at"))).max().orElse(0);
    wh.addProperty("not_sent_are_the_last", stored.stream().filter(e -> !hookKeys.contains(key(e))).allMatch(e -> ms(str(e, "at")) >= newestSent));
    if (delays.isEmpty()) wh.add("signed_after_newest_event_seconds", JsonNull.INSTANCE);
    else {
      JsonObject dl = new JsonObject();
      dl.addProperty("min", round(delays.get(0), 1));
      dl.addProperty("max", round(delays.get(delays.size() - 1), 1));
      wh.add("signed_after_newest_event_seconds", dl);
    }
    wh.addProperty("min_gap_seconds", sent.size() > 1 ? secs(minGap) : null);
    wh.addProperty("min_gap_before_stop_seconds", sent.size() > 2 ? secs(minGapBeforeStop) : null);
    wh.addProperty("max_events_per_batch", maxPerBatch);
    if (doneHook != null) {
      JsonObject dd = JsonParser.parseString(str(doneHook, "body")).getAsJsonObject().getAsJsonObject("data");
      JsonObject o = new JsonObject();
      o.add("status", dd.get("status"));
      o.add("progress", dd.get("progress"));
      o.add("error", dd.has("error") ? dd.get("error") : JsonNull.INSTANCE);
      wh.add("done", o);
    } else wh.add("done", JsonNull.INSTANCE);
    out.add("webhook", wh);

    JsonObject ws = new JsonObject();
    JsonObject wsSpec = JsonParser.parseString(WS_WATCH).getAsJsonObject();
    ws.add("page", wsSpec.get("url"));
    ws.add("filter", wsSpec.get("filter"));
    ws.addProperty("start_status", wsStart.status());
    ws.addProperty("status", str(wsJob, "status"));
    ws.addProperty("session_engine", wsSession == null ? null : str(wsSession, "engine"));
    ws.addProperty("events", wsEvents.size());
    ws.addProperty("in", wsIn.size());
    ws.addProperty("out", wsOut.size());
    ws.add("urls", GSON.toJsonTree(wsEvents.stream().map(e -> str(e, "url")).distinct().toList()));
    ws.addProperty("first_in_seconds", wsIn.isEmpty() ? null : secs(ms(str(wsIn.get(0), "at")) - tWs));
    ws.add("sent", GSON.toJsonTree(wsOut.stream().map(e -> str(e, "data")).toList()));
    ws.add("received", GSON.toJsonTree(inKeys));
    if (mempool.isEmpty()) ws.add("mempool_txs", JsonNull.INSTANCE);
    else {
      JsonObject mp = new JsonObject();
      mp.add("first", mempool.get(0));
      mp.add("last", mempool.get(mempool.size() - 1));
      mp.addProperty("updates", mempool.size());
      ws.add("mempool_txs", mp);
    }
    ws.add("billed_seconds", wsJob.has("billed_seconds") ? wsJob.get("billed_seconds") : JsonNull.INSTANCE);
    out.add("ws", ws);
    out.addProperty("measured_at", LocalDate.now(ZoneOffset.UTC).toString());
    System.out.println(GSON.toJson(out));
  }
}
```

```csharp
// .NET 9, no packages. env: CDPFLEET_API_KEY (a paid plan: jobs aren't on Community)
using System.Security.Cryptography;
using System.Text;
using System.Text.Encodings.Web;
using System.Text.Json;
using System.Text.Json.Nodes;
using System.Text.RegularExpressions;

var key = Environment.GetEnvironmentVariable("CDPFLEET_API_KEY")!;
const string Api = "https://cdpfleet.com";
// A public page that changes on its own: it asks its server for the ISS position about once a second
// (XHR to /w/ajax/realtime) and rewrites a small table with it.
const string Page = "https://wheretheiss.at/";
const int StopAfterS = 40; // DELETE the job this long after starting it
// A second, short watch for WebSocket messages: mempool.space pushes Bitcoin mempool stats to its page over one.
// No browser given: watch defaults to Chrome (WebSocket capture needs a Chromium-based browser).
var wsWatch = new JsonObject { ["url"] = "https://mempool.space/", ["duration"] = 120, ["filter"] = new JsonObject { ["url"] = new JsonArray("*/api/v1/ws*"), ["types"] = new JsonArray("ws") } };
const int WsStopAfterS = 45;
string[] ended = ["done", "failed", "stopped", "interrupted"];
string[] types = ["xhr", "fetch", "ws", "dom"];
var uuidRe = new Regex("^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$");
var http = new HttpClient { Timeout = TimeSpan.FromSeconds(240) };

async Task<(int Status, JsonObject Json, string Text)> Send(HttpMethod method, string url, JsonNode? body = null, bool withKey = true)
{
    var req = new HttpRequestMessage(method, url);
    if (withKey) req.Headers.Add("x-api-key", key);
    if (body != null) req.Content = new StringContent(body.ToJsonString(), Encoding.UTF8, "application/json");
    var res = await http.SendAsync(req);
    var text = Encoding.UTF8.GetString(await res.Content.ReadAsByteArrayAsync());
    JsonObject json;
    try { json = JsonNode.Parse(text)!.AsObject(); } catch { json = new JsonObject(); }
    return ((int)res.StatusCode, json, text);
}
async Task<(int Status, JsonObject Json, string Text)> CallApi(HttpMethod method, string path, JsonNode? body = null) => await Send(method, Api + path, body);
string? Str(JsonNode? n) => n is JsonValue v && v.TryGetValue<string>(out var s) ? s : null;
double Num(JsonNode? n) => n is JsonValue v && v.TryGetValue<double>(out var d) ? d : 0;
double Secs(double ms) => Math.Round(ms / 100, MidpointRounding.AwayFromZero) / 10;
double Round(double x, int d) => Math.Round(x, d, MidpointRounding.AwayFromZero);
double Ms(JsonNode? iso) => DateTimeOffset.Parse(Str(iso)!).ToUnixTimeMilliseconds(); // an API timestamp
double NowMs() => DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
string Key(JsonNode e) => $"{Str(e["type"])} {Str(e["at"])} {Str(e["url"]) ?? Str(e["selector"])}";
JsonObject ByType(IEnumerable<JsonNode> events)
{
    var o = new JsonObject();
    foreach (var t in types) { var n = events.Count(e => Str(e["type"]) == t); if (n > 0) o[t] = n; }
    return o;
}

// x-cdpfleet-signature: t=<unix>,v1=<hex HMAC-SHA256(secret, "<t>.<raw body>")>, checked on the raw bytes.
bool Verify(string secret, string? header, string rawBody)
{
    var parts = (header ?? "").Split(',').Select(kv => kv.Split('=', 2)).Where(p => p.Length == 2).ToDictionary(p => p[0], p => p[1]);
    if (!parts.TryGetValue("t", out var t) || !parts.TryGetValue("v1", out var v1)) return false;
    var expected = HMACSHA256.HashData(Encoding.UTF8.GetBytes(secret), Encoding.UTF8.GetBytes($"{t}.{rawBody}"));
    byte[] given;
    try { given = Convert.FromHexString(v1); } catch { return false; }
    return CryptographicOperations.FixedTimeEquals(expected, given);
}

// Great-circle distance in km.
double Km(double lat1, double lon1, double lat2, double lon2)
{
    double R(double d) => d * Math.PI / 180;
    var h = Math.Pow(Math.Sin(R(lat2 - lat1) / 2), 2) + Math.Cos(R(lat1)) * Math.Cos(R(lat2)) * Math.Pow(Math.Sin(R(lon2 - lon1) / 2), 2);
    return 2 * 6371 * Math.Asin(Math.Sqrt(h));
}

// The webhook receiver here is cdpfleet.com's own test inbox: it keeps deliveries an hour (raw body and
// headers) and reads them back with GET. Use your own public https endpoint.
var hook = $"{Api}/e2e/hook/{Convert.ToBase64String(RandomNumberGenerator.GetBytes(18)).Replace('+', '-').Replace('/', '_').TrimEnd('=')}";
var secret = Str((await CallApi(HttpMethod.Get, "/v1/me/webhook-secret")).Json["secret"])!;

// 1. Start the watch: XHR/fetch calls matching the glob, and changes under the selector, with a webhook.
var watch = new JsonObject
{
    ["url"] = Page, ["browser"] = "chrome",
    ["duration"] = 300, // a safety net: the program stops the job itself
    ["filter"] = new JsonObject { ["url"] = new JsonArray("*/w/ajax/realtime*"), ["selector"] = ".span3 table.table-striped", ["types"] = new JsonArray("xhr", "fetch", "dom") },
};
var body = watch.DeepClone().AsObject();
body["deliver"] = new JsonObject { ["webhook"] = new JsonObject { ["url"] = hook, ["events"] = new JsonArray("event", "done") } };
var t0 = NowMs();
var start = await CallApi(HttpMethod.Post, "/v1/functions/watch", body);
if (start.Status != 202) throw new Exception($"watch {start.Status} {start.Text}");
var id = Str(start.Json["job_id"])!;

// 2. Follow it live on the SSE stream; stop it with DELETE after StopAfterS seconds.
JsonObject? stop = null;
double stopAt = 0;
using var cancelStop = new CancellationTokenSource();
var stopper = Task.Run(async () =>
{
    await Task.Delay(StopAfterS * 1000, cancelStop.Token);
    stopAt = NowMs();
    var r = await CallApi(HttpMethod.Delete, $"/v1/jobs/{id}");
    stop = new JsonObject { ["status"] = r.Status, ["reply"] = Str(r.Json["status"]) };
});

var live = new List<(JsonNode Ev, double Received)>(); // every event as it arrived
var seen = new Dictionary<string, int> { ["progress"] = 0, ["part"] = 0, ["event"] = 0, ["done"] = 0 };
JsonNode? doneData = null;
var sseReq = new HttpRequestMessage(HttpMethod.Get, $"{Api}/v1/jobs/{id}/events");
sseReq.Headers.Add("x-api-key", key);
sseReq.Headers.Add("accept", "text/event-stream");
using (var sse = await http.SendAsync(sseReq, HttpCompletionOption.ResponseHeadersRead))
using (var reader = new StreamReader(await sse.Content.ReadAsStreamAsync()))
{
    string? name = null;
    var data = new List<string>();
    while (await reader.ReadLineAsync() is string line)
    {
        if (line.Length > 0)
        {
            if (line.StartsWith("event:")) name = line[6..].Trim();
            else if (line.StartsWith("data:")) data.Add(line[5..].Trim());
            continue;
        }
        if (name != null) // a blank line ends a message; a keep-alive comment has no event name
        {
            seen[name] = seen.GetValueOrDefault(name) + 1;
            var payload = data.Count > 0 ? JsonNode.Parse(string.Join("\n", data)) : new JsonObject();
            if (name == "event") foreach (var ev in payload!["events"]!.AsArray()) live.Add((ev!.DeepClone(), NowMs()));
            if (name == "done") { doneData = payload; break; }
        }
        name = null;
        data.Clear();
    }
}
cancelStop.Cancel();
try { await stopper; } catch (TaskCanceledException) { /* the job ended first */ }
var tEnd = NowMs();

// 3. The job as it ended, and its events as stored.
var job = (await CallApi(HttpMethod.Get, $"/v1/jobs/{id}")).Json;
for (var i = 0; i < 20 && !ended.Contains(Str(job["status"])); i++) { await Task.Delay(2000); job = (await CallApi(HttpMethod.Get, $"/v1/jobs/{id}")).Json; }
var stored = new List<JsonNode>();
for (long cursor = 0; ;)
{
    var r = (await CallApi(HttpMethod.Get, $"/v1/jobs/{id}/results?cursor={cursor}")).Json;
    stored.AddRange(r["results"]!.AsArray().Select(x => x!.DeepClone()));
    if (r["next_cursor"] is null || (long)Num(r["next_cursor"]) == cursor) break;
    cursor = (long)Num(r["next_cursor"]);
}

// 4. The webhook deliveries: job.event batches (at most one per 2 s) and job.done. Wait for job.done.
var items = new List<JsonNode>();
for (var end = NowMs() + 90000; NowMs() < end; await Task.Delay(2000))
{
    items = JsonNode.Parse((await Send(HttpMethod.Get, hook, withKey: false)).Text)!["items"]!.AsArray().Select(x => x!.DeepClone()).ToList();
    if (items.Any(d => Str(d["event"]) == "job.done")) break;
}
var hooks = items.Select(d => (D: d, Json: JsonNode.Parse(Str(d["body"])!)!)).ToList();
var batches = hooks.Where(h => Str(h.D["event"]) == "job.event").ToList();
var doneHook = hooks.FirstOrDefault(h => Str(h.D["event"]) == "job.done");
var hookEvents = batches.SelectMany(h => h.Json["data"]!["events"]!.AsArray().Select(e => e!)).ToList();
var sent = batches.Select(h => Ms(h.Json["created_at"])).OrderBy(x => x).ToList();
var gaps = sent.Skip(1).Select((t, i) => t - sent[i]).ToList();
// How long after its newest event each batch was signed (the signature's t, whole seconds).
var delays = batches.Select(h => long.Parse(Regex.Match(Str(h.D["signature"]) ?? "", @"t=(\d+)").Groups[1].Value)
    - h.Json["data"]!["events"]!.AsArray().Max(e => Ms(e!["at"])) / 1000).OrderBy(x => x).ToList();
var storedKeys = stored.Select(Key).ToHashSet();
var hookKeys = hookEvents.Select(Key).ToHashSet();

// 5. What the events say: the ISS position from each XHR response, the table text from each DOM change.
// An XHR still in flight when the page reloads is reported without a response body.
var positions = stored.Where(e => Str(e["type"]) == "xhr" && Str(e["response_body"]) != null).Select(e =>
{
    var d = JsonNode.Parse(Str(e["response_body"])!)!["data"]!;
    return (At: Ms(e["at"]), Lat: Num(d["lat"]), Lon: Num(d["lon"]), Time: Str(d["time"]));
}).ToList();
var lags = live.Select(x => (x.Received - Ms(x.Ev["at"])) / 1000).OrderBy(x => x).ToList();
var dom = stored.Where(e => Str(e["type"]) == "dom").ToList();
JsonNode? FirstOf(string t) => live.Where(x => Str(x.Ev["type"]) == t).Select(x => (JsonNode?)Secs(x.Received - t0)).FirstOrDefault();

// 6. WebSocket messages: start the second watch, stop it after WsStopAfterS, read what it stored.
var tWs = NowMs();
var wsStart = await CallApi(HttpMethod.Post, "/v1/functions/watch", wsWatch);
if (wsStart.Status != 202) throw new Exception($"ws watch {wsStart.Status} {wsStart.Text}");
var wsId = Str(wsStart.Json["job_id"])!;
await Task.Delay(TimeSpan.FromMilliseconds(Math.Max(0, WsStopAfterS * 1000 - (NowMs() - tWs))));
await CallApi(HttpMethod.Delete, $"/v1/jobs/{wsId}");
var wsJob = (await CallApi(HttpMethod.Get, $"/v1/jobs/{wsId}")).Json;
for (var i = 0; i < 20 && !ended.Contains(Str(wsJob["status"])); i++) { await Task.Delay(2000); wsJob = (await CallApi(HttpMethod.Get, $"/v1/jobs/{wsId}")).Json; }
var wsEvents = new List<JsonNode>();
for (long cursor = 0; ;)
{
    var r = (await CallApi(HttpMethod.Get, $"/v1/jobs/{wsId}/results?cursor={cursor}")).Json;
    wsEvents.AddRange(r["results"]!.AsArray().Select(x => x!.DeepClone()));
    if (r["next_cursor"] is null || (long)Num(r["next_cursor"]) == cursor) break;
    cursor = (long)Num(r["next_cursor"]);
}
// Each message is the frame's text: here JSON, cut at 64 KB (a cut one no longer parses).
JsonObject? ParseMsg(string? text) { try { return JsonNode.Parse(text ?? "") as JsonObject; } catch (JsonException) { return null; } }
var wsIn = wsEvents.Where(e => Str(e["type"]) == "ws" && Str(e["direction"]) == "in").ToList();
var wsOut = wsEvents.Where(e => Str(e["type"]) == "ws" && Str(e["direction"]) == "out").ToList();
var inKeys = new Dictionary<string, int>();
var mempool = new List<JsonNode?>(); // mempoolInfo.size of each update
foreach (var e in wsIn)
{
    var msg = ParseMsg(Str(e["data"]));
    var k = msg != null && msg.Count > 0 ? msg.First().Key : "(cut at 64 KB)";
    inKeys[k] = inKeys.GetValueOrDefault(k) + 1;
    if (msg?["mempoolInfo"] is JsonObject info) mempool.Add(info["size"]?.DeepClone());
}

// 7. The bill: a watch runs in your own browser session, one thread for a headless Chrome.
var since = Uri.EscapeDataString(DateTimeOffset.FromUnixTimeMilliseconds((long)t0 - 60000).UtcDateTime.ToString("yyyy-MM-ddTHH:mm:ss.fffZ"));
var sessions = (await CallApi(HttpMethod.Get, $"/v1/me/sessions?from={since}&limit=200")).Json["sessions"]?.AsArray() ?? new JsonArray();
var sessionId = Str(job["session_id"]);
var session = sessions.FirstOrDefault(s => sessionId != null && (Str(s!["session_id"]) == sessionId || Str(s!["id"]) == sessionId));
var wsSessionId = Str(wsJob["session_id"]);
var wsSession = sessions.FirstOrDefault(s => wsSessionId != null && (Str(s!["session_id"]) == wsSessionId || Str(s!["id"]) == wsSessionId));

var watchOut = watch.DeepClone().AsObject();
watchOut["stop_after_s"] = StopAfterS;
watchOut["start_status"] = start.Status;
var first = positions.Count > 0 ? positions[0] : default;
var last = positions.Count > 0 ? positions[^1] : default;
var dist = positions.Count > 0 ? Km(first.Lat, first.Lon, last.Lat, last.Lon) : 0;
var output = new JsonObject
{
    ["language"] = "csharp",
    ["page"] = Page,
    ["watch"] = watchOut,
    ["sse"] = new JsonObject
    {
        ["events"] = live.Count, ["by_type"] = ByType(live.Select(x => x.Ev)),
        ["messages"] = new JsonObject(seen.Select(kv => KeyValuePair.Create(kv.Key, (JsonNode?)kv.Value))),
        ["first_event_seconds"] = live.Count > 0 ? Secs(live[0].Received - t0) : null,
        ["first_xhr_seconds"] = FirstOf("xhr"), ["first_dom_seconds"] = FirstOf("dom"),
        ["lag_seconds"] = lags.Count > 0 ? new JsonObject { ["min"] = Round(lags[0], 2), ["median"] = Round(lags[lags.Count / 2], 2), ["max"] = Round(lags[^1], 2) } : null,
        ["done_status"] = Str(doneData?["status"]), ["stream_seconds"] = Secs(tEnd - t0),
    },
    ["stop"] = stop == null ? null : new JsonObject
    {
        ["after_seconds"] = Secs(stopAt - t0), ["status"] = stop["status"]!.DeepClone(), ["reply"] = stop["reply"]?.DeepClone(),
        ["seconds_to_ended"] = Secs(Ms(job["ended_at"]) - stopAt),
    },
    ["job"] = new JsonObject
    {
        ["status"] = job["status"]?.DeepClone(), ["own_session"] = job["own_session"]?.DeepClone(), ["concurrency"] = job["concurrency"]?.DeepClone(),
        ["runs"] = job["runs"]?.DeepClone(), ["progress"] = job["progress"]?.DeepClone(), ["billed_seconds"] = job["billed_seconds"]?.DeepClone(),
        ["run_seconds"] = Secs(Ms(job["ended_at"]) - Ms(job["started_at"])),
        ["results"] = job["results"] is JsonObject res ? new JsonObject { ["parts"] = res["parts"]?.DeepClone(), ["bytes"] = res["bytes"]?.DeepClone() } : null,
        ["error"] = job["error"]?.DeepClone(),
    },
    ["session"] = session == null ? null : new JsonObject
    {
        ["engine"] = session["engine"]?.DeepClone(), ["weight"] = session["weight"]?.DeepClone(), ["seconds"] = Secs(Ms(session["ended_at"]) - Ms(session["started_at"])),
    },
    ["stored"] = new JsonObject
    {
        ["events"] = stored.Count, ["by_type"] = ByType(stored),
        ["same_as_live"] = stored.Count == live.Count && live.All(x => storedKeys.Contains(Key(x.Ev))),
        ["xhr_urls_match_glob"] = stored.Where(e => Str(e["type"]) == "xhr").All(e => Str(e["url"])!.Contains("/w/ajax/realtime") && Num(e["status"]) == 200 && Str(e["method"]) == "GET"),
        ["xhr_without_body"] = stored.Count(e => Str(e["type"]) == "xhr" && Str(e["response_body"]) == null),
        ["positions"] = positions.Count, ["distinct_positions"] = positions.Select(p => $"{p.Lat},{p.Lon}").Distinct().Count(),
        ["first"] = positions.Count > 0 ? new JsonObject { ["lat"] = Round(first.Lat, 3), ["lon"] = Round(first.Lon, 3), ["time"] = first.Time } : null,
        ["last"] = positions.Count > 0 ? new JsonObject { ["lat"] = Round(last.Lat, 3), ["lon"] = Round(last.Lon, 3), ["time"] = last.Time } : null,
        ["ground_km"] = positions.Count > 0 ? Round(dist, 1) : null,
        ["ground_km_per_s"] = positions.Count > 0 && last.At > first.At ? Round(dist / ((last.At - first.At) / 1000), 2) : null,
        ["dom_events"] = dom.Count,
        ["dom_text_sample"] = dom.Count > 0 ? Regex.Replace(Str(dom[0]["text"])!, @"\s+", " ").Trim() : null,
        ["dom_html_bytes"] = dom.Count > 0 ? Encoding.UTF8.GetByteCount(Str(dom[0]["html"])!) : null,
    },
    ["webhook"] = new JsonObject
    {
        ["deliveries"] = hooks.Count, ["inbox_full"] = items.Count >= 20,
        ["by_event"] = new JsonObject(hooks.GroupBy(h => Str(h.D["event"])!).Select(g => KeyValuePair.Create(g.Key, (JsonNode?)g.Count()))),
        ["signatures_valid"] = hooks.Count(h => Verify(secret, Str(h.D["signature"]), Str(h.D["body"])!)),
        ["tampered_body_rejected"] = hooks.Count > 0 && !Verify(secret, Str(hooks[0].D["signature"]), Str(hooks[0].D["body"]) + " "),
        ["delivery_ids_unique_uuids"] = hooks.Select(h => Str(h.D["delivery"])).Distinct().Count() == hooks.Count && hooks.All(h => uuidRe.IsMatch(Str(h.D["delivery"]) ?? "")),
        ["job_id_matches"] = hooks.All(h => Str(h.Json["job_id"]) == id),
        ["events"] = hookEvents.Count, ["events_by_type"] = ByType(hookEvents), ["events_all_stored"] = hookEvents.All(e => storedKeys.Contains(Key(e))),
        ["stored_not_in_webhooks"] = stored.Count(e => !hookKeys.Contains(Key(e))),
        ["not_sent_are_the_last"] = stored.Where(e => !hookKeys.Contains(Key(e))).All(e => hookEvents.All(h => Ms(h["at"]) <= Ms(e["at"]))),
        ["signed_after_newest_event_seconds"] = delays.Count > 0 ? new JsonObject { ["min"] = Round(delays[0], 1), ["max"] = Round(delays[^1], 1) } : null,
        ["min_gap_seconds"] = gaps.Count > 0 ? Secs(gaps.Min()) : null,
        ["min_gap_before_stop_seconds"] = gaps.Count > 1 ? Secs(gaps.SkipLast(1).Min()) : null,
        ["max_events_per_batch"] = batches.Select(h => h.Json["data"]!["events"]!.AsArray().Count).DefaultIfEmpty(0).Max(),
        ["done"] = doneHook.Json is JsonNode dj ? new JsonObject
        {
            ["status"] = dj["data"]?["status"]?.DeepClone(), ["progress"] = dj["data"]?["progress"]?.DeepClone(), ["error"] = dj["data"]?["error"]?.DeepClone(),
        } : null,
    },
    ["ws"] = new JsonObject
    {
        ["page"] = wsWatch["url"]!.DeepClone(), ["filter"] = wsWatch["filter"]!.DeepClone(), ["start_status"] = wsStart.Status,
        ["status"] = Str(wsJob["status"]), ["session_engine"] = Str(wsSession?["engine"]),
        ["events"] = wsEvents.Count, ["in"] = wsIn.Count, ["out"] = wsOut.Count,
        ["urls"] = new JsonArray(wsEvents.Select(e => Str(e["url"])).Distinct().Select(u => (JsonNode?)u).ToArray()),
        ["first_in_seconds"] = wsIn.Count > 0 ? Secs(Ms(wsIn[0]["at"]) - tWs) : null,
        ["sent"] = new JsonArray(wsOut.Select(e => (JsonNode?)Str(e["data"])).ToArray()), ["received"] = new JsonObject(inKeys.Select(kv => KeyValuePair.Create(kv.Key, (JsonNode?)kv.Value))),
        ["mempool_txs"] = mempool.Count > 0 ? new JsonObject { ["first"] = mempool[0]?.DeepClone(), ["last"] = mempool[^1]?.DeepClone(), ["updates"] = mempool.Count } : null,
        ["billed_seconds"] = wsJob["billed_seconds"]?.DeepClone(),
    },
    ["measured_at"] = DateTime.UtcNow.ToString("yyyy-MM-dd"),
};
Console.WriteLine(output.ToJsonString(new JsonSerializerOptions { WriteIndented = true, Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping }));
```

```go
// Go 1.22, standard library only. env: CDPFLEET_API_KEY (a paid plan: jobs aren't on Community)
package main

import (
	"bufio"
	"bytes"
	"crypto/hmac"
	"crypto/rand"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"math"
	"net/http"
	"net/url"
	"os"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
)

const api = "https://cdpfleet.com"

// A public page that changes on its own: it asks its server for the ISS position about once a second
// (XHR to /w/ajax/realtime) and rewrites a small table with it.
const page = "https://wheretheiss.at/"
const stopAfterS = 40 // DELETE the job this long after starting it

// A second, short watch for WebSocket messages: mempool.space pushes Bitcoin mempool stats to its page over one.
// No browser given: watch defaults to Chrome (WebSocket capture needs a Chromium-based browser).
var wsWatch = M{"url": "https://mempool.space/", "duration": 120, "filter": M{"url": []string{"*/api/v1/ws*"}, "types": []string{"ws"}}}

const wsStopAfterS = 45

var key = os.Getenv("CDPFLEET_API_KEY")
var ended = map[string]bool{"done": true, "failed": true, "stopped": true, "interrupted": true}
var types = []string{"xhr", "fetch", "ws", "dom"}
var uuidRe = regexp.MustCompile(`^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$`)

type M = map[string]any

func send(method, u string, body any, withKey bool) (int, []byte) {
	var rd io.Reader
	if body != nil {
		b, _ := json.Marshal(body)
		rd = bytes.NewReader(b)
	}
	req, _ := http.NewRequest(method, u, rd)
	if withKey {
		req.Header.Set("x-api-key", key)
	}
	if body != nil {
		req.Header.Set("content-type", "application/json")
	}
	client := &http.Client{Timeout: 120 * time.Second}
	res, err := client.Do(req)
	if err != nil {
		panic(err)
	}
	defer res.Body.Close()
	b, _ := io.ReadAll(res.Body)
	return res.StatusCode, b
}

func call(method, path string, body any) (int, M) {
	status, b := send(method, api+path, body, true)
	var m M
	json.Unmarshal(b, &m)
	return status, m
}

func str(v any) string { s, _ := v.(string); return s }
func num(v any) float64 { f, _ := v.(float64); return f }
func obj(v any) M { m, _ := v.(M); return m }
func arr(v any) []any { a, _ := v.([]any); return a }
func nowMs() float64 { return float64(time.Now().UnixMilli()) }
func secs(ms float64) float64 { return math.Round(ms/100) / 10 }
func round(x float64, d int) float64 { p := math.Pow(10, float64(d)); return math.Round(x*p) / p }
func ms(iso any) float64 { // an API timestamp
	t, _ := time.Parse(time.RFC3339Nano, str(iso))
	return float64(t.UnixMilli())
}
func evKey(e M) string {
	u := str(e["url"])
	if u == "" {
		u = str(e["selector"])
	}
	return str(e["type"]) + " " + str(e["at"]) + " " + u
}
func byType(events []M) M {
	o := M{}
	for _, t := range types {
		n := 0
		for _, e := range events {
			if str(e["type"]) == t {
				n++
			}
		}
		if n > 0 {
			o[t] = n
		}
	}
	return o
}

// x-cdpfleet-signature: t=<unix>,v1=<hex HMAC-SHA256(secret, "<t>.<raw body>")>, checked on the raw bytes.
func verify(secret, header, rawBody string) bool {
	parts := map[string]string{}
	for _, kv := range strings.Split(header, ",") {
		if k, v, ok := strings.Cut(kv, "="); ok {
			parts[k] = v
		}
	}
	given, err := hex.DecodeString(parts["v1"])
	if parts["t"] == "" || err != nil {
		return false
	}
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(parts["t"] + "." + rawBody))
	return hmac.Equal(mac.Sum(nil), given)
}

func contains(list []string, s string) bool {
	for _, x := range list {
		if x == s {
			return true
		}
	}
	return false
}

// Great-circle distance in km.
func km(lat1, lon1, lat2, lon2 float64) float64 {
	r := func(d float64) float64 { return d * math.Pi / 180 }
	h := math.Pow(math.Sin(r(lat2-lat1)/2), 2) + math.Cos(r(lat1))*math.Cos(r(lat2))*math.Pow(math.Sin(r(lon2-lon1)/2), 2)
	return 2 * 6371 * math.Asin(math.Sqrt(h))
}

type live struct {
	ev       M
	received float64
}

func main() {
	// The webhook receiver here is cdpfleet.com's own test inbox: it keeps deliveries an hour (raw body and
	// headers) and reads them back with GET. Use your own public https endpoint.
	tok := make([]byte, 18)
	rand.Read(tok)
	hook := api + "/e2e/hook/" + base64.RawURLEncoding.EncodeToString(tok)
	_, sec := call("GET", "/v1/me/webhook-secret", nil)
	secret := str(sec["secret"])

	// 1. Start the watch: XHR/fetch calls matching the glob, and changes under the selector, with a webhook.
	watch := M{
		"url": page, "browser": "chrome",
		"duration": 300, // a safety net: the program stops the job itself
		"filter":   M{"url": []string{"*/w/ajax/realtime*"}, "selector": ".span3 table.table-striped", "types": []string{"xhr", "fetch", "dom"}},
	}
	body := M{"deliver": M{"webhook": M{"url": hook, "events": []string{"event", "done"}}}}
	for k, v := range watch {
		body[k] = v
	}
	t0 := nowMs()
	startStatus, start := call("POST", "/v1/functions/watch", body)
	if startStatus != 202 {
		panic(fmt.Sprintf("watch %d %v", startStatus, start))
	}
	id := str(start["job_id"])

	// 2. Follow it live on the SSE stream; stop it with DELETE after stopAfterS seconds.
	var mu sync.Mutex
	var stop M
	var stopAt float64
	timer := time.AfterFunc(stopAfterS*time.Second, func() {
		t := nowMs()
		status, r := call("DELETE", "/v1/jobs/"+id, nil)
		mu.Lock()
		stop, stopAt = M{"status": status, "reply": r["status"]}, t
		mu.Unlock()
	})

	var events []live
	seen := M{"progress": 0, "part": 0, "event": 0, "done": 0}
	var doneData M
	req, _ := http.NewRequest("GET", api+"/v1/jobs/"+id+"/events", nil)
	req.Header.Set("x-api-key", key)
	req.Header.Set("accept", "text/event-stream")
	sse, err := (&http.Client{Timeout: 240 * time.Second}).Do(req)
	if err != nil {
		panic(err)
	}
	sc := bufio.NewScanner(sse.Body)
	sc.Buffer(make([]byte, 1<<20), 4<<20)
	name, data := "", []string{}
	for sc.Scan() {
		line := sc.Text()
		if line != "" {
			if strings.HasPrefix(line, "event:") {
				name = strings.TrimSpace(line[6:])
			} else if strings.HasPrefix(line, "data:") {
				data = append(data, strings.TrimSpace(line[5:]))
			}
			continue
		}
		if name != "" { // a blank line ends a message; a keep-alive comment has no event name
			n, _ := seen[name].(int)
			seen[name] = n + 1
			var payload M
			json.Unmarshal([]byte(strings.Join(data, "\n")), &payload)
			if name == "event" {
				for _, ev := range arr(payload["events"]) {
					events = append(events, live{obj(ev), nowMs()})
				}
			}
			if name == "done" {
				doneData = payload
				break
			}
		}
		name, data = "", nil
	}
	sse.Body.Close()
	timer.Stop()
	tEnd := nowMs()

	// 3. The job as it ended, and its events as stored.
	_, job := call("GET", "/v1/jobs/"+id, nil)
	for i := 0; i < 20 && !ended[str(job["status"])]; i++ {
		time.Sleep(2 * time.Second)
		_, job = call("GET", "/v1/jobs/"+id, nil)
	}
	var stored []M
	for cursor := 0.0; ; {
		_, r := call("GET", fmt.Sprintf("/v1/jobs/%s/results?cursor=%d", id, int64(cursor)), nil)
		for _, x := range arr(r["results"]) {
			stored = append(stored, obj(x))
		}
		if r["next_cursor"] == nil || num(r["next_cursor"]) == cursor {
			break
		}
		cursor = num(r["next_cursor"])
	}

	// 4. The webhook deliveries: job.event batches (at most one per 2 s) and job.done. Wait for job.done.
	var items []M
	for end := time.Now().Add(90 * time.Second); time.Now().Before(end); time.Sleep(2 * time.Second) {
		_, b := send("GET", hook, nil, false)
		var box struct{ Items []M }
		json.Unmarshal(b, &box)
		items = box.Items
		found := false
		for _, d := range items {
			found = found || str(d["event"]) == "job.done"
		}
		if found {
			break
		}
	}
	var hookEvents []M
	var sent, delays []float64
	var doneHook M
	byEvent := M{}
	valid, maxPerBatch := 0, 0
	ids := map[string]bool{}
	allUUID, jobMatches := true, true
	sigRe := regexp.MustCompile(`t=(\d+)`)
	for _, d := range items {
		var b M
		json.Unmarshal([]byte(str(d["body"])), &b)
		ev := str(d["event"])
		n, _ := byEvent[ev].(int)
		byEvent[ev] = n + 1
		if verify(secret, str(d["signature"]), str(d["body"])) {
			valid++
		}
		ids[str(d["delivery"])] = true
		allUUID = allUUID && uuidRe.MatchString(str(d["delivery"]))
		jobMatches = jobMatches && str(b["job_id"]) == id
		if ev == "job.done" {
			doneHook = obj(b["data"])
		}
		if ev != "job.event" {
			continue
		}
		evs := arr(obj(b["data"])["events"])
		newest := 0.0
		for _, e := range evs {
			hookEvents = append(hookEvents, obj(e))
			newest = math.Max(newest, ms(obj(e)["at"]))
		}
		maxPerBatch = max(maxPerBatch, len(evs))
		sent = append(sent, ms(b["created_at"]))
		// How long after its newest event each batch was signed (the signature's t, whole seconds).
		if m := sigRe.FindStringSubmatch(str(d["signature"])); m != nil {
			t, _ := strconv.ParseFloat(m[1], 64)
			delays = append(delays, t-newest/1000)
		}
	}
	sort.Float64s(sent)
	sort.Float64s(delays)
	var minGap, minGapBeforeStop any
	for i := 1; i < len(sent); i++ {
		g := secs(sent[i] - sent[i-1])
		if minGap == nil || g < minGap.(float64) {
			minGap = g
		}
		if i < len(sent)-1 && (minGapBeforeStop == nil || g < minGapBeforeStop.(float64)) {
			minGapBeforeStop = g
		}
	}
	storedKeys, hookKeys := map[string]bool{}, map[string]bool{}
	for _, e := range stored {
		storedKeys[evKey(e)] = true
	}
	for _, e := range hookEvents {
		hookKeys[evKey(e)] = true
	}

	// 5. What the events say: the ISS position from each XHR response, the table text from each DOM change.
	type pos struct {
		at, lat, lon float64
		time         string
	}
	var positions []pos
	var dom []M
	globOK, noBody := true, 0
	for _, e := range stored {
		switch str(e["type"]) {
		case "xhr":
			globOK = globOK && strings.Contains(str(e["url"]), "/w/ajax/realtime") && num(e["status"]) == 200 && str(e["method"]) == "GET"
			if str(e["response_body"]) == "" { // an XHR still in flight when the page reloads has no body
				noBody++
				continue
			}
			var r M
			json.Unmarshal([]byte(str(e["response_body"])), &r)
			d := obj(r["data"])
			positions = append(positions, pos{ms(e["at"]), num(d["lat"]), num(d["lon"]), str(d["time"])})
		case "dom":
			dom = append(dom, e)
		}
	}
	var lags []float64
	liveEvents := make([]M, len(events))
	sameAsLive := len(stored) == len(events)
	for i, x := range events {
		lags = append(lags, (x.received-ms(x.ev["at"]))/1000)
		liveEvents[i] = x.ev
		sameAsLive = sameAsLive && storedKeys[evKey(x.ev)]
	}
	sort.Float64s(lags)
	firstOf := func(t string) any {
		for _, x := range events {
			if str(x.ev["type"]) == t {
				return secs(x.received - t0)
			}
		}
		return nil
	}

	// 6. WebSocket messages: start the second watch, stop it after wsStopAfterS, read what it stored.
	tWs := nowMs()
	wsStartStatus, wsStart := call("POST", "/v1/functions/watch", wsWatch)
	if wsStartStatus != 202 {
		panic(fmt.Sprintf("ws watch %d %v", wsStartStatus, wsStart))
	}
	wsID := str(wsStart["job_id"])
	time.Sleep(time.Duration(wsStopAfterS*1000-(nowMs()-tWs)) * time.Millisecond)
	call("DELETE", "/v1/jobs/"+wsID, nil)
	_, wsJob := call("GET", "/v1/jobs/"+wsID, nil)
	for i := 0; i < 20 && !ended[str(wsJob["status"])]; i++ {
		time.Sleep(2 * time.Second)
		_, wsJob = call("GET", "/v1/jobs/"+wsID, nil)
	}
	var wsEvents []M
	for cursor := 0.0; ; {
		_, r := call("GET", fmt.Sprintf("/v1/jobs/%s/results?cursor=%d", wsID, int64(cursor)), nil)
		for _, x := range arr(r["results"]) {
			wsEvents = append(wsEvents, obj(x))
		}
		if r["next_cursor"] == nil || num(r["next_cursor"]) == cursor {
			break
		}
		cursor = num(r["next_cursor"])
	}
	// Each message is the frame's text: here JSON, cut at 64 KB (a cut one no longer parses).
	inKeys := M{}
	sentMsgs, wsURLs := []string{}, []string{}
	wsIn, wsOut := 0, 0
	var firstIn, mempoolFirst, mempoolLast any
	mempoolUpdates := 0
	for _, e := range wsEvents {
		if !contains(wsURLs, str(e["url"])) {
			wsURLs = append(wsURLs, str(e["url"]))
		}
		if str(e["direction"]) == "out" {
			wsOut++
			sentMsgs = append(sentMsgs, str(e["data"]))
			continue
		}
		wsIn++
		if firstIn == nil {
			firstIn = secs(ms(e["at"]) - tWs)
		}
		k := "(cut at 64 KB)"
		if data := str(e["data"]); json.Valid([]byte(data)) {
			dec := json.NewDecoder(strings.NewReader(data))
			if t, _ := dec.Token(); t == json.Delim('{') {
				if t, _ := dec.Token(); t != nil {
					k = str(t)
				}
			}
			var msg M
			json.Unmarshal([]byte(data), &msg)
			if info := obj(msg["mempoolInfo"]); info != nil {
				if mempoolFirst == nil {
					mempoolFirst = info["size"]
				}
				mempoolLast = info["size"]
				mempoolUpdates++
			}
		}
		n, _ := inKeys[k].(int)
		inKeys[k] = n + 1
	}
	var mempoolOut any
	if mempoolUpdates > 0 {
		mempoolOut = M{"first": mempoolFirst, "last": mempoolLast, "updates": mempoolUpdates}
	}

	// 7. The bill: a watch runs in your own browser session, one thread for a headless Chrome.
	since := time.UnixMilli(int64(t0) - 60000).UTC().Format("2006-01-02T15:04:05.000Z")
	_, ss := call("GET", "/v1/me/sessions?from="+url.QueryEscape(since)+"&limit=200", nil)
	var session, wsSessionEngine any
	for _, s := range arr(ss["sessions"]) {
		so := obj(s)
		if sid := str(job["session_id"]); sid != "" && (str(so["session_id"]) == sid || str(so["id"]) == sid) {
			session = M{"engine": so["engine"], "weight": so["weight"], "seconds": secs(ms(so["ended_at"]) - ms(so["started_at"]))}
		}
		if sid := str(wsJob["session_id"]); sid != "" && (str(so["session_id"]) == sid || str(so["id"]) == sid) {
			wsSessionEngine = so["engine"]
		}
	}

	watchOut := M{"stop_after_s": stopAfterS, "start_status": startStatus}
	for k, v := range watch {
		watchOut[k] = v
	}
	sseOut := M{
		"events": len(events), "by_type": byType(liveEvents), "messages": seen,
		"first_event_seconds": nil, "first_xhr_seconds": firstOf("xhr"), "first_dom_seconds": firstOf("dom"),
		"lag_seconds": nil, "done_status": doneData["status"], "stream_seconds": secs(tEnd - t0),
	}
	if len(events) > 0 {
		sseOut["first_event_seconds"] = secs(events[0].received - t0)
		sseOut["lag_seconds"] = M{"min": round(lags[0], 2), "median": round(lags[len(lags)/2], 2), "max": round(lags[len(lags)-1], 2)}
	}
	var stopOut any
	mu.Lock()
	if stop != nil {
		stopOut = M{"after_seconds": secs(stopAt - t0), "status": stop["status"], "reply": stop["reply"], "seconds_to_ended": secs(ms(job["ended_at"]) - stopAt)}
	}
	mu.Unlock()
	var results any
	if r := obj(job["results"]); r != nil {
		results = M{"parts": r["parts"], "bytes": r["bytes"]}
	}
	distinct := map[string]bool{}
	for _, p := range positions {
		distinct[fmt.Sprint(p.lat, ",", p.lon)] = true
	}
	storedOut := M{
		"events": len(stored), "by_type": byType(stored), "same_as_live": sameAsLive, "xhr_urls_match_glob": globOK,
		"xhr_without_body": noBody, "positions": len(positions), "distinct_positions": len(distinct),
		"first": nil, "last": nil, "ground_km": nil, "ground_km_per_s": nil,
		"dom_events": len(dom), "dom_text_sample": nil, "dom_html_bytes": nil,
	}
	if len(positions) > 0 {
		f, l := positions[0], positions[len(positions)-1]
		dist := km(f.lat, f.lon, l.lat, l.lon)
		storedOut["first"] = M{"lat": round(f.lat, 3), "lon": round(f.lon, 3), "time": f.time}
		storedOut["last"] = M{"lat": round(l.lat, 3), "lon": round(l.lon, 3), "time": l.time}
		storedOut["ground_km"] = round(dist, 1)
		if l.at > f.at {
			storedOut["ground_km_per_s"] = round(dist/((l.at-f.at)/1000), 2)
		}
	}
	if len(dom) > 0 {
		storedOut["dom_text_sample"] = strings.Join(strings.Fields(str(dom[0]["text"])), " ")
		storedOut["dom_html_bytes"] = len(str(dom[0]["html"]))
	}
	newestSent := 0.0
	for _, e := range hookEvents {
		newestSent = math.Max(newestSent, ms(e["at"]))
	}
	notInHooks, notSentLast := 0, true
	for _, e := range stored {
		if !hookKeys[evKey(e)] {
			notInHooks++
			notSentLast = notSentLast && ms(e["at"]) >= newestSent
		}
	}
	allStored := true
	for _, e := range hookEvents {
		allStored = allStored && storedKeys[evKey(e)]
	}
	var delayOut, doneOut any
	if len(delays) > 0 {
		delayOut = M{"min": round(delays[0], 1), "max": round(delays[len(delays)-1], 1)}
	}
	if doneHook != nil {
		doneOut = M{"status": doneHook["status"], "progress": doneHook["progress"], "error": doneHook["error"]}
	}
	tampered := len(items) > 0 && !verify(secret, str(items[0]["signature"]), str(items[0]["body"])+" ")

	out := M{
		"language": "go",
		"page":     page,
		"watch":    watchOut,
		"sse":      sseOut,
		"stop":     stopOut,
		"job": M{
			"status": job["status"], "own_session": job["own_session"], "concurrency": job["concurrency"], "runs": job["runs"],
			"progress": job["progress"], "billed_seconds": job["billed_seconds"],
			"run_seconds": secs(ms(job["ended_at"]) - ms(job["started_at"])), "results": results, "error": job["error"],
		},
		"session": session,
		"stored":  storedOut,
		"webhook": M{
			"deliveries": len(items), "inbox_full": len(items) >= 20, "by_event": byEvent,
			"signatures_valid": valid, "tampered_body_rejected": tampered,
			"delivery_ids_unique_uuids": len(ids) == len(items) && allUUID, "job_id_matches": jobMatches,
			"events": len(hookEvents), "events_by_type": byType(hookEvents), "events_all_stored": allStored,
			"stored_not_in_webhooks": notInHooks, "not_sent_are_the_last": notSentLast, "signed_after_newest_event_seconds": delayOut,
			"min_gap_seconds": minGap, "min_gap_before_stop_seconds": minGapBeforeStop, "max_events_per_batch": maxPerBatch, "done": doneOut,
		},
		"ws": M{
			"page": wsWatch["url"], "filter": wsWatch["filter"], "start_status": wsStartStatus, "status": wsJob["status"],
			"session_engine": wsSessionEngine, "events": len(wsEvents), "in": wsIn, "out": wsOut, "urls": wsURLs,
			"first_in_seconds": firstIn, "sent": sentMsgs, "received": inKeys, "mempool_txs": mempoolOut,
			"billed_seconds": wsJob["billed_seconds"],
		},
		"measured_at": time.Now().UTC().Format("2006-01-02"),
	}
	enc := json.NewEncoder(os.Stdout)
	enc.SetEscapeHTML(false)
	enc.SetIndent("", "  ")
	enc.Encode(out)
}
```

## What we got

| Step | What happened | Measured |
| --- | --- | --- |
| Start | 202: Chrome on https://wheretheiss.at/, XHR/fetch matching \*/w/ajax/realtime\* and changes under \`.span3 table.table-striped\` | 2026-11-10 |
| SSE | 60 events (30 xhr + 30 dom) in 24 messages; first XHR 12.1 s and first DOM change 12.1 s after the start; 0.53 s (max 1.01 s) from an event to its arrival | 2026-11-10 |
| Stop | DELETE at 40.1 s → stopping; job stopped 0.5 s later, SSE \`done\` with stopped | 2026-11-10 |
| Stored | 60 events (30 xhr + 30 dom), the same as arrived live: true | 2026-11-10 |
| Webhooks | 17 deliveries (16 job.event, 1 job.done), 17 of 17 signatures valid; 60 events in the batches, 0 stored events not sent; batches ≥ 2 s apart until the stop; job.done says stopped | 2026-11-10 |
| The data | ISS from -24.513, 71.937 (2:15:23 am PST) to -23.144, 73.19 (2:15:52 am PST): 198.5 km of ground track, 6.92 km/s, 30 distinct positions in 30 responses | 2026-11-10 |
| DOM event | Time 2:15:23 am PST Altitude 262.23 miles Velocity 17,135.44 mph Latitude -24.513271274056 Longitude 71.937013051503 The ISS is in daylight | 2026-11-10 |
| WebSocket | https://mempool.space/: 4 messages sent, 31 received (27 mempoolInfo, 1 blocks, 1 rbfLatestSummary, 1 (cut at 64 KB), 1 live-2h-chart), first 18.1 s after the start; chrome by default, 45 thread-seconds | 2026-11-10 |
| Cost | 40 thread-seconds for 40.6 s of running (own session: chrome, weight 1) | 2026-11-10 |

From the Node.js run on 2026-11-10. IP addresses are replaced with placeholders (203.0.113.x); equal addresses stay equal. The other languages produced the same findings.

Raw output (Node.js)

```
{
  "language": "node",
  "page": "https://wheretheiss.at/",
  "watch": {
    "url": "https://wheretheiss.at/",
    "browser": "chrome",
    "duration": 300,
    "filter": {
      "url": [
        "*/w/ajax/realtime*"
      ],
      "selector": ".span3 table.table-striped",
      "types": [
        "xhr",
        "fetch",
        "dom"
      ]
    },
    "stop_after_s": 40,
    "start_status": 202
  },
  "sse": {
    "events": 60,
    "by_type": {
      "xhr": 30,
      "dom": 30
    },
    "messages": {
      "progress": 4,
      "part": 1,
      "event": 24,
      "done": 1
    },
    "first_event_seconds": 12.1,
    "first_xhr_seconds": 12.1,
    "first_dom_seconds": 12.1,
    "lag_seconds": {
      "min": 0.02,
      "median": 0.53,
      "max": 1.01
    },
    "done_status": "stopped",
    "stream_seconds": 42.2
  },
  "stop": {
    "after_seconds": 40.1,
    "status": 200,
    "reply": "stopping",
    "seconds_to_ended": 0.5
  },
  "job": {
    "status": "stopped",
    "own_session": true,
    "concurrency": 1,
    "runs": 1,
    "progress": {
      "events": 60
    },
    "billed_seconds": 40,
    "run_seconds": 40.6,
    "results": {
      "parts": 1,
      "bytes": 52937
    },
    "error": null
  },
  "session": {
    "engine": "chrome",
    "weight": 1,
    "seconds": 40
  },
  "stored": {
    "events": 60,
    "by_type": {
      "xhr": 30,
      "dom": 30
    },
    "same_as_live": true,
    "xhr_urls_match_glob": true,
    "xhr_without_body": 0,
    "positions": 30,
    "distinct_positions": 30,
    "first": {
      "lat": -24.513,
      "lon": 71.937,
      "time": "2:15:23 am PST"
    },
    "last": {
      "lat": -23.144,
      "lon": 73.19,
      "time": "2:15:52 am PST"
    },
    "ground_km": 198.5,
    "ground_km_per_s": 6.92,
    "dom_events": 30,
    "dom_text_sample": "Time 2:15:23 am PST Altitude 262.23 miles Velocity 17,135.44 mph Latitude -24.513271274056 Longitude 71.937013051503 The ISS is in daylight",
    "dom_html_bytes": 1038
  },
  "webhook": {
    "deliveries": 17,
    "inbox_full": false,
    "by_event": {
      "job.event": 16,
      "job.done": 1
    },
    "signatures_valid": 17,
    "tampered_body_rejected": true,
    "delivery_ids_unique_uuids": true,
    "job_id_matches": true,
    "events": 60,
    "events_by_type": {
      "xhr": 30,
      "dom": 30
    },
    "events_all_stored": true,
    "stored_not_in_webhooks": 0,
    "not_sent_are_the_last": true,
    "signed_after_newest_event_seconds": {
      "min": -0.6,
      "max": 0.4
    },
    "min_gap_seconds": 0.4,
    "min_gap_before_stop_seconds": 2,
    "max_events_per_batch": 6,
    "done": {
      "status": "stopped",
      "progress": {
        "events": 60
      },
      "error": null
    }
  },
  "ws": {
    "page": "https://mempool.space/",
    "filter": {
      "url": [
        "*/api/v1/ws*"
      ],
      "types": [
        "ws"
      ]
    },
    "start_status": 202,
    "status": "stopped",
    "session_engine": "chrome",
    "events": 35,
    "in": 31,
    "out": 4,
    "urls": [
      "wss://mempool.space/api/v1/ws"
    ],
    "first_in_seconds": 18.1,
    "sent": [
      "{\"action\":\"init\"}",
      "{\"action\":\"want\",\"data\":[\"blocks\",\"stats\",\"mempool-blocks\",\"live-2h-chart\"]}",
      "{\"track-rbf-summary\":true}",
      "{\"track-mempool-block\":0}"
    ],
    "received": {
      "mempoolInfo": 27,
      "blocks": 1,
      "rbfLatestSummary": 1,
      "(cut at 64 KB)": 1,
      "live-2h-chart": 1
    },
    "mempool_txs": {
      "first": 75815,
      "last": 75936,
      "updates": 28
    },
    "billed_seconds": 45
  },
  "measured_at": "2026-11-10"
}
```

## Takeaways

-   **The whole loop works, in all five languages.** On 2026-11-10 (re-run after the Functions fixes of that day) each run started the watch, followed it on SSE, stopped it after 40 s and found the same events three ways: 50–60 of them per run (25–30 XHR responses and as many DOM changes, give or take one), stored with the job exactly as they had arrived live, and every one of them in a signed webhook.
-   **Live means within a second.** The first event arrived 8.3–17.3 s after the `POST` (starting a Chrome and loading the page); after that each event reached the SSE stream a median 0.52–0.55 s and at most 1.02 s after it happened — the service sends events in batches of up to 1 s. The page asks for the position about once a second, and every response was a new position: 25–30 per run, the ISS moving 164–226 km at 6.92–6.94 km/s over the ground. The data comes from the page's own JSON, without parsing its HTML or calling its API yourself.
-   **The filter decides what you pay attention to.** With `filter.url: ["*/w/ajax/realtime*"]` every reported call was the position request (`GET`, 200, with its `response_body`), and nothing else the page loads was reported.
-   **DOM events start with the page.** A DOM event carries the element's `text` and `html` (about 1 KB here); the first one came in the same second as the first XHR in every run, and then one per position update — no `reload_every` needed. (Before the 2026-11-10 fix the observer only started at the first reload, 11–15 s later; `reload_every` is now only for pages that need refreshing.)
-   **Stop is immediate.** `DELETE` answered `stopping`; the job was `stopped` 0.4–0.7 s later and the SSE stream ended with `done` and status `stopped`.
-   **Webhooks carry everything, in batches.** Each run received 13–16 `job.event` batches with up to 6 events each, plus one `job.done` with `status: "stopped"`; all 79 deliveries had a valid signature (a changed body failed) and a unique delivery id, and each batch was signed within 1.3 s of its newest event. Batches came at least 2.0 s apart until the stop; the last one, flushed at the stop, followed the previous after 0.4–2.0 s. Every stored event was also in a webhook — the last batch before a stop is no longer lost.
-   **WebSocket messages, both ways.** On mempool.space the watch captured the page's own subscription — the 4 messages it sent (`{"action":"init"}`, `{"action":"want","data":["blocks","stats","mempool-blocks","live-2h-chart"]}` and two more) — and 24–38 messages it received in 45 s, the first 6.4–24.0 s after the `POST`: one `blocks` snapshot and then a `mempoolInfo` update about every second (the mempool at 75,815–77,337 transactions). `data` is the frame's text, cut at 64 KB: one large message per run no longer parsed as JSON — check before you parse. The job ran in Chrome without asking for it. One more attempt (C#, not kept) ended `failed` with `error: "navigation_failed"` after 31 s, and was billed those 31 s: check a job's `status` and `error`, not only its events.
-   **A watch is paid by the second, not by the event: 40 thread-seconds for 40 s.** It runs in your own browser session (`own_session: true`; a `chrome` session of weight 1 in the session history), and `billed_seconds` matched its running time (40.4–40.7 s); the 45 s WebSocket watch was billed 45–46 — 3,600 thread-seconds an hour for a headless Chrome, however many events. That is what makes it the tool for data that changes by the second; for a page that changes a few times a day, a [monitor](https://cdpfleet.com/docs/cases/monitor-end-to-end) costs a short scrape per check.

---
Source: https://cdpfleet.com/docs/cases/watch-a-page-live · Markdown: https://cdpfleet.com/docs/cases/watch-a-page-live.md · All docs for AI tools: https://cdpfleet.com/llms.txt