Start defining the coordinator

coordinator/Cargo.toml

@@ -14,13 +14,19 @@ all-features = true
 rustdoc-args = ["--cfg", "docsrs"]
 
 [dependencies]
+log = "0.4"
+
 blake2 = "0.10"
 
 frost = { package = "modular-frost", path = "../crypto/frost" }
 
+serai-db = { path = "../common/db" }
+
 processor-messages = { package = "processor-messages", path = "../processor/messages" }
 tributary = { package = "tributary-chain", path = "./tributary" }
 
+serai-client = { path = "../substrate/client", features = ["serai"] }
+
 tokio = { version = "1", features = ["full"] }
 
 [dev-dependencies]

coordinator/src/main.rs

@@ -1,7 +1,33 @@
+#![allow(dead_code)]
+#![allow(unused_variables)]
+#![allow(unused_mut)]
+
+use serai_db::Db;
+
+use serai_client::Serai;
+
 mod transaction;
+mod substrate;
 
 #[cfg(test)]
 mod tests;
 
+async fn run<D: Db>(db: D, serai: Serai) {
+  let mut last_substrate_block = 0; // TODO: Load from DB
+
+  loop {
+    match substrate::handle_new_blocks(&serai, &mut last_substrate_block).await {
+      Ok(()) => {}
+      Err(e) => log::error!("couldn't communicate with serai node: {e}"),
+    }
+
+    // Handle all messages from tributaries
+
+    // Handle all messages from processors
+  }
+}
+
 #[tokio::main]
-async fn main() {}
+async fn main() {
+  // Open the database
+}

coordinator/src/substrate.rs

@@ -0,0 +1,52 @@
+use serai_client::{SeraiError, Block, Serai};
+
+async fn handle_block(serai: &Serai, block: Block) -> Result<Vec<()>, SeraiError> {
+  let hash = block.hash();
+  let mut actions = vec![];
+
+  // If a new validator set was activated, create tributary/inform processor to do a DKG
+  for new_set in serai.get_new_set_events(hash).await? {
+    todo!()
+  }
+
+  // If a key pair was confirmed, inform the processor
+  for key_gen in serai.get_key_gen_events(hash).await? {
+    todo!()
+  }
+
+  // If batch, tell processor of block acknowledged/burns
+  for new_set in serai.get_batch_events(hash).await? {
+    todo!()
+  }
+
+  Ok(actions)
+}
+
+pub(crate) async fn handle_new_blocks(
+  serai: &Serai,
+  last_substrate_block: &mut u64,
+) -> Result<(), SeraiError> {
+  // Check if there's been a new Substrate block
+  let latest = serai.get_latest_block().await?;
+  let latest_number = latest.number();
+  if latest_number == *last_substrate_block {
+    return Ok(());
+  }
+  let mut latest = Some(latest);
+
+  for b in (*last_substrate_block + 1) ..= latest_number {
+    let actions = handle_block(
+      serai,
+      if b == latest_number {
+        latest.take().unwrap()
+      } else {
+        serai.get_block_by_number(b).await?.unwrap()
+      },
+    )
+    .await?;
+    // TODO: Handle actions, update the DB
+    *last_substrate_block += 1;
+  }
+
+  Ok(())
+}

substrate/client/src/serai/mod.rs

@@ -16,7 +16,7 @@ use subxt::{
     extrinsic_params::{BaseExtrinsicParams, BaseExtrinsicParamsBuilder},
   },
   tx::{Signer, Payload, TxClient},
-  rpc::types::ChainBlock,
+  rpc::types::{ChainBlock, ChainBlockExtrinsic},
   Config as SubxtConfig, OnlineClient,
 };
 
@@ -56,7 +56,32 @@ impl SubxtConfig for SeraiConfig {
   type ExtrinsicParams = BaseExtrinsicParams<SeraiConfig, Tip>;
 }
 
-pub type Block = ChainBlock<SeraiConfig>;
+#[derive(Debug)]
+pub struct Block(ChainBlock<SeraiConfig>);
+impl Block {
+  pub fn hash(&self) -> [u8; 32] {
+    self.0.header.hash().into()
+  }
+  pub fn number(&self) -> u64 {
+    self.0.header.number
+  }
+
+  pub fn header(&self) -> &Header {
+    &self.0.header
+  }
+  pub fn transactions(&self) -> &[ChainBlockExtrinsic] {
+    &self.0.extrinsics
+  }
+}
+
+impl Clone for Block {
+  fn clone(&self) -> Block {
+    Block(ChainBlock::<SeraiConfig> {
+      header: self.0.header.clone(),
+      extrinsics: self.0.extrinsics.clone(),
+    })
+  }
+}
 
 #[derive(Error, Debug)]
 pub enum SeraiError {
@@ -120,6 +145,19 @@ impl Serai {
     Ok(self.0.rpc().finalized_head().await.map_err(SeraiError::RpcError)?.into())
   }
 
+  pub async fn get_latest_block(&self) -> Result<Block, SeraiError> {
+    Ok(Block(
+      self
+        .0
+        .rpc()
+        .block(Some(self.0.rpc().finalized_head().await.map_err(SeraiError::RpcError)?))
+        .await
+        .map_err(SeraiError::RpcError)?
+        .ok_or(SeraiError::InvalidNode)?
+        .block,
+    ))
+  }
+
   // There is no provided method for this
   // TODO: Add one to Serai
   pub async fn is_finalized(&self, header: &Header) -> Result<Option<bool>, SeraiError> {
@@ -169,7 +207,7 @@ impl Serai {
       return Ok(None);
     }
 
-    Ok(Some(res.block))
+    Ok(Some(Block(res.block)))
   }
 
   // Ideally, this would be get_block_hash, not get_block_by_number

substrate/client/tests/common/tx.rs

@@ -2,21 +2,16 @@ use core::time::Duration;
 
 use tokio::time::sleep;
 
-use serai_client::subxt::{config::Header, utils::Encoded};
+use serai_client::subxt::utils::Encoded;
 
 use crate::common::serai;
 
 #[allow(dead_code)]
 pub async fn publish_tx(tx: &Encoded) -> [u8; 32] {
   let serai = serai().await;
 
-  let mut latest = serai
-    .get_block(serai.get_latest_block_hash().await.unwrap())
-    .await
-    .unwrap()
-    .unwrap()
-    .header
-    .number();
+  let mut latest =
+    serai.get_block(serai.get_latest_block_hash().await.unwrap()).await.unwrap().unwrap().number();
 
   serai.publish(tx).await.unwrap();
 
@@ -42,9 +37,9 @@ pub async fn publish_tx(tx: &Encoded) -> [u8; 32] {
       block.unwrap()
     };
 
-    for extrinsic in block.extrinsics {
-      if extrinsic.0 == tx.0[2 ..] {
-        return block.header.hash().into();
+    for transaction in block.transactions() {
+      if transaction.0 == tx.0[2 ..] {
+        return block.hash();
       }
     }
   }

substrate/client/tests/validator_sets.rs

@@ -10,7 +10,6 @@ use serai_client::{
     primitives::{Session, ValidatorSet},
     ValidatorSetsEvent,
   },
-  subxt::config::Header,
   Serai,
 };
 
@@ -38,9 +37,7 @@ serai_test!(
     // Make sure the genesis is as expected
     assert_eq!(
       serai
-        .get_new_set_events(
-          serai.get_block_by_number(0).await.unwrap().unwrap().header.hash().into()
-        )
+        .get_new_set_events(serai.get_block_by_number(0).await.unwrap().unwrap().hash())
         .await
         .unwrap(),
       [BITCOIN_NET_ID, ETHEREUM_NET_ID, MONERO_NET_ID]