TON Mempool Stream

GET /v1/stream HTTP/1.1
Host: mempool.tonnode.io
Connection: Upgrade
Upgrade: websocket
Authorization: Bearer tnmp_0123456789ab_000102…1e1f

POST https://mempool.tonnode.io/v1/ticket
Authorization: Bearer tnmp_…

{"v":1,"ticket":"tnmt_5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a","expires_at":1791322908,
 "subprotocols":["tnmp.v1","tnmp.ticket.tnmt_5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a"]}

server → hello
client → resume            (optional, only before the first subscribe)
client → subscribe
server → subscribed
server → msg | gap …        (in seq order)
client → subscribe          (any time: replaces the filters, no replay)
client → ping / server → pong (any time)

{"v":1,"type":"hello","contract":"1.1","epoch":"9f3c1a7be2d04c11","head_seq":123456,"oldest_seq":118020,
 "ts_gw_us":1791322891600000,"key_id":"0123456789ab",
 "limits":{"max_conn":3,"max_filters":null,"delay_ms":0,"replay_limit":null}}

{"v":1,"type":"subscribe",
 "filters":{"dest":["0:2a18ac1734f34579e6a6c2a9e738bb5799d21e7a73ef106895c00c34549a9167"],
            "out_dest":["EQB3ncyBUTjZUA5EnFKR5_EnOMI9V1tTEAAPaiU71gc4TiUt"],
            "op":["0x0f8a7ea5"],
            "pool":["0:a0d1bc3a139cbc6b39a3e3633f74476be441343f9d0fd88b949fca0327b65a75"]},
 "include_unverified":false}

{"v":1,"type":"subscribed","filters":4,"include_unverified":false,"from_seq":123451}

{"v":1,"type":"msg","epoch":"9f3c1a7be2d04c11","seq":123456,
 "ts_node_us":1791322891482113,"ts_gw_us":1791322891503407,
 "boc":"te6cckECBAEAAVUAAeGIAFQxWC5p5orzzU2FU85xdq8zpDz0594g0SuAGGipNSLOB7z7QMUV59oCWxVn53cDBK/Ybxi+GoEjqPHLcC+fXbQCYeqmIcanyqPs/5OAt/hG6HYhZ3NqEZqmTz1Fj1tHUAlNTRi7VitY4AAHOzgADAEBaGIAVjFm774yMmB6BL/Il5hbRtPkPw4x11PgvVI0Bdo6vuYgpuScAAAAAAAAAAAAAAAAAAECAbAPin6lasVrDVlsH/9AL68ICADvO5kConGyoByJOKUjz+JOcYR6rramIAAe1Ep3rA5wnQAKhisFzTzRXnmpsKp5zi7V5nSHnpz7xBolcAMNFSakWcgjw0YBAwCZJZOFYYAKdbrk697uQeRUm/diojQYsc971Seu+rZ8qyEAmbxZCEtZ5WY4BwAKhisFzTzRXnmpsKp5zi7V5nSHnpz7xBolcAMNFSakWdAwKqiS",
 "size":357,
 "cell_hash":"c296499f76b7d670e8b880d697daa407dce698ec4866fa611d32d052bacc0fa3",
 "norm_hash":"c59a7c88a4b4d8e8fa8499960be4d784235c58953fbccf3ca5ff897da1f24e35",
 "sha256":"43da45c22cc6596c43e9203bdcfc89c1155bb1b008af8592d26fc1f5946ef96a",
 "dest":"0:2a18ac1734f34579e6a6c2a9e738bb5799d21e7a73ef106895c00c34549a9167",
 "unverified":false,"path":"checked",
 "parsed":{"wallet":"v4r2","valid_until":1791322908,"seqno":59239,
   "out":[{"dest":"0:ac62cddf7c6464c0f4097f912f30b68da7c87e1c63aea7c17aa4680bb4757dcc","value":"350000000","mode":1,
           "op":"0x0f8a7ea5",
           "jetton":{"amount":"50000000","to":"0:779dcc815138d9500e449c5291e7f12738c23d575b5310000f6a253bd607384e"},
           "dex":{"kind":"stonfi_swap"}}]}}

{"v":1,"type":"gap","epoch":"9f3c1a7be2d04c11","from_seq":130001,"to_seq":130460,"reason":"slow"}

{"v":1,"type":"resume","epoch":"9f3c1a7be2d04c11","last_seq":123450}

{"v":1,"type":"ping","id":"42"}

{"v":1,"type":"pong","id":"42","ts_gw_us":1791322891600000}

{"v":1,"type":"error","code":"too_many_filters","message":"12000 entries > limit 10000","fatal":false}

// npm install ws
import WebSocket from "ws";

const URL = process.env.MEMPOOL_URL ?? "wss://mempool.tonnode.io/v1/stream";
const KEY = process.env.TONNODE_MEMPOOL_KEY; // tnmp_<key_id>_<secret>

let last = null; // { epoch, seq } of the last event handled, for resume
let backoff = 500;
let stopped = false; // the key was refused: reconnecting will not help

function connect() {
  const ws = new WebSocket(URL, { headers: { Authorization: `Bearer ${KEY}` } });

  // The server may refuse the upgrade itself with an HTTP status. terminate()
  // ends the attempt and still emits "close", which retries anything else.
  ws.on("unexpected-response", (req, res) => {
    if (res.statusCode === 401 || res.statusCode === 403) {
      console.error(`refused with HTTP ${res.statusCode}: check the key in the console`);
      stopped = true;
    }
    ws.terminate();
  });

  ws.on("message", (data) => {
    const m = JSON.parse(data.toString());
    switch (m.type) {
      case "hello":
        backoff = 500;
        console.log(`hello epoch=${m.epoch} head_seq=${m.head_seq} limits=${JSON.stringify(m.limits)}`);
        // resume is only valid before the first subscribe on a connection
        if (last) ws.send(JSON.stringify({ v: 1, type: "resume", epoch: last.epoch, last_seq: last.seq }));
        ws.send(JSON.stringify({ v: 1, type: "subscribe", filters: {} }));
        break;
      case "subscribed":
        console.log(`subscribed filters=${m.filters} include_unverified=${m.include_unverified} from_seq=${m.from_seq}`);
        break;
      case "msg": {
        last = { epoch: m.epoch, seq: m.seq };
        const out = m.parsed?.out?.[0];
        console.log(`${m.seq} ${m.path}${m.unverified ? " (unverified)" : ""} dest=${m.dest} hash=${m.cell_hash.slice(0, 16)}… ${out?.op ?? ""} ${out?.dex?.kind ?? ""}`);
        break;
      }
      case "gap":
        last = { epoch: m.epoch, seq: m.to_seq };
        console.warn(`gap ${m.reason} ${m.from_seq}..${m.to_seq}`);
        break;
      case "error":
        console.error(`error ${m.code}: ${m.message} (fatal=${m.fatal})`);
        break;
    }
  });

  ws.on("close", (code, reason) => {
    if (stopped) return;
    // 4401 / 4403: unknown, revoked, suspended or expired key. Retrying will not help.
    if (code === 4401 || code === 4403) {
      console.error(`closed ${code} ${reason}: check the key in the console`);
      return;
    }
    const wait = Math.min(backoff, 30_000) * (0.5 + Math.random());
    backoff = Math.min(backoff * 2, 30_000);
    console.warn(`closed ${code}, reconnecting in ${Math.round(wait)} ms`);
    setTimeout(connect, wait);
  });

  ws.on("error", (err) => console.error(`websocket error: ${err.message}`));
}

connect();

// Your backend (Node 18+, no dependencies). It holds the key and hands the
// browser a one-time ticket; the key itself never reaches the page.
import { createServer } from "node:http";

const TICKET_URL = process.env.MEMPOOL_TICKET_URL ?? "https://mempool.tonnode.io/v1/ticket";
const KEY = process.env.TONNODE_MEMPOOL_KEY;

createServer(async (req, res) => {
  if (req.method === "POST" && req.url === "/mempool-ticket") {
    // Check your own user's session here before handing out a ticket.
    const r = await fetch(TICKET_URL, { method: "POST", headers: { Authorization: `Bearer ${KEY}` } });
    res.writeHead(r.status, { "content-type": "application/json" });
    res.end(await r.text()); // {"v":1,"ticket":"tnmt_…","expires_at":…,"subprotocols":["tnmp.v1","tnmp.ticket.tnmt_…"]}
    return;
  }
  res.writeHead(404).end();
}).listen(Number(process.env.PORT ?? 8080));

const STREAM_URL = "wss://mempool.tonnode.io/v1/stream";

let last = null; // { epoch, seq } of the last event handled, for resume
let backoff = 500;

async function connect() {
  // A fresh ticket for every connection: it is valid for 30 s and one upgrade.
  const r = await fetch("/mempool-ticket", { method: "POST" });
  if (r.status === 401 || r.status === 403) return console.error("key refused", await r.text());
  if (!r.ok) throw new Error(`ticket: HTTP ${r.status}`);
  const { subprotocols } = await r.json();
  const ws = new WebSocket(STREAM_URL, subprotocols);

  ws.onmessage = (ev) => {
    const m = JSON.parse(ev.data);
    if (m.type === "hello") {
      backoff = 500;
      if (last) ws.send(JSON.stringify({ v: 1, type: "resume", epoch: last.epoch, last_seq: last.seq }));
      ws.send(JSON.stringify({ v: 1, type: "subscribe", filters: {}, include_unverified: false }));
    } else if (m.type === "msg") {
      last = { epoch: m.epoch, seq: m.seq };
      console.log(m.seq, m.path, m.dest, m.cell_hash);
    } else if (m.type === "gap") {
      last = { epoch: m.epoch, seq: m.to_seq };
      console.warn("gap", m.reason, m.from_seq, m.to_seq);
    } else if (m.type === "error") {
      console.warn("error", m.code, m.message);
    }
  };
  ws.onclose = (ev) => {
    if (ev.code === 4401 || ev.code === 4403) return console.error("key refused", ev.code, ev.reason);
    reconnect();
  };
}

function reconnect() {
  const wait = Math.min(backoff, 30_000) * (0.5 + Math.random());
  backoff = Math.min(backoff * 2, 30_000);
  setTimeout(() => connect().catch(reconnect), wait);
}

connect().catch(reconnect);

# pip install "websockets>=14"
import asyncio
import json
import os
import random

import websockets

URL = os.environ.get("MEMPOOL_URL", "wss://mempool.tonnode.io/v1/stream")
KEY = os.environ["TONNODE_MEMPOOL_KEY"]  # tnmp_<key_id>_<secret>


async def main():
    last = None  # (epoch, seq) of the last event handled, for resume
    backoff = 0.5
    while True:
        try:
            async with websockets.connect(URL, additional_headers={"Authorization": f"Bearer {KEY}"}) as ws:
                async for raw in ws:
                    m = json.loads(raw)
                    kind = m["type"]
                    if kind == "hello":
                        backoff = 0.5
                        print("hello", m["epoch"], "head_seq", m["head_seq"], "limits", m["limits"])
                        if last:  # resume is only valid before the first subscribe
                            await ws.send(json.dumps({"v": 1, "type": "resume", "epoch": last[0], "last_seq": last[1]}))
                        await ws.send(json.dumps({"v": 1, "type": "subscribe", "filters": {}, "include_unverified": False}))
                    elif kind == "subscribed":
                        print("subscribed", m)
                    elif kind == "msg":
                        last = (m["epoch"], m["seq"])
                        outs = m.get("parsed", {}).get("out", [])
                        print(m["seq"], m["path"], m["dest"], m["cell_hash"][:16], [o.get("op") for o in outs])
                    elif kind == "gap":
                        last = (m["epoch"], m["to_seq"])
                        print("gap", m["reason"], m["from_seq"], m["to_seq"])
                    elif kind == "error":
                        print("error", m["code"], m.get("message"), "fatal" if m.get("fatal") else "")
        except websockets.ConnectionClosedError as e:
            if e.rcvd and e.rcvd.code in (4401, 4403):  # bad, revoked, suspended or expired key
                print("key refused:", e.rcvd.code, e.rcvd.reason)
                return
            print("connection closed:", e)
        except websockets.InvalidStatus as e:  # the upgrade itself was refused
            if e.response.status_code in (401, 403):
                print("key refused: HTTP", e.response.status_code)
                return
            print("handshake failed:", e)
        except OSError as e:
            print("connect failed:", e)
        # also reached after a normal close (1000, or 1001 when the server restarts)
        await asyncio.sleep(min(backoff, 30) * random.uniform(0.5, 1.5))
        backoff = min(backoff * 2, 30)


asyncio.run(main())

// go get github.com/gorilla/websocket
package main

import (
	"errors"
	"log"
	"math/rand"
	"net/http"
	"os"
	"time"

	"github.com/gorilla/websocket"
)

type Out struct {
	Dest  string `json:"dest"`
	Value string `json:"value"`
	Op    string `json:"op"`
}

type Event struct {
	Type       string `json:"type"`
	Epoch      string `json:"epoch"`
	Seq        int64  `json:"seq"`
	HeadSeq    int64  `json:"head_seq"`
	FromSeq    int64  `json:"from_seq"`
	ToSeq      int64  `json:"to_seq"`
	Reason     string `json:"reason"`
	Code       string `json:"code"`
	Message    string `json:"message"`
	Dest       string `json:"dest"`
	CellHash   string `json:"cell_hash"`
	Path       string `json:"path"`
	Unverified bool   `json:"unverified"`
	Boc        string `json:"boc"`
	Parsed     *struct {
		Wallet string `json:"wallet"`
		Out    []Out  `json:"out"`
	} `json:"parsed"`
}

func main() {
	url := os.Getenv("MEMPOOL_URL")
	if url == "" {
		url = "wss://mempool.tonnode.io/v1/stream"
	}
	header := http.Header{"Authorization": {"Bearer " + os.Getenv("TONNODE_MEMPOOL_KEY")}}

	var epoch string // last event handled, for resume
	var seq int64
	backoff := 500 * time.Millisecond
	for {
		conn, resp, err := websocket.DefaultDialer.Dial(url, header)
		if err != nil {
			if resp != nil && (resp.StatusCode == 401 || resp.StatusCode == 403) {
				log.Fatalf("key refused: HTTP %d", resp.StatusCode)
			}
			log.Printf("dial: %v", err)
		} else {
			if fatal := read(conn, &epoch, &seq, &backoff); fatal {
				return
			}
		}
		time.Sleep(time.Duration(float64(backoff) * (0.5 + rand.Float64())))
		backoff = min(backoff*2, 30*time.Second)
	}
}

// read handles one connection until it closes; true means do not reconnect.
func read(conn *websocket.Conn, epoch *string, seq *int64, backoff *time.Duration) bool {
	defer conn.Close()
	for {
		var ev Event
		if err := conn.ReadJSON(&ev); err != nil {
			var ce *websocket.CloseError
			if errors.As(err, &ce) && (ce.Code == 4401 || ce.Code == 4403) {
				log.Printf("key refused: %d %s", ce.Code, ce.Text)
				return true
			}
			log.Printf("read: %v", err)
			return false
		}
		switch ev.Type {
		case "hello":
			*backoff = 500 * time.Millisecond
			log.Printf("hello epoch=%s head_seq=%d", ev.Epoch, ev.HeadSeq)
			if *epoch != "" { // resume is only valid before the first subscribe
				conn.WriteJSON(map[string]any{"v": 1, "type": "resume", "epoch": *epoch, "last_seq": *seq})
			}
			conn.WriteJSON(map[string]any{"v": 1, "type": "subscribe", "filters": map[string]any{}})
		case "msg":
			*epoch, *seq = ev.Epoch, ev.Seq
			op := ""
			if ev.Parsed != nil && len(ev.Parsed.Out) > 0 {
				op = ev.Parsed.Out[0].Op
			}
			log.Printf("%d %s unverified=%v dest=%s hash=%s op=%s", ev.Seq, ev.Path, ev.Unverified, ev.Dest, ev.CellHash[:16], op)
		case "gap":
			*epoch, *seq = ev.Epoch, ev.ToSeq
			log.Printf("gap %s %d..%d", ev.Reason, ev.FromSeq, ev.ToSeq)
		case "error":
			log.Printf("error %s: %s", ev.Code, ev.Message)
		}
	}
}

Build a long-running service in <language> that consumes the TONNode TON mempool stream.

Protocol (authoritative, do not invent anything else):
- WebSocket wss://mempool.tonnode.io/v1/stream. Authenticate with the HTTP header
  "Authorization: Bearer <key>"; read the key from the env var TONNODE_MEMPOOL_KEY and never log it.
- Every message is one JSON object with "v":1 and a "type". The server sends "hello" first
  (fields: epoch, head_seq, oldest_seq, key_id, limits). Nothing is delivered until the client sends
  {"v":1,"type":"subscribe","filters":{}}; the server answers "subscribed".
- Events: "msg" (fields: epoch, seq, ts_node_us, ts_gw_us, boc (base64), size, cell_hash, norm_hash,
  sha256, dest, unverified, path "early"|"checked", optional parsed {wallet, valid_until, seqno,
  out:[{dest, value, mode, op, jetton?, dex?}]}) and "gap" (epoch, from_seq, to_seq, reason).
  Ignore unknown fields and types.
- Remember the epoch and the last seq handled (a msg's seq or a gap's to_seq). After a reconnect,
  send {"v":1,"type":"resume","epoch":E,"last_seq":S} BEFORE the first subscribe.
- Reconnect with exponential backoff: start 0.5 s, cap 30 s, random jitter. Close code 1001 means a
  server restart: reconnect and resume. Close codes 4401 and 4403 mean the key is wrong, revoked,
  suspended or expired: stop and report, do not retry in a loop. 4429 means all connections of the
  key are in use: back off.
- "error" messages have code, message, fatal. Non-fatal errors keep the connection.
- Messages with "unverified": true are written before the node's signature check and may be garbage
  or forgeries. Keep them out of any automated decision unless a "checked" message with the same
  cell_hash arrives or the chain confirms it.
- The same cell_hash can arrive more than once (early then checked, or again after a reconnect):
  de-duplicate by (cell_hash, path) with a TTL of a few minutes.
- Coverage is partial and pending messages may never be included; never treat absence from the
  stream as proof that something did not happen.

Deliverables: the service, structured logs (seq, path, dest, cell_hash, op), a counter of gaps by
reason, graceful shutdown, and a short README explaining how to run it.

Write a <language> program that watches TON wallets in the TONNode mempool stream and prints every
pending outgoing transfer from them.

- Connect to wss://mempool.tonnode.io/v1/stream with "Authorization: Bearer $TONNODE_MEMPOOL_KEY".
- Wait for {"type":"hello"}, then send
  {"v":1,"type":"subscribe","filters":{"dest":[<addresses>]},"include_unverified":false}.
  Addresses may be raw "0:<64 hex>" or 48-character user-friendly strings; the server normalises them.
  "dest" matches the wallet that sends the external message. A subscription may hold at most
  10,000 entries; on {"type":"error","code":"too_many_filters"} or "bad_request" the previous
  subscription stays in force.
- For each {"type":"msg"}: print dest, cell_hash, and for each parsed.out[] item its dest, value
  (nanotons, a decimal string: use a big-integer type) and, if present, jetton.amount and jetton.to.
  parsed may be absent; then print only dest and cell_hash.
- Track epoch and the last seq (msg.seq or gap.to_seq); on reconnect send
  {"v":1,"type":"resume","epoch":...,"last_seq":...} before subscribing again. Backoff 0.5 s to 30 s
  with jitter; stop on close codes 4401/4403.
- A pending message is not a confirmed transaction: label output "pending" and never mark anything
  as paid from the stream alone.

Build a small web page plus backend that shows the live TONNode TON mempool stream in a browser.

- Backend (<framework>): an authenticated route POST /mempool-ticket that calls
  POST https://mempool.tonnode.io/v1/ticket with "Authorization: Bearer $TONNODE_MEMPOOL_KEY" and
  returns the JSON unchanged: {"v":1,"ticket":"tnmt_…","expires_at":<unix s>,"subprotocols":[…]}.
  The key must never reach the browser.
- Page: fetch a fresh ticket for every connection (a ticket is valid 30 s and for one connection),
  then new WebSocket("wss://mempool.tonnode.io/v1/stream", subprotocols). On "hello", send
  {"v":1,"type":"subscribe","filters":{},"include_unverified":false} (or resume first after a
  reconnect, using the last epoch and seq).
- Render a rolling table of the last 200 "msg" events: time from ts_gw_us (microseconds), dest,
  short cell_hash, parsed.wallet, the first parsed.out[] op and dex.kind. Show "gap" events as a
  separator row with the reason.
- Reconnect with backoff (0.5 s to 30 s, jitter); on close code 4401/4403 show
  "key refused" and stop.
- Label the view "pending messages — not confirmed, coverage partial".