WebSocket

import WebSocket from "ws";
import { randomBytes } from "node:crypto";

// One id per running client, the same on every reconnect: it is how the server tells
// "this client coming back after a network drop" from "a second client".
const clientId = randomBytes(16).toString("base64url");
const takeover = false;   // true = replace whichever API client holds this account's stream

function connect(first) {
  const ws = new WebSocket(`wss://api.pmwallets.com/v1/ws?clientId=${clientId}${first && takeover ? "&takeover=1" : ""}`, {
    headers: {
      "x-api-key": process.env.PMW_KEY,     // a handshake cannot set Authorization
    },
  });
  ws.on("close", (code) => {
    if (code === 4411) return stop("another API client holds this account's stream");   // refused: do not retry
    if (code === 1000) return stop("replaced by another client's takeover");            // do not fight back
    setTimeout(() => connect(false), code === 1012 ? 0 : backoff());   // 1012 = we are restarting: come right back
  });
  return ws;
}
const ws = connect(true);

// Persist these across reconnects — comparing against a session you just overwrote
// can never detect anything.
let session = load("session"), seq = load("seq") ?? 0;
let lastBlock = load("block") ?? 0, lastLogIndex = load("logIndex") ?? 0;

// One message at a time: a replay is awaited, and a second frame arriving meanwhile
// would advance the cursor past the gap being repaired. Note the catch — chaining
// onto a rejected promise skips every later .then(), so a single failed replay would
// leave the socket open and silently processing nothing at all.
let queue = Promise.resolve();
ws.onmessage = (e) => {
  queue = queue
    .then(() => handleFrame(JSON.parse(e.data)))
    .catch((err) => { report(err); ws.close(); });   // reconnect and replay rather than go quiet
};

async function handleFrame(m) {
  if (m.type === "hello") {
    // A NEW session means frames were missed while the socket was down. Replay BEFORE
    // adopting it — overwriting session here is what silently swallows the outage.
    // a failing replay must not be swallowed: leave the cursor where it is and throw, so the
    // handler above closes the socket and the next connection tries again from the same place
    if (session !== null && m.session !== session) await replayFrom(lastBlock, lastLogIndex);
    session = m.session; seq = m.seq; save();
    return;
  }
  if (m.type !== "fill") return;

  if (m.session !== session || m.seq !== seq + 1) {   // dropped frame, or a session we never saw
    await replayFrom(lastBlock, lastLogIndex);
    session = m.session;
  }
  seq = m.seq;

  if (!seen.has(m.data.eventId)) { seen.add(m.data.eventId); handle(m.data); }
  lastBlock = m.data.block; lastLogIndex = m.data.logIndex;
  save();
}

// walk forward until the page is short; every row carries the same eventId as the stream
async function replayFrom(block, logIndex) {
  for (;;) {
    const r = await fetch(
      `https://api.pmwallets.com/v1/account/fills?sinceBlock=${block}&sinceLogIndex=${logIndex}&limit=500`,
      { headers: { "x-api-key": PMW_KEY } },
    );
    if (!r.ok) throw new Error(`replay failed: HTTP ${r.status}`);   // do NOT advance the cursor
    const page = await r.json();
    // fills are kept for 60 days: an older gap is gone, not empty
    if (page.truncated) console.warn(`fills before block ${page.retainedFrom} are no longer kept`);

    for (const row of page.rows) if (!seen.has(row.eventId)) { seen.add(row.eventId); handle(row); }
    if (!page.next) return;
    ({ sinceBlock: block, sinceLogIndex: logIndex } = page.next);
  }
}