Follow a contract's events

// Node 18+: save as follow.mjs and run `node follow.mjs` (Ctrl-C to stop)
const { ACTIVITY_URL, POOL, LUMEUS_API_KEY } = process.env;
const headers = { Authorization: `Bearer ${LUMEUS_API_KEY}` };
const sleep = (s) => new Promise((resolve) => setTimeout(resolve, s * 1000));
const RETRYABLE = ['rate_limited', 'busy', 'index_not_ready', 'unavailable'];

// GET one page. Retries only the codes that are safe to repeat, at most 5 times.
async function get(href) {
  for (let attempt = 1; ; attempt++) {
    const response = await fetch(new URL(href, ACTIVITY_URL), { headers });
    const page = await response.json().catch(() => ({}));
    if (response.ok) return page;
    const wait = Number(response.headers.get('retry-after') ?? 2);
    if (!RETRYABLE.includes(page.code) || attempt === 5 || wait > 60) {
      // query_timeout lands here: lower limit instead of repeating the URL.
      throw new Error(`${response.status} ${page.code}: ${page.error} (${page.request_id})`);
    }
    await sleep(wait);
  }
}

// Start at the tip. Use a saved ledger instead to resume after a restart.
let scanned = (await get(`/contracts/${POOL}/activity?since_ledger=2147483647`)).scanned_through_ledger;
console.log(`following ${POOL} from ledger ${scanned + 1}`);

for (;;) {
  let href = `/contracts/${POOL}/activity?order=asc&limit=200&since_ledger=${scanned + 1}`;
  while (href) {
    const page = await get(href);
    for (const e of page._embedded.records) {
      if (e.kind === 'contract_event') console.log(e.ledger, e.type, JSON.stringify(e.topics.slice(1)));
    }
    if (page.range_complete) scanned = page.scanned_through_ledger;
    href = page.range_complete ? null : page._links.next.href;
  }
  await sleep(30);
}

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

import requests

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


RETRYABLE = {"rate_limited", "busy", "index_not_ready", "unavailable"}


def get(href):
    """GET one page. Retries only the codes that are safe to repeat, at most 5 times."""
    for attempt in range(1, 6):
        response = requests.get(base + href, headers=headers, timeout=60)
        try:
            page = response.json()
        except ValueError:
            page = {}
        if response.ok:
            return page
        wait = int(response.headers.get("Retry-After", "2"))
        if page.get("code") not in RETRYABLE or attempt == 5 or wait > 60:
            # query_timeout lands here: lower limit instead of repeating the URL.
            raise SystemExit(f"{response.status_code} {page.get('code')}: {page.get('error')} ({page.get('request_id')})")
        time.sleep(wait)


# Start at the tip. Use a saved ledger instead to resume after a restart.
scanned = get(f"/contracts/{pool}/activity?since_ledger=2147483647")["scanned_through_ledger"]
print(f"following {pool} from ledger {scanned + 1}", flush=True)

while True:
    href = f"/contracts/{pool}/activity?order=asc&limit=200&since_ledger={scanned + 1}"
    while href:
        page = get(href)
        for e in page["_embedded"]["records"]:
            if e["kind"] == "contract_event":
                print(e["ledger"], e["type"], json.dumps(e["topics"][1:]), flush=True)
        if page["range_complete"]:
            scanned = page["scanned_through_ledger"]
        href = None if page["range_complete"] else page["_links"]["next"]["href"]
    time.sleep(30)