feat: store chain timestamp proof data in db and read in pipeline for conclusion events

Author: dav1do

CONTRIBUTING.md

@@ -15,6 +15,18 @@ Using the makefile is not necessary during your development cycle, feel free to
 
 However running `make` before publishing a PR will provide a good signal if you PR will pass CI.
 
+### Migrations
+
+If you need to add to the sqlite database schema, you will need to add a migration using the sqlx CLI. 
+
+```sh
+cargo install sqlx-cli
+# use the name of the migration and the source directory
+sqlx migrate add -r "chain_proof" --source ./migrations/sqlite
+```
+
+After the up and down files are generated, write the apply/revert SQL in the up/down files. This will be applied automatically at startup.
+
 ### Testing Specific Changes
 
 The above `make` targets test changes as a whole.

event-svc/src/blockchain/eth_rpc/http.rs

@@ -5,7 +5,9 @@ use std::{
     sync::{Arc, Mutex},
 };
 
-use crate::eth_rpc::{ChainInclusion, ChainInclusionProof, Error};
+use crate::eth_rpc::{
+    types::ChainProofMetadata, ChainInclusion, ChainInclusionProof, Error, Timestamp,
+};
 use alloy::{
     hex,
     primitives::{BlockHash, TxHash},
@@ -194,11 +196,6 @@ fn get_root_cid_from_input(input: &str, tx_type: EthProofType) -> anyhow::Result
     }
 }
 
-/// Get the expected transaction hash for a given root CID (this is v1 proof type)
-fn expected_tx_hash(cid: Cid) -> anyhow::Result<TxHash> {
-    Ok(TxHash::from_str(&hex::encode(cid.hash().digest()))?)
-}
-
 #[async_trait::async_trait]
 impl ChainInclusion for HttpEthRpc {
     fn chain_id(&self) -> &caip2::ChainId {
@@ -208,7 +205,7 @@ impl ChainInclusion for HttpEthRpc {
     /// Get the block chain transaction if it exists with the block timestamp information
     async fn get_chain_inclusion_proof(&self, input: &AnchorProof) -> Result<ChainInclusionProof> {
         // transaction to blockHash, blockNumber, input
-        let tx_hash = expected_tx_hash(input.tx_hash())
+        let tx_hash = crate::blockchain::tx_hash_try_from_cid(input.tx_hash())
             .map_err(|e| Error::InvalidArgument(format!("invalid transaction hash: {}", e)))?;
         let tx_hash_res = match self.eth_transaction_by_hash(tx_hash).await? {
             Some(tx) => tx,
@@ -239,7 +236,8 @@ impl ChainInclusion for HttpEthRpc {
             trace!(?blk_hash_res, "blockByHash response");
             let tx_type = EthProofType::from_str(input.tx_type())
                 .map_err(|e| Error::InvalidProof(e.to_string()))?;
-            let root_cid = get_root_cid_from_input(&tx_hash_res.input.to_string(), tx_type)
+            let tx_input = tx_hash_res.input.to_string();
+            let root_cid = get_root_cid_from_input(&tx_input, tx_type)
                 .map_err(|e| Error::InvalidProof(e.to_string()))?;
 
             if let Some(threshold) = BLOCK_THRESHHOLDS.get(self.chain_id()) {
@@ -252,8 +250,14 @@ impl ChainInclusion for HttpEthRpc {
             }
 
             Ok(ChainInclusionProof {
-                timestamp: blk_hash_res.header.timestamp,
+                timestamp: Timestamp::from_unix_ts(blk_hash_res.header.timestamp),
+                block_hash: block_hash.to_string(),
                 root_cid,
+                meta_data: ChainProofMetadata {
+                    chain_id: self.chain_id.clone(),
+                    tx_hash: tx_hash.to_string(),
+                    tx_input,
+                },
             })
         } else {
             Err(Error::TxNotMined {

event-svc/src/blockchain/eth_rpc/mod.rs

@@ -2,4 +2,7 @@ mod http;
 mod types;
 
 pub use http::HttpEthRpc;
-pub use types::{BlockHash, ChainInclusion, ChainInclusionProof, Error, EthProofType, TxHash};
+pub use types::{
+    BlockHash, ChainInclusion, ChainInclusionProof, ChainProofMetadata, Error, EthProofType,
+    Timestamp, TxHash,
+};

event-svc/src/blockchain/eth_rpc/types.rs

@@ -92,13 +92,44 @@ impl FromStr for EthProofType {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+/// A timestamp that is able to provide seconds since the unix epoch
+pub struct Timestamp(u64);
+
+impl Timestamp {
+    /// Create a timestamp from a unix epoch timestamp
+    pub const fn from_unix_ts(ts: u64) -> Self {
+        Self(ts)
+    }
+
+    /// A unix epoch timestamp
+    pub fn as_unix_ts(&self) -> u64 {
+        self.0
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 /// A proof of time derived from state on the blockchain
 pub struct ChainInclusionProof {
     /// The timestamp the proof was recorded
-    pub timestamp: u64,
+    pub timestamp: Timestamp,
+    /// The block hash in hex form with '0x' prefix
+    pub block_hash: String,
     /// The root CID of the proof
     pub root_cid: Cid,
+    /// The metadata about the proof and where it's stored
+    pub meta_data: ChainProofMetadata,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+/// Metadata about the proof and where it's stored
+pub struct ChainProofMetadata {
+    /// Chain ID of the proof
+    pub chain_id: ssi::caip2::ChainId,
+    /// The transaction hash in hex form with '0x' prefix
+    pub tx_hash: String,
+    /// The transaction input in hex form with '0x' prefix
+    pub tx_input: String,
 }
 
 #[async_trait::async_trait]

event-svc/src/blockchain/mod.rs

@@ -1,2 +1,12 @@
+use std::str::FromStr as _;
+
+use alloy::primitives::TxHash;
+use ceramic_core::Cid;
+
 /// The ethereum RPC provider module
 pub mod eth_rpc;
+
+/// Get the expected transaction hash for a given root CID (this is v1 proof type)
+pub(crate) fn tx_hash_try_from_cid(cid: Cid) -> anyhow::Result<TxHash> {
+    Ok(TxHash::from_str(&hex::encode(cid.hash().digest()))?)
+}

event-svc/src/event/service.rs

@@ -21,8 +21,8 @@ use itertools::Itertools;
 use recon::ReconItem;
 use tracing::{trace, warn};
 
-use crate::event::validator::ChainInclusionProvider;
-use crate::store::{EventAccess, EventInsertable, EventRowDelivered};
+use crate::store::{ChainProof, EventAccess, EventInsertable, EventRowDelivered};
+use crate::{blockchain::tx_hash_try_from_cid, event::validator::ChainInclusionProvider};
 use crate::{Error, Result};
 
 /// How many events to select at once to see if they've become deliverable when we have downtime
@@ -269,6 +269,7 @@ impl EventService {
             valid,
             unvalidated,
             invalid,
+            proofs,
         } = self
             .event_validator
             .validate_events(validation_requirement, to_validate)
@@ -279,6 +280,7 @@ impl EventService {
             valid,
             unvalidated,
             invalid: invalid_events,
+            proofs,
         })
     }
 
@@ -293,6 +295,7 @@ impl EventService {
             valid,
             unvalidated,
             mut invalid,
+            proofs,
         } = self.validate_events(items, validation_req.as_ref()).await?;
 
         let to_insert: Vec<EventInsertable> = valid
@@ -307,6 +310,14 @@ impl EventService {
             self.track_pending(unvalidated);
         }
 
+        // Someday, we may want to have the validation/proof inclusion logic have knowledge of the database and persist/read
+        // from it directly, rather than only keeping proofs in memory + RPC calls. But for now, it's simpler to persist everything once here
+        // and then the pipeline is able to read from this table and use the timestamps for conclusion events etc.
+        let proofs = proofs.into_iter().map(|p| p.into()).collect::<Vec<_>>();
+        self.event_access
+            .persist_chain_inclusion_proofs(&proofs)
+            .await?;
+
         let (new, existed) = self
             .persist_events(to_insert, deliverable_req, &mut invalid)
             .await?;
@@ -391,16 +402,48 @@ impl EventService {
 
         match event {
             ceramic_event::unvalidated::Event::Time(time_event) => {
+                let proof = match self.discover_chain_proof(&time_event).await {
+                    Ok(proof) => {
+                        if proof.is_none() {
+                            return Err(Error::new_app(anyhow::anyhow!(
+                                "Missing chain proof for time event: {}",
+                                event_cid
+                            )));
+                        }
+                        proof
+                    }
+                    Err(e) => {
+                        return Err(Error::new_app(anyhow::anyhow!(
+                            "Failed to discover chain proof for time event: {}. Error: {}",
+                            event_cid,
+                            e
+                        )));
+                    }
+                };
+
                 Ok(ConclusionEvent::Time(ConclusionTime {
                     event_cid,
                     init,
                     previous: vec![*time_event.prev()],
                     order: delivered as u64,
-                    before: todo!(),
-                    chain_id: todo!(),
-                    tx_hash: todo!(),
-                    tx_type: todo!(),
-                    root: todo!(),
+                    before: proof
+                        .as_ref()
+                        .and_then(|p| {
+                            p.timestamp
+                                .map(|ts| ts.try_into().expect("conclusion timestamp overflow"))
+                        })
+                        .unwrap_or_default(),
+                    chain_id: time_event.proof().chain_id().to_string(),
+                    tx_hash: time_event.proof().tx_hash().to_string(),
+                    tx_type: time_event.proof().tx_type().to_string(),
+                    root: time_event.proof().root(),
+                    tx_input: proof
+                        .as_ref()
+                        .map_or("".to_string(), |p| p.transaction_input.clone()),
+                    block_hash: proof
+                        .as_ref()
+                        .and_then(|p| p.block_hash.clone())
+                        .unwrap_or_default(),
                 }))
             }
             ceramic_event::unvalidated::Event::Signed(signed_event) => {
@@ -523,6 +566,42 @@ impl EventService {
             Ok(())
         }
     }
+
+    /// This is a helper function for migrations to get the chain proof for a given event from the database,
+    /// or to validate and store it if it doesn't exist.
+    async fn discover_chain_proof(
+        &self,
+        event: &ceramic_event::unvalidated::TimeEvent,
+    ) -> anyhow::Result<Option<ChainProof>> {
+        let tx_hash = event.proof().tx_hash();
+        let tx_hash = tx_hash_try_from_cid(tx_hash).unwrap().to_string();
+        let proof = self
+            .event_access
+            .get_chain_proof(event.proof().chain_id(), &tx_hash)
+            .await?;
+
+        if let Some(proof) = proof {
+            return Ok(Some(proof));
+        }
+
+        // try using the RPC provider and store the proof afterward
+        let proof = match self
+            .event_validator
+            .time_event_verifier()
+            .validate_chain_inclusion(event)
+            .await
+        {
+            Ok(proof) => proof,
+            Err(e) => anyhow::bail!("Failed to validate chain inclusion proof: {:?}", e),
+        };
+
+        let proof = ChainProof::from(proof);
+        self.event_access
+            .persist_chain_inclusion_proofs(&[proof.clone()])
+            .await?;
+
+        Ok(Some(proof))
+    }
 }
 
 // Small wrapper container around the data field to hold other mutable metadata for the

event-svc/src/event/validator/event.rs

@@ -6,19 +6,19 @@ use ipld_core::ipld::Ipld;
 use recon::ReconItem;
 use tokio::try_join;
 
-use crate::blockchain::eth_rpc;
-use crate::event::validator::ChainInclusionProvider;
-use crate::store::EventAccess;
 use crate::{
+    blockchain::eth_rpc,
+    eth_rpc::ChainInclusionProof,
     event::{
         service::{ValidationError, ValidationRequirement},
         validator::{
             grouped::{GroupedEvents, SignedValidationBatch, TimeValidationBatch},
             signed::SignedEventValidator,
             time::TimeEventValidator,
+            ChainInclusionProvider,
         },
     },
-    store::EventInsertable,
+    store::{EventAccess, EventInsertable},
     Result,
 };
 
@@ -31,6 +31,8 @@ pub struct ValidatedEvents {
     pub unvalidated: Vec<UnvalidatedEvent>,
     /// Events that failed validation
     pub invalid: Vec<ValidationError>,
+    /// The proofs for the validated time events that can be stored for future time stamping
+    pub proofs: Vec<ChainInclusionProof>,
 }
 
 #[derive(Debug)]
@@ -107,6 +109,7 @@ impl ValidatedEvents {
             valid: Vec::with_capacity(valid),
             unvalidated: Vec::with_capacity(valid / 4),
             invalid: Vec::new(),
+            proofs: Vec::new(),
         }
     }
 
@@ -141,6 +144,11 @@ impl EventValidator {
         })
     }
 
+    /// Get the time event verifier
+    pub(crate) fn time_event_verifier(&self) -> &TimeEventValidator {
+        &self.time_event_verifier
+    }
+
     /// Validates the events with the given validation requirement
     /// If the [`ValidationRequirement`] is None, it just returns every event as valid
     pub(crate) async fn validate_events(
@@ -159,6 +167,7 @@ impl EventValidator {
                     .collect(),
                 unvalidated: Vec::new(),
                 invalid: Vec::new(),
+                proofs: Vec::new(),
             });
         };
 
@@ -211,8 +220,8 @@ impl EventValidator {
                 .validate_chain_inclusion(time_event.as_time())
                 .await
             {
-                Ok(_t) => {
-                    // TODO(AES-345): Someday, we will use `t.as_unix_ts()` and care about the actual timestamp, but for now we just consider it valid
+                Ok(t) => {
+                    validated_events.proofs.push(t);
                     validated_events.valid.push(time_event.into());
                 }
                 Err(err) => {