Stream finalised blocks

const RPC_URL = process.env.SIGIL_RPC_URL ?? 'https://testnet-rpc.sigil.ml/rpc';
const POLL_INTERVAL_MS = 1_500;

async function rpc<T>(method: string, params: unknown = {}): Promise<T> {
  const res = await fetch(RPC_URL, {
    method: 'POST',
    headers: { 'content-type': 'application/json' },
    body: JSON.stringify({ jsonrpc: '2.0', id: 1, method, params }),
  });
  const body = await res.json() as { result?: T; error?: { code: number; message: string } };
  if (body.error) throw new Error(`${body.error.code}: ${body.error.message}`);
  if (body.result === undefined) throw new Error('missing result');
  return body.result;
}

interface Block {
  height: number;
  hash: string;
  timestamp: string;
  transactions: Array<{ tx_hash: string; tx_type: string }>;
}

async function* finalisedBlocks(startFrom: number): AsyncGenerator<Block> {
  let cursor = startFrom;
  while (true) {
    const head = await rpc<Block>('sigil_getLatestBlock', {});
    const finality = await rpc<{ finalized_height: number }>('tower_getFinalityStatus', {});

    const safeTarget = Math.min(head.height, finality.finalized_height);
    while (cursor <= safeTarget) {
      const block = await rpc<Block>('sigil_getBlock', { height: cursor });
      yield block;
      cursor += 1;
    }
    await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS));
  }
}

async function main() {
  // Resume from durable storage; fall back to head minus 100 on first run.
  const resumeFrom = Number(process.env.RESUME_HEIGHT ?? '0');
  let cursor = resumeFrom;
  if (cursor === 0) {
    const head = await rpc<{ height: number }>('sigil_getLatestBlock', {});
    cursor = Math.max(0, head.height - 100);
  }

  for await (const block of finalisedBlocks(cursor)) {
    // Process the block exactly once.
    console.log(`#${block.height} ${block.hash} (${block.transactions.length} txs)`);
    // PERSIST cursor = block.height + 1 here, transactionally with downstream effects.
  }
}

main().catch((err) => { console.error(err); process.exit(1); });

use anyhow::Context;
use serde::Deserialize;
use serde_json::json;
use std::time::Duration;
use tokio::time::sleep;

#[derive(Deserialize)]
struct Block {
    height: u64,
    hash: String,
    transactions: Vec<serde_json::Value>,
}

#[derive(Deserialize)]
struct Finality {
    finalized_height: u64,
}

async fn rpc<T: for<'de> Deserialize<'de>>(
    rpc_url: &str,
    method: &str,
    params: serde_json::Value,
) -> anyhow::Result<T> {
    let body = json!({ "jsonrpc": "2.0", "id": 1, "method": method, "params": params });
    let res: serde_json::Value = reqwest::Client::new()
        .post(rpc_url).json(&body).send().await?.json().await?;
    if let Some(err) = res.get("error") {
        anyhow::bail!("rpc {method}: {err}");
    }
    Ok(serde_json::from_value(res["result"].clone())?)
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let rpc_url = std::env::var("SIGIL_RPC_URL")
        .unwrap_or_else(|_| "https://testnet-rpc.sigil.ml/rpc".into());

    let mut cursor: u64 = std::env::var("RESUME_HEIGHT").ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or_else(|| 0);
    if cursor == 0 {
        let head: Block = rpc(&rpc_url, "sigil_getLatestBlock", json!({})).await?;
        cursor = head.height.saturating_sub(100);
    }

    loop {
        let head: Block = rpc(&rpc_url, "sigil_getLatestBlock", json!({})).await?;
        let finality: Finality = rpc(&rpc_url, "tower_getFinalityStatus", json!({})).await
            .context("tower finality")?;
        let safe_target = head.height.min(finality.finalized_height);
        while cursor <= safe_target {
            let block: Block = rpc(&rpc_url, "sigil_getBlock", json!({ "height": cursor })).await?;
            println!("#{} {} ({} txs)", block.height, block.hash, block.transactions.len());
            // Persist cursor = block.height + 1 atomically with downstream effects.
            cursor = block.height.saturating_add(1);
        }
        sleep(Duration::from_millis(1_500)).await;
    }
}