Watch a Solana wallet over WebSocket

wscat -c "wss://triport.io/ws/sol" -H "x-token: $TRIPORT_API_KEY"
> {"jsonrpc":"2.0","id":1,"method":"accountSubscribe","params":["5tzFkiKscXHK5ZXCGbXZxdw7gTjjD1mBwuoFbhUvuAi9",{"encoding":"base64","commitment":"confirmed"}]}

{"jsonrpc":"2.0","result":1,"id":1}
{"jsonrpc":"2.0","method":"accountNotification","params":{"result":{"context":{"slot":449504578},"value":{"lamports":1450455511477811,"data":["","base64"],"owner":"11111111111111111111111111111111","executable":false,"rentEpoch":18446744073709551615,"space":0}},"subscription":1}}

import WebSocket from "ws";

const KEY = process.env.TRIPORT_API_KEY!;
const WALLET = process.argv[2] ?? "5tzFkiKscXHK5ZXCGbXZxdw7gTjjD1mBwuoFbhUvuAi9";
const RPC = "https://triport.io/sol";
const PAGE = 100;

interface SigInfo { signature: string; slot: number; err: unknown; blockTime: number | null }

let started = false;             // false until the first catch-up has run
let lastSeen: string | undefined; // newest processed signature; persist it in production

async function rpc<T>(method: string, params: unknown[]): Promise<T> {
  for (;;) {
    const res = await fetch(RPC, {
      method: "POST",
      headers: { "content-type": "application/json", "x-token": KEY },
      body: JSON.stringify({ jsonrpc: "2.0", id: 1, method, params }),
    });
    if (res.status === 429) {
      // Plan limit reached: wait as long as the server says, then retry.
      const wait = Number(res.headers.get("retry-after") ?? "1");
      await new Promise((r) => setTimeout(r, wait * 1000));
      continue;
    }
    const body = await res.json();
    if (!res.ok) throw new Error(`${res.status} ${body.error}: ${body.message}`);
    if (body.error) throw new Error(`RPC ${body.error.code}: ${body.error.message}`);
    return body.result as T;
  }
}

async function catchUp(): Promise<void> {
  if (!started) {
    // First run: start from "now" instead of replaying the wallet's history.
    const [newest] = await rpc<SigInfo[]>("getSignaturesForAddress", [WALLET, { limit: 1 }]);
    lastSeen = newest?.signature;
    started = true;
    return;
  }
  // Walk back from the newest signature until the cursor, a page at a time.
  const fresh: SigInfo[] = [];
  let before: string | undefined;
  for (;;) {
    const page = await rpc<SigInfo[]>("getSignaturesForAddress", [
      WALLET,
      { limit: PAGE, until: lastSeen, before, commitment: "confirmed" },
    ]);
    fresh.push(...page);
    if (page.length < PAGE) break;
    before = page[page.length - 1].signature;
  }
  // Process oldest first, and advance the cursor only after each one is handled.
  for (const s of fresh.reverse()) {
    const tx = await rpc<any>("getTransaction", [
      s.signature,
      { encoding: "jsonParsed", maxSupportedTransactionVersion: 1, commitment: "confirmed" },
    ]);
    if (tx === null) return; // not readable yet: stop here, the next trigger retries
    const i = tx.transaction.message.accountKeys.findIndex((k: any) => k.pubkey === WALLET);
    const delta = (tx.meta.postBalances[i] - tx.meta.preBalances[i]) / 1e9;
    console.log(`${s.signature} slot=${s.slot} ok=${s.err === null} SOL change=${delta}`);
    lastSeen = s.signature;
  }
}

// Run catch-ups one at a time; a trigger that arrives mid-run schedules one more.
let running = false;
let again = false;
async function trigger(): Promise<void> {
  if (running) { again = true; return; }
  running = true;
  try {
    do { again = false; await catchUp(); } while (again);
  } catch (e) {
    console.error("catch-up failed:", (e as Error).message);
  } finally {
    running = false;
  }
}

function connect(): void {
  const ws = new WebSocket("wss://triport.io/ws/sol", { headers: { "x-token": KEY } });
  let ping: NodeJS.Timeout | undefined;

  ws.on("open", () => {
    ws.send(JSON.stringify({
      jsonrpc: "2.0", id: 1, method: "accountSubscribe",
      params: [WALLET, { encoding: "base64", commitment: "confirmed" }],
    }));
    ping = setInterval(() => ws.ping(), 30_000);
    void trigger(); // close the gap left by a previous connection
  });

  ws.on("message", (data) => {
    const msg = JSON.parse(data.toString());
    if (msg.id === 1 && typeof msg.result === "number") {
      console.log("subscribed, id", msg.result);
    } else if (msg.method === "accountNotification") {
      const { context, value } = msg.params.result;
      console.log(`account changed at slot ${context.slot}, lamports ${value.lamports}`);
      void trigger();
    } else if (msg.error) {
      console.error("server:", JSON.stringify(msg)); // error frame before a close
    }
  });

  // A bad key is refused before the upgrade, with an HTTP status and a JSON body.
  ws.on("unexpected-response", (_req, res) => {
    let body = "";
    res.on("data", (c) => (body += c));
    res.on("end", () => console.error(`refused: HTTP ${res.statusCode} ${body}`));
  });

  ws.on("close", (code, reason) => {
    clearInterval(ping);
    console.log(`closed ${code} ${reason.toString()}`);
    if ([4001, 4003, 4030].includes(code)) return; // key, plan or trial problem: retrying will not help
    setTimeout(connect, 2_000);
  });
  ws.on("error", (e) => console.error("socket:", e.message));
}

connect();

subscribed, id 1
account changed at slot 449506827, lamports 8814763864917
account changed at slot 449506828, lamports 8814768910946
ZooBgrSopWEu5hbzc5Vgsju5CVtxoj4WzjBVv52DtG7bTFvq2kT4gUHJ1FCjASgnfze1P11UTXkBf2VcJWwe7Ph slot=449506796 ok=true SOL change=0.000045939
4XVTGtvHzkZRhaE2DjbvFbBqnyhFcYhPWDd3DzinBbNd2g8rgbhndhMCfPHUQ4P4ckw96J4LjhF7nsd7n4xwD1UC slot=449506796 ok=true SOL change=0.002934474

{"code":-32015,"message":"Transaction version (1) is not supported by the requesting client. Please try the request again with the following configuration parameter: \"maxSupportedTransactionVersion\": 1"}

import WebSocket from "ws";


export function waitFinalized(signature: string, timeoutMs = 120_000): Promise<unknown> {
  return new Promise((resolve, reject) => {
    const ws = new WebSocket("wss://triport.io/ws/sol", {
      headers: { "x-token": process.env.TRIPORT_API_KEY! },
    });
    const timer = setTimeout(() => { ws.close(); reject(new Error("no notification")); }, timeoutMs);
    ws.on("open", () => ws.send(JSON.stringify({
      jsonrpc: "2.0", id: 1, method: "signatureSubscribe",
      params: [signature, { commitment: "finalized" }],
    })));
    ws.on("message", (data) => {
      const msg = JSON.parse(data.toString());
      if (msg.method === "signatureNotification") {
        clearTimeout(timer);
        ws.close();
        resolve(msg.params.result.value.err); // null means it succeeded
      }
    });
    ws.on("error", (e) => { clearTimeout(timer); reject(e); });
  });
}

{"jsonrpc":"2.0","result":2,"id":2}
{"jsonrpc":"2.0","method":"signatureNotification","params":{"result":{"context":{"slot":449504545},"value":{"err":null}},"subscription":2}}

{"error":"unauthorized","message":"Missing or unknown API key. Pass your key as ?api-key=<key> or the header Authorization: Bearer <key>. Create or copy a key at https://triport.io/app/keys","keys_url":"https://triport.io/app/keys"}

{
  "error": "rate_limited",
  "message": "Rate limit reached: your free plan allows 20 requests/s for sol_read_rpc. Retry in 1s (see the Retry-After header) or upgrade for a higher limit at https://triport.io/app/pricing",
  "current_tier": "free",
  "category": "sol_read_rpc",
  "limit_rps": 20,
  "burst_capacity": 40,
  "retry_after_sec": 1,
  "upgrade_url": "https://triport.io/app/pricing",
  "docs_url": "https://triport.io/errors/http-429-rate-limit"
}