Fill stream

async function streamFills(token) {
  const res = await fetch(`${API_BASE}/fut/stream/user?openOnly=true`, {
    headers: { Authorization: `Bearer ${token}`, Accept: "text/event-stream" },
  });
  const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
  let buf = "", lastSeq = null;

  for (;;) {
    const { value, done } = await reader.read();
    if (done) return reconnect();
    buf += value;
    let i;
    while ((i = buf.indexOf("\n\n")) >= 0) {
      const frame = buf.slice(0, i);
      buf = buf.slice(i + 2);
      const event = /^event:\s?(.*)$/m.exec(frame)?.[1];
      const data = /^data:\s?(.*)$/m.exec(frame)?.[1];
      if (!data) continue; // heartbeat comment

      const { eventSeq, payload } = JSON.parse(data);
      if (event === "snapshot") { lastSeq = eventSeq; initPositions(payload.positions); continue; }
      if (event === "delta") {
        if (lastSeq !== null && eventSeq !== lastSeq + 1) return reconnect(); // gap
        lastSeq = eventSeq;
        for (const t of payload.trades ?? []) onFill(t);                 // carries marketId
        for (const p of payload.positions?.upserts ?? []) onPosition(p);
      }
    }
  }
}