// You provide: loadLastBlock, saveLastBlock, upsertFill, recordGap, backoffWithJitter, resetBackoff, onFatal.
let lastBlock = await loadLastBlock();   // last fully processed block, or null on first run
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 lastSeq = 0;                       // seq restarts at 1 on every new subscription
  let stopped = false;                   // set on any failure or close
  let queue = Promise.resolve();         // process messages strictly one at a time
 
  function stop(err) {
    if (stopped) return;
    stopped = true;
    if (err.fatal) {                     // e.g. invalid_params or limit_reached: retrying won't help
      halted = true;
      onFatal(err.error);
    }
    ws.close();                          // otherwise reconnect and replay from lastBlock + 1
  }
 
  ws.onmessage = ({ data }) => {
    queue = queue.then(() => handle(JSON.parse(data))).catch(stop);
  };
 
  async function handle(msg) {
    if (stopped) return;
    if (msg.channel === "connected") {
      ws.send(JSON.stringify({
        method: "subscribe",
        subscriptionId: "wallets",
        subscription: {
          type: "userFills",
          users: ["0x654086857e1fad6dcf05cf6695cce51ea3984268"],
          // First run: start from block 0. The server answers with a gap up to its oldest
          // replayable block, which becomes the starting checkpoint, then replays from there.
          fromBlock: lastBlock == null ? 0 : Number(lastBlock) + 1,
        },
      }));
      return;
    }
 
    // Errors come as a subscription error (with our ID) or a connection-level error (no ID).
    if (msg.error && (msg.channel === "error" || msg.subscriptionId === "wallets")) {
      if (msg.error.code === "fills_not_ready" || msg.error.code === "going_away") throw new Error(msg.error.code);  // temporary: reconnect with backoff
      throw Object.assign(new Error(msg.error.code), { fatal: true, error: msg.error });
    }
    if (msg.subscriptionId !== "wallets") return;  // other subscriptions
    if (msg.seq === undefined) return resetBackoff();  // acknowledgement: the connection is healthy
 
    if (msg.seq !== lastSeq + 1) throw new Error("missed message");
    lastSeq = msg.seq;
 
    if (msg.gap) {
      if (msg.gap.fromBlock > 0) await recordGap(msg.gap);  // range no longer replayable (0 = first-run start)
      if (stopped) return;               // connection replaced mid-message: let the new one checkpoint
      await saveLastBlock(msg.gap.toBlock);
      lastBlock = msg.gap.toBlock;
      return;
    }
 
    for (const [user, fill] of msg.data.fills) {
      await upsertFill(`${fill.tid}|${user}`, user, fill);  // idempotent, keyed on tid + user
    }
    if (stopped) return;
    await saveLastBlock(msg.data.blockNumber);  // only after the block's fills are stored
    lastBlock = msg.data.blockNumber;
  }
 
  ws.onclose = () => {
    stopped = true;
    if (!halted) setTimeout(connect, backoffWithJitter());
  };
}
connect();