Refactor monitor_blocktime into sdk

bot/src/client.rs

@@ -0,0 +1,35 @@
+use {
+    crate::env,
+    solana_client_helpers::{Client, ClientResult, RpcClient},
+    solana_sdk::{
+        commitment_config::CommitmentConfig, instruction::Instruction, signature::read_keypair,
+        signature::Signature, transaction::Transaction,
+    },
+    std::fs::File,
+};
+
+pub trait RPCClient {
+    fn new() -> Client;
+    fn sign_and_submit(&self, ixs: &[Instruction], memo: &str) -> ClientResult<Signature>;
+}
+
+impl RPCClient for Client {
+    fn new() -> Client {
+        let payer = read_keypair(&mut File::open(env::keypath().as_str()).unwrap()).unwrap();
+        let client = RpcClient::new_with_commitment::<String>(
+            env::rpc_endpoint().as_str().into(),
+            CommitmentConfig::processed(),
+        );
+        Client { client, payer }
+    }
+
+    fn sign_and_submit(&self, ixs: &[Instruction], memo: &str) -> ClientResult<Signature> {
+        println!("{}", memo);
+        let payer = self.payer_pubkey();
+        let mut tx = Transaction::new_with_payer(ixs, Some(&payer));
+        tx.sign(&vec![&self.payer], self.latest_blockhash()?);
+        let sig = self.send_and_confirm_transaction(&tx)?;
+        println!("✅ {:?}", sig);
+        Ok(sig)
+    }
+}

bot/src/exec.rs

@@ -1,9 +1,5 @@
 use {
-    crate::{
-        bucket::Bucket,
-        cache::TaskCache,
-        utils::{monitor_blocktime, sign_and_submit},
-    },
+    crate::{bucket::Bucket, cache::TaskCache, client::RPCClient, env},
     cronos_sdk::account::*,
     solana_client_helpers::Client,
     solana_sdk::{instruction::AccountMeta, pubkey::Pubkey},
@@ -20,7 +16,7 @@ pub fn execute_tasks(
     cache: Arc<RwLock<TaskCache>>,
     bucket: Arc<Mutex<Bucket>>,
 ) {
-    let blocktime_receiver = monitor_blocktime();
+    let blocktime_receiver = cronos_sdk::monitor_blocktime(env::wss_endpoint());
     for blocktime in blocktime_receiver {
         println!("⏳ Blocktime: {}", blocktime);
         let tcache = cache.clone();
@@ -112,8 +108,7 @@ fn execute_task(
             .push(AccountMeta::new_readonly(task.ix.program_id, false));
 
         // Sign and submit
-        let res = sign_and_submit(
-            &client,
+        let res = client.sign_and_submit(
             &[ix_exec],
             format!("🤖 Executing task: {} {}", key, task.schedule.exec_at).as_str(),
         );

bot/src/main.rs

@@ -2,16 +2,16 @@ use std::sync::{Arc, Mutex, RwLock};
 
 use bucket::Bucket;
 use cache::TaskCache;
+use client::RPCClient;
 use dotenv::dotenv;
-use solana_client_helpers::ClientResult;
-use utils::new_rpc_client;
+use solana_client_helpers::{Client, ClientResult};
 
 mod bucket;
 mod cache;
+mod client;
 mod env;
 mod exec;
 mod replicate;
-mod utils;
 
 use {exec::*, replicate::*};
 
@@ -20,7 +20,7 @@ fn main() -> ClientResult<()> {
     dotenv().ok();
 
     // Load resources
-    let client = Arc::new(new_rpc_client());
+    let client = Arc::new(Client::new());
     let cache = Arc::new(RwLock::new(TaskCache::new()));
     let bucket = Arc::new(Mutex::new(Bucket::default()));
 

cli/src/processor/blocktime/get.rs

@@ -6,7 +6,7 @@ use crate::error::CliError;
 
 pub fn get(client: &Arc<Client>) -> Result<(), CliError> {
     let blocktime =
-        cronos_sdk::blocktime(client).map_err(|err| CliError::BadClient(err.to_string()))?;
+        cronos_sdk::get_blocktime(client).map_err(|err| CliError::BadClient(err.to_string()))?;
     println!("Blocktime: {}", blocktime);
     Ok(())
 }

cli/src/processor/health/get.rs

@@ -12,7 +12,7 @@ pub fn get(client: &Arc<Client>) -> Result<(), CliError> {
     let health_data = cronos_sdk::account::Health::try_from(data)
         .map_err(|_err| CliError::AccountDataNotParsable(health_addr.to_string()))?;
     let blocktime =
-        cronos_sdk::blocktime(client).map_err(|err| CliError::BadClient(err.to_string()))?;
+        cronos_sdk::get_blocktime(client).map_err(|err| CliError::BadClient(err.to_string()))?;
 
     println!("  Block time: {}", blocktime);
     println!("   Last ping: {} sec", blocktime - health_data.last_ping);

cli/src/processor/task/new.rs

@@ -28,7 +28,7 @@ pub fn new(
     let exec_at = match exec_at {
         Some(v) => v,
         None => {
-            cronos_sdk::blocktime(client).map_err(|err| CliError::BadClient(err.to_string()))?
+            cronos_sdk::get_blocktime(client).map_err(|err| CliError::BadClient(err.to_string()))?
         }
     };
     let recurr = match recurr {
@@ -48,15 +48,15 @@ pub fn new(
     let schedule = cronos_sdk::account::TaskSchedule {
         exec_at,
         stop_at,
-        recurr
+        recurr,
     };
     let task_ix = cronos_sdk::instruction::task_create(
         task_pda,
         config_addr,
         daemon_addr,
         owner,
         ix,
-        schedule
+        schedule,
     );
 
     // Sign and submit

sdk/Cargo.toml

@@ -15,6 +15,8 @@ keywords = ["solana", "cronos", "sdk"]
 [dependencies]
 anchor-client = { version = "0.22.0", features = ["debug"] }
 cronos-program = { path = "../programs/cronos", features = ["no-entrypoint"], version = "0.1.6" }
+solana-account-decoder = "1.9.9"
+solana-client = "1.9.9"
 solana-client-helpers = "1.1.0"
 solana-program = "1.9.5"
 solana-sdk = "1.9.8"

sdk/src/blocktime.rs

@@ -1,24 +1,64 @@
-use std::{str::FromStr, sync::Arc};
-
 use anchor_client::ClientError;
+use solana_account_decoder::UiAccountEncoding;
+use solana_client::{pubsub_client::PubsubClient, rpc_config::RpcAccountInfoConfig};
 use solana_client_helpers::Client;
 use solana_sdk::{
+    account::Account,
     clock::{Clock, Epoch, Slot, UnixTimestamp},
+    commitment_config::CommitmentConfig,
     pubkey::Pubkey,
 };
+use std::{
+    str::FromStr,
+    sync::{
+        mpsc::{self, Receiver},
+        Arc,
+    },
+    thread,
+};
 
-pub fn blocktime(client: &Arc<Client>) -> Result<i64, ClientError> {
+pub fn get_blocktime(client: &Arc<Client>) -> Result<i64, ClientError> {
     let clock = fetch_clock_sysvar(client).unwrap();
     Ok(clock.unix_timestamp)
 }
 
+pub fn monitor_blocktime(url: String) -> Receiver<i64> {
+    let (blocktime_sender, blocktime_receiver) = mpsc::channel::<i64>();
+    thread::spawn(move || {
+        let mut latest_blocktime: i64 = 0;
+        let clock_addr = Pubkey::from_str("SysvarC1ock11111111111111111111111111111111").unwrap();
+        let (_ws_client, clock_receiver) = PubsubClient::account_subscribe(
+            url.as_str(),
+            &clock_addr,
+            Some(RpcAccountInfoConfig {
+                encoding: Some(UiAccountEncoding::Base64),
+                commitment: Some(CommitmentConfig::processed()),
+                data_slice: None,
+            }),
+        )
+        .unwrap();
+
+        for ui_account_response in clock_receiver {
+            let ui_account = ui_account_response.value;
+            let account = ui_account.decode::<Account>().unwrap();
+            let clock = deserialize_clock(account.data);
+            let blocktime = clock.unix_timestamp;
+            if blocktime > latest_blocktime {
+                latest_blocktime = blocktime;
+                blocktime_sender.send(blocktime).unwrap()
+            }
+        }
+    });
+    return blocktime_receiver;
+}
+
 fn fetch_clock_sysvar(client: &Arc<Client>) -> Result<Clock, ClientError> {
     let clock_addr = Pubkey::from_str("SysvarC1ock11111111111111111111111111111111").unwrap();
     let data = client.get_account_data(&clock_addr)?;
-    Ok(get_clock_from_data(data))
+    Ok(deserialize_clock(data))
 }
 
-fn get_clock_from_data(data: Vec<u8>) -> Clock {
+fn deserialize_clock(data: Vec<u8>) -> Clock {
     Clock {
         slot: Slot::from_le_bytes(data.as_slice()[0..8].try_into().unwrap()),
         epoch_start_timestamp: UnixTimestamp::from_le_bytes(

sdk/src/lib.rs

@@ -1,7 +1,7 @@
 mod blocktime;
 pub mod instruction;
 
-pub use blocktime::blocktime;
+pub use blocktime::*;
 pub use cronos_program::errors;
 pub use cronos_program::pda;
 pub use cronos_program::state as account;