Stream executions

{ "token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..." }

/**
 * Stream one execution (or batch) to completion.
 * @param {"execution"|"batch"} resourceType
 * @param {string} resourceId  e.g. "exec_aabbccdd" or "batch_11223344"
 * @param {() => Promise<string>} getApiKey  resolves to your raw API key (the X-API-Key header value)
 * @param {(event: {type: string, data: any}) => void} onEvent
 */
async function streamResource(resourceType, resourceId, getApiKey, onEvent) {
  const BASE = "https://api.getdialed.ai/v1";
  const streamPath =
    resourceType === "execution"
      ? `/flows/executions/${resourceId}/stream`
      : `/flows/batches/${resourceId}/stream`;

  const seen = new Set(); // "(entity_id, sequence)" pairs already rendered
  let lastEventId = null; // "{entity_id}#{sequence}" of the last frame processed
  let es = null;
  let done = false;

  // Mint a fresh 60s stream token for this resource.
  async function mintToken() {
    const res = await fetch(`${BASE}/account/auth/stream-token`, {
      method: "POST",
      headers: {
        "X-API-Key": await getApiKey(),
        "Content-Type": "application/json",
      },
      body: JSON.stringify({
        resource_type: resourceType,
        resource_id: resourceId,
      }),
    });
    if (!res.ok) throw new Error(`stream-token mint failed: ${res.status}`);
    return (await res.json()).token;
  }

  const TERMINAL = new Set(["execution.completed", "execution.failed"]);
  const EVENT_TYPES = [
    "execution.created",
    "step.started",
    "step.completed",
    "step.failed",
    "execution.completed",
    "execution.failed",
  ];

  function handleFrame(type, evt) {
    // Frame id is "{entity_id}#{sequence}" — the Last-Event-ID resume anchor.
    lastEventId = evt.lastEventId;
    const key = evt.lastEventId; // already "(entity_id, sequence)" in one string
    if (seen.has(key)) return; // dedup cross-reconnect repeats (batch stream)
    seen.add(key);

    const data = JSON.parse(evt.data);
    onEvent({ type, data });

    // Terminal event: stop. A browser EventSource otherwise reconnects forever.
    if (TERMINAL.has(type)) {
      done = true;
      es.close();
    }
  }

  async function connect() {
    if (done) return;
    const token = await mintToken();
    es = new EventSource(`${BASE}${streamPath}?token=${encodeURIComponent(token)}`);

    for (const type of EVENT_TYPES) {
      es.addEventListener(type, (evt) => handleFrame(type, evt));
    }

    es.onerror = () => {
      if (done) return;
      // readyState CLOSED = fatal (token expired -> 401). CONNECTING = the
      // browser is auto-reconnecting within the TTL (Last-Event-ID header sent
      // for us) — let it. On a fatal close, re-mint and reopen; the full
      // backlog replays and (entity_id, sequence) dedup drops the repeats.
      if (es.readyState === EventSource.CLOSED) {
        es.close();
        connect().catch((err) => onEvent({ type: "error", data: String(err) }));
      }
    };
  }

  await connect();
}