fix: review comments

Author: smrz2001

event-svc/src/event/service.rs

@@ -541,8 +541,8 @@ impl EventService {
         }
     }
 
-    /// This is a helper function to get the chain proof for a given event from the database, or to validate and store
-    /// it if it doesn't exist.
+    /// This is a helper function to get the chain proof for a given event from the database, or throw an error if it
+    /// wasn't found.
     async fn get_chain_proof(
         &self,
         event: &ceramic_event::unvalidated::TimeEvent,

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

@@ -4,6 +4,7 @@ use ceramic_core::{Cid, EventId, NodeId};
 use ceramic_event::unvalidated;
 use ipld_core::ipld::Ipld;
 use recon::ReconItem;
+use tokio::try_join;
 
 use crate::{
     blockchain::eth_rpc,
@@ -157,22 +158,14 @@ impl EventValidator {
         // Partition the events by type of validation needed and delegate to validators
         let grouped = GroupedEvents::from(parsed_events);
 
-        // Time events are always validated
-        let validated_time = self.validate_time_events(grouped.time_batch).await?;
+        let (validated_signed, validated_time) = try_join!(
+            self.validate_signed_events(grouped.signed_batch, validation_req),
+            // Time events are always validated
+            self.validate_time_events(grouped.time_batch)
+        )?;
+        validated.extend_with(validated_signed);
         validated.extend_with(validated_time);
 
-        if let Some(req) = validation_req {
-            let validated_signed = self
-                .validate_signed_events(grouped.signed_batch, req)
-                .await?;
-            validated.extend_with(validated_signed);
-        } else {
-            // Return all data events as valid
-            validated
-                .valid
-                .extend(Vec::<ValidatedEvent>::from(grouped.signed_batch));
-        };
-
         if !validated.invalid.is_empty() {
             tracing::warn!(count=%validated.invalid.len(), "invalid events discovered");
         }
@@ -183,23 +176,33 @@ impl EventValidator {
     async fn validate_signed_events(
         &self,
         events: SignedValidationBatch,
-        validation_req: &ValidationRequirement,
+        validation_req: Option<&ValidationRequirement>,
     ) -> Result<ValidatedEvents> {
-        let opts = if validation_req.check_exp {
-            ceramic_validation::VerifyJwsOpts::default()
+        if let Some(req) = validation_req {
+            let opts = if req.check_exp {
+                ceramic_validation::VerifyJwsOpts::default()
+            } else {
+                ceramic_validation::VerifyJwsOpts {
+                    at_time: ceramic_validation::AtTime::SkipTimeChecks,
+                    ..Default::default()
+                }
+            };
+            SignedEventValidator::validate_events(
+                Arc::clone(&self.event_access),
+                &opts,
+                events,
+                req.require_local_init,
+            )
+            .await
         } else {
-            ceramic_validation::VerifyJwsOpts {
-                at_time: ceramic_validation::AtTime::SkipTimeChecks,
-                ..Default::default()
-            }
-        };
-        SignedEventValidator::validate_events(
-            Arc::clone(&self.event_access),
-            &opts,
-            events,
-            validation_req.require_local_init,
-        )
-        .await
+            // If no validation requirement is specified, we assume all events are valid.
+            Ok(ValidatedEvents {
+                valid: events.into(),
+                unvalidated: Vec::new(),
+                invalid: Vec::new(),
+                proofs: Vec::new(),
+            })
+        }
     }
 
     async fn validate_time_events(&self, events: TimeValidationBatch) -> Result<ValidatedEvents> {

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

@@ -17,7 +17,7 @@ pub struct TimeEventValidator {
     /// we could support multiple providers for each chain (to get around rate limits)
     /// but we'll just force people to run a light client if they really need the throughput
     chain_providers: HashMap<caip2::ChainId, ChainInclusionProvider>,
-    // add a sql connection / block table access
+    // TODO: add a sql connection / block table access
 }
 
 impl std::fmt::Debug for TimeEventValidator {

event-svc/src/store/sql/entities/chain_proof.rs

@@ -8,7 +8,7 @@ pub struct ChainProof {
     pub chain_id: String,
     /// The transaction hash of the proof
     pub transaction_hash: String,
-    /// The transaction input of the proof
+    /// The transaction input of the proof (i.e. the CID of the root of the Merkle tree being anchored)
     pub transaction_input: String,
     /// The block hash of the proof
     pub block_hash: String,

pipeline/src/resolver/mod.rs

@@ -441,11 +441,11 @@ pub struct StreamStateMsg {
     pub id: StreamId,
 }
 impl Message for StreamStateMsg {
-    type Result = anyhow::Result<Option<StreamState>>;
+    type Result = anyhow::Result<Option<ResolvedStreamState>>;
 }
 
-/// State of a single stream
-pub struct StreamState {
+/// State of a single stream after conflict resolution
+pub struct ResolvedStreamState {
     /// Multibase encoding of the stream id
     pub id: StreamId,
 
@@ -462,7 +462,7 @@ pub struct StreamState {
     pub data: Vec<u8>,
 }
 
-impl std::fmt::Debug for StreamState {
+impl std::fmt::Debug for ResolvedStreamState {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         if !f.alternate() {
             f.debug_struct("StreamState")
@@ -583,7 +583,7 @@ impl Handler<StreamStateMsg> for Resolver {
             let value = values.value(i);
             dimensions.insert(key.to_string(), value.to_vec());
         }
-        Ok(Some(StreamState {
+        Ok(Some(ResolvedStreamState {
             id,
             event_cid: Cid::read_bytes(event_cid.value(0)).context("event_cid as a CID")?,
             controller: controller.value(0).to_string(),