Sync history to a database

// Node 22.13+ (built-in node:sqlite): save as sync.mjs and run `node sync.mjs`
import { DatabaseSync } from 'node:sqlite';

const { ACTIVITY_URL, WALLET, LUMEUS_API_KEY } = process.env;
const LIMIT = 100; // lower it if a page fails with query_timeout
const RETRYABLE = ['rate_limited', 'busy', 'index_not_ready', 'unavailable'];
const sleep = (s) => new Promise((resolve) => setTimeout(resolve, s * 1000));

// 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: { Authorization: `Bearer ${LUMEUS_API_KEY}` },
    });
    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: rerun with a lower LIMIT instead of repeating the URL.
      throw new Error(`${response.status} ${page.code}: ${page.error} (${page.request_id})`);
    }
    await sleep(wait);
  }
}

const db = new DatabaseSync('activity.db');
db.exec(`
  CREATE TABLE IF NOT EXISTS activity_events (
    network TEXT NOT NULL, id TEXT NOT NULL, account TEXT NOT NULL,
    ledger BIGINT NOT NULL, type TEXT NOT NULL, body TEXT NOT NULL,
    PRIMARY KEY (network, id));
  CREATE TABLE IF NOT EXISTS activity_sync (
    account TEXT PRIMARY KEY, next_href TEXT, scanned_through_ledger BIGINT);`);

const insert = db.prepare(`INSERT INTO activity_events VALUES (?, ?, ?, ?, ?, ?)
  ON CONFLICT DO NOTHING`);
const save = db.prepare(`INSERT INTO activity_sync VALUES (?, ?, ?)
  ON CONFLICT (account) DO UPDATE SET next_href = excluded.next_href,
    scanned_through_ledger = excluded.scanned_through_ledger`);
const state = db.prepare('SELECT * FROM activity_sync WHERE account = ?').get(WALLET);

let href = state?.next_href
  ?? `/accounts/${WALLET}/activity?order=asc&limit=${LIMIT}`
    + (state?.scanned_through_ledger ? `&since_ledger=${state.scanned_through_ledger + 1}` : '');
let scanned = state?.scanned_through_ledger ?? null;

while (href) {
  const page = await get(href);
  const next = page.range_complete ? null : page._links.next.href;
  if (page.range_complete) scanned = page.scanned_through_ledger;

  db.exec('BEGIN');
  for (const r of page._embedded.records) {
    insert.run(r.network, r.id, WALLET, r.ledger, r.type, JSON.stringify(r));
  }
  save.run(WALLET, next, scanned);
  db.exec('COMMIT');

  console.log(`stored ${page._embedded.records.length} records, complete=${page.range_complete}`);
  href = next;
}
const { n } = db.prepare('SELECT count(*) AS n FROM activity_events WHERE account = ?').get(WALLET);
console.log(`${n} records in activity.db, synced through ledger ${scanned}`);

# pip install requests; save as sync.py and run `python sync.py`
import json
import os
import sqlite3
import time

import requests

base, wallet = os.environ["ACTIVITY_URL"], os.environ["WALLET"]
headers = {"Authorization": f"Bearer {os.environ['LUMEUS_API_KEY']}"}
LIMIT = 100  # lower it if a page fails with query_timeout
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: rerun with a 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)


db = sqlite3.connect("activity.db")
db.executescript("""
  CREATE TABLE IF NOT EXISTS activity_events (
    network TEXT NOT NULL, id TEXT NOT NULL, account TEXT NOT NULL,
    ledger BIGINT NOT NULL, type TEXT NOT NULL, body TEXT NOT NULL,
    PRIMARY KEY (network, id));
  CREATE TABLE IF NOT EXISTS activity_sync (
    account TEXT PRIMARY KEY, next_href TEXT, scanned_through_ledger BIGINT);
""")
state = db.execute(
    "SELECT next_href, scanned_through_ledger FROM activity_sync WHERE account = ?", (wallet,)
).fetchone()
next_href, scanned = state if state else (None, None)

href = next_href or f"/accounts/{wallet}/activity?order=asc&limit={LIMIT}" + (
    f"&since_ledger={scanned + 1}" if scanned else ""
)

while href:
    page = get(href)
    href = None if page["range_complete"] else page["_links"]["next"]["href"]
    if page["range_complete"]:
        scanned = page["scanned_through_ledger"]

    with db:  # one transaction: records and progress together
        db.executemany(
            "INSERT INTO activity_events VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING",
            [(r["network"], r["id"], wallet, r["ledger"], r["type"], json.dumps(r))
             for r in page["_embedded"]["records"]],
        )
        db.execute(
            "INSERT INTO activity_sync VALUES (?, ?, ?) ON CONFLICT (account) DO UPDATE SET "
            "next_href = excluded.next_href, scanned_through_ledger = excluded.scanned_through_ledger",
            (wallet, href, scanned),
        )
    print(f"stored {len(page['_embedded']['records'])} records, complete={page['range_complete']}")

(n,) = db.execute("SELECT count(*) FROM activity_events WHERE account = ?", (wallet,)).fetchone()
print(f"{n} records in activity.db, synced through ledger {scanned}")