// You provide: backoffWithJitter, resetBackoff, onFatal.
const N_LEVELS = 20;
const books = {};      // coin -> { bids: Map<px, level>, asks: Map<px, level> }
let subId = null;      // current subscription; messages for any other ID are ignored
let generation = 0;
let lastSeq = 0;
let ready = false;     // true once the current subscription's snapshot has arrived
let halted = false;    // set when an error needs a fix, not a retry
 
function connect() {
  const ws = new WebSocket("wss://hypercore-mainnet.g.alchemy.com/v2/<API_KEY>");
  let closed = false;
 
  // Start a fresh subscription. A new ID means late messages from the old one are ignored.
  function subscribe() {
    if (closed) return;
    if (subId) ws.send(JSON.stringify({ method: "unsubscribe", subscriptionId: subId }));
    subId = `book-${++generation}`;
    lastSeq = 0;
    resetBooks();
    ws.send(JSON.stringify({
      method: "subscribe",
      subscriptionId: subId,
      subscription: { type: "l2BookDiff", coins: ["BTC", "ETH"], nLevels: N_LEVELS },
    }));
  }
 
  ws.onmessage = ({ data }) => {
    if (closed) return;
    let msg;
    try {
      msg = JSON.parse(data);
    } catch {
      return ws.close();                       // malformed message: reconnect
    }
    if (msg.channel === "connected") {
      subId = null;                            // new connection, nothing to unsubscribe
      return subscribe();
    }
 
    // Errors come as a subscription error (with our ID) or a connection-level error (no ID).
    if (msg.error && (msg.channel === "error" || msg.subscriptionId === subId)) {
      if (msg.error.code === "book_reset") return subscribe();      // start over from a full snapshot
      if (msg.error.code === "book_not_ready" || msg.error.code === "going_away") return ws.close();   // temporary: reconnect with backoff
      halted = true;                           // e.g. invalid_params or limit_reached
      onFatal(msg.error);
      return ws.close();
    }
    if (msg.subscriptionId !== subId) return;  // stale or unrelated subscription
    if (msg.seq === undefined) return resetBackoff();  // acknowledgement: the connection is healthy
 
    if (msg.seq !== lastSeq + 1) return subscribe();  // missed a message: start over
    lastSeq = msg.seq;
 
    if (msg.data.isSnapshot) ready = true;
    if (!ready) return;                        // never apply diffs without a snapshot
 
    try {
      for (const { coin, levels } of msg.data.books) {
        if (msg.data.isSnapshot) books[coin] = { bids: new Map(), asks: new Map() };
        else if (!books[coin]) return ws.close();    // diff for a market with no snapshot: reconnect with backoff
        const [bids, asks] = levels;
        apply(books[coin].bids, bids, (a, b) => b - a);  // bids: high to low
        apply(books[coin].asks, asks, (a, b) => a - b);  // asks: low to high
      }
    } catch {
      return ws.close();                       // unexpected message shape: reconnect with backoff
    }
  };
 
  ws.onclose = () => {
    closed = true;
    resetBooks();                              // stale until the next snapshot
    if (!halted) setTimeout(connect, backoffWithJitter());
  };
}
 
function resetBooks() {
  for (const coin in books) delete books[coin];
  ready = false;
}
 
function apply(side, levels, order) {
  for (const l of levels) {
    if (Number(l.sz) === 0 || l.n === 0) side.delete(l.px);
    else side.set(l.px, l);
  }
  // Keep only the top N_LEVELS, so a level that leaves the window can't linger.
  const prices = [...side.keys()].sort((a, b) => order(Number(a), Number(b)));
  for (const px of prices.slice(N_LEVELS)) side.delete(px);
}
 
connect();