chore: eigen costs

e2e/helpers/src/committer.rs

@@ -379,6 +379,19 @@ impl CommitterProcess {
         Ok(response)
     }
 
+    pub async fn latest_costs(&self) -> anyhow::Result<Vec<BundleCost>> {
+        let response = reqwest::get(format!(
+            "http://localhost:{}/v1/costs?variant=latest",
+            self.port,
+        ))
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+
+        Ok(response)
+    }
+
     pub fn port(&self) -> u16 {
         self.port
     }

e2e/tests/tests/eigen_costs.rs

@@ -0,0 +1,66 @@
+use anyhow::Result;
+use e2e_helpers::whole_stack::{
+    create_and_fund_kms_signers, deploy_contract, start_db, start_eigen_committer, start_eth,
+    start_fuel_node, start_kms,
+};
+use k256::ecdsa::SigningKey as K256SigningKey;
+use std::time::Duration;
+use tracing::info;
+
+#[tokio::test]
+async fn test_eigen_costs() -> Result<()> {
+    // Start required services
+    let logs = true;
+    let kms = start_kms(logs).await?;
+    let eth_node = start_eth(logs).await?;
+    let eth_signers = create_and_fund_kms_signers(&kms, &eth_node).await?;
+
+    // Get Eigen key from environment and inject into KMS
+    let eigen_key_hex = std::env::var("EIGEN_KEY")
+        .expect("EIGEN_KEY environment variable must be set for Eigen tests");
+
+    // Convert hex string to bytes
+    let key_bytes: [u8; 32] = hex::decode(&eigen_key_hex)
+        .expect("Failed to decode EIGEN_KEY hex")
+        .try_into()
+        .expect("EIGEN_KEY must be 32 bytes");
+
+    // Create signing key and inject into KMS
+    let secret_key = k256::elliptic_curve::SecretKey::from_slice(&key_bytes)?;
+    let k256_signing_key = K256SigningKey::from(&secret_key);
+    let kms_key_id = kms.inject_secp256k1_key(&k256_signing_key).await?;
+
+    // Deploy contract and start services
+    let request_timeout = Duration::from_secs(50);
+    let max_fee = 1_000_000_000_000;
+    let (_contract_args, deployed_contract) =
+        deploy_contract(&eth_node, eth_signers.clone(), max_fee, request_timeout).await?;
+    let db = start_db().await?;
+    let fuel_node = start_fuel_node(logs, Some(Duration::from_millis(200))).await?;
+
+    // Start Eigen committer with KMS key
+    let logs = true;
+    let committer = start_eigen_committer(
+        logs,
+        db.clone(),
+        &eth_node,
+        fuel_node.url(),
+        &deployed_contract,
+        eth_signers.main,
+        kms_key_id, // Use the KMS key ID instead of raw EIGEN_KEY
+        "1 KB",
+    )
+    .await?;
+
+    // Wait for some blocks to be processed
+    tokio::time::sleep(Duration::from_secs(100)).await;
+
+    // Check if committer has updated the costs
+    let costs = committer.latest_costs().await?;
+
+    info!("received {costs:?}");
+
+    assert!(costs.iter().all(|cost| cost.size > 0));
+
+    Ok(())
+}

packages/adapters/storage/migrations/0010_eigen_indexes.sql

@@ -0,0 +1,8 @@
+BEGIN;
+
+CREATE INDEX idx_eigen_submission_fragments_id_desc ON eigen_submission_fragments (id DESC);
+CREATE INDEX idx_eigen_submission_status ON eigen_submission (status);
+CREATE INDEX idx_l1_fragments_bundle_id ON l1_fragments (bundle_id);
+CREATE INDEX idx_eigen_submission_status_id ON eigen_submission (status, id);
+
+COMMIT;

packages/adapters/storage/src/lib.rs

@@ -69,6 +69,10 @@ impl services::state_listener::port::Storage for Postgres {
             .await
             .map_err(Into::into)
     }
+
+    async fn update_eigen_costs(&self) -> services::Result<()> {
+        self._update_eigen_costs().await.map_err(Into::into)
+    }
 }
 
 impl services::cost_reporter::port::Storage for Postgres {

packages/adapters/storage/src/postgres.rs

@@ -1468,6 +1468,41 @@ impl Postgres {
         Ok(())
     }
 
+    pub(crate) async fn _update_eigen_costs(&self) -> Result<()> {
+        let mut tx = self.connection_pool.begin().await?;
+
+        sqlx::query!(
+            r#"
+           INSERT INTO bundle_cost (bundle_id, da_block_height, cost, size, is_finalized)
+           SELECT DISTINCT
+               b.id as bundle_id,
+               0 as da_block_height,
+               0 as cost,
+               COALESCE(SUM(lf.total_bytes), 0) as size,
+               true as is_finalized
+           FROM eigen_submission_fragments esf
+           JOIN eigen_submission es ON esf.submission_id = es.id
+           JOIN l1_fragments lf ON esf.fragment_id = lf.id
+           JOIN bundles b ON lf.bundle_id = b.id
+           LEFT JOIN bundle_cost bc ON b.id = bc.bundle_id
+           WHERE es.status = $1
+             AND bc.bundle_id IS NULL  -- Don't insert duplicates
+             AND esf.id IN (
+                 SELECT id FROM eigen_submission_fragments
+                 ORDER BY id DESC
+             )
+           GROUP BY b.id;
+           "#,
+            i16::from(eigen_tables::SubmissionStatus::Finalized),
+        )
+        .execute(&mut *tx)
+        .await?;
+
+        tx.commit().await?;
+
+        Ok(())
+    }
+
     pub(crate) async fn _get_latest_block_height(&self) -> Result<u32> {
         let result = sqlx::query!(r#"SELECT MAX(height) as max_height FROM fuel_blocks"#)
             .fetch_one(&self.connection_pool)

packages/adapters/storage/src/test_instance.rs

@@ -230,6 +230,10 @@ impl services::state_listener::port::Storage for DbWithProcess {
     ) -> services::Result<()> {
         unimplemented!();
     }
+
+    async fn update_eigen_costs(&self) -> services::Result<()> {
+        unimplemented!()
+    }
 }
 
 impl block_importer::port::Storage for DbWithProcess {

packages/services/src/state_listener/eigen_service.rs

@@ -129,6 +129,8 @@ where
 
         self.storage.update_eigen_submissions(changes).await?;
 
+        self.storage.update_eigen_costs().await?;
+
         Ok(())
     }
 }

packages/services/src/state_listener/port.rs

@@ -56,6 +56,7 @@ pub trait Storage: Sync {
         request_id: &EigenDARequestId,
     ) -> Result<Option<DateTime<Utc>>>;
     async fn update_eigen_submissions(&self, changes: Vec<(u32, DispersalStatus)>) -> Result<()>;
+    async fn update_eigen_costs(&self) -> Result<()>;
 }
 
 pub trait Clock {