observer/crates/blackbeard-data/tests/store.rs

//! Integration tests against a real Postgres.
//!
//! Skipped unless `TEST_DATABASE_URL` is set, so `cargo test` stays useful on a
//! machine with no database. CI and the dev loop set it to a throwaway server:
//!
//! ```sh
//! podman run -d --rm --name bb-pg -e POSTGRES_PASSWORD=dev \
//!   -e POSTGRES_DB=blackbeard -p 55432:5432 docker.io/library/postgres:18-alpine
//! TEST_DATABASE_URL='postgres://postgres:dev@127.0.0.1:55432/blackbeard' cargo test
//! ```
//!
//! Each test gets its own schema rather than its own database — same isolation,
//! no `create database` race when tests run concurrently — and drops it on the
//! way out.
//!
//! These exist because the queries they exercise are the ones a unit test
//! cannot reach: an `unnest` batch upsert, the reorg semantics of the primary
//! key, `date_bin` bucketing, and the height bucketing behind the chain series.
//! Each has a failure mode that is silent rather than loud — wrong rows, not an
//! error.

use std::time::Duration;

use blackbeard_core::attribution::NodeKey;
use blackbeard_data::store::{BlockRecord, ChainRecord, Store};
use blackbeard_entities::{ChainId, MinerId};
use chrono::{DateTime, TimeZone, Utc};
use sqlx::postgres::PgPoolOptions;

/// A store on a private schema, plus the pool so the schema can be dropped.
struct Fixture {
    store: Store,
    pool: sqlx::PgPool,
    schema: String,
}

impl Fixture {
    async fn new(name: &str) -> Option<Self> {
        // An empty value counts as unset. `export A=x B="$A"` in one statement
        // expands `$A` before assigning it, which is an easy way to end up with
        // an empty TEST_DATABASE_URL and a confusing parse error instead of a
        // skip.
        let url = std::env::var("TEST_DATABASE_URL")
            .ok()
            .filter(|u| !u.trim().is_empty())?;
        let schema = format!("test_{name}");
        let pool = PgPoolOptions::new()
            .max_connections(2)
            // Every connection in this pool lands in the test's own schema, so
            // the migrations and the queries under test agree on where the
            // tables are without any query needing to qualify a name.
            .after_connect({
                let schema = schema.clone();
                move |conn, _| {
                    let schema = schema.clone();
                    Box::pin(async move {
                        sqlx::query(&format!("set search_path to {schema}"))
                            .execute(conn)
                            .await?;
                        Ok(())
                    })
                }
            })
            .connect(&url)
            .await
            .expect("TEST_DATABASE_URL is set but unreachable");

        sqlx::query(&format!("drop schema if exists {schema} cascade"))
            .execute(&pool)
            .await
            .unwrap();
        sqlx::query(&format!("create schema {schema}"))
            .execute(&pool)
            .await
            .unwrap();

        let store = Store::from_pool(pool.clone());
        store.migrate().await.expect("migrations apply");
        Some(Self {
            store,
            pool,
            schema,
        })
    }

    async fn cleanup(self) {
        sqlx::query(&format!("drop schema {} cascade", self.schema))
            .execute(&self.pool)
            .await
            .unwrap();
    }
}

fn chain() -> ChainId {
    ChainId("planck".into())
}

fn miner(n: u8) -> MinerId {
    MinerId(format!("0x{n:064x}"))
}

fn at(secs: i64) -> DateTime<Utc> {
    Utc.timestamp_opt(1_757_000_000 + secs, 0).unwrap()
}

fn block(height: u64, m: u8, secs: i64, difficulty: &str) -> BlockRecord {
    BlockRecord {
        chain: chain(),
        height,
        hash: format!("0x{height:064x}"),
        miner: miner(m),
        authored_at: Some(at(secs - 1)),
        observed_at: at(secs),
        difficulty: Some(difficulty.to_owned()),
        at_tip: true,
    }
}

async fn seed_chain(store: &Store) {
    store
        .upsert_chain(&ChainRecord {
            chain: chain(),
            display_name: "Planck Testnet".into(),
            mainnet: false,
            genesis: Some("0x4901".into()),
            token_symbol: Some("PLK".into()),
            token_decimals: Some(12),
            target_block_time: 6.0,
        })
        .await
        .unwrap();
}

macro_rules! fixture {
    ($name:literal) => {
        match Fixture::new($name).await {
            Some(f) => f,
            None => {
                eprintln!("skipping: TEST_DATABASE_URL not set");
                return;
            }
        }
    };
}

#[tokio::test]
async fn blocks_round_trip_through_the_batch_upsert() {
    let f = fixture!("round_trip");
    seed_chain(&f.store).await;

    let blocks: Vec<_> = (1..=50)
        .map(|h| block(h, (h % 3) as u8, h as i64 * 6, "41650875256"))
        .collect();
    assert_eq!(f.store.record_blocks(&blocks).await.unwrap(), 50);
    assert_eq!(f.store.max_height(&chain()).await.unwrap(), Some(50));

    // Oldest first, which is the order the rolling window replays in.
    let recent = f.store.recent_blocks(&chain(), 10).await.unwrap();
    assert_eq!(recent.len(), 10);
    assert_eq!(recent.first().unwrap().height, 41);
    assert_eq!(recent.last().unwrap().height, 50);

    f.cleanup().await;
}

#[tokio::test]
async fn a_reorg_replaces_the_block_at_that_height() {
    // The primary key is (chain, height), not the hash, precisely so this
    // happens. A schema keyed on the hash would accumulate both sides of every
    // fork and inflate the loser's leaderboard row forever.
    let f = fixture!("reorg");
    seed_chain(&f.store).await;

    f.store
        .record_blocks(&[block(100, 1, 600, "1000")])
        .await
        .unwrap();
    let mut replacement = block(100, 2, 601, "1000");
    replacement.hash = "0xdifferent".into();
    f.store.record_blocks(&[replacement]).await.unwrap();

    let rows = f.store.recent_blocks(&chain(), 10).await.unwrap();
    assert_eq!(rows.len(), 1, "one block per height, not one per fork");
    assert_eq!(rows[0].miner, miner(2));

    let (blocks, _, _, _) = f.store.miner_totals(&chain(), &miner(1)).await.unwrap();
    assert_eq!(blocks, 0, "the orphaned author keeps no credit");

    f.cleanup().await;
}

#[tokio::test]
async fn an_empty_batch_is_a_no_op_not_an_error() {
    let f = fixture!("empty_batch");
    seed_chain(&f.store).await;
    assert_eq!(f.store.record_blocks(&[]).await.unwrap(), 0);
    assert_eq!(f.store.max_height(&chain()).await.unwrap(), None);
    f.cleanup().await;
}

#[tokio::test]
async fn cumulative_work_sums_difficulty_beyond_any_integer_type() {
    // Difficulty is a U512. If `difficulty` were bigint — or if the sum were
    // read back through an i64 — this would overflow rather than answer.
    let f = fixture!("work");
    seed_chain(&f.store).await;

    let huge = "1000000000000000000000000000000000000000";
    f.store
        .record_blocks(&[block(1, 1, 6, huge), block(2, 1, 12, huge)])
        .await
        .unwrap();

    let (blocks, first, last, work) = f.store.miner_totals(&chain(), &miner(1)).await.unwrap();
    assert_eq!(blocks, 2);
    assert_eq!(first, Some(at(6)));
    assert_eq!(last, Some(at(12)));
    assert_eq!(work, "2000000000000000000000000000000000000000");

    f.cleanup().await;
}

#[tokio::test]
async fn a_miner_with_no_blocks_totals_zero_rather_than_failing() {
    let f = fixture!("absent_miner");
    seed_chain(&f.store).await;
    let (blocks, first, last, work) = f.store.miner_totals(&chain(), &miner(9)).await.unwrap();
    assert_eq!((blocks, first, last, work.as_str()), (0, None, None, "0"));
    f.cleanup().await;
}

#[tokio::test]
async fn the_series_buckets_on_fixed_boundaries() {
    let f = fixture!("series");
    seed_chain(&f.store).await;

    // Two hours of blocks, one every ten minutes, alternating authors.
    let blocks: Vec<_> = (0..12)
        .map(|i| block(i + 1, (i % 2) as u8, i as i64 * 600, "100"))
        .collect();
    f.store.record_blocks(&blocks).await.unwrap();

    let series = f
        .store
        .miner_series(&chain(), &miner(0), at(0), Duration::from_secs(3600))
        .await
        .unwrap();

    // Anchored to the epoch, so every boundary lands on the hour regardless of
    // when the range starts or when the query is made. A chart bucketed from
    // the range's own start would re-cut itself on every refresh and appear to
    // shift under the reader — which is why this assertion, not a tidier one
    // about the bucket count, is the one worth making. Two hours of blocks
    // starting mid-hour therefore straddle three buckets, not two.
    assert!(series.len() >= 2);
    for (bucket, _, _) in &series {
        assert_eq!(bucket.timestamp() % 3600, 0, "buckets start on the hour");
    }
    assert_eq!(series.iter().map(|(_, mine, _)| mine).sum::<u32>(), 6);
    assert_eq!(series.iter().map(|(_, _, total)| total).sum::<u32>(), 12);
    // Buckets are contiguous and ascending.
    for pair in series.windows(2) {
        assert!(pair[1].0 > pair[0].0);
    }

    f.cleanup().await;
}

#[tokio::test]
async fn attributions_survive_a_restart() {
    let f = fixture!("attribution");
    seed_chain(&f.store).await;

    f.store
        .save_attributions(
            &chain(),
            &[
                (
                    miner(1),
                    NodeKey::Peer("12D3KooWalice".into()),
                    40,
                    38,
                    Some("alice".into()),
                ),
                (miner(2), NodeKey::Node(77), 12, 9, None),
            ],
        )
        .await
        .unwrap();

    let mut loaded = f.store.load_attributions(&chain()).await.unwrap();
    loaded.sort_by(|a, b| a.miner.cmp(&b.miner));
    assert_eq!(loaded.len(), 2);
    assert_eq!(loaded[0].key, NodeKey::Peer("12D3KooWalice".into()));
    assert_eq!(loaded[0].name.as_deref(), Some("alice"));
    assert_eq!(loaded[0].attempts, 40);
    assert_eq!(loaded[1].key, NodeKey::Node(77));

    // A later save with no resolvable name must not erase the one we had — a
    // telemetry hiccup would otherwise cost every miner its name permanently.
    f.store
        .save_attributions(
            &chain(),
            &[(
                miner(1),
                NodeKey::Peer("12D3KooWalice".into()),
                41,
                39,
                None,
            )],
        )
        .await
        .unwrap();
    let reloaded = f.store.load_attributions(&chain()).await.unwrap();
    let alice = reloaded.iter().find(|a| a.miner == miner(1)).unwrap();
    assert_eq!(alice.name.as_deref(), Some("alice"));
    assert_eq!(alice.attempts, 41);

    f.cleanup().await;
}

#[tokio::test]
async fn a_failed_poll_does_not_blank_out_discovered_chain_properties() {
    let f = fixture!("chain_upsert");
    seed_chain(&f.store).await;

    // A poll that reached the node for nothing but the name.
    f.store
        .upsert_chain(&ChainRecord {
            chain: chain(),
            display_name: "Planck Testnet".into(),
            mainnet: false,
            genesis: None,
            token_symbol: None,
            token_decimals: None,
            target_block_time: 6.0,
        })
        .await
        .unwrap();

    let row = sqlx::query_scalar::<_, Option<String>>("select genesis from chain where id = $1")
        .bind(chain().as_str())
        .fetch_one(&f.pool)
        .await
        .unwrap();
    assert_eq!(row.as_deref(), Some("0x4901"));

    f.cleanup().await;
}

#[tokio::test]
async fn the_chain_series_buckets_on_absolute_height() {
    // Bucket boundaries are `height / bucket_blocks` on the absolute height, so
    // they are the same on every request forever. Anchored to the newest block
    // instead they would slide by one on every block, and the chart would shift
    // under a reader watching it.
    let f = fixture!("series_buckets");
    seed_chain(&f.store).await;

    // Heights 100..=139: four buckets of ten, indices 10 through 13.
    let blocks: Vec<_> = (100..140)
        .map(|h| block(h, (h % 4) as u8, h as i64 * 15, "1200000000"))
        .collect();
    f.store.record_blocks(&blocks).await.unwrap();

    let buckets = f.store.chain_series(&chain(), 10, 8).await.unwrap();
    assert_eq!(
        buckets.iter().map(|b| b.index).collect::<Vec<_>>(),
        vec![10, 11, 12, 13]
    );
    assert!(buckets.iter().all(|b| b.blocks == 10));
    assert_eq!(buckets[0].last_height, 109);
    assert_eq!(buckets[3].last_height, 139);

    // Difficulty is a mean, carried as an integer decimal string — never a
    // float, and never with the fractional part `avg` would otherwise give it.
    assert_eq!(buckets[0].difficulty.as_deref(), Some("1200000000"));

    // Authors come back as sets so the caller can union adjacent buckets; four
    // rotating miners over ten blocks is four distinct, not ten.
    let mut miners = buckets[0].miners.clone();
    miners.sort();
    assert_eq!(miners.len(), 4);

    // Timing bounds are the author timestamps and the heights that carry them,
    // which is what makes the interval a height difference rather than a
    // sample count.
    assert_eq!(buckets[0].timed_low, Some(100));
    assert_eq!(buckets[0].timed_high, Some(109));
    assert_eq!(
        (buckets[0].last_authored.unwrap() - buckets[0].first_authored.unwrap()).num_seconds(),
        9 * 15
    );

    f.cleanup().await;
}

#[tokio::test]
async fn a_bucket_range_the_observer_missed_is_absent_rather_than_empty() {
    // Postgres returns no row at all for a height range with no blocks. The
    // series builder puts the point back; what matters here is that the query
    // does not invent one, and that the window is counted back from the tip in
    // buckets rather than in rows.
    let f = fixture!("series_gap");
    seed_chain(&f.store).await;

    let mut blocks: Vec<_> = (0..10)
        .map(|h| block(h, 1, h as i64 * 15, "1000"))
        .collect();
    // Nothing in 10..29 — two whole buckets of outage.
    blocks.extend((30..40).map(|h| block(h, 2, h as i64 * 15, "1000")));
    f.store.record_blocks(&blocks).await.unwrap();

    let buckets = f.store.chain_series(&chain(), 10, 4).await.unwrap();
    assert_eq!(
        buckets.iter().map(|b| b.index).collect::<Vec<_>>(),
        vec![0, 3],
        "the missing buckets are absent, not zero-filled"
    );

    f.cleanup().await;
}

#[tokio::test]
async fn a_block_with_no_difficulty_leaves_the_bucket_mean_to_the_rest() {
    // A gap fill against a pruning node cannot learn the difficulty of an old
    // block and records `None`. That must not poison the bucket: the mean is
    // over the blocks that have one, and only a bucket with none at all is
    // itself unknown.
    let f = fixture!("series_null_difficulty");
    seed_chain(&f.store).await;

    let mut blocks: Vec<_> = (0..10)
        .map(|h| block(h, 1, h as i64 * 15, "1000"))
        .collect();
    for b in blocks.iter_mut().take(5) {
        b.difficulty = None;
    }
    // A whole bucket with nothing known.
    blocks.extend((10..20).map(|h| {
        let mut b = block(h, 1, h as i64 * 15, "1000");
        b.difficulty = None;
        b
    }));
    f.store.record_blocks(&blocks).await.unwrap();

    let buckets = f.store.chain_series(&chain(), 10, 4).await.unwrap();
    assert_eq!(buckets[0].difficulty.as_deref(), Some("1000"));
    assert_eq!(buckets[1].difficulty, None);

    f.cleanup().await;
}

#[tokio::test]
async fn re_recording_a_block_without_a_difficulty_keeps_the_one_already_known() {
    // A reorg replacement fetched from history has no difficulty to offer. The
    // upsert coalesces rather than overwriting, because a hole in the series is
    // a worse answer than the figure recorded when the height was at the tip.
    let f = fixture!("series_difficulty_kept");
    seed_chain(&f.store).await;

    f.store
        .record_blocks(&[block(7, 1, 100, "1234")])
        .await
        .unwrap();
    let mut replacement = block(7, 2, 101, "1234");
    replacement.difficulty = None;
    replacement.at_tip = false;
    f.store.record_blocks(&[replacement]).await.unwrap();

    let buckets = f.store.chain_series(&chain(), 10, 4).await.unwrap();
    assert_eq!(buckets[0].difficulty.as_deref(), Some("1234"));
    assert_eq!(buckets[0].miners, vec![miner(2).0], "the reorg still won");

    f.cleanup().await;
}

/// The displacement table, read back raw — there is no store method for it yet.
/// The read shape belongs with the route that will use it; what these tests
/// hold is that the write happens and captures the right side of the swap.
async fn displacements(pool: &sqlx::PgPool, height: i64) -> Vec<(String, String, Option<String>)> {
    sqlx::query_as::<_, (String, String, Option<String>)>(
        "select hash, miner, difficulty::text from block_displacement
         where chain = $1 and height = $2 order by hash",
    )
    .bind(chain().as_str())
    .bind(height)
    .fetch_all(pool)
    .await
    .unwrap()
}

#[tokio::test]
async fn a_reorg_records_what_it_displaced() {
    // The losing block is overwritten in `block` by design. Without this it
    // would vanish entirely — and the node cannot be asked for it later, since
    // non-canonical bodies are discarded once finality passes them. What is
    // captured here is all there will ever be, so it has to be the *old* row:
    // the author who lost the race, not the one who won it.
    let f = fixture!("displacement");
    seed_chain(&f.store).await;

    f.store
        .record_blocks(&[block(100, 1, 600, "1000")])
        .await
        .unwrap();

    let mut replacement = block(100, 2, 601, "2000");
    replacement.hash = "0xreplacement".into();
    f.store.record_blocks(&[replacement]).await.unwrap();

    // The canonical row is the winner.
    let rows = f.store.recent_blocks(&chain(), 10).await.unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].miner, miner(2));

    // The trace is the loser, with everything we knew about it.
    let gone = displacements(&f.pool, 100).await;
    assert_eq!(gone.len(), 1);
    assert_eq!(gone[0].0, format!("0x{:064x}", 100), "the displaced hash");
    assert_eq!(gone[0].1, miner(1).0, "the author who lost the race");
    assert_eq!(
        gone[0].2.as_deref(),
        Some("1000"),
        "the difficulty as recorded then, not the replacement's"
    );

    f.cleanup().await;
}

#[tokio::test]
async fn re_recording_the_same_block_displaces_nothing() {
    // Every head is written again by the gap fill that spans it, and the
    // startup backfill rewrites whatever it overlaps. If an identical rewrite
    // counted as a displacement, the table would fill with reorgs that never
    // happened and the block page would claim every height had been contested.
    let f = fixture!("displacement_noop");
    seed_chain(&f.store).await;

    for _ in 0..3 {
        f.store
            .record_blocks(&[block(100, 1, 600, "1000")])
            .await
            .unwrap();
    }

    assert!(displacements(&f.pool, 100).await.is_empty());

    f.cleanup().await;
}

#[tokio::test]
async fn a_batch_records_only_the_heights_that_actually_changed() {
    // The capture joins the whole incoming batch against `block`, so a gap fill
    // of a thousand blocks that replaces one of them must record one row.
    let f = fixture!("displacement_batch");
    seed_chain(&f.store).await;

    let first: Vec<_> = (1..=10)
        .map(|h| block(h, 1, h as i64 * 6, "1000"))
        .collect();
    f.store.record_blocks(&first).await.unwrap();

    let mut second = first.clone();
    second[4].hash = "0xforked".into();
    second[4].miner = miner(9);
    assert_eq!(f.store.record_blocks(&second).await.unwrap(), 10);

    assert!(displacements(&f.pool, 4).await.is_empty());
    let gone = displacements(&f.pool, 5).await;
    assert_eq!(gone.len(), 1);
    assert_eq!(gone[0].1, miner(1).0);
    assert!(displacements(&f.pool, 6).await.is_empty());

    f.cleanup().await;
}

#[tokio::test]
async fn a_chain_that_reorgs_back_and_forth_keeps_both_sides() {
    // A displaces nothing, B displaces A, A displaces B, B displaces A again.
    // The last step is a hash that has been displaced before: it must update
    // rather than collide, and the table must not grow a third row for a fork
    // with only two sides.
    let f = fixture!("displacement_flapping");
    seed_chain(&f.store).await;

    let side = |tag: &str, m: u8| {
        let mut b = block(100, m, 600, "1000");
        b.hash = format!("0x{tag}");
        b
    };

    f.store.record_blocks(&[side("aaa", 1)]).await.unwrap();
    f.store.record_blocks(&[side("bbb", 2)]).await.unwrap();
    f.store.record_blocks(&[side("aaa", 1)]).await.unwrap();

    let before_last: DateTime<Utc> = sqlx::query_scalar("select now()")
        .fetch_one(&f.pool)
        .await
        .unwrap();
    f.store.record_blocks(&[side("bbb", 2)]).await.unwrap();

    let gone = displacements(&f.pool, 100).await;
    assert_eq!(
        gone.iter().map(|g| g.0.as_str()).collect::<Vec<_>>(),
        vec!["0xaaa", "0xbbb"],
        "two sides of the fork, one row each"
    );

    // The repeat displacement of 0xaaa moved its timestamp rather than being
    // dropped, so the table answers "when was this last replaced".
    let latest: DateTime<Utc> = sqlx::query_scalar(
        "select displaced_at from block_displacement
         where chain = $1 and height = 100 and hash = '0xaaa'",
    )
    .bind(chain().as_str())
    .fetch_one(&f.pool)
    .await
    .unwrap();
    assert!(
        latest >= before_last,
        "displaced_at {latest} did not advance past {before_last}"
    );

    f.cleanup().await;
}