victorvobis/watcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bless

Browse Source
This commit is contained in:
victor 2025-09-18 18:55:38 +02:00
parent e5ffe31ca7
commit 2cb4e45235
47 changed files with 1243 additions and 465 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
native_client/Cargo.lock generated
View File

@ -696,6 +696,56 @@ dependencies = [
"cfg-if",
]
[[package]]
name = "crossbeam"
version = "0.8.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1137cd7e7fc0fb5d3c5a8678be38ec56e819125d8d7907411fe24ccb943faca8"
dependencies = [
"crossbeam-channel",
"crossbeam-deque",
"crossbeam-epoch",
"crossbeam-queue",
"crossbeam-utils",
]
[[package]]
name = "crossbeam-channel"
version = "0.5.15"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "82b8f8f868b36967f9606790d1903570de9ceaf870a7bf9fbbd3016d636a2cb2"
dependencies = [
"crossbeam-utils",
]
[[package]]
name = "crossbeam-deque"
version = "0.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9dd111b7b7f7d55b72c0a6ae361660ee5853c9af73f70c3c2ef6858b950e2e51"
dependencies = [
"crossbeam-epoch",
"crossbeam-utils",
]
[[package]]
name = "crossbeam-epoch"
version = "0.9.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e"
dependencies = [
"crossbeam-utils",
]
[[package]]
name = "crossbeam-queue"
version = "0.3.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0f58bbc28f91df819d0aa2a2c00cd19754769c2fad90579b3592b1c9ba7a3115"
dependencies = [
"crossbeam-utils",
]
[[package]]
name = "crossbeam-utils"
version = "0.8.21"
@ -2084,15 +2134,18 @@ name = "native_client"
version = "0.1.0"
dependencies = [
"bincode",
"crossbeam",
"eframe",
"egui",
"futures-util",
"hex",
"once_cell",
"serde",
"shared",
"thiserror 2.0.16",
"tokio",
"tokio-tungstenite",
"uuid",
"vlogger",
"watchlet",
]
@ -3128,6 +3181,7 @@ dependencies = [
"bincode",
"clap",
"hex",
"k256",
"serde",
"serde_json",
"sha2",
@ -3599,6 +3653,17 @@ version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
[[package]]
name = "uuid"
version = "1.18.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2f87b8aa10b915a06587d0dec516c282ff295b475d94abf425d62b57710070a2"
dependencies = [
"getrandom 0.3.3",
"js-sys",
"wasm-bindgen",
]
[[package]]
name = "version_check"
version = "0.9.5"

3
native_client/Cargo.toml
View File

@ -23,3 +23,6 @@ tokio = { version = "1.47.1", features = ["full", "rt-multi-thread"] }
tokio-tungstenite = "0.27.0"
thiserror = "2.0.16"
futures-util = "0.3.31"
uuid = { version = "1.18.1", features = ["v4"] }
crossbeam = "0.8.4"
once_cell = "1.21.3"

45
native_client/src/backend/wallet.rs
View File

@ -1,9 +1,8 @@
use shared::WsClientRequest;
use tokio::sync::mpsc::{Receiver, Sender};
use crossbeam::channel::Sender;
use vlogger::*;
use watchlet::{FileStorage, Wallet, WalletManager};
use crate::bus::event::{subscribe_system_event, SystemEvent};
use crate::messages::client::ClientMessage;
use crate::messages::error::SystemError;
use crate::messages::frontend::FrontendMessage;
@ -12,7 +11,7 @@ use crate::constants::DATA_DIR_PATH;
use std::path;
#[derive(Debug)]
#[derive(Debug, serde::Deserialize, serde::Serialize, Clone)]
pub struct WalletState {
pub address: String,
pub balance: u64,
@ -20,7 +19,7 @@ pub struct WalletState {
pub struct WalletService {
exit: bool,
receiver: Receiver<BackendMessage>,
receiver: crossbeam::channel::Receiver<BackendMessage>,
fr_sender: Sender<FrontendMessage>,
ws_sender: Sender<ClientMessage>,
wallet_manager: WalletManager<FileStorage>,
@ -28,7 +27,7 @@ pub struct WalletService {
impl WalletService {
pub fn new(
receiver: Receiver<BackendMessage>,
receiver: crossbeam::channel::Receiver<BackendMessage>,
fr_sender: Sender<FrontendMessage>,
ws_sender: Sender<ClientMessage>,
) -> Self {
@ -49,6 +48,11 @@ impl WalletService {
self.wallet_manager.wallet()
}
pub fn exit(&mut self) {
log!(INFO, "Shutting down WalletService",);
self.exit = true
}
pub fn update(&mut self) {
log!(DEBUG, "Dummy Update");
}
@ -64,28 +68,49 @@ impl WalletService {
}
pub async fn handle_cmd(&mut self, cmd: BackendMessage) {
log!(DEBUG, "Received Command: {:#?}", cmd);
match cmd {
BackendMessage::StateRequest => {
let msg = match self.get_state() {
Some(state) => FrontendMessage::StateResponse(state),
None => FrontendMessage::Error(SystemError::WalletNotLoaded),
};
if let Err(e) = self.sender.send(msg).await {
if let Err(e) = self.fr_sender.send(msg) {
log!(ERROR, "Failed to send FrontendMessage: {e}");
}
},
BackendMessage::Transaction(_transaction) => {
},
BackendMessage::Shutdown => self.exit = true,
}
}
pub async fn handle_system_event(&mut self, event: SystemEvent) {
match event {
SystemEvent::Shutdown => self.exit(),
_ => {}
}
}
pub async fn run(&mut self) {
let mut event_bus = subscribe_system_event();
while !self.exit {
if let Some(cmd) = self.receiver.recv().await {
tokio::select! {
result = tokio::task::spawn_blocking({
let rx = self.receiver.clone();
move || rx.recv()
}) => {
if let Ok(Ok(cmd)) = result {
self.handle_cmd(cmd).await;
}
}
result = event_bus.recv() => {
if let Ok(event) = result {
self.handle_system_event(event).await;
}
}
}
}
}
}

131
native_client/src/backend/ws_client.rs
View File

@ -1,47 +1,124 @@
use tokio::sync::mpsc::{Receiver, Sender};
use std::collections::HashMap;
use std::net::SocketAddr;
use tokio::net::TcpStream;
use crossbeam::channel::{Receiver, Sender};
use futures_util::{SinkExt, StreamExt};
use tokio_tungstenite::tungstenite::Message;
use shared::{ WsClientRequest, WsClientResponse };
use tokio_tungstenite::WebSocketStream;
use tokio_tungstenite::{tungstenite::{self, Message }, MaybeTlsStream};
use thiserror::Error;
use uuid::Uuid;
use vlogger::*;
use crate::bus::event::{publish_system_event, subscribe_system_event, SystemEvent};
use crate::messages::{backend::BackendMessage, client::ClientMessage };
use crate::messages::error::WsClientError;
pub async fn run_client(tx: Sender<BackendMessage>, mut rx: Receiver<ClientMessage>) -> Result<(), WsClientError> {
let (mut socket, _response) = tokio_tungstenite::connect_async("ws://localhost:9001").await?;
static BINCODE_CONFIG: bincode::config::Configuration = bincode::config::standard();
let msg = bincode::encode_to_vec(WsClientRequest::Ping, bincode::config::standard()).unwrap();
socket.send(msg.into());
#[derive(Debug, Error)]
pub enum WsClientError {
#[error("WS Error: {0}")]
WsError(#[from] tokio_tungstenite::tungstenite::Error),
#[error("Decode Error: {0}")]
Decode(#[from] bincode::error::DecodeError),
#[error("Encode Error: {0}")]
Encode(#[from] bincode::error::EncodeError),
}
let bincode_config = bincode::config::standard();
pub struct WsClient {
back_tx: Sender<BackendMessage>,
rx: Receiver<ClientMessage>,
nodes: HashMap<Uuid, SocketAddr>,
socket: WebSocketStream<MaybeTlsStream<TcpStream>>,
exit: bool,
}
loop {
tokio::select! {
cl_msg_res = rx.recv() => {
if let Some(cmd) = cl_msg_res {
impl WsClient {
pub async fn new(
back_tx: Sender<BackendMessage>,
rx: Receiver<ClientMessage>,
) -> Result<Self, WsClientError> {
let (socket, _response) = tokio_tungstenite::connect_async("ws://localhost:9001").await?;
publish_system_event(SystemEvent::WsClientConnect);
Ok(Self {
back_tx,
rx,
nodes: HashMap::new(),
socket,
exit: false,
})
}
}
ws_msg_res = socket.next() => {
if let Some(cmd) = ws_msg_res {
match cmd {
Ok(msg) => {
match msg {
Message::Text(b) => {
pub async fn handle_cmd(&mut self, cmd: ClientMessage) {
}
Message::Binary(mes_bytes) => {
let (msg, _) = bincode::decode_from_slice::<WsClientResponse, _>(&mes_bytes, bincode_config)?;
match msg {
WsClientResponse::Pong => {}
}
fn exit(&mut self) {
log!(INFO, "Shutting down Websocket Client");
self.exit = true;
}
pub async fn handle_system_event(&mut self, event: SystemEvent) {
match event {
SystemEvent::Shutdown => self.exit(),
_ => {}
}
}
pub async fn run(&mut self) -> Result<(), WsClientError> {
let mut event_bus = subscribe_system_event();
while !self.exit {
tokio::select! {
event_res = event_bus.recv() => {
if let Ok(event) = event_res {
self.handle_system_event(event).await;
}
}
cl_msg_res = tokio::task::spawn_blocking({
let rx = self.rx.clone();
move || rx.recv()
}) => {
if let Ok(Ok(cmd)) = cl_msg_res {
self.execute_command(cmd).await?;
}
}
ws_msg_res = self.socket.next() => {
if let Some(cmd) = ws_msg_res {
match cmd {
Ok(msg) => self.handle_ws_message(msg).await?,
Err(e) => self.handle_ws_error(e).await?,
}
}
}
}
}
Ok(self.socket.close(None).await?)
}
async fn handle_ws_error(&mut self, e: tungstenite::Error) -> Result<(), WsClientError> {
match e {
tungstenite::Error::ConnectionClosed => {
self.socket.close(None).await?;
publish_system_event(SystemEvent::WsClientDisconnect)
},
Err(e) => {}
tungstenite::Error::AlreadyClosed => {},
_ => {}
}
Ok(())
}
async fn handle_ws_message(&mut self, msg: Message) -> Result<(), WsClientError> {
Ok(())
}
async fn execute_command(&mut self, cmd: ClientMessage) -> Result<(), WsClientError> {
match cmd {
ClientMessage::BroadcastTransaction(tx) => {
let bin_tx = bincode::encode_to_vec(tx, BINCODE_CONFIG)?;
self.socket.send(bin_tx.into()).await?;
Ok(())
}
}
}
}

45
native_client/src/bus/event.rs Normal file
View File

@ -0,0 +1,45 @@
use tokio::sync::broadcast::{self, Sender, Receiver};
use once_cell::sync::Lazy;
use std::sync::Arc;
#[derive(Clone)]
pub enum SystemEvent {
WsClientConnect,
WsClientDisconnect,
Shutdown,
}
pub struct EventBus {
sender: Sender<SystemEvent>,
_receiver: Receiver<SystemEvent>,
}
impl EventBus {
pub fn new() -> Self{
let (tx, rx) = broadcast::channel::<SystemEvent>(100);
Self {
sender: tx,
_receiver: rx,
}
}
pub fn publish(&self, msg: SystemEvent) {
if let Err(e) = self.sender.send(msg) {
eprintln!("{e}")
}
}
pub fn subscribe(&self) -> broadcast::Receiver<SystemEvent> {
self.sender.subscribe()
}
}
static SYSTEM_EVENT_BUS: Lazy<Arc<EventBus>> = Lazy::new(|| Arc::new(EventBus::new()));
pub fn publish_system_event(event: SystemEvent) {
SYSTEM_EVENT_BUS.publish(event);
}
pub fn subscribe_system_event() -> broadcast::Receiver<SystemEvent> {
SYSTEM_EVENT_BUS.subscribe()
}

3
native_client/src/frontend/display.rs
View File

@ -1,6 +1,5 @@
use std::sync::mpsc::Sender;
use crate::messages::backend::BackendMessage;
pub trait DisplayPage {
fn show(&mut self, rx: &Sender<BackendMessage>, ctx: &egui::Context);
fn show(&mut self, ctx: &egui::Context) -> Option<BackendMessage>;
}

16
native_client/src/frontend/pages/home.rs
View File

@ -1,18 +1,25 @@
use std::sync::mpsc::Sender;
use crossbeam::channel::Sender;
use crate::{frontend::DisplayPage, messages::backend::BackendMessage};
#[derive(Default)]
#[derive(serde::Deserialize, serde::Serialize)]
pub struct HomePage {
label: String,
#[serde(skip)]
value: f32,
}
impl HomePage {
pub fn new() -> Self {
Self {
label: Default::default(),
value: Default::default(),
}
}
}
impl DisplayPage for HomePage {
fn show(&mut self, _rx: &Sender<BackendMessage>, ctx: &egui::Context) {
fn show(&mut self, ctx: &egui::Context) -> Option<BackendMessage> {
egui::CentralPanel::default().show(ctx, |ui| {
ui.heading("eframe template");
@ -37,5 +44,6 @@ impl DisplayPage for HomePage {
egui::warn_if_debug_build(ui);
});
});
None
}
}

21
native_client/src/frontend/pages/tx_page.rs
View File

@ -1,17 +1,25 @@
use std::sync::mpsc::Sender;
use crossbeam::channel::Sender;
use egui::{Align, Layout, RichText};
use crate::{frontend::DisplayPage, messages::backend::BackendMessage};
#[derive(Default)]
#[derive(serde::Serialize, serde::Deserialize)]
pub struct TransactionPage {
from_to_amount: [String; 3],
focused: usize,
}
#[derive(Debug)]
impl TransactionPage {
pub fn new() -> Self {
Self {
from_to_amount: Default::default(),
focused: 0,
}
}
}
#[derive(Debug, Clone)]
pub struct TransactionRequest {
}
@ -23,7 +31,9 @@ const TX_FIELDS: &[&str] = &[
];
impl DisplayPage for TransactionPage {
fn show(&mut self, tx: &Sender<BackendMessage>, ctx: &egui::Context) {
fn show(&mut self, ctx: &egui::Context) -> Option<BackendMessage> {
let mut msg = None;
ctx.input(|input| {
if input.key_pressed(egui::Key::Tab) {
self.focused = (self.focused + 1) % TX_FIELDS.len()
@ -44,8 +54,9 @@ impl DisplayPage for TransactionPage {
}
});
if ui.button("GetState").clicked() {
tx.send(BackendMessage::StateRequest).unwrap();
msg = Some(BackendMessage::StateRequest);
}
});
msg
}
}

52
native_client/src/frontend/pages/wallet_interface.rs Normal file
View File

@ -0,0 +1,52 @@
use crate::{backend::wallet::WalletState, frontend::DisplayPage, messages::backend::BackendMessage};
#[derive(serde::Serialize, serde::Deserialize)]
pub struct WalletInterface {
wallet: Option<WalletState>,
}
impl WalletInterface {
pub fn new(wallet: Option<WalletState>) -> Self{
Self {
wallet,
}
}
pub fn set_wallet(&mut self, wallet: WalletState) {
self.wallet = Some(wallet);
}
pub fn wallet(&self) -> &Option<WalletState> {
&self.wallet
}
}
impl DisplayPage for WalletInterface {
fn show(&mut self, ctx: &egui::Context) -> Option<BackendMessage> {
egui::CentralPanel::default().show(ctx, |ui| {
ui.heading("Wallet Interface");
ui.separator();
ui.add_space(20.0);
ui.vertical_centered(|ui| {
egui::Grid::new("wallet_grid")
.num_columns(2)
.spacing([20.0, 10.0])
.striped(true)
.show(ui, |ui| {
if let Some(wallet) = &self.wallet {
ui.strong("Address:");
ui.label(format!("0x{}", &wallet.address));
ui.end_row();
ui.strong("Balance:");
ui.label(wallet.balance.to_string());
ui.end_row();
} else {
ui.spinner();
ui.label("Loading Wallet...");
}
})
})
});
None
}
}

197
native_client/src/frontend/state.rs
View File

@ -1,45 +1,96 @@
use std::{process::exit, thread::sleep, time::Duration};
use egui::{Button, Color32, Key, RichText};
use crate::messages::{backend::BackendMessage, frontend::FrontendMessage};
use crate::{backend::wallet::WalletState, bus::event::{publish_system_event, subscribe_system_event, SystemEvent}, messages::{backend::BackendMessage, frontend::FrontendMessage}};
use vlogger::*;
use super::{pages::HomePage, DisplayPage, TransactionPage};
use std::sync::mpsc::{self, Receiver, Sender};
use super::{pages::HomePage, DisplayPage, TransactionPage, wallet_interface::WalletInterface};
use crossbeam::channel::{Receiver, Sender};
#[derive(serde::Deserialize, serde::Serialize)]
#[serde(default)]
pub struct State {
page: Page,
pages: Vec<Page>,
current_page_index: usize,
side_panel: bool,
connected: bool,
wallet: Option<WalletState>,
#[serde(skip)]
receiver: Receiver<FrontendMessage>,
#[serde(skip)]
sender: Sender<BackendMessage>,
sender: crossbeam::channel::Sender<BackendMessage>,
#[serde(skip)]
event_bus: tokio::sync::broadcast::Receiver<SystemEvent>,
#[serde(skip)]
callback: Option<PageCallback>,
#[serde(skip)]
command: Option<BackendMessage>,
}
#[derive(serde::Deserialize, serde::Serialize)]
pub enum Page {
Home(HomePage),
Transaction(TransactionPage),
Wallet(WalletInterface),
}
const HOME_PAGE_INDEX: usize = 0;
const TRANSACTION_PAGE_INDEX: usize = 1;
const WALLET_INTERFACE_INDEX: usize = 2;
type PageCallback = fn(&mut State);
static PAGE_CALLBACKS: &[PageCallback] = &[
|_| {},
|_| {},
|s| {
if let Page::Wallet(w) = &mut s.pages[s.current_page_index] {
if w.wallet().is_none() {
if let Some(wallet) = &s.wallet {
w.set_wallet(wallet.clone());
s.callback = None;
}
}
}
}
];
const PAGE_INDEXES: &[(usize, &str, Option<PageCallback>)] = &[
(HOME_PAGE_INDEX, "Home", None),
(TRANSACTION_PAGE_INDEX, "Transaction", None),
(WALLET_INTERFACE_INDEX, "Wallet", Some(PAGE_CALLBACKS[WALLET_INTERFACE_INDEX]))
];
impl DisplayPage for Page {
fn show(&mut self, rx: &Sender<BackendMessage>, ctx: &egui::Context) {
fn show(&mut self, ctx: &egui::Context) -> Option<BackendMessage> {
match self {
Page::Home(home) => home.show(rx, ctx),
Page::Transaction(tx) => tx.show(rx, ctx),
Page::Home(home) => home.show(ctx),
Page::Transaction(tx) => tx.show(ctx),
Page::Wallet(wallet) => wallet.show(ctx),
}
}
}
impl Default for State {
fn default() -> Self {
let (_, front_rx) = mpsc::channel::<FrontendMessage>();
let (back_tx, _) = mpsc::channel::<BackendMessage>();
let (_, front_rx) = crossbeam::channel::unbounded::<FrontendMessage>();
let (back_tx, _) = crossbeam::channel::unbounded::<BackendMessage>();
Self {
// Example stuff:
// page: Page::Home(HomePage::default()),
page: Page::Transaction(TransactionPage::default()),
pages: vec![
Page::Home(HomePage::new()),
Page::Transaction(TransactionPage::new()),
Page::Wallet(WalletInterface::new(None)),
],
current_page_index: HOME_PAGE_INDEX,
receiver: front_rx,
sender: back_tx,
event_bus: subscribe_system_event(),
wallet: None,
callback: None,
command: None,
side_panel: true,
connected: false,
}
}
}
@ -63,6 +114,75 @@ impl State {
}
}
}
pub fn run_callback(&mut self) {
if let Some(c) = self.callback {
c(self);
}
}
pub fn run_command(&mut self) {
if let Some(cmd) = &self.command {
if let Err(e) = self.sender.send((*cmd).clone()) {
log!(ERROR, "{e}");
} else {
self.command = None;
}
}
}
pub fn handle_key_input(&self, ctx: &egui::Context) {
ctx.input(|input| {
if input.key_pressed(Key::Escape) {
publish_system_event(SystemEvent::Shutdown);
sleep(Duration::from_millis(100));
exit(1);
};
});
}
pub fn home(&mut self) -> &mut Page {
&mut self.pages[HOME_PAGE_INDEX]
}
pub fn wallet(&mut self) -> &mut Page {
&mut self.pages[WALLET_INTERFACE_INDEX]
}
pub fn transaction(&mut self) -> &mut Page {
&mut self.pages[TRANSACTION_PAGE_INDEX]
}
pub fn current_page(&mut self) -> &mut Page {
&mut self.pages[self.current_page_index]
}
pub fn listen(&mut self) {
if let Ok(msg) = self.event_bus.try_recv() {
match msg {
SystemEvent::Shutdown => {
log!(DEBUG, "Received Wallet State Response");
}
SystemEvent::WsClientConnect => {
self.connected = true;
}
SystemEvent::WsClientDisconnect => {
self.connected = false;
}
}
}
if let Ok(msg) = self.receiver.try_recv() {
match msg {
FrontendMessage::StateResponse(state) => {
log!(DEBUG, "Received Wallet State Response");
self.wallet = Some(state)
}
FrontendMessage::Error(e) => {
log!(ERROR, "Received Error {e:?}")
}
}
}
}
}
impl eframe::App for State {
@ -71,49 +191,44 @@ impl eframe::App for State {
}
fn update(&mut self, ctx: &egui::Context, _frame: &mut eframe::Frame) {
if let Ok(msg) = self.receiver.try_recv() {
match msg {
FrontendMessage::StateResponse(state) => {
log!(DEBUG, "Recived: {:#?}", state);
self.handle_key_input(ctx);
self.run_command();
self.listen();
self.run_callback();
egui::TopBottomPanel::top("Menu").show(ctx, |ui| {
ui.set_height(30.);
ui.horizontal_centered(|ui| {
if ui.button("☰").clicked() {
self.side_panel = !self.side_panel;
}
FrontendMessage::Error(e) => {
log!(ERROR, "Received Error {e:?}")
ui.with_layout(egui::Layout::right_to_left(egui::Align::Center), |ui| {
if self.connected {
ui.label(RichText::new("Connected").color(Color32::GREEN))
} else {
ui.label(RichText::new("Disconnected").color(Color32::RED))
}
}
}
egui::TopBottomPanel::top("top_panel").show(ctx, |ui| {
ctx.input(|input| {
if input.key_pressed(Key::Escape) {
self.sender.send(BackendMessage::Shutdown).unwrap();
std::process::exit(1)
};
});
egui::MenuBar::new().ui(ui, |ui| {
egui::widgets::global_theme_preference_buttons(ui);
})
});
});
if self.side_panel {
egui::SidePanel::left("menu panel").show(ctx, |ui| {
ui.set_width(150.0);
ui.heading("Navigation");
let available_width = ui.available_width();
for (i, s, c) in PAGE_INDEXES {
if ui.add_sized(
[available_width, 24.0],
Button::new(RichText::new("Home").size(14.0))
Button::new(RichText::new(*s).size(14.0))
.right_text("")
.fill(Color32::TRANSPARENT)
).clicked() {
self.page = Page::Home(HomePage::default())
self.current_page_index = *i;
self.callback = *c;
}
if ui.add_sized(
[available_width, 24.0],
Button::new(RichText::new("Transaction").size(14.0))
.right_text("")
.fill(Color32::TRANSPARENT)
).clicked() {
self.page = Page::Transaction(TransactionPage::default())
}
});
self.page.show(&self.sender, ctx)
}
self.current_page().show(ctx);
}
}

8
native_client/src/lib.rs
View File

@ -7,11 +7,11 @@ pub mod frontend {
pub mod display;
pub use display::*;
pub mod message;
pub use pages::*;
pub mod pages {
pub mod wallet_interface;
pub mod tx_page;
pub use tx_page::*;
@ -27,6 +27,10 @@ pub mod messages {
pub mod client;
}
pub mod bus {
pub mod event;
}
pub mod backend {
pub mod wallet;
pub mod ws_client;

55
native_client/src/main.rs
View File

@ -2,7 +2,7 @@
#![cfg_attr(not(debug_assertions), windows_subsystem = "windows")]
use native_client::{backend::{wallet::WalletService, ws_client}, frontend, messages::{backend::BackendMessage, client::ClientMessage, frontend::FrontendMessage}};
use tokio::sync::mpsc::{self, Receiver, Sender};
use crossbeam::channel::{Sender, Receiver};
fn dispatch_frontend(
receiver: Receiver<FrontendMessage>,
@ -18,24 +18,49 @@ fn dispatch_frontend(
eframe::run_native(
"eframe template",
native_options,
Box::new(|cc| Ok(Box::new(frontend::State::new(cc, receiver, sender)))),
Box::new(|cc|
{
let mut style = (*cc.egui_ctx.style()).clone();
style.text_styles = [
(egui::TextStyle::Heading, egui::FontId::new(24.0, egui::FontFamily::Proportional)),
(egui::TextStyle::Body, egui::FontId::new(18.0, egui::FontFamily::Proportional)),
(egui::TextStyle::Button, egui::FontId::new(16.0, egui::FontFamily::Proportional)),
(egui::TextStyle::Small, egui::FontId::new(12.0, egui::FontFamily::Proportional)),
].into();
cc.egui_ctx.set_style(style);
Ok(Box::new(frontend::State::new(cc, receiver, sender)))
}
)
)
}
#[tokio::main]
async fn main() -> Result<(), std::io::Error> {
let (front_tx, front_rx) = mpsc::channel::<FrontendMessage>(100);
let (back_tx, back_rx) = mpsc::channel::<BackendMessage>(100);
let (client_tx, client_rx) = mpsc::channel::<ClientMessage>(100);
fn main() -> Result<(), std::io::Error> {
let (front_tx, front_rx) = crossbeam::channel::unbounded::<FrontendMessage>();
let (back_tx, back_rx) = crossbeam::channel::unbounded::<BackendMessage>();
let (client_tx, client_rx) = crossbeam::channel::unbounded::<ClientMessage>();
let backend_handle = std::thread::spawn(|| {
let mut wallet_service = WalletService::new(back_rx, front_tx);
wallet_service.run();
});
let ws_client_handle = std::thread::spawn({
let tx = back_tx.clone();
|| ws_client::run_client(tx, client_rx)
let backend_handle = std::thread::spawn({
let back_tx = back_tx.clone();
move || {
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on({
let back_tx = back_tx.clone();
let front_tx = front_tx.clone();
async {
tokio::join!(
async {
let mut wallet_service = WalletService::new(back_rx, front_tx, client_tx);
wallet_service.run().await;
},
async {
let mut ws_client = ws_client::WsClient::new(back_tx, client_rx).await.unwrap();
ws_client.run().await.unwrap();
}
)
}
})
}
});
let _res = dispatch_frontend(front_rx, back_tx);

3
native_client/src/messages/backend.rs
View File

@ -1,8 +1,7 @@
use crate::frontend::TransactionRequest;
#[derive(Debug)]
#[derive(Debug, Clone)]
pub enum BackendMessage {
StateRequest,
Transaction(TransactionRequest),
Shutdown,
}

4
native_client/src/messages/client.rs
View File

@ -1,3 +1,5 @@
use shared::blockchain_core::Transaction;
pub enum ClientMessage {
Shutdown,
BroadcastTransaction(Transaction),
}

10
native_client/src/messages/error.rs
View File

@ -4,13 +4,3 @@ use thiserror::Error;
pub enum SystemError {
WalletNotLoaded
}
#[derive(Debug, Error)]
pub enum WsClientError {
#[error("WS Error: {0}")]
WsError(#[from] tokio_tungstenite::tungstenite::Error),
#[error("Decode Error: {0}")]
DecodeError(#[from] bincode::error::DecodeError),
#[error("Encode Error: {0}")]
EncodeError(#[from] bincode::error::EncodeError),
}

12
node/Cargo.lock generated
View File

@ -310,7 +310,7 @@ dependencies = [
"ring",
"secp256k1",
"serde",
"serde_json",
"serde-big-array",
"sha2",
"shared",
"sled",
@ -2120,6 +2120,15 @@ dependencies = [
"serde_derive",
]
[[package]]
name = "serde-big-array"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "11fc7cc2c76d73e0f27ee52abbd64eec84d46f370c88371120433196934e4b7f"
dependencies = [
"serde",
]
[[package]]
name = "serde_derive"
version = "1.0.219"
@ -2183,6 +2192,7 @@ dependencies = [
"bincode",
"clap",
"hex",
"k256",
"serde",
"serde_json",
"sha2",

2
node/Cargo.toml
View File

@ -8,7 +8,6 @@ chrono = "0.4.41"
clap = { version = "4.5.45", features = ["derive"] }
hex = "0.4.3"
serde = { version = "1.0.219", features = ["derive"] }
serde_json = "1.0.143"
sha2 = "0.10.9"
thiserror = "2.0.16"
tokio = { version = "1.47.1", features = ["rt-multi-thread", "net", "sync", "time", "macros"] }
@ -33,3 +32,4 @@ shared = { path = "../shared", features = ["node"] }
watchlet = { path = "../watchlet" }
cli-renderer = { path = "../cli-renderer" }
tiny_http = "0.12.0"
serde-big-array = "0.5.1"

6
node/src/api/server.rs
View File

@ -1,10 +1,10 @@
#![allow(dead_code)]
use tiny_http::{ Method, Response, Header };
use thiserror::Error;
use crate::{ log, PROJECT_PATH };
use vlogger::*;
use std::io::Cursor;
use std::fs;
#[derive(Error, Debug)]
@ -61,7 +61,7 @@ fn internal_server_error(e: String) -> Response<std::io::Cursor<Vec<u8>>> {
type JSON = std::io::Cursor<Vec<u8>>;
fn handle_api(endpoint: &str) -> Result<Response<JSON>, HttpServerError>{
fn handle_api(_endpoint: &str) -> Result<Response<JSON>, HttpServerError>{
Ok(Response::from_string("Hello World".to_string()))
}

4
node/src/args.rs
View File

@ -1,6 +1,6 @@
use std::net::SocketAddr;
use shared::core;
use shared::blockchain_core;
use cli_renderer::RenderLayoutKind;
use clap::{Parser, Subcommand};
@ -44,7 +44,7 @@ pub enum CliCommand {
/// Make a Transaction
#[command(name = "tx")]
Transaction(core::Transaction),
Transaction(blockchain_core::Transaction),
/// Start new TcpListner on Addr
#[command(name = "listen")]

12
node/src/cli.rs
View File

@ -1,5 +1,6 @@
use crate::args::*;
use shared::core::ChainData;
use crate::network::NodeId;
use shared::blockchain_core::ChainData;
use vlogger::*;
use crate::executor::ExecutorCommand;
use crate::node::*;
@ -11,7 +12,7 @@ pub fn handle_peer_command(cmd: CliPeerCommand) -> NodeCommand {
match cmd {
CliPeerCommand::List => NodeCommand::ListPeers,
CliPeerCommand::Remove { id } => NodeCommand::RemovePeer {
peer_id: id.parse::<uuid::Uuid>().unwrap(),
peer_id: NodeId(*id.parse::<uuid::Uuid>().unwrap().as_bytes()),
},
CliPeerCommand::Connect { addr } => NodeCommand::ConnectTcpPeer(addr),
}
@ -37,7 +38,10 @@ fn handle_seed_command(cmd: CliSeedCommand) -> NodeCommand {
fn handle_ping(cmd: CliPingCommand) -> NodeCommand {
match cmd {
CliPingCommand::Id { id } => NodeCommand::PingId(id),
CliPingCommand::Id { id } => {
let id = NodeId(*id.parse::<uuid::Uuid>().unwrap().as_bytes());
NodeCommand::PingId(id)
},
CliPingCommand::Addr { addr } => NodeCommand::PingAddr(addr),
}
}
@ -53,7 +57,7 @@ pub fn cli(input: &str) -> ExecutorCommand {
CliCommand::Peer { peer_cmd } => ExecutorCommand::Node(handle_peer_command(peer_cmd)),
CliCommand::Block { block_cmd } => ExecutorCommand::Node(handle_block_command(block_cmd)),
CliCommand::Transaction(tx) => {
ExecutorCommand::Node(NodeCommand::ProcessChainData(ChainData::Transaction(tx)))
ExecutorCommand::Node(NodeCommand::ProcessChainData(ChainData::NodeTransaction(tx)))
}
CliCommand::Award { address, amount } => {
let mut bytes = [0u8; 20];

94
node/src/db/database.rs
View File

@ -2,7 +2,7 @@ use bincode::{self, config::Configuration};
use sled::{self, Batch};
use std::sync::Arc;
use vlogger::*;
use shared::core::{self, Block, ChainData, Hasher, Address};
use shared::blockchain_core::{self, Address, Block, ChainData, Hasher};
use crate::{
db::error::DatabaseError,
@ -42,7 +42,7 @@ pub struct ChainDb {
}
impl ChainDb {
pub fn new(path: Option<String>, db_temp: bool) -> Result<ChainDb, DatabaseError> {
pub fn open(path: Option<String>, db_temp: bool) -> Result<ChainDb, DatabaseError> {
let path = if path.is_some() {
&path.unwrap()
} else {
@ -75,6 +75,28 @@ impl ChainDb {
}
}
pub fn replace_chain(&mut self, blocks: &Vec<Block>, tx: &Vec<ChainData>) -> Result<(), DatabaseError> {
self.db.clear()?;
let mut batch = Batch::default();
for b in blocks {
self.add_block_batch(&mut batch, &b)?;
let block_txs = tx.iter().filter(|t| b.data.contains(&t.hash())).collect::<Vec<&ChainData>>();
for b_data in block_txs {
self.add_data_batch(&mut batch, b_data, b.head().block_hash())?
}
}
self.db.apply_batch(batch)?;
Ok(())
}
pub fn build_from(&mut self, blocks: &Vec<Block>, tx: &Vec<ChainData>) -> Result<(), DatabaseError> {
let chain_height = self.get_height()?;
if blocks.len() > chain_height as usize {
return self.replace_chain(blocks, tx);
}
Ok(())
}
pub fn recover_mempool(&self) -> Result<Vec<ChainData>, DatabaseError> {
let mem_tree = self.db.scan_prefix(MEMPOOL_PREFIX);
let mut mem_vec = vec![];
@ -149,11 +171,11 @@ impl ChainDb {
}
}
pub fn get_block_by_key(&self, block_hash: &[u8]) -> Result<Option<Arc<core::Block>>, DatabaseError> {
pub fn get_block_by_key(&self, block_hash: &[u8]) -> Result<Option<Arc<blockchain_core::Block>>, DatabaseError> {
let block_hash = prefix(&BLOCK_PREFIX, block_hash);
log(msg!(DEBUG, "{:?}", hex::encode(&block_hash)));
if let Some(bin_block) = self.db.get(block_hash)? {
let (block, _) = bincode::decode_from_slice::<core::Block, _>(&bin_block, BINCODE_CONFIG)
let (block, _) = bincode::decode_from_slice::<blockchain_core::Block, _>(&bin_block, BINCODE_CONFIG)
.map_err(|e| DatabaseError::Decode(e))?;
Ok(Some(block.into()))
} else {
@ -164,7 +186,7 @@ impl ChainDb {
pub fn get_block_by_height(
&self,
height: u64,
) -> Result<Option<Arc<core::Block>>, DatabaseError> {
) -> Result<Option<Arc<blockchain_core::Block>>, DatabaseError> {
if let Some(hash) = self.db.get(prefix(&HEIGHT_TO_HASH_PREFIX, &height.to_be_bytes()))? {
let (hash_str, _) = bincode::decode_from_slice::<[u8; 32], _>(&hash, BINCODE_CONFIG)?;
Ok(self.get_block_by_key(&hash_str)?)
@ -189,28 +211,49 @@ impl ChainDb {
Ok(Arc::new(chain_data))
}
pub fn get_all_blocks(&self) -> Result<Vec<Arc<core::Block>>, DatabaseError> {
pub fn get_height(&self) -> Result<u64, DatabaseError> {
if let Some(bin_height) = self.db.get(prefix(&METADATA_PREFIX, &HEIGHT_KEY))? {
let (height, _) = bincode::decode_from_slice(&bin_height, BINCODE_CONFIG)?;
Ok(height)
} else {
Err(DatabaseError::MissingData(format!("Height")))
}
}
pub fn get_all_blocks(&self) -> Result<Vec<Arc<blockchain_core::Block>>, DatabaseError> {
self
.db
.scan_prefix(BLOCK_PREFIX)
.map(|res| -> Result<Arc<core::Block>, DatabaseError> {
.map(|res| -> Result<Arc<blockchain_core::Block>, DatabaseError> {
let (_key, value) = res?;
let (block, _size) = bincode::decode_from_slice::<core::Block, _>(&value, BINCODE_CONFIG)
let (block, _size) = bincode::decode_from_slice::<blockchain_core::Block, _>(&value, BINCODE_CONFIG)
.map_err(|e| DatabaseError::Decode(e))?;
Ok(Arc::new(block))
})
.collect()
}
pub fn add_data(&self, data: &core::ChainData) -> Result<(), DatabaseError> {
pub fn add_data_batch(&self, batch: &mut Batch, data: &blockchain_core::ChainData, block: &[u8]) -> Result<(), DatabaseError> {
let bin_data = bincode::encode_to_vec(data, BINCODE_CONFIG)?;
let data_hash = prefix(&CHAIN_DATA_PREFIX, &Hasher::hash_chain_data(data));
self.db.insert(data_hash, bin_data)?;
batch.insert(data_hash, bin_data);
let data_to_block_hash = prefix(DATA_TO_BLOCK_PREFIX, block);
batch.insert(data_to_block_hash, block);
Ok(())
}
pub fn add_block(&self, block: &core::Block) -> Result<(), DatabaseError> {
let mut db_batch = Batch::default();
pub fn add_data(&self, data: &blockchain_core::ChainData, block: &[u8]) -> Result<(), DatabaseError> {
let mut batch = Batch::default();
let bin_data = bincode::encode_to_vec(data, BINCODE_CONFIG)?;
let data_hash = prefix(&CHAIN_DATA_PREFIX, &Hasher::hash_chain_data(data));
batch.insert(data_hash, bin_data);
let data_to_block_hash = prefix(DATA_TO_BLOCK_PREFIX, block);
batch.insert(data_to_block_hash, block);
self.db.apply_batch(batch)?;
Ok(())
}
fn add_block_batch(&self, db_batch: &mut Batch, block: &blockchain_core::Block) -> Result<(), DatabaseError> {
let bin_block = bincode::encode_to_vec(block, BINCODE_CONFIG)?;
let block_prefix = prefix(&BLOCK_PREFIX, block.head().block_hash());
log(msg!(DEBUG, "{:?}", hex::encode(&block_prefix)));
@ -226,10 +269,33 @@ impl ChainDb {
);
}
db_batch.insert(prefix(&METADATA_PREFIX, &TIP_KEY), block.head().block_hash());
let bin_head = bincode::encode_to_vec(&block.head().height, BINCODE_CONFIG)?;
db_batch.insert(
prefix(&METADATA_PREFIX, &HEIGHT_KEY),
bin_head
&block.head().height.to_be_bytes()
);
Ok(())
}
pub fn add_block(&self, block: &blockchain_core::Block) -> Result<(), DatabaseError> {
let mut db_batch = Batch::default();
let bin_block = bincode::encode_to_vec(block, BINCODE_CONFIG)?;
let block_prefix = prefix(&BLOCK_PREFIX, block.head().block_hash());
db_batch.insert(block_prefix, bin_block);
db_batch.insert(
prefix(&HEIGHT_TO_HASH_PREFIX, &block.head().height.to_be_bytes()),
block.head().block_hash(),
);
for data in block.data() {
db_batch.insert(
prefix(&DATA_TO_BLOCK_PREFIX, data),
block.head().block_hash(),
);
}
db_batch.insert(prefix(&METADATA_PREFIX, &TIP_KEY), block.head().block_hash());
db_batch.insert(
prefix(&METADATA_PREFIX, &HEIGHT_KEY),
&block.head().height.to_be_bytes()
);
self.db.apply_batch(db_batch)?;
Ok(())

4
node/src/db/error.rs
View File

@ -6,9 +6,9 @@ pub enum DatabaseError {
Init(#[from] std::io::Error),
#[error("Database operation failed: {0}")]
Operation(#[from] sled::Error),
#[error("Failed to serialize data: {0}")]
#[error("Failed to encode data: {0}")]
Encode(#[from] bincode::error::EncodeError),
#[error("Failed to deserialize data: {0}")]
#[error("Failed to decode data: {0}")]
Decode(#[from] bincode::error::DecodeError),
#[error("Chain data not found for hash: {0}")]
MissingData(String),

1
node/src/error.rs
View File

@ -1,4 +1,3 @@
use crate::log;
use thiserror::Error;
#[derive(Debug, Error)]

15
node/src/lib.rs
View File

@ -81,6 +81,21 @@ pub mod seeds_constants;
use cli_renderer::{RenderCommand, RenderTarget};
pub fn print_error_chain(err: &anyhow::Error) {
let mut err_string = String::from(format!("Error: {}\n", err));
let mut source = err.source();
let mut level = 1;
while let Some(err) = source {
err_string.push_str(format!(" {}: {}\n", level, err).as_str());
source = err.source();
level += 1;
}
log(err_string)
}
pub fn log(msg: String) {
crate::bus::publish_watcher_event(watcher::WatcherCommand::Render(RenderCommand::StringToPaneId {
pane: RenderTarget::CliOutput,

13
node/src/network/connection.rs
View File

@ -1,5 +1,6 @@
use crate::executor::ExecutorCommand;
use crate::log;
use crate::network::NodeId;
use crate::node::node;
use super::ProtocolMessage;
use tokio::net;
@ -12,8 +13,8 @@ use vlogger::*;
#[allow(dead_code)]
#[derive(Debug)]
pub struct Connection {
node_id: uuid::Uuid,
peer_id: uuid::Uuid,
node_id: NodeId,
peer_id: NodeId,
stream: net::TcpStream,
exec_tx: mpsc::Sender<ExecutorCommand>,
rx: mpsc::Receiver<ProtocolMessage>,
@ -21,8 +22,8 @@ pub struct Connection {
impl Connection {
pub fn new(
node_id: uuid::Uuid,
peer_id: uuid::Uuid,
node_id: NodeId,
peer_id: NodeId,
stream: net::TcpStream,
exec_tx: mpsc::Sender<ExecutorCommand>,
rx: mpsc::Receiver<ProtocolMessage>,
@ -46,7 +47,7 @@ impl Connection {
match response_result {
Some(response) => {
if let Err(e) = Connector::send_message(&mut self.stream, &response).await {
log(msg!(ERROR, "Failed to send response to {}: {}", self.peer_id, e));
log(msg!(ERROR, "Failed to send response to {}: {}", self.peer_id.clone(), e));
break;
}
},
@ -63,7 +64,7 @@ impl Connection {
log(msg!(DEBUG, "Received Message from {}", self.peer_id));
let command = ExecutorCommand::Node(node::NodeCommand::ProcessMessage {
peer_id: self.peer_id,
peer_id: self.peer_id.clone(),
message: message.clone()
});

27
node/src/network/connector.rs
View File

@ -7,7 +7,9 @@ use vlogger::*;
use shared::print_error_chain;
use thiserror::*;
use crate::db::BINCODE_CONFIG;
use crate::log;
use crate::network::NodeId;
use super::Connection;
use crate::bus::*;
use crate::executor::ExecutorCommand;
@ -22,7 +24,7 @@ pub enum ConnectorCommand {
}
pub struct Connector {
node_id: uuid::Uuid,
node_id: NodeId,
addr: SocketAddr,
exec_tx: mpsc::Sender<ExecutorCommand>,
rx: mpsc::Receiver<ConnectorCommand>,
@ -39,7 +41,7 @@ const MAX_LISTNER_TRIES: usize = 5;
impl Connector {
pub fn new(
node_id: uuid::Uuid,
node_id: NodeId,
addr: SocketAddr,
exec_tx: mpsc::Sender<ExecutorCommand>,
rx: mpsc::Receiver<ConnectorCommand>,
@ -146,7 +148,7 @@ impl Connector {
addr: SocketAddr,
) {
let handshake = ProtocolMessage::Handshake {
peer_id: self.node_id,
peer_id: self.node_id.clone(),
version: "".to_string(),
};
match Connector::send_message(&mut stream, &handshake).await {
@ -168,7 +170,7 @@ impl Connector {
let cmd = ExecutorCommand::Node(node::NodeCommand::AddPeer(peer.clone()));
publish_network_event(NetworkEvent::SeedConnected(addr.to_string()));
let _ = self.exec_tx.send(cmd).await;
Connection::new(self.node_id, peer.id, stream, self.exec_tx.clone(), ch_rx)
Connection::new(self.node_id.clone(), peer.id, stream, self.exec_tx.clone(), ch_rx)
.start()
.await;
}
@ -183,7 +185,7 @@ impl Connector {
addr: SocketAddr,
) {
let handshake = ProtocolMessage::Handshake {
peer_id: self.node_id,
peer_id: self.node_id.clone(),
version: "".to_string(),
};
match Connector::send_message(&mut stream, &handshake).await {
@ -204,7 +206,7 @@ impl Connector {
};
let cmd = ExecutorCommand::Node(node::NodeCommand::AddPeer(peer.clone()));
let _ = self.exec_tx.send(cmd).await;
Connection::new(self.node_id, peer.id, stream, self.exec_tx.clone(), ch_rx)
Connection::new(self.node_id.clone(), peer.id, stream, self.exec_tx.clone(), ch_rx)
.start()
.await;
}
@ -223,7 +225,7 @@ impl Connector {
let peer = match mes {
ProtocolMessage::Handshake { peer_id, .. } => {
let ack = ProtocolMessage::HandshakeAck {
peer_id: self.node_id,
peer_id: self.node_id.clone(),
version: "".to_string(),
};
match Connector::send_message(&mut stream, &ack).await {
@ -241,7 +243,7 @@ impl Connector {
};
let cmd = ExecutorCommand::Node(node::NodeCommand::AddPeer(peer.clone()));
let _ = self.exec_tx.send(cmd).await;
Connection::new(self.node_id, peer.id, stream, self.exec_tx.clone(), ch_rx)
Connection::new(self.node_id.clone(), peer.id, stream, self.exec_tx.clone(), ch_rx)
.start()
.await;
}
@ -251,8 +253,7 @@ impl Connector {
stream: &mut net::TcpStream,
message: &ProtocolMessage,
) -> Result<(), NetworkError> {
let json = serde_json::to_string(message).map_err(|_e| NetworkError::TODO)?;
let data = json.as_bytes();
let data = bincode::encode_to_vec(message, BINCODE_CONFIG)?;
let len = data.len() as u32;
stream
@ -261,7 +262,7 @@ impl Connector {
.map_err(|_e| NetworkError::TODO)?;
stream
.write_all(data)
.write_all(&data)
.await
.map_err(|_e| NetworkError::TODO)?;
stream.flush().await.map_err(|_e| NetworkError::TODO)?;
@ -289,9 +290,7 @@ impl Connector {
.await
.map_err(|_e| NetworkError::TODO)?;
let json = String::from_utf8(data).map_err(|_e| NetworkError::TODO)?;
let message: ProtocolMessage = serde_json::from_str(&json).map_err(|_e| NetworkError::TODO)?;
let (message, _): (ProtocolMessage, usize) = bincode::decode_from_slice(&data, BINCODE_CONFIG)?;
Ok(message)
}

38
node/src/network/message.rs
View File

@ -1,52 +1,62 @@
use shared::core::{self, ChainData};
use shared::blockchain_core::{self, ChainData};
use std::fmt;
use std::net::SocketAddr;
pub const MAX_MESSAGE_SIZE: usize = 1_000_000;
#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
#[derive(Debug, Clone, bincode::Encode, bincode::Decode, Hash, PartialEq, Eq)]
pub struct NodeId(pub [u8; 16]);
#[derive(Debug, Clone, bincode::Encode, bincode::Decode)]
pub enum ProtocolMessage {
BootstrapRequest {
peer_id: uuid::Uuid,
peer_id: NodeId,
version: String,
},
BootstrapResponse {
blocks: Option<String>,
blocks: Vec<u8>,
},
GetPeersRequest {
peer_id: uuid::Uuid,
peer_id: NodeId,
},
GetPeersResponse {
peer_addresses: Vec<SocketAddr>,
},
Handshake {
peer_id: uuid::Uuid,
peer_id: NodeId,
version: String,
},
HandshakeAck {
peer_id: uuid::Uuid,
peer_id: NodeId,
version: String,
},
Block {
peer_id: uuid::Uuid,
peer_id: NodeId,
height: u64,
block: core::Block,
block: blockchain_core::Block,
},
ChainData {
peer_id: uuid::Uuid,
peer_id: NodeId,
data: ChainData,
},
Ping {
peer_id: uuid::Uuid,
peer_id: NodeId,
},
Pong {
peer_id: uuid::Uuid,
peer_id: NodeId,
},
Disconnect {
peer_id: uuid::Uuid,
peer_id: NodeId,
},
}
impl fmt::Display for NodeId {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let msg = self.to_string();
write!(f, "{}", msg)
}
}
impl fmt::Display for ProtocolMessage {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
@ -57,7 +67,7 @@ impl fmt::Display for ProtocolMessage {
write!(
f,
"BootstrapResponse with {:?} blocks",
blocks.clone().unwrap_or_default().len()
blocks.len()
)
}
ProtocolMessage::GetPeersRequest { peer_id } => {

46
node/src/network/ws_server.rs
View File

@ -1,6 +1,8 @@
#![allow(dead_code)]
use std::collections::HashMap;
use std::net::SocketAddr;
use shared::blockchain_core::validator::{ValidationError, Validator};
use tokio::io::AsyncWriteExt;
use tokio::net::TcpStream;
use futures::{SinkExt, StreamExt};
@ -8,10 +10,12 @@ use tokio::sync::mpsc::{self, Receiver, Sender};
use thiserror::Error;
use vlogger::*;
use shared::ws_protocol::{ WsClientRequest, WsClientResponse };
use watchlet::WalletError;
use crate::db::BINCODE_CONFIG;
use crate::executor::ExecutorCommand;
use crate::log;
use crate::node::NodeCommand;
use crate::seeds_constants::WS_LISTEN_ADDRESS;
#[derive(Debug, bincode::Encode, bincode::Decode)]
@ -23,21 +27,21 @@ pub enum WsCommand {
}
#[derive(Debug, Error)]
pub enum WsError {
pub enum WsServerError {
#[error("Socker Error: {0}")]
Socket(#[from] std::io::Error),
#[error("Tungstenite Error: {0}")]
Connection(#[from] tokio_tungstenite::tungstenite::Error),
#[error("Encoding Error: {0}")]
Encode(#[from] bincode::error::EncodeError),
#[error("Decoding Error: {0}")]
Decode(#[from] bincode::error::DecodeError),
#[error("Crypto Error: {0}")]
Crypto(#[from] WalletError),
#[error("MPSC Send Error: {0}")]
MpscSend(#[from] mpsc::error::SendError<WsClientResponse>)
MpscSend(#[from] mpsc::error::SendError<WsClientResponse>),
#[error("Validation Error: {0}")]
Validation(#[from] ValidationError),
}
pub struct WsServer {
@ -46,11 +50,28 @@ pub struct WsServer {
clients: HashMap<SocketAddr, Sender<WsClientResponse>>,
}
async fn handle_ws_client_request(
req: WsClientRequest,
_tx: Sender<ExecutorCommand>,
) -> Result<(), WsServerError> {
match req {
WsClientRequest::Ping => {
log(msg!(DEBUG, "Received Ping from client"));
let _response = bincode::encode_to_vec(WsClientResponse::Pong, BINCODE_CONFIG)?;
}
WsClientRequest::BroadcastTransaction(sign_tx) => {
Validator::verify_signature(&sign_tx)?;
let _cmd = ExecutorCommand::Node(NodeCommand::BroadcastTransaction(sign_tx));
}
}
Ok(())
}
async fn ws_connection(
stream: TcpStream,
mut rx: Receiver<WsClientResponse>,
_tx: Sender<ExecutorCommand>,
) -> Result<(), WsError> {
) -> Result<(), WsServerError> {
let ws_server = tokio_tungstenite::accept_async(stream).await.unwrap();
let (mut write, mut read) = ws_server.split();
@ -60,14 +81,7 @@ async fn ws_connection(
Some(Ok(msg)) => {
log(msg!(DEBUG, "msg: {:#?}", msg));
if msg.is_text() || msg.is_binary() {
let (message, _size): (WsClientRequest, usize) = bincode::decode_from_slice(msg.to_string().as_bytes(), BINCODE_CONFIG)?;
match message {
WsClientRequest::Ping => {
log(msg!(DEBUG, "Received Ping from client"));
let response = bincode::encode_to_vec(WsClientResponse::Pong, BINCODE_CONFIG)?;
write.send(response.into()).await?;
}
}
let (_message, _size): (WsClientRequest, usize) = bincode::decode_from_slice(msg.to_string().as_bytes(), BINCODE_CONFIG)?;
}
}
_ => {}
@ -95,7 +109,7 @@ impl WsServer {
}
}
pub async fn run(&mut self) -> Result<(), WsError> {
pub async fn run(&mut self) -> Result<(), WsServerError> {
let listener = tokio::net::TcpListener::bind(*WS_LISTEN_ADDRESS).await?;
let mut tasks = Vec::new();

92
node/src/node/blockchain.rs
View File

@ -1,19 +1,21 @@
use std::sync::Arc;
use std::time::UNIX_EPOCH;
use shared::core::Address;
use shared::blockchain_core::validator::ValidationError;
use shared::blockchain_core::{self, Address};
use shared::print_error_chain;
use thiserror::*;
use vlogger::*;
use shared::core;
use shared::core::ChainData;
use shared::blockchain_core::Block;
use shared::blockchain_core::ChainData;
use crate::db;
use crate::db::DatabaseError;
use crate::db::database;
use crate::db::BINCODE_CONFIG;
use crate::log;
use shared::core::Hasher;
use shared::blockchain_core::Hasher;
#[allow(dead_code)]
#[derive(Debug)]
pub struct Blockchain {
@ -22,40 +24,33 @@ pub struct Blockchain {
db: database::ChainDb,
}
#[derive(Debug, bincode::Encode, bincode::Decode)]
pub struct ChainBootStrap {
blocks: Vec<Block>,
mempool: Vec<ChainData>,
}
#[allow(dead_code)]
#[derive(Error, Debug)]
pub enum BlockchainError {
#[error("Failed to serialize data: {0}")]
Encode(#[from] bincode::error::EncodeError),
#[error("Failed to deserialize data: {0}")]
Decode(#[from] bincode::error::DecodeError),
#[error("Database operation failed")]
Database(#[from] DatabaseError),
#[error("invalid account creation")]
InvalidAccountCreation,
#[error("Transactional error")]
Transaction(#[from] shared::core::TransactionError),
Transaction(#[from] shared::blockchain_core::TransactionError),
#[error("Validation Error")]
Validation(#[from] ValidationError),
#[error("Insufficient fonds on address {0}")]
InsufficientFunds(String),
#[error("Block Creation Error")]
BlockCreation,
}
const BLOCKCHAIN_ID: &str = "watch-chain";
#[derive(Debug, thiserror::Error)]
pub enum ValidationError {
#[error("Invalid Block Hash Detected at height {0}")]
InvalidBlockHash(u64),
#[error("Previous Block Hash doesn't match at height {0}")]
InvalidPreviousBlockHash(u64),
#[error("Invalid Block JSON: {0}")]
InvalidBlockJson(#[from] serde_json::Error),
}
#[allow(dead_code)]
impl Blockchain {
fn hash_transaction_pool(&self) -> Vec<[u8; 32]> {
@ -76,9 +71,8 @@ impl Blockchain {
Ok(self.db.set_balance(&address, new_balance)?)
}
pub fn create_block(&mut self) -> Result<Arc<core::Block>, BlockchainError> {
match self.blocks() {
Ok(blocks) => {
pub fn create_block(&mut self) -> Result<Arc<blockchain_core::Block>, BlockchainError> {
let blocks = self.db.get_all_blocks()?;
let previous_hash = if blocks.len() > 0 {
<[u8; 32]>::try_from(blocks.last().unwrap().head().block_hash()).unwrap()
} else {
@ -93,7 +87,7 @@ impl Blockchain {
.as_secs();
let nonce = 0;
let mut new_head = core::BlockHeader {
let mut new_head = blockchain_core::BlockHeader {
previous_hash: <[u8; 32]>::from(previous_hash),
merkle_root,
timestamp,
@ -110,15 +104,12 @@ impl Blockchain {
new_head.block_hash = block_hash;
let new_block = Arc::new(core::Block::new(new_head, tx_hashes));
let new_block = Arc::new(blockchain_core::Block::new(new_head, tx_hashes));
self.add_block(new_block.clone())?;
Ok(new_block)
}
Err(_) => Err(BlockchainError::BlockCreation),
}
}
fn apply_transaction(&mut self, tx: &core::Transaction) -> Result<(), BlockchainError> {
fn apply_transaction(&mut self, tx: &blockchain_core::Transaction) -> Result<(), BlockchainError> {
tx.validate()?;
let from = tx.from();
let to = tx.to();
@ -146,6 +137,10 @@ impl Blockchain {
pub fn apply_chain_data(&mut self, data: ChainData) -> Result<(), BlockchainError> {
match &data {
ChainData::Transaction(tx) => {
self.apply_transaction(tx.data())?;
self.mempool.push(data);
}
ChainData::NodeTransaction(tx) => {
self.apply_transaction(tx)?;
self.mempool.push(data);
}
@ -164,16 +159,16 @@ impl Blockchain {
Ok(ret)
}
pub fn blocks(&self) -> Result<Vec<std::sync::Arc<core::Block>>, BlockchainError> {
pub fn blocks(&self) -> Result<Vec<std::sync::Arc<blockchain_core::Block>>, BlockchainError> {
Ok(self.db.get_all_blocks()?)
}
fn insert_block(&self, block: &core::Block) -> Result<(), BlockchainError> {
fn insert_block(&self, block: &blockchain_core::Block) -> Result<(), BlockchainError> {
self.db.add_block(block)?;
Ok(())
}
pub fn add_block(&mut self, block: Arc<core::Block>) -> Result<(), BlockchainError> {
pub fn add_block(&mut self, block: Arc<blockchain_core::Block>) -> Result<(), BlockchainError> {
match self.validate_block(&block) {
Ok(()) => Ok(self.insert_block(&block)?),
Err(e) => Err(BlockchainError::Validation(e)),
@ -195,7 +190,7 @@ impl Blockchain {
}
}
fn validate_block(&self, block: &core::Block) -> Result<(), ValidationError> {
fn validate_block(&self, block: &blockchain_core::Block) -> Result<(), ValidationError> {
let head = block.head();
let hash = Hasher::calculate_block_hash(block.head());
if hash != head.block_hash() {
@ -238,14 +233,26 @@ impl Blockchain {
}
}
pub fn bootstrap(&self) -> Result<Vec<u8>, BlockchainError> {
let blocks = self.blocks()?;
let mempool = self.mempool.clone();
let bs = ChainBootStrap {
blocks: blocks.iter().map(|b| (**b).clone()).collect::<Vec<Block>>(),
mempool
};
let bin_bs = bincode::encode_to_vec(&bs, BINCODE_CONFIG)?;
Ok(bin_bs)
}
pub async fn shutdown(&self) -> Result<(), BlockchainError> {
self.db.dump_mempool(&self.mempool)?;
self.db.shutdown().await?;
Ok(())
}
pub fn build(path: Option<String>, db_temp: bool) -> Result<Blockchain, BlockchainError> {
let db = db::ChainDb::new(path, db_temp).or_else(|e| Err(BlockchainError::Database(e)))?;
pub fn new(path: Option<String>, db_temp: bool) -> Result<Blockchain, BlockchainError> {
let db = db::ChainDb::open(path, db_temp).or_else(|e| Err(BlockchainError::Database(e)))?;
let mempool = db.recover_mempool()?;
let chain = Blockchain {
@ -258,4 +265,15 @@ impl Blockchain {
.or_else(|e| return Err(BlockchainError::Validation(e)))?;
Ok(chain)
}
pub fn build(&mut self, bin_data: Vec<u8>) -> Result<(), BlockchainError> {
let (data, _) = bincode::decode_from_slice::<ChainBootStrap, _>(&bin_data, BINCODE_CONFIG)?;
self.db.build_from(&data.blocks, &data.mempool)?;
self.mempool = data.mempool;
self
.validate_chain()
.or_else(|e| return Err(BlockchainError::Validation(e)))?;
Ok(())
}
}

6
node/src/node/error.rs
View File

@ -1,7 +1,11 @@
use thiserror::Error;
#[derive(Debug, Clone, Error)]
#[derive(Debug, Error)]
pub enum NetworkError {
#[error("Implement NetworkError Enum: ({})", file!())]
TODO,
#[error("Decode Error: {0}")]
Decode(#[from] bincode::error::DecodeError),
#[error("Encode Error: {0}")]
Encode(#[from] bincode::error::EncodeError),
}

101
node/src/node/node.rs
View File

@ -1,14 +1,14 @@
use crate::bus::{publish_system_event, publish_watcher_event, subscribe_system_event, SystemEvent};
use shared::core::{self, ChainData};
use shared::print_error_chain;
use shared::blockchain_core::{self, ChainData, SignedTransaction, validator::ValidationError};
use crate::print_error_chain;
use crate::executor::ExecutorCommand;
use crate::log;
use crate::network::ProtocolMessage;
use crate::network::{NodeId, ProtocolMessage};
use crate::network::{Connector, ConnectorCommand};
use crate::seeds_constants::SEED_NODES;
use crate::watcher::{WatcherCommand, WatcherMode};
use crate::network::ws_server::{WsCommand, WsServer};
use super::{ Blockchain, BlockchainError, ValidationError };
use super::{ Blockchain, BlockchainError };
use std::collections::HashMap;
use std::net::SocketAddr;
@ -22,14 +22,14 @@ use vlogger::*;
#[derive(Debug, Clone)]
pub struct TcpPeer {
pub id: Uuid,
pub id: NodeId,
pub addr: SocketAddr,
pub sender: tokio::sync::mpsc::Sender<ProtocolMessage>,
}
impl TcpPeer {
pub fn new(
id: Uuid,
id: NodeId,
addr: SocketAddr,
sender: tokio::sync::mpsc::Sender<ProtocolMessage>,
) -> Self {
@ -40,9 +40,9 @@ impl TcpPeer {
#[allow(dead_code)]
pub struct Node {
pub tcp_connector: Option<mpsc::Sender<ConnectorCommand>>,
pub id: Uuid,
pub id: NodeId,
pub addr: Option<SocketAddr>,
pub tcp_peers: HashMap<Uuid, TcpPeer>,
pub tcp_peers: HashMap<NodeId, TcpPeer>,
chain: Blockchain,
listner_handle: Option<tokio::task::JoinHandle<()>>,
exec_tx: mpsc::Sender<ExecutorCommand>,
@ -58,18 +58,19 @@ pub enum NodeError {
#[derive(Debug, Clone)]
pub enum NodeCommand {
BroadcastTransaction(SignedTransaction),
AddPeer(TcpPeer),
RemovePeer {
peer_id: Uuid,
peer_id: NodeId,
},
ProcessMessage {
peer_id: Uuid,
peer_id: NodeId,
message: ProtocolMessage,
},
ProcessChainData(ChainData),
StartListner(SocketAddr),
PingAddr(String),
PingId(String),
PingId(NodeId),
CreateBlock,
DisplayBlockInteractive,
DisplayBlockByKey(String),
@ -84,6 +85,7 @@ pub enum NodeCommand {
Exit,
}
#[allow(dead_code)]
impl Node {
pub fn peer_addresses(&self) -> Vec<SocketAddr> {
let mut addr: Vec<SocketAddr> = self
@ -109,18 +111,18 @@ impl Node {
log(msg!(DEBUG, "Node Id: {}", self.id))
}
async fn remove_tcp_peer(&mut self, peer_id: Uuid) {
async fn remove_tcp_peer(&mut self, peer_id: NodeId) {
log(msg!(DEBUG, "Removing Peer {peer_id}"));
self.tcp_peers.remove_entry(&peer_id);
}
async fn add_tcp_peer(&mut self, peer: TcpPeer) {
log(msg!(DEBUG, "Added Peer from address: {}", peer.addr));
self.tcp_peers.insert(peer.id, peer);
self.tcp_peers.insert(peer.id.clone(), peer);
}
pub async fn new_with_id(
id: uuid::Uuid,
id: NodeId,
exec_tx: mpsc::Sender<ExecutorCommand>,
addr: Option<SocketAddr>,
chain: Blockchain,
@ -146,7 +148,7 @@ impl Node {
) -> Self {
let (tx, rx) = mpsc::channel::<NodeCommand>(100);
Self {
id: Uuid::new_v4(),
id: NodeId(*Uuid::new_v4().as_bytes()),
tcp_peers: HashMap::new(),
addr,
exec_tx,
@ -168,43 +170,23 @@ impl Node {
let _ = self.chain.shutdown().await;
}
fn get_blocks(&self) -> Result<Vec<Arc<core::Block>>, NodeError> {
fn get_blocks(&self) -> Result<Vec<Arc<blockchain_core::Block>>, NodeError> {
Ok(self.chain.blocks()?)
}
pub async fn process_message(&mut self, peer_id: uuid::Uuid, message: ProtocolMessage) {
pub async fn process_message(&mut self, peer_id: NodeId, message: ProtocolMessage) -> Result<(), NodeError> {
match message {
ProtocolMessage::BootstrapRequest { .. } => {
log(msg!(DEBUG, "Received BootstrapRequest from {peer_id}"));
let peer = &self.tcp_peers[&peer_id];
let resp = ProtocolMessage::BootstrapResponse {
blocks: {
if let Ok(blocks) = self.get_blocks() {
serde_json::to_string(
&blocks
.iter()
.map(|f| (**f).clone())
.collect::<Vec<core::Block>>(),
)
.map_err(|e| {
log(msg!(
ERROR,
"Failed to serde Chain for BootstrapResponse: {e}"
));
e
})
.ok()
} else {
None
}
},
};
let blocks = self.chain.bootstrap()?;
let resp = ProtocolMessage::BootstrapResponse { blocks };
peer.sender.send(resp).await.unwrap();
log(msg!(DEBUG, "Send BootstrapResponse to {peer_id}"));
}
ProtocolMessage::BootstrapResponse { blocks } => {
log(msg!(DEBUG, "Received BootstrapResponse from seed"));
self.chain = Blockchain::build(blocks, true).unwrap();
self.chain.build(blocks).unwrap();
}
ProtocolMessage::Pong { peer_id } => {
log(msg!(DEBUG, "Received Pong from {peer_id}"));
@ -245,6 +227,7 @@ impl Node {
log(msg!(DEBUG, "TODO: implement this message type"));
}
}
Ok(())
}
pub async fn send_message_to_peer_addr(&self, addr: SocketAddr, msg: ProtocolMessage) {
@ -261,7 +244,7 @@ impl Node {
}
}
pub async fn send_message_to_peer_id(&self, id: Uuid, msg: ProtocolMessage) {
pub async fn send_message_to_peer_id(&self, id: NodeId, msg: ProtocolMessage) {
if let Some(peer) = self.tcp_peers.get(&id) {
if let Err(e) = peer.sender.send(msg).await {
log(msg!(ERROR, "Error Sending message to peer: {e}"));
@ -286,7 +269,7 @@ impl Node {
log(msg!(DEBUG, "Bootstrapping"));
let message = ProtocolMessage::BootstrapRequest {
peer_id: self.id,
peer_id: self.id.clone(),
version: "".to_string(),
};
self.send_message_to_seed(message).await;
@ -297,7 +280,7 @@ impl Node {
async fn broadcast_network_data(&self, data: ChainData) {
for (id, peer) in &self.tcp_peers {
let message = ProtocolMessage::ChainData {
peer_id: self.id,
peer_id: self.id.clone(),
data: data.clone(),
};
peer.sender.send(message).await.unwrap();
@ -305,10 +288,10 @@ impl Node {
}
}
async fn broadcast_block(&self, block: &core::Block) {
async fn broadcast_block(&self, block: &blockchain_core::Block) {
for (id, peer) in &self.tcp_peers {
let message = ProtocolMessage::Block {
peer_id: self.id,
peer_id: self.id.clone(),
height: block.head().height as u64,
block: block.clone(),
};
@ -342,7 +325,7 @@ impl Node {
self.tcp_connector = Some(con_tx);
self.listner_handle = Some(tokio::spawn({
let mut connector = Connector::new(self.id, addr, self.exec_tx(), con_rx);
let mut connector = Connector::new(self.id.clone(), addr, self.exec_tx(), con_rx);
log(msg!(DEBUG, "Connector Build"));
async move { connector.start().await }
}));
@ -361,6 +344,9 @@ impl Node {
log(msg!(DEBUG, "Received NodeCommand::BootStrap"));
let _ = self.bootstrap().await;
}
NodeCommand::BroadcastTransaction(sign_tx) => {
self.broadcast_network_data(ChainData::Transaction(sign_tx)).await;
}
NodeCommand::StartListner(addr) => {
self.start_connection_listner(addr).await;
}
@ -378,19 +364,15 @@ impl Node {
}
NodeCommand::PingAddr(addr) => {
if let Ok(addr_sock) = addr.parse::<SocketAddr>() {
let mes = ProtocolMessage::Ping { peer_id: self.id };
let mes = ProtocolMessage::Ping { peer_id: self.id.clone() };
self.send_message_to_peer_addr(addr_sock, mes).await;
} else {
log(msg!(ERROR, "Failed to Parse to sock_addr: {addr}"));
}
}
NodeCommand::PingId(id) => {
if let Ok(id) = id.parse::<Uuid>() {
let mes = ProtocolMessage::Ping { peer_id: self.id };
let mes = ProtocolMessage::Ping { peer_id: self.id.clone() };
self.send_message_to_peer_id(id, mes).await;
} else {
log(msg!(ERROR, "Failed to Parse to sock_addr: {id}"));
}
}
NodeCommand::AddPeer(peer) => {
self.add_tcp_peer(peer).await;
@ -400,7 +382,7 @@ impl Node {
}
NodeCommand::ProcessMessage { peer_id, message } => {
self.process_message(peer_id, message).await;
self.process_message(peer_id, message).await.unwrap();
}
NodeCommand::AwardCurrency { address, amount } => {
if let Err(e) = self.chain.award_currency(address, amount) {
@ -415,7 +397,8 @@ impl Node {
}
NodeCommand::CreateBlock => {
log(msg!(DEBUG, "Received CreateBlock Command"));
if let Ok(block) = self.chain.create_block() {
match self.chain.create_block() {
Ok(block) => {
log(msg!(
INFO,
"Created Block with hash {}",
@ -423,6 +406,8 @@ impl Node {
));
self.broadcast_block(&block).await;
}
Err(e) => print_error_chain(&e.into()),
}
}
NodeCommand::DisplayBlockInteractive => {
let blocks = match self.chain.list_blocks() {
@ -482,12 +467,14 @@ impl Node {
let _ = crate::api::server::start_server().await;
});
let (tx, rx) = mpsc::channel::<WsCommand>(100);
let (_tx, rx) = mpsc::channel::<WsCommand>(100);
let mut ws_server = WsServer::new(rx, self.exec_tx());
let ws_handle = tokio::spawn(async move {
ws_server.run().await;
let _ws_handle = tokio::spawn(async move {
if let Err(e) = ws_server.run().await {
print_error_chain(&e.into());
}
});
let mut system_rx = subscribe_system_event();

2
node/src/watcher/builder.rs
View File

@ -76,7 +76,7 @@ impl WatcherBuilder {
self.addr = Some(crate::seeds_constants::SEED_NODES[0]);
}
let chain = node::Blockchain::build(None, self.temporary).unwrap();
let chain = node::Blockchain::new(None, self.temporary).unwrap();
let mut node = Node::new(self.addr.clone(), exec_tx.clone(), chain);
log(msg!(INFO, "Built Node"));

2
node/src/watcher/watcher.rs
View File

@ -226,7 +226,7 @@ impl Watcher {
pub async fn log_memory() {
tokio::spawn(async move {
let mut id = format!("{}_{}", current_timestamp(), std::process::id());
let id = format!("{}_{}", current_timestamp(), std::process::id());
let id = id.replace(":", "_");
let mut path = std::path::PathBuf::new();
path.push("proc");

219
shared/Cargo.lock generated
View File

@ -58,6 +58,18 @@ version = "1.0.99"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b0674a1ddeecb70197781e945de4b3b8ffb61fa939a5597bcf48503737663100"
[[package]]
name = "base16ct"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4c7f02d4ea65f2c1853089ffd8d2787bdbc63de2f0d29dedbcf8ccdfa0ccd4cf"
[[package]]
name = "base64ct"
version = "1.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "55248b47b0caf0546f7988906588779981c43bb1bc9d0c44087278f80cdb44ba"
[[package]]
name = "bincode"
version = "2.0.1"
@ -139,6 +151,12 @@ version = "1.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b05b61dc5112cbb17e4b6cd61790d9845d13888356391624cbe7e41efeac1e75"
[[package]]
name = "const-oid"
version = "0.9.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8"
[[package]]
name = "cpufeatures"
version = "0.2.17"
@ -148,6 +166,18 @@ dependencies = [
"libc",
]
[[package]]
name = "crypto-bigint"
version = "0.5.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0dc92fb57ca44df6db8059111ab3af99a63d5d0f8375d9972e319a379c6bab76"
dependencies = [
"generic-array",
"rand_core",
"subtle",
"zeroize",
]
[[package]]
name = "crypto-common"
version = "0.1.6"
@ -158,6 +188,16 @@ dependencies = [
"typenum",
]
[[package]]
name = "der"
version = "0.7.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb"
dependencies = [
"const-oid",
"zeroize",
]
[[package]]
name = "digest"
version = "0.10.7"
@ -165,7 +205,52 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
dependencies = [
"block-buffer",
"const-oid",
"crypto-common",
"subtle",
]
[[package]]
name = "ecdsa"
version = "0.16.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ee27f32b5c5292967d2d4a9d7f1e0b0aed2c15daded5a60300e4abb9d8020bca"
dependencies = [
"der",
"digest",
"elliptic-curve",
"rfc6979",
"signature",
"spki",
]
[[package]]
name = "elliptic-curve"
version = "0.13.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b5e6043086bf7973472e0c7dff2142ea0b680d30e18d9cc40f267efbf222bd47"
dependencies = [
"base16ct",
"crypto-bigint",
"digest",
"ff",
"generic-array",
"group",
"pkcs8",
"rand_core",
"sec1",
"subtle",
"zeroize",
]
[[package]]
name = "ff"
version = "0.13.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c0b50bfb653653f9ca9095b427bed08ab8d75a137839d9ad64eb11810d5b6393"
dependencies = [
"rand_core",
"subtle",
]
[[package]]
@ -176,6 +261,29 @@ checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a"
dependencies = [
"typenum",
"version_check",
"zeroize",
]
[[package]]
name = "getrandom"
version = "0.2.16"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "335ff9f135e4384c8150d6f27c6daed433577f86b4750418338c01a1a2528592"
dependencies = [
"cfg-if",
"libc",
"wasi",
]
[[package]]
name = "group"
version = "0.13.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f0f9ef7462f7c099f518d754361858f86d8a07af53ba9af0fe635bbccb151a63"
dependencies = [
"ff",
"rand_core",
"subtle",
]
[[package]]
@ -190,6 +298,15 @@ version = "0.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70"
[[package]]
name = "hmac"
version = "0.12.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e"
dependencies = [
"digest",
]
[[package]]
name = "is_terminal_polyfill"
version = "1.70.1"
@ -202,6 +319,20 @@ version = "1.0.15"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4a5f13b858c8d314ee3e8f639011f7ccefe71f97f96e50151fb991f267928e2c"
[[package]]
name = "k256"
version = "0.13.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f6e3919bbaa2945715f0bb6d3934a173d1e9a59ac23767fbaaef277265a7411b"
dependencies = [
"cfg-if",
"ecdsa",
"elliptic-curve",
"once_cell",
"sha2",
"signature",
]
[[package]]
name = "keccak"
version = "0.1.5"
@ -223,12 +354,28 @@ version = "2.7.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32a282da65faaf38286cf3be983213fcf1d2e2a58700e808f83f4ea9a4804bc0"
[[package]]
name = "once_cell"
version = "1.21.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d"
[[package]]
name = "once_cell_polyfill"
version = "1.70.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a4895175b425cb1f87721b59f0f286c2092bd4af812243672510e1ac53e2e0ad"
[[package]]
name = "pkcs8"
version = "0.10.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f950b2377845cebe5cf8b5165cb3cc1a5e0fa5cfa3e1f7f55707d8fd82e0a7b7"
dependencies = [
"der",
"spki",
]
[[package]]
name = "proc-macro2"
version = "1.0.101"
@ -247,12 +394,45 @@ dependencies = [
"proc-macro2",
]
[[package]]
name = "rand_core"
version = "0.6.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c"
dependencies = [
"getrandom",
]
[[package]]
name = "rfc6979"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f8dd2a808d456c4a54e300a23e9f5a67e122c3024119acbfd73e3bf664491cb2"
dependencies = [
"hmac",
"subtle",
]
[[package]]
name = "ryu"
version = "1.0.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "28d3b2b1366ec20994f1fd18c3c594f05c5dd4bc44d8bb0c1c632c8d6829481f"
[[package]]
name = "sec1"
version = "0.7.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d3e97a565f76233a6003f9f5c54be1d9c5bdfa3eccfb189469f11ec4901c47dc"
dependencies = [
"base16ct",
"der",
"generic-array",
"pkcs8",
"subtle",
"zeroize",
]
[[package]]
name = "serde"
version = "1.0.219"
@ -314,6 +494,7 @@ dependencies = [
"bincode",
"clap",
"hex",
"k256",
"serde",
"serde_json",
"sha2",
@ -321,12 +502,38 @@ dependencies = [
"thiserror",
]
[[package]]
name = "signature"
version = "2.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "77549399552de45a898a580c1b41d445bf730df867cc44e6c0233bbc4b8329de"
dependencies = [
"digest",
"rand_core",
]
[[package]]
name = "spki"
version = "0.7.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d91ed6c858b01f942cd56b37a94b3e0a1798290327d1236e4d9cf4eaca44d29d"
dependencies = [
"base64ct",
"der",
]
[[package]]
name = "strsim"
version = "0.11.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f"
[[package]]
name = "subtle"
version = "2.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
[[package]]
name = "syn"
version = "2.0.106"
@ -394,6 +601,12 @@ version = "0.0.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "051eb1abcf10076295e815102942cc58f9d5e3b4560e46e53c21e8ff6f3af7b1"
[[package]]
name = "wasi"
version = "0.11.1+wasi-snapshot-preview1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
[[package]]
name = "windows-link"
version = "0.1.3"
@ -473,3 +686,9 @@ name = "windows_x86_64_msvc"
version = "0.53.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "271414315aff87387382ec3d271b52d7ae78726f5d44ac98b4f4030c91880486"
[[package]]
name = "zeroize"
version = "1.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ced3678a2879b30306d323f4542626697a464a97c0a07c9aebf7ebca65cd4dde"

1
shared/Cargo.toml
View File

@ -15,6 +15,7 @@ bincode = "2.0.1"
clap = { version = "4.5.47", features = ["derive"] }
# getrandom = { version = "0.3.3", features = ["wasm_js"] }
hex = "0.4.3"
k256 = { version = "0.13.4", features = ["ecdsa-core"] }
serde = { version = "1.0.219", features = ["derive"] }
serde_json = "1.0.143"
sha2 = "0.10.9"

0
shared/src/core/address.rs → shared/src/blockchain_core/address.rs
View File

0
shared/src/core/block.rs → shared/src/blockchain_core/block.rs
View File

18
shared/src/blockchain_core/data.rs Normal file
View File

@ -0,0 +1,18 @@
use bincode::{Decode, Encode};
use crate::blockchain_core::{SignedTransaction, Transaction};
#[derive(Encode, Decode, Debug, Clone)]
pub enum ChainData {
Transaction(SignedTransaction),
NodeTransaction(Transaction),
}
impl ChainData {
pub fn hash(&self) -> [u8; 32] {
match self {
Self::Transaction(tx) => tx.hash(),
Self::NodeTransaction(tx) => tx.hash(),
}
}
}

28
shared/src/core/hasher.rs → shared/src/blockchain_core/hasher.rs
View File

@ -2,25 +2,35 @@ use sha2::Digest;
use sha2::Sha256;
use sha3::Keccak256;
use crate::blockchain_core::Transaction;
use super::{BlockHeader, ChainData};
pub struct Hasher {}
impl Hasher {
pub fn hash_chain_data(data: &ChainData) -> [u8; 32] {
pub fn hash_transaction_data(data: &Transaction) -> [u8; 32] {
let mut hasher = Keccak256::new();
match data {
ChainData::Transaction(tx) => {
hasher.update(tx.to());
hasher.update(tx.from());
hasher.update(tx.value().to_be_bytes());
hasher.update(tx.data());
}
}
hasher.update(data.to());
hasher.update(data.from());
hasher.update(data.value().to_be_bytes());
hasher.update(data.data());
let res = hasher.finalize();
res.into()
}
pub fn hash_chain_data(data: &ChainData) -> [u8; 32] {
match data {
ChainData::Transaction(signed_transaction) => {
signed_transaction.hash()
}
ChainData::NodeTransaction(tx) => {
tx.hash()
}
}
}
pub fn calculate_next_level(level: &[[u8; 32]]) -> Vec<[u8; 32]> {
let mut next_level = Vec::new();

14
shared/src/core/tx.rs → shared/src/blockchain_core/tx.rs
View File

@ -1,3 +1,5 @@
use std::time::UNIX_EPOCH;
use super::Address;
use thiserror::Error;
@ -55,10 +57,11 @@ pub struct Transaction {
to: Address,
value: u64,
data: String,
timestamp: u64,
nonce: u64,
}
#[derive(Debug, Clone)]
#[derive(Debug, Clone, bincode::Decode, bincode::Encode)]
pub struct SignedTransaction {
tx: Transaction,
signature: [u8; 64],
@ -79,11 +82,15 @@ impl SignedTransaction {
&self.signature
}
pub fn hash(&self) -> [u8; 32] {
self.data().hash()
}
pub fn recovery_id(&self) -> u8 {
self.recovery_id
}
pub fn tx(&self) -> &Transaction {
pub fn data(&self) -> &Transaction {
&self.tx
}
}
@ -96,6 +103,7 @@ impl Transaction {
value,
data,
nonce,
timestamp: std::time::SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs(),
}
}
@ -111,7 +119,7 @@ impl Transaction {
}
pub fn hash(&self) -> [u8; 32] {
super::Hasher::hash_chain_data(&super::ChainData::Transaction(self.clone()))
super::Hasher::hash_transaction_data(self)
}
pub fn from(&self) -> &Address {

65
shared/src/blockchain_core/validator.rs Normal file
View File

@ -0,0 +1,65 @@
use k256::ecdsa::{
VerifyingKey,
RecoveryId,
Signature,
signature::Verifier,
};
use crate::blockchain_core::{Hasher, SignedTransaction};
use crate::blockchain_core::{ Block, ChainData };
#[derive(Debug, thiserror::Error)]
pub enum ValidationError {
#[error("Invalid Block Hash Detected at height {0}")]
InvalidBlockHash(u64),
#[error("Invalid Block Hash Difficulty Detected at height {0}")]
InvalidBlockHashDifficulty(u64),
#[error("Previous Block Hash doesn't match at height {0}")]
InvalidPreviousBlockHash(u64),
#[error("Invalid Block Data Hash Count at height: {0}!\nexpected: {1}\nfound: {2}\n")]
BlockDataCount(u64, usize, usize),
#[error("Ecdsa Error: {0}")]
K256(#[from] k256::ecdsa::Error),
#[error("Invalid Recovery ID: {0}")]
InvalidRecoveryId(u8),
}
pub struct Validator {}
impl Validator {
pub fn validate_chain(blocks: &[Block], data: &[ChainData]) -> Result<(), ValidationError> {
let data_hashes = data.iter().map(|d| d.hash()).collect::<Vec<[u8; 32]>>();
for block in blocks {
let block_data_hashes = data_hashes.iter().filter(|d| block.data().contains(d)).collect::<Vec<_>>();
Self::validate_block(block, &block_data_hashes)?
}
Ok(())
}
pub fn verify_signature(sign_tx: &SignedTransaction) -> Result<(), ValidationError>{
if let Some(rec_id) = RecoveryId::from_byte(sign_tx.recovery_id()) {
let sig = Signature::from_slice(sign_tx.signature())?;
let hash = sign_tx.hash();
let pub_key = VerifyingKey::recover_from_msg(&hash, &sig, rec_id).unwrap();
Ok(pub_key.verify(&hash, &sig).unwrap())
} else {
Err(ValidationError::InvalidRecoveryId(sign_tx.recovery_id()))
}
}
pub fn validate_block(block: &Block, data: &[&[u8; 32]]) -> Result<(), ValidationError> {
let block_hash = Hasher::calculate_block_hash(block.head());
if !block_hash.starts_with(b"0") {
return Err(ValidationError::InvalidBlockHashDifficulty(block.head().height));
}
if block_hash != block.head().block_hash() {
return Err(ValidationError::InvalidBlockHash(block.head().height))
}
if data.len() != block.data().len() {
return Err(ValidationError::BlockDataCount(block.head().height, block.data().len(), data.len()));
}
Ok(())
}
}

8
shared/src/core/data.rs
View File

@ -1,8 +0,0 @@
use bincode::{Decode, Encode};
use super::Transaction;
#[derive(serde::Deserialize, serde::Serialize, Encode, Decode, Debug, Clone)]
pub enum ChainData {
Transaction(Transaction),
}

3
shared/src/lib.rs
View File

@ -1,4 +1,4 @@
pub mod core {
pub mod blockchain_core {
pub mod block;
pub use block::*;
@ -14,6 +14,7 @@ pub mod core {
pub mod address;
pub use address::*;
pub mod validator;
}
pub mod ws_protocol;

7
shared/src/ws_protocol.rs
View File

@ -1,6 +1,9 @@
#[derive(Debug, bincode::Encode, bincode::Decode)]
use crate::blockchain_core::SignedTransaction;
#[derive(Debug, Clone, bincode::Encode, bincode::Decode)]
pub enum WsClientRequest {
Ping
Ping,
BroadcastTransaction(SignedTransaction),
}
#[derive(Debug, bincode::Encode, bincode::Decode)]

1
watchlet/Cargo.lock generated
View File

@ -1399,6 +1399,7 @@ dependencies = [
"bincode",
"clap",
"hex",
"k256",
"serde",
"serde_json",
"sha2",

4
watchlet/src/wallet/manager.rs
View File

@ -1,5 +1,5 @@
use age::secrecy::SecretString;
use shared::core::{SignedTransaction, Transaction};
use shared::blockchain_core::{validator::Validator, SignedTransaction, Transaction};
use super::{
wallet::{ Wallet, WalletError },
@ -45,7 +45,7 @@ impl<S: WalletStorage> WalletManager<S> {
pub fn verify_transaction(&self, transaction: SignedTransaction) -> Result<(), WalletError> {
match &self.wallet {
Some(wallet) => Ok(wallet.verify_signature(&transaction)?),
Some(_wallet) => Ok(Validator::verify_signature(&transaction)?),
None => Err(WalletError::NoPrivateKeyProvided),
}
}

99
watchlet/src/wallet/wallet.rs
View File

@ -4,13 +4,11 @@ use k256::ecdsa::{
self,
SigningKey,
VerifyingKey,
RecoveryId,
Signature,
signature::Verifier,
};
use shared::core::{ Transaction, SignedTransaction, Address, };
use k256::elliptic_curve::rand_core::OsRng;
use sha3::Keccak256;
use k256::elliptic_curve::rand_core::OsRng;
use shared::blockchain_core::validator::ValidationError;
use shared::blockchain_core::{ Transaction, SignedTransaction, Address, };
#[derive(Debug, thiserror::Error)]
pub enum WalletError {
@ -26,9 +24,6 @@ pub enum WalletError {
#[error("Decryption Error: {0}")]
DecryptionError(#[from] age::DecryptError),
#[error("Provided Recovery ID is invalid: {0}")]
InvalidRecoveryId(u8),
#[error("I/O error: {0}")]
IO(#[from] std::io::Error),
@ -43,6 +38,9 @@ pub enum WalletError {
#[error("Encode Error: {0}")]
EncodeError(#[from] bincode::error::EncodeError),
#[error("Validation Error: {0}")]
Validation(#[from] ValidationError)
}
#[derive(Debug, Encode, Decode)]
@ -55,7 +53,6 @@ pub struct Wallet {
impl Wallet {
#[cfg(not(target_arch = "wasm32"))]
pub fn new() -> Self {
let pk = Wallet::generate_private_key();
let address = Self::public_key_to_address(&pk.verifying_key());
@ -67,18 +64,6 @@ impl Wallet {
}
}
#[cfg(target_arch = "wasm32")]
pub fn new() -> Result<Self, WalletError> {
let pk = Wallet::generate_private_key()?;
let address = Self::public_key_to_address(&pk.verifying_key());
Self {
nonce: 0,
balance: 0,
address,
private_key: pk.to_bytes().into()
}
}
pub fn from_private_key(private_key: [u8; 32]) -> Result<Self, WalletError> {
let pk = SigningKey::from_bytes(&private_key.into())?;
let address = Self::public_key_to_address(&pk.verifying_key());
@ -90,38 +75,6 @@ impl Wallet {
})
}
pub fn verify_signature(&self, sign_tx: &SignedTransaction) -> Result<(), WalletError>{
if let Some(rec_id) = RecoveryId::from_byte(sign_tx.recovery_id()) {
let sig = Signature::from_slice(sign_tx.signature())?;
let hash = sign_tx.tx().hash();
let pub_key = VerifyingKey::recover_from_msg(&hash, &sig, rec_id).unwrap();
Ok(pub_key.verify(&hash, &sig).unwrap())
} else {
Err(WalletError::InvalidRecoveryId(sign_tx.recovery_id()))
}
}
#[cfg(target_arch = "wasm32")]
pub fn generate_private_key() -> Result<SigningKey, WalletError> {
let mut bytes = [0u8; 32];
let crypto = web_sys::window()
.ok_or_else(|| WalletError::WasmError("No window object".to_string()))?
.crypto()
.map_err(|_| WalletError::WasmError("No crypto object".to_string()))?;
let array = Uint8Array::new_with_length(32);
crypto.get_random_values_with_u8_array(&mut array.view_mut()[..])
.map_err(|_| WalletError::WasmError("Failed to get random values".to_string()))?;
array.copy_to(&mut bytes);
SigningKey::from_bytes(&bytes.into())
.map_err(|e| WalletError::SignatureError(e))
}
#[cfg(not(target_arch = "wasm32"))]
pub fn generate_private_key() -> SigningKey {
SigningKey::random(&mut OsRng)
}
@ -171,46 +124,6 @@ impl Wallet {
}
}
#[cfg(target_arch = "wasm32")]
#[wasm_bindgen]
impl Wallet {
#[wasm_bindgen(constructor)]
pub fn new_js() -> Result<Wallet, JsValue> {
Wallet::new().map_err(|e| JsValue::from_str(&e.to_string()))
}
#[wasm_bindgen]
pub fn from_private_key_js(private_key: &[u8]) -> Result<Wallet, JsValue> {
if private_key.len() != 32 {
return Err(JsValue::from_str("Private key must be 32 bytes"));
}
let mut key_array = [0u8; 32];
key_array.copy_from_slice(private_key);
Wallet::from_private_key(key_array)
.map_err(|e| JsValue::from_str(&e.to_string()))
}
#[wasm_bindgen(getter)]
pub fn address_js(&self) -> Vec<u8> {
self.address().to_vec()
}
#[wasm_bindgen(getter)]
pub fn nonce_js(&self) -> u64 {
self.nonce()
}
#[wasm_bindgen]
pub fn address_hex(&self) -> String {
format!("0x{}", hex::encode(self.address()))
}
#[wasm_bindgen]
pub fn private_key_hex(&self) -> String {
hex::encode(self.private_key)
}
}
#[test]
fn acc_new_sign_no_key() {
let wallet = Wallet::new();