Page through RPC events

rpc() {
  curl --fail-with-body -sS "$RPC_URL" \
    -H "Authorization: Bearer $LUMEUS_API_KEY" \
    -H 'Content-Type: application/json' \
    --data "$1"
}

# Start 100 ledgers (about 8 minutes) back from the newest ledger.
LATEST=$(rpc '{"jsonrpc":"2.0","id":1,"method":"getHealth"}' | jq .result.latestLedger)

PAGE=$(rpc "$(jq -nc --argjson start "$((LATEST - 100))" --arg pool "$POOL" '{
  jsonrpc: "2.0", id: 2, method: "getEvents",
  params: {startLedger: $start, xdrFormat: "json",
           filters: [{type: "contract", contractIds: [$pool]}], pagination: {limit: 5}}}')")
echo "$PAGE" | jq -c '.result.events[] | [.ledger, .topicJson[0].symbol, .id]'

# Next page: the same filters, the cursor instead of startLedger.
CURSOR=$(echo "$PAGE" | jq -r .result.cursor)
rpc "$(jq -nc --arg cursor "$CURSOR" --arg pool "$POOL" '{
  jsonrpc: "2.0", id: 3, method: "getEvents",
  params: {xdrFormat: "json", filters: [{type: "contract", contractIds: [$pool]}],
           pagination: {cursor: $cursor, limit: 5}}}')" \
  | jq -c '.result.events[] | [.ledger, .topicJson[0].symbol, .id]'

// Node 18+: save as rpc-events.mjs and run `node rpc-events.mjs` (Ctrl-C to stop)
const { RPC_URL, POOL, LUMEUS_API_KEY } = process.env;
const sleep = (s) => new Promise((resolve) => setTimeout(resolve, s * 1000));

// One JSON-RPC call. Retries 429 and 503 after Retry-After, at most 5 times.
async function rpc(method, params) {
  for (let attempt = 1; ; attempt++) {
    const response = await fetch(RPC_URL, {
      method: 'POST',
      headers: { Authorization: `Bearer ${LUMEUS_API_KEY}`, 'Content-Type': 'application/json' },
      body: JSON.stringify({ jsonrpc: '2.0', id: 1, method, params }),
    });
    const body = await response.json().catch(() => ({}));
    if (response.ok && !body.error) return body.result;
    const wait = Number(response.headers.get('retry-after') ?? 2);
    const retryable = response.status === 429 || response.status === 503;
    // A wait over 60 s is the daily Testnet cap (resets at 00:00 UTC): stop instead.
    if (!retryable || attempt === 5 || wait > 60) {
      throw new Error(`${response.status} ${body.error?.data}: ${body.error?.message} (${body.error?.request_id})`);
    }
    await sleep(wait);
  }
}

const filters = [{ type: 'contract', contractIds: [POOL] }];
const { latestLedger } = await rpc('getHealth');
let params = { startLedger: latestLedger - 100, filters, xdrFormat: 'json', pagination: { limit: 100 } };

for (;;) {
  const page = await rpc('getEvents', params);
  for (const e of page.events) console.log(e.ledger, e.topicJson[0].symbol, e.id);
  params = { filters, xdrFormat: 'json', pagination: { cursor: page.cursor, limit: 100 } };
  // A short page means you have caught up: wait for new ledgers (about 5 s each).
  if (page.events.length < 100) await sleep(30);
}

# pip install requests; save as rpc_events.py and run `python rpc_events.py` (Ctrl-C to stop)
import os
import time

import requests

rpc_url, pool = os.environ["RPC_URL"], os.environ["POOL"]
headers = {"Authorization": f"Bearer {os.environ['LUMEUS_API_KEY']}"}


def rpc(method, params=None):
    """One JSON-RPC call. Retries 429 and 503 after Retry-After, at most 5 times."""
    body = {"jsonrpc": "2.0", "id": 1, "method": method}
    if params is not None:
        body["params"] = params
    for attempt in range(1, 6):
        response = requests.post(rpc_url, json=body, headers=headers, timeout=60)
        try:
            reply = response.json()
        except ValueError:
            reply = {}
        if response.ok and "error" not in reply:
            return reply["result"]
        wait = int(response.headers.get("Retry-After", "2"))
        # A wait over 60 s is the daily Testnet cap (resets at 00:00 UTC): stop instead.
        if response.status_code not in (429, 503) or attempt == 5 or wait > 60:
            error = reply.get("error", {})
            raise SystemExit(f"{response.status_code} {error.get('data')}: {error.get('message')} ({error.get('request_id')})")
        time.sleep(wait)


filters = [{"type": "contract", "contractIds": [pool]}]
latest = rpc("getHealth")["latestLedger"]
params = {"startLedger": latest - 100, "filters": filters, "xdrFormat": "json", "pagination": {"limit": 100}}

while True:
    page = rpc("getEvents", params)
    for e in page["events"]:
        print(e["ledger"], e["topicJson"][0]["symbol"], e["id"], flush=True)
    params = {"filters": filters, "xdrFormat": "json", "pagination": {"cursor": page["cursor"], "limit": 100}}
    # A short page means you have caught up: wait for new ledgers (about 5 s each).
    if len(page["events"]) < 100:
        time.sleep(30)