feat: index voyager messages in hasura

Cargo.toml

@@ -49,6 +49,7 @@ protos = { path = "generated/rust", default-features = false }
 unionlabs = { path = "lib/unionlabs", default-features = false }
 beacon-api = { path = "lib/beacon-api", default-features = false }
 chain-utils = { path = "lib/chain-utils", default-features = false }
+hubble = { path = "hubble" }
 
 token-factory-api = { path = "cosmwasm/token-factory-api", default-features = false }
 ucs01-relay-api = { path = "cosmwasm/ucs01-relay-api", default-features = false }

hubble/src/hasura.rs

@@ -14,13 +14,19 @@ pub trait Datastore {
         <Q as GraphQLQuery>::Variables: 'static;
 }
 
-#[derive(Clone)]
+#[derive(Debug, Clone)]
 pub struct HasuraDataStore {
     client: Client,
     url: Url,
     secret: String,
 }
 
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct HasuraConfig {
+    pub url: Url,
+    pub secret: String,
+}
+
 impl Datastore for HasuraDataStore {
     /// Performs a GraphQL post request (which may query or mutate).
     /// It injects the x-hasura-admin-secret header.

lib/chain-utils/Cargo.toml

@@ -25,3 +25,6 @@ typenum = { version = "1.16.0", default-features = false, features = ["const-gen
 prost = "*"
 sha2 = "0.10.6"
 chrono = { version = "0.4.26", default-features = false, features = ["alloc"] }
+hubble.workspace = true
+reqwest = "0.11.20"
+serde_json = "1.0.107"

lib/chain-utils/src/evm.rs

@@ -19,12 +19,13 @@ use ethers::{
     signers::{LocalWallet, Wallet},
     utils::secret_key_to_address,
 };
-use futures::{stream, Future, FutureExt, Stream, StreamExt};
+use futures::{stream, Future, FutureExt, Stream, StreamExt, TryStreamExt};
+use hubble::hasura::{insert_demo_tx, Datastore, HasuraConfig, HasuraDataStore, InsertDemoTx};
 use serde::{Deserialize, Serialize};
 use typenum::Unsigned;
 use unionlabs::{
     ethereum::{Address, H256, U256},
-    ethereum_consts_traits::ChainSpec,
+    ethereum_consts_traits::{ChainSpec, Mainnet, Minimal},
     events::{
         AcknowledgePacket, ChannelOpenAck, ChannelOpenConfirm, ChannelOpenInit, ChannelOpenTry,
         ConnectionOpenAck, ConnectionOpenConfirm, ConnectionOpenInit, ConnectionOpenTry,
@@ -57,6 +58,8 @@ pub struct Evm<C: ChainSpec> {
     pub ibc_handler: IBCHandler<CometblsMiddleware>,
     pub provider: Provider<Ws>,
     pub beacon_api_client: BeaconApiClient<C>,
+
+    pub hasura_client: HasuraDataStore,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -71,6 +74,8 @@ pub struct Config {
     pub eth_rpc_api: String,
     /// The RPC endpoint for the beacon chain.
     pub eth_beacon_rpc_api: String,
+
+    pub hasura_config: HasuraConfig,
 }
 
 impl<C: ChainSpec> Chain for Evm<C> {
@@ -277,18 +282,24 @@ impl<C: ChainSpec> Evm<C> {
             provider,
             beacon_api_client: BeaconApiClient::new(config.eth_beacon_rpc_api).await,
             wallet,
+            hasura_client: HasuraDataStore::new(
+                reqwest::Client::new(),
+                config.hasura_config.url,
+                config.hasura_config.secret,
+            ),
         }
     }
 
     // TODO: Change to take a beacon slot instead of a height
-    // TODO: Change to return a block number, not a height
     pub async fn execution_height(&self, beacon_height: Height) -> u64 {
-        let height = self
+        let response = self
             .beacon_api_client
             .block(beacon_api::client::BlockId::Slot(
                 beacon_height.revision_height,
             ))
-            .await
+            .await;
+
+        let height = response
             .unwrap()
             .data
             .message
@@ -426,6 +437,14 @@ impl<C: ChainSpec> EventSource for Evm<C> {
         _seed: Self::Seed,
     ) -> impl Stream<Item = Result<Self::Event, Self::Error>> + '_ {
         async move {
+            let genesis_time = self
+                .beacon_api_client
+                .genesis()
+                .await
+                .unwrap()
+                .data
+                .genesis_time;
+
             stream::unfold(
                 self.query_latest_height().await,
                 move |previous_beacon_height| async move {
@@ -825,6 +844,26 @@ impl<C: ChainSpec> EventSource for Evm<C> {
                 },
             )
             .flatten()
+            .then(move |event| async move {
+                if let Ok(ref event) = event {
+                    let current_slot = event.height.revision_height;
+
+                    let next_epoch_ts = next_epoch_timestamp::<C>(current_slot, genesis_time);
+
+                    self.hasura_client
+                        .do_post::<InsertDemoTx>(insert_demo_tx::Variables {
+                            data: serde_json::json! {{
+                                "latest_execution_block_hash": event.block_hash,
+                                "timestamp": next_epoch_ts,
+                            }},
+                        })
+                        .await
+                        .unwrap();
+                }
+
+                // pass it back through
+                event
+            })
         }
         .flatten_stream()
         .inspect(|x| {
@@ -904,3 +943,21 @@ impl_eth_call_ext! {
     GetChannelCall                -> GetChannelReturn;
     GetHashedPacketCommitmentCall -> GetHashedPacketCommitmentReturn;
 }
+
+pub fn next_epoch_timestamp<C: ChainSpec>(slot: u64, genesis_timestamp: u64) -> u64 {
+    let next_epoch_slot = slot + (C::SLOTS_PER_EPOCH::U64 - (slot % C::SLOTS_PER_EPOCH::U64));
+    genesis_timestamp + (next_epoch_slot * C::SECONDS_PER_SLOT::U64)
+}
+
+#[test]
+fn next_epoch_ts() {
+    dbg!(next_epoch_timestamp::<Mainnet>(6, 0));
+    dbg!(next_epoch_timestamp::<Mainnet>(7, 0));
+    dbg!(next_epoch_timestamp::<Mainnet>(8, 0));
+    dbg!(next_epoch_timestamp::<Mainnet>(9, 0));
+
+    dbg!(next_epoch_timestamp::<Minimal>(6, 0));
+    // dbg!(next_epoch::<Minimal>(48, 0));
+    // dbg!(next_epoch::<Minimal>(49, 0));
+    // dbg!(next_epoch::<Minimal>(47, 0));
+}

voyager-config.json

@@ -9,7 +9,11 @@
         "raw": "0x4e9444a6efd6d42725a250b650a781da2737ea308c839eaccb0f7f3dbd2fea77"
       },
       "eth_rpc_api": "ws://localhost:8546",
-      "eth_beacon_rpc_api": "http://localhost:9596"
+      "eth_beacon_rpc_api": "http://localhost:9596",
+      "hasura_config": {
+        "url": "https://graphql.union.build/v1/graphql",
+        "secret": "3N5Mt2f4Y1AC7dE663AsGqRy66yiHBuZ3RMgUjM6X4Q3Ma8G2jihgfchsdasdsadasda"
+      }
     },
     "union-devnet": {
       "chain_type": "union",

voyager/Cargo.toml

@@ -47,6 +47,7 @@ frunk = "0.4.2"
 bitvec = "1.0.1"
 displaydoc = { version = "0.2.4", default-features = false }
 frame-support-procedural = "18.0.0"
+hubble.workspace = true
 
 [features]
-eth-mainnet = [ "unionlabs/eth-mainnet" ]
+eth-mainnet = [ "unionlabs/eth-mainnet" ]

voyager/src/chain.rs

@@ -45,6 +45,7 @@ impl AnyChain {
                     signer: evm.signer,
                     eth_rpc_api: evm.eth_rpc_api,
                     eth_beacon_rpc_api: evm.eth_beacon_rpc_api,
+                    hasura_config: evm.hasura_config,
                 })
                 .await,
             ),
@@ -54,6 +55,7 @@ impl AnyChain {
                     signer: evm.signer,
                     eth_rpc_api: evm.eth_rpc_api,
                     eth_beacon_rpc_api: evm.eth_beacon_rpc_api,
+                    hasura_config: evm.hasura_config,
                 })
                 .await,
             ),

voyager/src/chain/evm.rs

@@ -29,9 +29,11 @@ use ethers::{
 use frame_support_procedural::{CloneNoBound, DebugNoBound, PartialEqNoBound};
 use frunk::{hlist_pat, HList};
 use futures::Future;
+use hubble::hasura::{insert_demo_tx, Datastore, InsertDemoTx};
 use prost::Message;
 use protos::union::ibc::lightclients::ethereum::v1 as ethereum_v1;
 use serde::{Deserialize, Serialize};
+use serde_json::json;
 use typenum::Unsigned;
 use unionlabs::{
     ethereum::{