/* This file is part of DarkFi (https://dark.fi) * * Copyright (C) 2020-2024 Dyne.org foundation * * This program is free software: you can redistribute it and/or modify * it under the terms of the GNU Affero General Public License as * published by the Free Software Foundation, either version 3 of the * License, or (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU Affero General Public License for more details. * * You should have received a copy of the GNU Affero General Public License * along with this program. If not, see . */ use std::sync::Arc; use log::{error, info, warn}; use tinyjson::JsonValue; use url::Url; use darkfi::{ blockchain::BlockInfo, rpc::{ client::RpcClient, jsonrpc::{ ErrorCode::{InternalError, InvalidParams, ParseError}, JsonError, JsonRequest, JsonResponse, JsonResult, }, }, system::{Publisher, StoppableTask, StoppableTaskPtr}, util::encoding::base64, Error, Result, }; use darkfi_serial::deserialize_async; use crate::Explorerd; impl Explorerd { // Queries darkfid for a block with given height. async fn get_block_by_height(&self, height: u32) -> Result { let params = self .darkfid_daemon_request( "blockchain.get_block", &JsonValue::Array(vec![JsonValue::String(height.to_string())]), ) .await?; let param = params.get::().unwrap(); let bytes = base64::decode(param).unwrap(); let block = deserialize_async(&bytes).await?; Ok(block) } /// Syncs the blockchain starting from the last synced block. /// If reset flag is provided, all tables are reset, and start syncing from beginning. pub async fn sync_blocks(&self, reset: bool) -> Result<()> { // Grab last synced block height let mut height = match self.db.last_block() { Ok(Some((height, _))) => height, Ok(None) => 0, Err(e) => { return Err(Error::DatabaseError(format!( "[sync_blocks] Retrieving last synced block failed: {:?}", e ))); } }; // If last synced block is genesis (0) or reset flag // has been provided we reset, otherwise continue with // the next block height if height == 0 || reset { self.db.reset_blocks()?; height = 0; } else { height += 1; }; loop { let rep = match self .darkfid_daemon_request("blockchain.last_known_block", &JsonValue::Array(vec![])) .await { Ok(r) => r, Err(e) => { let error_message = format!("[sync_blocks] RPC client request failed: {:?}", e); error!(target: "blockchain-explorer::rpc_blocks::sync_blocks", "{}", error_message); return Err(Error::DatabaseError(error_message)); } }; let last = *rep.get::().unwrap() as u32; info!(target: "blockchain-explorer::rpc_blocks::sync_blocks", "Requested to sync from block number: {height}"); info!(target: "blockchain-explorer::rpc_blocks::sync_blocks", "Last known block number reported by darkfid: {last}"); // Already synced last known block if height > last { return Ok(()) } while height <= last { let block = match self.get_block_by_height(height).await { Ok(r) => r, Err(e) => { let error_message = format!("[sync_blocks] RPC client request failed: {:?}", e); error!(target: "blockchain-explorer::rpc_blocks::sync_blocks", "{}", error_message); return Err(Error::DatabaseError(error_message)); } }; if let Err(e) = self.db.put_block(&block).await { let error_message = format!("[sync_blocks] Put block failed: {:?}", e); error!(target: "blockchain-explorer::rpc_blocks::sync_blocks", "{}", error_message); return Err(Error::DatabaseError(error_message)); }; info!(target: "blockchain-explorer::rpc_blocks::sync_blocks", "Synced block {height}"); height += 1; } } } // RPCAPI: // Queries the database to retrieve last N blocks. // Returns an array of readable blocks upon success. // // **Params:** // * `array[0]`: `u16` Number of blocks to retrieve (as string) // // **Returns:** // * Array of `BlockRecord` encoded into a JSON. // // --> {"jsonrpc": "2.0", "method": "blocks.get_last_n_blocks", "params": ["10"], "id": 1} // <-- {"jsonrpc": "2.0", "result": {...}, "id": 1} pub async fn blocks_get_last_n_blocks(&self, id: u16, params: JsonValue) -> JsonResult { let params = params.get::>().unwrap(); if params.len() != 1 || !params[0].is_string() { return JsonError::new(InvalidParams, None, id).into() } // Extract the number of last blocks to retrieve from parameters let n = match params[0].get::().unwrap().parse::() { Ok(v) => v, Err(_) => return JsonError::new(ParseError, None, id).into(), }; // Fetch the blocks and handle potential errors let blocks_result = match self.db.get_last_n(n) { Ok(blocks) => blocks, Err(e) => { error!(target: "blockchain-explorer::rpc_blocks::blocks_get_last_n_blocks", "Failed fetching blocks: {}", e); return JsonError::new(InternalError, None, id).into(); } }; // Transform blocks to json and return result if blocks_result.is_empty() { JsonResponse::new(JsonValue::Array(vec![]), id).into() } else { let json_blocks: Vec = blocks_result.into_iter().map(|block| block.to_json_array()).collect(); JsonResponse::new(JsonValue::Array(json_blocks), id).into() } } // RPCAPI: // Queries the database to retrieve blocks in provided heights range. // Returns an array of readable blocks upon success. // // **Params:** // * `array[0]`: `u32` Starting height (as string) // * `array[1]`: `u32` Ending height range (as string) // // **Returns:** // * Array of `BlockRecord` encoded into a JSON. // // --> {"jsonrpc": "2.0", "method": "blocks.get_blocks_in_heights_range", "params": ["10", "15"], "id": 1} // <-- {"jsonrpc": "2.0", "result": {...}, "id": 1} pub async fn blocks_get_blocks_in_heights_range( &self, id: u16, params: JsonValue, ) -> JsonResult { let params = params.get::>().unwrap(); if params.len() != 2 || !params[0].is_string() || !params[1].is_string() { return JsonError::new(InvalidParams, None, id).into() } let start = match params[0].get::().unwrap().parse::() { Ok(v) => v, Err(_) => return JsonError::new(ParseError, None, id).into(), }; let end = match params[1].get::().unwrap().parse::() { Ok(v) => v, Err(_) => return JsonError::new(ParseError, None, id).into(), }; if start > end { return JsonError::new(ParseError, None, id).into() } // Fetch the blocks and handle potential errors let blocks_result = match self.db.get_by_range(start, end) { Ok(blocks) => blocks, Err(e) => { error!(target: "blockchain-explorer::rpc_blocks::blocks_get_blocks_in_height_range", "Failed fetching blocks: {}", e); return JsonError::new(InternalError, None, id).into(); } }; // Transform blocks to json and return result if blocks_result.is_empty() { JsonResponse::new(JsonValue::Array(vec![]), id).into() } else { let json_blocks: Vec = blocks_result.into_iter().map(|block| block.to_json_array()).collect(); JsonResponse::new(JsonValue::Array(json_blocks), id).into() } } // RPCAPI: // Queries the database to retrieve the block corresponding to the provided hash. // Returns the readable block upon success. // // **Params:** // * `array[0]`: `String` Block header hash // // **Returns:** // * `BlockRecord` encoded into a JSON. // // --> {"jsonrpc": "2.0", "method": "blocks.get_block_by_hash", "params": ["5cc...2f9"], "id": 1} // <-- {"jsonrpc": "2.0", "result": {...}, "id": 1} pub async fn blocks_get_block_by_hash(&self, id: u16, params: JsonValue) -> JsonResult { let params = params.get::>().unwrap(); if params.len() != 1 || !params[0].is_string() { return JsonError::new(InvalidParams, None, id).into() } // Extract header hash from params, returning error if not provided let header_hash = match params[0].get::() { Some(hash) => hash, None => return JsonError::new(InvalidParams, None, id).into(), }; // Fetch and transform block to json, handling any errors and returning the result match self.db.get_block_by_hash(header_hash) { Ok(Some(block)) => JsonResponse::new(block.to_json_array(), id).into(), Ok(None) => JsonResponse::new(JsonValue::Array(vec![]), id).into(), Err(e) => { error!(target: "blockchain-explorer::rpc_blocks", "Failed fetching block: {:?}", e); JsonError::new(InternalError, None, id).into() } } } } /// Subscribes to darkfid's JSON-RPC notification endpoint that serves /// new finalized blocks. Upon receiving them, store them to the database. pub async fn subscribe_blocks( explorer: Arc, endpoint: Url, ex: Arc>, ) -> Result<(StoppableTaskPtr, StoppableTaskPtr)> { let rep = explorer .darkfid_daemon_request("blockchain.last_known_block", &JsonValue::Array(vec![])) .await?; let last_known = *rep.get::().unwrap() as u32; let last_synced = match explorer.db.last_block() { Ok(Some((height, _))) => height, Ok(None) => 0, Err(e) => { return Err(Error::DatabaseError(format!( "[subscribe_blocks] Retrieving last synced block failed: {e:?}" ))) } }; if last_known != last_synced { warn!(target: "blockchain-explorer::rpc_blocks::subscribe_blocks", "Warning: Last synced block is not the last known block."); warn!(target: "blockchain-explorer::rpc_blocks::subscribe_blocks", "You should first fully sync the blockchain, and then subscribe"); return Err(Error::DatabaseError( "[subscribe_blocks] Blockchain not fully synced".to_string(), )) } info!(target: "blockchain-explorer::rpc_blocks::subscribe_blocks", "Subscribing to receive notifications of incoming blocks"); let publisher = Publisher::new(); let subscription = publisher.clone().subscribe().await; let _ex = ex.clone(); let subscriber_task = StoppableTask::new(); subscriber_task.clone().start( // Weird hack to prevent lifetimes hell async move { let ex = _ex.clone(); let rpc_client = RpcClient::new(endpoint, ex).await?; let req = JsonRequest::new("blockchain.subscribe_blocks", JsonValue::Array(vec![])); rpc_client.subscribe(req, publisher).await }, |res| async move { match res { Ok(()) => { /* Do nothing */ } Err(e) => error!(target: "blockchain-explorer::rpc_blocks::subscribe_blocks", "[subscribe_blocks] JSON-RPC server error: {e:?}"), } }, Error::RpcServerStopped, ex.clone(), ); info!(target: "blockchain-explorer::rpc_blocks::subscribe_blocks", "Detached subscription to background"); info!(target: "blockchain-explorer::rpc_blocks::subscribe_blocks", "All is good. Waiting for block notifications..."); let listener_task = StoppableTask::new(); listener_task.clone().start( // Weird hack to prevent lifetimes hell async move { loop { match subscription.receive().await { JsonResult::Notification(n) => { info!(target: "blockchain-explorer::rpc_blocks::subscribe_blocks", "Got Block notification from darkfid subscription"); if n.method != "blockchain.subscribe_blocks" { return Err(Error::UnexpectedJsonRpc(format!( "Got foreign notification from darkfid: {}", n.method ))) } // Verify parameters if !n.params.is_array() { return Err(Error::UnexpectedJsonRpc( "Received notification params are not an array".to_string(), )) } let params = n.params.get::>().unwrap(); if params.is_empty() { return Err(Error::UnexpectedJsonRpc( "Notification parameters are empty".to_string(), )) } for param in params { let param = param.get::().unwrap(); let bytes = base64::decode(param).unwrap(); let block_data: BlockInfo = match deserialize_async(&bytes).await { Ok(b) => b, Err(e) => { return Err(Error::UnexpectedJsonRpc(format!( "[subscribe_blocks] Deserializing block failed: {e:?}" ))) }, }; let header_hash = block_data.hash().to_string(); info!(target: "blockchain-explorer::rpc_blocks::subscribe_blocks", "======================================="); info!(target: "blockchain-explorer::rpc_blocks::subscribe_blocks", "Block header: {header_hash}"); info!(target: "blockchain-explorer::rpc_blocks::subscribe_blocks", "======================================="); info!(target: "blockchain-explorer::rpc_blocks::subscribe_blocks", "Deserialized successfully. Storing block..."); if let Err(e) = explorer.db.put_block(&block_data).await { return Err(Error::DatabaseError(format!( "[subscribe_blocks] Put block failed: {e:?}" ))) } } } JsonResult::Error(e) => { // Some error happened in the transmission return Err(Error::UnexpectedJsonRpc(format!("Got error from JSON-RPC: {e:?}"))) } x => { // And this is weird return Err(Error::UnexpectedJsonRpc(format!( "Got unexpected data from JSON-RPC: {x:?}" ))) } } }; }, |res| async move { match res { Ok(()) => { /* Do nothing */ } Err(e) => error!(target: "blockchain-explorer::rpc_blocks::subscribe_blocks", "[subscribe_blocks] JSON-RPC server error: {e:?}"), } }, Error::RpcServerStopped, ex, ); Ok((subscriber_task, listener_task)) }