Production-Grade Reserve Streaming

// resilient-stream.mjs: production-grade reserve streaming. Node 18+, no deps.
// Reconnects with real backoff, detects missed blocks, classifies errors,
// and watches its own heartbeat.

const SUBSCRIPTIONS = [
  { chain: "base", address: "0xb2cc224c1c9fee385f8ad6a55b4d94e92359dc59", method: "pool_reserves" },
  { chain: "solana", address: "8sjV1AqBFvFuADBCQHhotaRq5DFFYSjjg1jMyVWMqXvZ", method: "pool_reserves" },
];

const STALE_MS = 30_000; // two missed pings = reconnect
const MAX_BACKOFF_MS = 30_000;

// Capacity errors clear on their own; subscription errors never will.
function isRetryable(message) {
  return /limit|rate|too many|capacity|timeout|temporar/i.test(message);
}

const lastBlock = new Map(); // subscription index -> last seen block

function handleEvent(idx, update) {
  // Absolute state: total_reserve_usd is the pool's liquidity NOW, so state
  // self-heals after any gap. Only the per-block ledger can have holes.
  const prev = lastBlock.get(idx);
  if (prev !== undefined && update.previous_block !== prev) {
    console.warn(
      `[gap] sub ${idx}: missed blocks ${prev} -> ${update.previous_block}. ` +
        "Reset any delta-window aggregation for this subscription.",
    );
  }
  lastBlock.set(idx, update.block);

  console.log(
    `sub ${idx} block ${update.block}: $${Math.round(update.total_reserve_usd).toLocaleString("en-US")} ` +
      `(${update.total_delta_usd >= 0 ? "+" : "-"}$${Math.abs(update.total_delta_usd).toFixed(2)})`,
  );
}

async function connectOnce(signal) {
  const res = await fetch("https://streaming.dexpaprika.com/sse/reserves", {
    method: "POST",
    headers: { "content-type": "application/json" },
    body: JSON.stringify(SUBSCRIPTIONS),
    signal,
  });
  if (!res.ok) {
    const body = await res.text();
    // 4xx other than 408/429 means the request itself is wrong: do not retry.
    const fatal = res.status >= 400 && res.status < 500 && ![408, 429].includes(res.status);
    throw Object.assign(new Error(`${res.status}: ${body}`), { fatal });
  }

  let sawEvent = false;
  let lastActivity = Date.now();
  const controller = new AbortController();
  signal.addEventListener("abort", () => controller.abort(), { once: true });
  const watchdog = setInterval(() => {
    if (Date.now() - lastActivity > STALE_MS) {
      console.warn("[watchdog] no events or pings for 30s, forcing reconnect");
      controller.abort(); // tears down the fetch, the outer loop reconnects
    }
  }, 5_000);

  try {
    let buffer = "";
    for await (const chunk of res.body.pipeThrough(new TextDecoderStream())) {
      lastActivity = Date.now();
      buffer += chunk;
      const frames = buffer.split("\n\n");
      buffer = frames.pop();

      for (const frame of frames) {
        const event = frame.match(/^event: (.*)$/m)?.[1];
        const id = frame.match(/^request_id: (.*)$/m)?.[1];
        const data = frame.match(/^data: (.*)$/m)?.[1];
        if (!data) continue;
        sawEvent = true; // proof the stream works, not just HTTP 200

        if (event === "error") {
          const message = JSON.parse(data).message ?? data;
          throw Object.assign(new Error(message), { fatal: !isRetryable(message) });
        }
        if (event === "warning") {
          console.warn(`[server] ${JSON.parse(data).message ?? data}`);
          continue;
        }
        if (event === "ping") continue; // counts as activity, nothing else
        if (event === "pool_reserves" || event === "token_reserves") {
          handleEvent(Number(id), JSON.parse(data));
        }
      }
    }
  } finally {
    clearInterval(watchdog);
  }
  return sawEvent;
}

let backoff = 1_000;
while (true) {
  const session = new AbortController();
  try {
    console.log("connecting…");
    const healthy = await connectOnce(session.signal);
    if (healthy) backoff = 1_000; // reset only after a proven-working stream
  } catch (err) {
    if (err.fatal) {
      console.error(`fatal, fix the subscription list: ${err.message}`);
      process.exit(1);
    }
    console.error(`transient: ${err.message}`);
  }
  await new Promise((r) => setTimeout(r, backoff + Math.random() * 1_000)); // jitter
  backoff = Math.min(backoff * 2, MAX_BACKOFF_MS);
}