feat: Add epoch activity REST endpoints with optional historical storage

modules/epoch_activity_counter/Cargo.toml

@@ -16,6 +16,9 @@ anyhow = "1.0"
 tokio = { version = "1", features = ["full"] }
 config = "0.15.11"
 tracing = "0.1.40"
+serde = { version = "1.0.214", features = ["derive"] }
+serde_json = "1.0.132"
+hex = "0.4.3"
 
 [lib]
 path = "src/epoch_activity_counter.rs"

modules/epoch_activity_counter/src/epoch_activity_counter.rs

@@ -3,6 +3,7 @@
 
 use acropolis_common::{
     messages::{CardanoMessage, Message},
+    rest_helper::{handle_rest, handle_rest_with_parameter},
     Era,
 };
 use anyhow::Result;
@@ -15,10 +16,16 @@ use tracing::{error, info};
 
 mod state;
 use state::State;
+mod rest;
+use rest::{handle_epoch, handle_historical_epoch};
 
 const DEFAULT_SUBSCRIBE_HEADERS_TOPIC: &str = "cardano.block.header";
 const DEFAULT_SUBSCRIBE_FEES_TOPIC: &str = "cardano.block.fees";
 const DEFAULT_PUBLISH_TOPIC: &str = "cardano.epoch.activity";
+const DEFAULT_HANDLE_CURRENT_TOPIC: (&str, &str) = ("handle-topic-current-epoch", "rest.get.epoch");
+const DEFAULT_HANDLE_HISTORICAL_TOPIC: (&str, &str) =
+    ("handle-topic-historical-epoch", "rest.get.epochs.*");
+const DEFAULT_STORE_HISTORY: (&str, bool) = ("store-history", false);
 
 /// Epoch activity counter module
 #[module(
@@ -33,17 +40,14 @@ impl EpochActivityCounter {
     async fn run(
         context: Arc<Context<Message>>,
         config: Arc<Config>,
+        state: Arc<Mutex<State>>,
         mut headers_subscription: Box<dyn Subscription<Message>>,
         mut fees_subscription: Box<dyn Subscription<Message>>,
     ) -> Result<()> {
         let publish_topic =
             config.get_string("publish-topic").unwrap_or(DEFAULT_PUBLISH_TOPIC.to_string());
         info!("Publishing on '{publish_topic}'");
 
-        // Create state
-        // TODO!  Handling rollbacks with StateHistory
-        let state = Arc::new(Mutex::new(State::new()));
-
         loop {
             // Read both topics in parallel
             let headers_message_f = headers_subscription.read();
@@ -118,16 +122,53 @@ impl EpochActivityCounter {
             .unwrap_or(DEFAULT_SUBSCRIBE_FEES_TOPIC.to_string());
         info!("Creating subscriber for fees on '{subscribe_fees_topic}'");
 
+        let store_history =
+            config.get_bool(DEFAULT_STORE_HISTORY.0).unwrap_or(DEFAULT_STORE_HISTORY.1);
+
+        // REST handler topics
+        let handle_current_topic = config
+            .get_string(DEFAULT_HANDLE_CURRENT_TOPIC.0)
+            .unwrap_or(DEFAULT_HANDLE_CURRENT_TOPIC.1.to_string());
+        info!("Creating request handler on '{}'", handle_current_topic);
+
+        let handle_historical_topic = config
+            .get_string(DEFAULT_HANDLE_HISTORICAL_TOPIC.0)
+            .unwrap_or(DEFAULT_HANDLE_HISTORICAL_TOPIC.1.to_string());
+        info!("Creating request handler on '{}'", handle_historical_topic);
+
         // Subscribe
         let headers_subscription = context.subscribe(&subscribe_headers_topic).await?;
         let fees_subscription = context.subscribe(&subscribe_fees_topic).await?;
 
+        // Create state
+        // TODO!  Handling rollbacks with StateHistory
+        let state = Arc::new(Mutex::new(State::new(store_history)));
+
+        handle_rest(context.clone(), &handle_current_topic, {
+            let state = state.clone();
+            move || {
+                let state = state.clone();
+                async move { handle_epoch(state).await }
+            }
+        });
+
+        handle_rest_with_parameter(context.clone(), &handle_historical_topic, {
+            let state = state.clone();
+            move |param| handle_historical_epoch(state.clone(), param[0].to_string())
+        });
+
         // Start run task
         let run_context = context.clone();
         context.run(async move {
-            Self::run(run_context, config, headers_subscription, fees_subscription)
-                .await
-                .unwrap_or_else(|e| error!("Failed: {e}"));
+            Self::run(
+                run_context,
+                config,
+                state,
+                headers_subscription,
+                fees_subscription,
+            )
+            .await
+            .unwrap_or_else(|e| error!("Failed: {e}"));
         });
 
         Ok(())

modules/epoch_activity_counter/src/rest.rs

@@ -0,0 +1,105 @@
+use crate::state::State;
+use acropolis_common::messages::RESTResponse;
+use anyhow::Result;
+use serde::Serialize;
+use std::sync::Arc;
+use tokio::sync::Mutex;
+
+#[derive(Serialize)]
+pub struct EpochActivityRest {
+    pub epoch: u64,
+    pub total_blocks: usize,
+    pub total_fees: u64,
+    pub vrf_vkey_hashes: Vec<VRFKeyCount>,
+}
+#[derive(Serialize)]
+pub struct VRFKeyCount {
+    pub vrf_key_hash: String,
+    pub block_count: usize,
+}
+
+/// Handles /epoch
+pub async fn handle_epoch(state: Arc<Mutex<State>>) -> Result<RESTResponse> {
+    let locked = state.lock().await;
+    let epoch_data = locked.get_current_epoch();
+
+    let response = EpochActivityRest {
+        epoch: epoch_data.epoch,
+        total_blocks: epoch_data.total_blocks,
+        total_fees: epoch_data.total_fees,
+        vrf_vkey_hashes: epoch_data
+            .vrf_vkey_hashes
+            .iter()
+            .map(|(key, count)| VRFKeyCount {
+                vrf_key_hash: hex::encode(key),
+                block_count: *count,
+            })
+            .collect(),
+    };
+
+    let json = match serde_json::to_string(&response) {
+        Ok(j) => j,
+        Err(e) => {
+            return Ok(RESTResponse::with_text(
+                500,
+                &format!("Internal server error while retrieving current epoch: {e}"),
+            ));
+        }
+    };
+    Ok(RESTResponse::with_json(200, &json))
+}
+
+/// Handles /epochs/{epoch}
+pub async fn handle_historical_epoch(
+    state: Arc<Mutex<State>>,
+    epoch: String,
+) -> Result<RESTResponse> {
+    let parsed_epoch = match epoch.parse::<u64>() {
+        Ok(v) => v,
+        Err(_) => {
+            return Ok(RESTResponse::with_text(
+                400,
+                &format!("Invalid epoch number: {epoch}"),
+            ))
+        }
+    };
+
+    let locked = state.lock().await;
+    match locked.get_historical_epoch(parsed_epoch) {
+        Err(_) => Ok(RESTResponse::with_text(
+            501,
+            "Historical epoch storage not enabled",
+        )),
+        Ok(Some(epoch_data)) => {
+            let response = EpochActivityRest {
+                epoch: epoch_data.epoch,
+                total_blocks: epoch_data.total_blocks,
+                total_fees: epoch_data.total_fees,
+                vrf_vkey_hashes: epoch_data
+                    .vrf_vkey_hashes
+                    .iter()
+                    .map(|(key, count)| VRFKeyCount {
+                        vrf_key_hash: hex::encode(key),
+                        block_count: *count,
+                    })
+                    .collect(),
+            };
+            let json = match serde_json::to_string(&response) {
+                Ok(j) => j,
+                Err(e) => {
+                    return Ok(RESTResponse::with_text(
+                        500,
+                        &format!(
+                            "Internal server error while retrieving epoch {parsed_epoch}: {e}"
+                        ),
+                    ));
+                }
+            };
+            Ok(RESTResponse::with_json(200, &json))
+        }
+        Ok(None) => Ok(RESTResponse::with_text(
+            404,
+            &format!("Epoch {parsed_epoch} not found"),
+        )),
+    }
+}