feat(minibf): implement /epochs/{number}/stakes endpoint

crates/minibf/src/lib.rs

@@ -409,6 +409,10 @@ where
             "/epochs/{epoch}/parameters",
             get(routes::epochs::by_number_parameters::<D>),
         )
+        .route(
+            "/epochs/{epoch}/stakes",
+            get(routes::epochs::by_number_stakes::<D>),
+        )
         .route(
             "/epochs/latest/parameters",
             get(routes::epochs::latest_parameters::<D>),

crates/minibf/src/routes/epochs/mod.rs

@@ -3,14 +3,24 @@ use axum::{
     http::StatusCode,
     Json,
 };
-use blockfrost_openapi::models::epoch_param_content::EpochParamContent;
-use pallas::ledger::{primitives::Epoch, traverse::MultiEraBlock};
+use blockfrost_openapi::models::{
+    epoch_param_content::EpochParamContent, epoch_stake_content_inner::EpochStakeContentInner,
+};
+use pallas::{
+    codec::minicbor,
+    ledger::{
+        primitives::{Epoch, StakeCredential},
+        traverse::MultiEraBlock,
+    },
+};
 
-use dolos_core::{archive::Skippable as _, ArchiveStore, Domain};
+use dolos_cardano::model::{AccountStakeLog, FixedNamespace as _};
+use dolos_core::{archive::Skippable as _, ArchiveStore, Domain, EntityKey, LogKey, TemporalKey};
 
 use crate::{
     error::Error,
-    mapping::IntoModel as _,
+    log_and_500,
+    mapping::{bech32_pool, stake_cred_to_address, IntoModel as _},
     pagination::{Order, Pagination, PaginationParameters},
     Facade,
 };
@@ -106,6 +116,78 @@ pub async fn by_number_blocks<D: Domain>(
     }))
 }
 
+pub async fn by_number_stakes<D: Domain>(
+    Path(epoch): Path<u64>,
+    Query(params): Query<PaginationParameters>,
+    State(domain): State<Facade<D>>,
+) -> Result<Json<Vec<EpochStakeContentInner>>, Error> {
+    let pagination = Pagination::try_from(params)?;
+
+    let tip = domain.get_tip_slot()?;
+    let summary = domain.get_chain_summary()?;
+    let (current, _) = summary.slot_epoch(tip);
+
+    // Blockfrost 404s epochs that don't exist yet; an epoch within range
+    // that simply has no logged distribution (pre-upgrade history, current
+    // epoch before its RUPD ran) returns an empty page instead.
+    if epoch > current {
+        return Err(StatusCode::NOT_FOUND.into());
+    }
+
+    let network = domain.get_network_id()?;
+
+    // Every row of an epoch's distribution shares the epoch-start temporal
+    // key, so the scan range is exactly one slot wide.
+    let start = summary.epoch_start(epoch);
+    let range = LogKey::from(TemporalKey::from(start))..LogKey::from(TemporalKey::from(start + 1));
+
+    let inner = domain.inner.clone();
+    let skip = pagination.skip();
+    let count = pagination.count;
+
+    let page = tokio::task::spawn_blocking(
+        move || -> Result<Vec<(LogKey, AccountStakeLog)>, StatusCode> {
+            let iter = inner
+                .archive()
+                .iter_logs_typed::<AccountStakeLog>(AccountStakeLog::NS, Some(range))
+                .map_err(log_and_500("failed to iterate account stake logs"))?;
+
+            // The log keeps zero-stake delegators (so row counts match
+            // `StakeLog.delegators_count`), but Blockfrost's epoch_stake
+            // excludes them — filter before paginating for parity.
+            iter.filter(|entry| !matches!(entry, Ok((_, log)) if log.amount == 0))
+                .skip(skip)
+                .take(count)
+                .collect::<Result<Vec<_>, _>>()
+                .map_err(log_and_500("failed to read account stake log"))
+        },
+    )
+    .await
+    .map_err(log_and_500("account stake scan task failed"))??;
+
+    let out = page
+        .into_iter()
+        .map(|(key, log)| {
+            let entity = EntityKey::from(key);
+            let credential: StakeCredential = minicbor::decode(entity.as_ref()).map_err(
+                log_and_500("failed to decode stake credential from log key"),
+            )?;
+
+            let stake_address = stake_cred_to_address(&credential, network)
+                .to_bech32()
+                .map_err(log_and_500("failed to encode stake address"))?;
+
+            Ok(EpochStakeContentInner {
+                stake_address,
+                pool_id: bech32_pool(&log.pool_id)?,
+                amount: log.amount.to_string(),
+            })
+        })
+        .collect::<Result<Vec<_>, StatusCode>>()?;
+
+    Ok(Json(out))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -181,4 +263,98 @@ mod tests {
         let path = "/epochs/latest/parameters";
         assert_status(&app, path, StatusCode::INTERNAL_SERVER_ERROR).await;
     }
+
+    #[tokio::test]
+    async fn epochs_stakes_happy_path() {
+        let app = TestApp::new();
+        let epoch = app.tip_epoch() - 1;
+        let path = format!("/epochs/{epoch}/stakes");
+        let (status, bytes) = app.get_bytes(&path).await;
+
+        assert_eq!(
+            status,
+            StatusCode::OK,
+            "unexpected status {status} with body: {}",
+            String::from_utf8_lossy(&bytes)
+        );
+
+        let stakes: Vec<EpochStakeContentInner> =
+            serde_json::from_slice(&bytes).expect("failed to parse epoch stakes");
+
+        // The seeder writes the vectors' account plus one synthetic script
+        // credential (both delegated to the vectors' pool) and one
+        // zero-stake credential, which must be excluded for Blockfrost
+        // parity.
+        assert_eq!(stakes.len(), 2);
+        assert!(stakes.iter().all(|x| x.amount != "0"));
+
+        let seeded = stakes
+            .iter()
+            .find(|x| x.stake_address == app.vectors().stake_address)
+            .expect("seeded stake address missing from distribution");
+
+        assert_eq!(seeded.pool_id, app.vectors().pool_id);
+        assert_eq!(seeded.amount, "7000000");
+    }
+
+    #[tokio::test]
+    async fn epochs_stakes_paginated() {
+        let app = TestApp::new();
+        let epoch = app.tip_epoch() - 1;
+
+        let (status_1, bytes_1) = app
+            .get_bytes(&format!("/epochs/{epoch}/stakes?count=1&page=1"))
+            .await;
+        let (status_2, bytes_2) = app
+            .get_bytes(&format!("/epochs/{epoch}/stakes?count=1&page=2"))
+            .await;
+
+        assert_eq!(status_1, StatusCode::OK);
+        assert_eq!(status_2, StatusCode::OK);
+
+        let page_1: Vec<EpochStakeContentInner> =
+            serde_json::from_slice(&bytes_1).expect("failed to parse stakes page 1");
+        let page_2: Vec<EpochStakeContentInner> =
+            serde_json::from_slice(&bytes_2).expect("failed to parse stakes page 2");
+
+        assert_eq!(page_1.len(), 1);
+        assert_eq!(page_2.len(), 1);
+        assert_ne!(page_1[0].stake_address, page_2[0].stake_address);
+    }
+
+    #[tokio::test]
+    async fn epochs_stakes_empty_epoch() {
+        let app = TestApp::new();
+        // Epoch 0 is in range but nothing is seeded there.
+        let (status, bytes) = app.get_bytes("/epochs/0/stakes").await;
+
+        assert_eq!(status, StatusCode::OK);
+        let stakes: Vec<EpochStakeContentInner> =
+            serde_json::from_slice(&bytes).expect("failed to parse empty stakes");
+        assert!(stakes.is_empty());
+    }
+
+    #[tokio::test]
+    async fn epochs_stakes_bad_request() {
+        let app = TestApp::new();
+        assert_status(&app, "/epochs/not-a-number/stakes", StatusCode::BAD_REQUEST).await;
+    }
+
+    #[tokio::test]
+    async fn epochs_stakes_not_found() {
+        let app = TestApp::new();
+        assert_status(&app, "/epochs/999999/stakes", StatusCode::NOT_FOUND).await;
+    }
+
+    #[tokio::test]
+    async fn epochs_stakes_internal_error() {
+        let app = TestApp::new_with_fault(Some(TestFault::StateStoreError));
+        assert_status(&app, "/epochs/0/stakes", StatusCode::INTERNAL_SERVER_ERROR).await;
+    }
+
+    #[tokio::test]
+    async fn epochs_stakes_archive_error() {
+        let app = TestApp::new_with_fault(Some(TestFault::ArchiveStoreError));
+        assert_status(&app, "/epochs/0/stakes", StatusCode::INTERNAL_SERVER_ERROR).await;
+    }
 }

crates/minibf/src/test_support.rs

@@ -12,8 +12,8 @@ use dolos_core::{
 };
 use dolos_testing::{
     synthetic::{
-        build_synthetic_blocks, seed_epoch_logs, seed_reward_logs, SyntheticBlockConfig,
-        SyntheticVectors,
+        build_synthetic_blocks, seed_account_stake_logs, seed_epoch_logs, seed_reward_logs,
+        SyntheticBlockConfig, SyntheticVectors,
     },
     toy_domain::ToyDomain,
 };
@@ -74,6 +74,15 @@ impl TestDomainBuilder {
             )
             .expect("failed to seed reward logs");
         }
+        if epoch >= 1 {
+            seed_account_stake_logs(
+                &domain,
+                &vectors.stake_address,
+                &vectors.pool_id,
+                &[epoch - 1],
+            )
+            .expect("failed to seed account stake logs");
+        }
 
         Self { domain, vectors }
     }
@@ -189,4 +198,24 @@ impl TestApp {
     pub fn vectors(&self) -> &SyntheticVectors {
         &self.vectors
     }
+
+    /// Epoch at the domain's tip. Only usable on fault-free apps — fault
+    /// wrappers make the underlying state reads fail.
+    pub fn tip_epoch(&self) -> u64 {
+        let summary =
+            dolos_cardano::eras::load_era_summary::<dolos_testing::faults::FaultyToyDomain>(
+                self._domain.state(),
+            )
+            .expect("era summary");
+
+        let tip = self
+            ._domain
+            .state()
+            .read_cursor()
+            .expect("cursor read failed")
+            .expect("missing tip")
+            .slot();
+
+        summary.slot_epoch(tip).0
+    }
 }

crates/testing/src/synthetic.rs

@@ -8,7 +8,7 @@ use dolos_core::{
 use crate::{tx_sequence_to_hash, utxo_with_value};
 
 use bech32::{FromBase32, ToBase32, Variant};
-use dolos_cardano::model::MemberRewardLog;
+use dolos_cardano::model::{AccountStakeLog, MemberRewardLog};
 use dolos_cardano::rupd::credential_to_key;
 use pallas::codec::utils::{CborWrap, Int, Nullable};
 use pallas::codec::{minicbor, utils::KeepRaw};
@@ -505,6 +505,56 @@ pub fn seed_reward_logs<D: Domain>(
     Ok(())
 }
 
+/// Seed per-account `AccountStakeLog` entries for the given epochs.
+///
+/// Writes one row for `stake_address` and one for a fixed synthetic script
+/// credential, both delegated to `pool_id` — two distinct keys so pagination
+/// over the epoch's stake distribution is testable. A third credential is
+/// seeded with zero stake, which endpoints must exclude (Blockfrost's
+/// epoch_stake has no zero-amount rows).
+pub fn seed_account_stake_logs<D: Domain>(
+    domain: &D,
+    stake_address: &str,
+    pool_id: &str,
+    epochs: &[u64],
+) -> Result<(), ChainError> {
+    let address = Address::from_bech32(stake_address)?;
+    let (stake_cred, _) = dolos_cardano::pallas_extras::address_as_stake_cred(&address)
+        .ok_or(ChainError::InvalidPoolParams)?;
+    let pool_keyhash =
+        pool_keyhash_from_bech32(pool_id).map_err(|_| ChainError::InvalidPoolParams)?;
+
+    let extra_cred = StakeCredential::ScriptHash(Hash::from([0xAA; 28]));
+    let zero_cred = StakeCredential::ScriptHash(Hash::from([0xBB; 28]));
+
+    let summary = dolos_cardano::eras::load_era_summary::<D>(domain.state())?;
+    let writer = domain.archive().start_writer()?;
+
+    for epoch in epochs {
+        let slot = summary.epoch_start(*epoch);
+
+        let entries = [
+            (&stake_cred, 7_000_000u64),
+            (&extra_cred, 3_000_000),
+            (&zero_cred, 0),
+        ];
+
+        for (credential, amount) in entries {
+            let log_key: LogKey = (TemporalKey::from(slot), credential_to_key(credential)).into();
+            let log = AccountStakeLog {
+                amount,
+                pool_id: pool_keyhash.as_ref().to_vec(),
+            };
+            writer
+                .write_log_typed(&log_key, &log)
+                .map_err(ChainError::from)?;
+        }
+    }
+
+    writer.commit().map_err(ChainError::from)?;
+    Ok(())
+}
+
 pub fn seed_epoch_logs<D: Domain>(domain: &D, epochs: &[u64]) -> Result<(), ChainError> {
     let summary = dolos_cardano::eras::load_era_summary::<D>(domain.state())?;
     let base = dolos_cardano::load_epoch::<D>(domain.state())?;

docs/content/apis/minibf.mdx

@@ -132,6 +132,7 @@ Dolos provides many, but not all of the Blockfrost endpoints. The following list
 | `/epochs/latest/parameters`                | Get latest epoch parameters                              |
 | `/epochs/{epoch}/blocks`                   | Get blocks in a specific epoch                           |
 | `/epochs/{epoch}/parameters`               | Get epoch parameters                                     |
+| `/epochs/{epoch}/stakes`                   | Get epoch stake distribution                             |
 | `/genesis`                                 | Get genesis information                                  |
 | `/governance/dreps/{drep_id}`              | Get DRep information                                     |
 | `/metadata/txs/labels/{label}`             | Get metadata for transactions with a specific label      |