Made the ledger more stable

crates/backend/src/broadcast.rs

@@ -1,7 +1,7 @@
 use futures_util::{stream::SplitSink, SinkExt};
 use hyper::upgrade::Upgraded;
 use hyper_util::rt::TokioIo;
-use immutable_bank_model::ledger_entry::LedgerEntry;
+use immutable_bank_model::header::LedgerMessage;
 use tokio_tungstenite::{tungstenite::Message, WebSocketStream};
 
 use crate::general_state::GeneralState;
@@ -24,8 +24,8 @@ impl GeneralState {
         });
     }
 
-    pub fn broadcast(&self, entry: LedgerEntry) {
-        let data = match bincode::serialize(&entry) {
+    pub fn broadcast(&self, msh: LedgerMessage) {
+        let data = match bincode::serialize(&msh) {
             Ok(d) => d,
             Err(err) => {
                 tracing::error!("failed to serialize entry to broadcast - {}", err);

crates/backend/src/process.rs

@@ -1,65 +1,65 @@
-use immutable_bank_model::{ledger_entry::LedgerEntry, ledger_type::LedgerType};
+use immutable_bank_model::{header::LedgerMessage, ledger_type::LedgerEntry};
 
 use crate::general_state::GeneralStateInner;
 
 impl GeneralStateInner {
-    pub fn process(&mut self, entry: LedgerEntry) {
-        match entry.entry.clone() {
-            LedgerType::NewBank(bank) => {
+    pub fn process(&mut self, msg: LedgerMessage) {
+        tracing::warn!("Duplicate message: {:?}", msg.header);
+        if self.ledger.entries.contains_key(&msg.header) {
+            return;
+        }
+        tracing::info!("Received Msg: {:?}", msg);
+        match msg.entry.clone() {
+            LedgerEntry::NewBank(bank) => {
                 if !self.existing_banks.contains(&bank.owner) {
                     let owner = bank.owner.clone();
-                    self.ledger.new_bank(entry.id, bank);
+                    self.ledger.new_bank(msg.header, bank);
                     self.existing_banks.insert(owner);
                 } else {
-                    self.ledger.entries.push(LedgerEntry {
-                        id: entry.id,
-                        entry: LedgerType::Error(format!("Bank already exist ({})", bank.owner)),
-                    });
+                    self.ledger.entries.insert(
+                        msg.header,
+                        LedgerEntry::Error(format!("Bank already exist ({})", bank.owner)),
+                    );
                 }
             }
-            LedgerType::UpdateBank(bank) => {
+            LedgerEntry::UpdateBank(bank) => {
                 if self.existing_banks.contains(&bank.owner) {
-                    self.ledger.save_bank(entry.id, bank);
+                    self.ledger.save_bank(msg.header, bank);
                 } else {
-                    self.ledger.entries.push(LedgerEntry {
-                        id: entry.id,
-                        entry: LedgerType::Error(format!(
-                            "Foreign bank does not exist ({})",
-                            bank.owner
-                        )),
-                    });
+                    self.ledger.entries.insert(
+                        msg.header,
+                        LedgerEntry::Error(format!("Foreign bank does not exist ({})", bank.owner)),
+                    );
                 }
             }
-            LedgerType::Transfer {
+            LedgerEntry::Transfer {
                 local_bank,
                 transaction,
             } => {
                 if self.existing_banks.contains(&local_bank) {
-                    match self.ledger.transfer(entry.id, local_bank, transaction) {
+                    match self
+                        .ledger
+                        .transfer(msg.header.clone(), local_bank, transaction)
+                    {
                         Ok(()) => {}
                         Err(err) => {
-                            self.ledger.entries.push(LedgerEntry {
-                                id: entry.id,
-                                entry: LedgerType::Error(err.to_string()),
-                            });
+                            self.ledger
+                                .entries
+                                .insert(msg.header, LedgerEntry::Error(err.to_string()));
                         }
                     }
                 } else {
-                    self.ledger.entries.push(LedgerEntry {
-                        id: entry.id,
-                        entry: LedgerType::Error(format!(
-                            "Foreign bank does not exist ({})",
-                            local_bank
-                        )),
-                    });
+                    self.ledger.entries.insert(
+                        msg.header,
+                        LedgerEntry::Error(format!("Foreign bank does not exist ({})", local_bank)),
+                    );
                 }
             }
-            LedgerType::Error(err) => {
-                tracing::debug!("ledger error (id={}) - {}", entry.id, err);
-                self.ledger.entries.push(LedgerEntry {
-                    id: entry.id,
-                    entry: LedgerType::Error(err),
-                });
+            LedgerEntry::Error(err) => {
+                tracing::debug!("ledger error (id={}) - {}", msg.header.id, err);
+                self.ledger
+                    .entries
+                    .insert(msg.header, LedgerEntry::Error(err));
             }
         }
     }

crates/backend/src/ws_handler.rs

@@ -1,7 +1,7 @@
 use anyhow::Context;
-use futures_util::StreamExt;
+use futures_util::{SinkExt, StreamExt};
 use hyper_tungstenite::HyperWebsocket;
-use immutable_bank_model::ledger_entry::LedgerEntry;
+use immutable_bank_model::header::{LedgerHeader, LedgerMessage};
 use tokio_tungstenite::tungstenite::Message;
 
 use crate::general_state::GeneralState;
@@ -11,7 +11,25 @@ pub async fn handle_ws(stream: HyperWebsocket, state: GeneralState) -> anyhow::R
         .await
         .context("Error during the websocket handshake occurred")?;
 
-    let (tx, mut rx) = ws_stream.split();
+    let (mut tx, mut rx) = ws_stream.split();
+
+    // Before we do anything we send the entire ledger to the client
+    let entire_ledger = {
+        let guard = state.inner.lock().await;
+        guard
+            .ledger
+            .entries
+            .iter()
+            .map(|(a, b)| LedgerMessage {
+                header: a.clone(),
+                entry: b.clone(),
+            })
+            .collect::<Vec<_>>()
+    };
+    for msg in entire_ledger {
+        let data = bincode::serialize(&msg)?;
+        tx.send(Message::binary(data)).await?;
+    }
 
     // Subscribe
     state.subscribe(tx);
@@ -24,18 +42,28 @@ pub async fn handle_ws(stream: HyperWebsocket, state: GeneralState) -> anyhow::R
                 tracing::debug!("WS message text - {}", txt);
             }
             Message::Binary(data) => {
-                let entry: LedgerEntry = bincode::deserialize(&data)?;
+                let msg: LedgerMessage = bincode::deserialize(&data)?;
 
                 let broadcast = {
                     let mut guard = state.lock().await;
-                    let start = guard.ledger.entries.len();
-                    guard.process(entry);
+                    let start = guard
+                        .ledger
+                        .entries
+                        .last_key_value()
+                        .map(|e| LedgerHeader {
+                            id: e.0.id,
+                            signature: 0,
+                        })
+                        .unwrap_or(LedgerHeader::ZERO);
+                    guard.process(msg);
                     guard
                         .ledger
                         .entries
-                        .iter()
-                        .skip(start)
-                        .cloned()
+                        .range(start..)
+                        .map(|a| LedgerMessage {
+                            header: a.0.clone(),
+                            entry: a.1.clone(),
+                        })
                         .collect::<Vec<_>>()
                 };
                 for broadcast in broadcast {

crates/model/src/bank.rs

@@ -4,13 +4,15 @@ use serde::{Deserialize, Serialize};
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct Bank {
     pub owner: String,
+    pub secret: String,
     pub accounts: Vec<Account>,
 }
 
 impl Bank {
-    pub fn new(owner: String) -> Self {
+    pub fn new(owner: String, password_hash: String) -> Self {
         Bank {
             owner,
+            secret: password_hash,
             accounts: vec![
                 Account {
                     type_: AccountType::Wallet,

crates/model/src/header.rs

@@ -0,0 +1,22 @@
+use serde::{Deserialize, Serialize};
+
+use crate::ledger_type::LedgerEntry;
+
+#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct LedgerHeader {
+    pub id: u64,
+    pub signature: u64,
+}
+
+impl LedgerHeader {
+    pub const ZERO: LedgerHeader = LedgerHeader {
+        id: 0,
+        signature: 0,
+    };
+}
+
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct LedgerMessage {
+    pub header: LedgerHeader,
+    pub entry: LedgerEntry,
+}

crates/model/src/ledger.rs

@@ -1,8 +1,10 @@
+use std::collections::BTreeMap;
+
 use serde::{Deserialize, Serialize};
 
-use crate::ledger_entry::LedgerEntry;
+use crate::{header::LedgerHeader, ledger_type::LedgerEntry};
 
 #[derive(Debug, Clone, Default, Deserialize, Serialize)]
 pub struct Ledger {
-    pub entries: Vec<LedgerEntry>,
+    pub entries: BTreeMap<LedgerHeader, LedgerEntry>,
 }

crates/model/src/ledger_type.rs

@@ -3,7 +3,7 @@ use serde::{Deserialize, Serialize};
 use crate::{bank::Bank, transaction::Transaction};
 
 #[derive(Debug, Clone, Deserialize, Serialize)]
-pub enum LedgerType {
+pub enum LedgerEntry {
     NewBank(Bank),
     UpdateBank(Bank),
     Transfer {

crates/model/src/lib.rs

@@ -2,6 +2,6 @@ pub mod account;
 pub mod bank;
 pub mod ledger_type;
 pub mod transaction;
-pub mod ledger_entry;
+pub mod header;
 pub mod ledger;
 pub mod logic;

crates/model/src/logic/new_bank.rs

@@ -1,41 +1,17 @@
+use rand::RngCore;
+
 use crate::{
     account::{AccountRef, AccountType},
     bank::Bank,
+    header::LedgerHeader,
     ledger::Ledger,
-    ledger_type::LedgerType,
+    ledger_type::LedgerEntry,
     transaction::Transaction,
 };
 
 impl Ledger {
-    pub fn new_bank(&mut self, entry_id: u64, mut bank: Bank) {
-        self.add(LedgerType::Transfer {
-            transaction: Transaction {
-                from: AccountRef::Foreign {
-                    bank: "John".to_string(),
-                    account: AccountType::Printer,
-                },
-                to: AccountRef::Local {
-                    account: AccountType::Wallet,
-                },
-                description: "Donation from money printer".to_string(),
-                amount_cents: 1_000_000_00,
-            },
-            local_bank: bank.owner.clone(),
-        });
-
-        self.add(LedgerType::Transfer {
-            transaction: Transaction {
-                from: AccountRef::Local {
-                    account: AccountType::Wallet,
-                },
-                to: AccountRef::Local {
-                    account: AccountType::Savings,
-                },
-                description: "Squirreling some money away".to_string(),
-                amount_cents: 999_000_00,
-            },
-            local_bank: bank.owner.clone(),
-        });
+    pub fn new_bank(&mut self, header: LedgerHeader, mut bank: Bank) {
+        let mut rand = rand::thread_rng();
 
         for acc in bank.accounts.iter_mut() {
             match acc.type_ {
@@ -49,6 +25,47 @@ impl Ledger {
             }
         }
 
-        self.add_with_id(entry_id, LedgerType::UpdateBank(bank.clone()));
+        self.add_with_header(
+            LedgerHeader {
+                id: header.id,
+                signature: rand.next_u64(),
+            },
+            LedgerEntry::Transfer {
+                transaction: Transaction {
+                    from: AccountRef::Foreign {
+                        bank: "John".to_string(),
+                        account: AccountType::Printer,
+                    },
+                    to: AccountRef::Local {
+                        account: AccountType::Wallet,
+                    },
+                    description: "Donation from money printer".to_string(),
+                    amount_cents: 1_000_000_00,
+                },
+                local_bank: bank.owner.clone(),
+            },
+        );
+
+        self.add_with_header(
+            LedgerHeader {
+                id: header.id,
+                signature: rand.next_u64(),
+            },
+            LedgerEntry::Transfer {
+                transaction: Transaction {
+                    from: AccountRef::Local {
+                        account: AccountType::Wallet,
+                    },
+                    to: AccountRef::Local {
+                        account: AccountType::Savings,
+                    },
+                    description: "Squirreling some money away".to_string(),
+                    amount_cents: 999_000_00,
+                },
+                local_bank: bank.owner.clone(),
+            },
+        );
+
+        self.add_with_header(header, LedgerEntry::NewBank(bank.clone()));
     }
 }