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

@@ -12,7 +12,7 @@ use super::{
 };
 use async_trait::async_trait;
 use ceramic_core::{EventId, Network, NodeId, SerializeExt};
-use ceramic_pipeline::{ConclusionData, ConclusionEvent, ConclusionInit, ConclusionTime};
+use ceramic_pipeline::{concluder::TimeProof, ConclusionData, ConclusionEvent, ConclusionInit, ConclusionTime};
 use ceramic_sql::sqlite::SqlitePool;
 use cid::Cid;
 use futures::stream::BoxStream;
@@ -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,30 @@ impl EventService {
 
         match event {
             ceramic_event::unvalidated::Event::Time(time_event) => {
+                let proof = match self.discover_chain_proof(&time_event).await {
+                    Ok(proof) => Some(proof),
+                    Err(error) => {
+                        tracing::warn!(
+                            ?event_cid,
+                            ?error,
+                            "Failed to discover chain proof for time event"
+                        );
+                        None
+                    }
+                };
+
                 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!(),
+                    time_proof: proof.map(|p| TimeProof {
+                        before: p
+                            .timestamp
+                            .try_into()
+                            .expect("conclusion timestamp overflow"),
+                        chain_id: p.chain_id,
+                    }),
                 }))
             }
             ceramic_event::unvalidated::Event::Signed(signed_event) => {
@@ -523,6 +548,40 @@ 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,
+    ) -> std::result::Result<ChainProof, crate::eth_rpc::Error> {
+        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
+            .map_err(|e| crate::eth_rpc::Error::Application(e.into()))?;
+
+        if let Some(proof) = proof {
+            return Ok(proof);
+        }
+
+        // try using the RPC provider and store the proof afterward
+        let proof = self
+            .event_validator
+            .time_event_verifier()
+            .validate_chain_inclusion(event)
+            .await?;
+
+        let proof = ChainProof::from(proof);
+        self.event_access
+            .persist_chain_inclusion_proofs(&[proof.clone()])
+            .await
+            .map_err(|e| crate::eth_rpc::Error::Application(e.into()))?;
+
+        Ok(proof)
+    }
 }
 
 // Small wrapper container around the data field to hold other mutable metadata for the
@@ -579,6 +638,12 @@ pub enum ValidationError {
         key: EventId,
         reason: String,
     },
+    /// 'Soft error' -> should not kill recon conversation but should not be persisted
+    /// A time event could not be validated because no RPC provider was available
+    SoftError {
+        key: EventId,
+        reason: String,
+    },
 }
 
 #[derive(Debug, PartialEq, Eq, Default)]

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) => {