feat: add sqlite peer_db (#6963)

Description
---
- Added an sqlite implementation of peer_db.
- Removed the lmdb implementation of peer_db.

Motivation and Context
---
The lmdb peer_db was memory and resource intensive.

How Has This Been Tested?
---
New unit tests
Existing unit tests depending on the new peer_db implementation pass
System-level testing passed

What process can a PR reviewer use to test or verify this change?
---
Code review
System-level testing 

<!-- Checklist -->
<!-- 1. Is the title of your PR in the form that would make nice release
notes? The title, excluding the conventional commit
tag, will be included exactly as is in the CHANGELOG, so please think
about it carefully. -->


Breaking Changes
---

- [ ] None
- [ ] Requires data directory on base node to be deleted
- [ ] Requires hard fork
- [X] Other - Please specify

<!-- Does this include a breaking change? If so, include this line as a
footer -->
BREAKING CHANGES: 
- The existing lmbd peer_db will be discarded (left intact but not used
anymore).
- In PeerManager, the method `fn closest_peers` is replaced with `fn
closest_n_active_peers` as amore flexible alternative.


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

## Summary by CodeRabbit

- **New Features**
- Transitioned peer storage backend from LMDB to SQLite, enhancing
reliability and simplifying setup.
- Added test utilities for creating peers with address stats and recent
last-seen timestamps.
- Extended peer and address data structures with constructors,
conversion methods, and reset functions.
- **Bug Fixes**
- Unified peer add/update operations into `add_or_update_peer` for
consistency across modules.
- Corrected configuration comments to specify SQLite database paths
instead of LMDB.
- **Refactor**
- Removed legacy peer query framework and migration code; replaced with
async wrappers over SQLite.
- Enhanced peer selection, filtering, and banning logic to leverage new
database queries.
- Simplified batch peer retrieval in wallet and FFI layers with direct
SQL queries.
- Updated test setups to use SQLite, replacing in-memory or LMDB
dependencies.
- Replaced direct sync calls with async wrappers in discovery, network,
and connection modules.
- Improved asynchronous handling and batch processing for peer
management.
- **Chores**
- Removed LMDB and related dependencies; added SQLite, Diesel, and
migration crates.
- Created SQL migration scripts for node identity, peers, and addresses
schemas.
  - Managed temporary database cleanup in tests and connection pools.
- Cleaned imports and streamlined peer manager initialization in
examples and tests.
- **Documentation**
- Updated configuration comments to clarify transition from LMDB to
SQLite storage paths.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: Stan Bondi <sdbondi@users.noreply.github.com>

Author: hansieodendaal

Cargo.lock

@@ -80,7 +80,7 @@ use tari_common_types::{
 use tari_comms::{
     connectivity::{ConnectivityEvent, ConnectivityRequester},
     multiaddr::Multiaddr,
-    peer_manager::{Peer, PeerQuery},
+    peer_manager::Peer,
     types::CommsPublicKey,
 };
 use tari_comms_dht::{envelope::NodeDestination, DhtDiscoveryRequester};
@@ -2600,11 +2600,10 @@ pub async fn command_runner(
                     .map_err(|e| CommandError::General(e.to_string()))?;
                     // config
 
-                    let query = PeerQuery::new().select_where(|p| p.is_seed());
                     let peer_seeds = wallet
                         .comms
                         .peer_manager()
-                        .perform_query(query)
+                        .get_seed_peers()
                         .await
                         .map_err(|e| CommandError::General(e.to_string()))?;
                     // config

applications/minotari_console_wallet/src/automation/commands.rs

@@ -1205,15 +1205,29 @@ impl wallet_server::Wallet for WalletGrpcServer {
             .await
             .map_err(|err| Status::internal(err.to_string()))?;
 
-        let mut peers = Vec::with_capacity(connected_peers.len());
-        for conn in connected_peers {
-            peers.push(
-                peer_manager
-                    .find_by_node_id(conn.peer_node_id())
-                    .await
-                    .map_err(|err| Status::internal(err.to_string()))?
-                    .ok_or_else(|| Status::not_found(format!("Peer '{}' not found", conn.peer_node_id())))?,
-            );
+        let node_ids = connected_peers
+            .iter()
+            .map(|c| c.peer_node_id())
+            .cloned()
+            .collect::<Vec<_>>();
+        let peers = peer_manager
+            .get_peers_by_node_ids(&node_ids)
+            .await
+            .map_err(|err| Status::internal(err.to_string()))?;
+        if peers.len() != node_ids.len() {
+            let mut error_response = Vec::new();
+            node_ids.iter().for_each(|node_id| {
+                if !peers.iter().any(|p| p.node_id == *node_id) {
+                    warn!(target: LOG_TARGET, "Peer '{}' not found", node_id);
+                    error_response.push(format!("'{}'", node_id));
+                }
+            });
+            if !error_response.is_empty() {
+                return Err(Status::not_found(format!(
+                    "Peer(s) not found: {}",
+                    error_response.join(", ")
+                )));
+            }
         }
 
         let resp = tari_rpc::ListConnectedPeersResponse {

applications/minotari_console_wallet/src/grpc/wallet_grpc_server.rs

@@ -61,7 +61,7 @@ use tari_common_types::{
 };
 use tari_comms::{
     multiaddr::Multiaddr,
-    peer_manager::{Peer, PeerFeatures, PeerQuery},
+    peer_manager::{Peer, PeerFeatures},
     types::CommsPublicKey,
     NodeIdentity,
 };
@@ -347,11 +347,10 @@ pub async fn set_peer_and_get_base_node_peer_config(
             }
         }
     }
-    let query = PeerQuery::new().select_where(|p| p.is_seed());
-    let peer_seeds = wallet.comms.peer_manager().perform_query(query).await.map_err(|err| {
+    let peer_seeds = wallet.comms.peer_manager().get_seed_peers().await.map_err(|err| {
         ExitError::new(
             ExitCode::InterfaceError,
-            format!("Could net get seed peers from peer manager: {}", err),
+            format!("Could not get seed peers from peer manager: {}", err),
         )
     })?;
     // config

applications/minotari_console_wallet/src/init/mod.rs

@@ -119,7 +119,7 @@ pub async fn wallet_recovery(
         );
         peer_public_keys.push(peer.public_key.clone());
         peer_manager
-            .add_peer(peer)
+            .add_or_update_peer(peer)
             .await
             .map_err(|err| ExitError::new(ExitCode::NetworkError, err))?;
     }

applications/minotari_console_wallet/src/recovery.rs

@@ -942,13 +942,15 @@ impl AppStateInner {
         self.refresh_network_id().await?;
         let connections = self.wallet.comms.connectivity().get_active_connections().await?;
         let peer_manager = self.wallet.comms.peer_manager();
-        let mut peers = Vec::with_capacity(connections.len());
-        for c in &connections {
-            if let Ok(Some(p)) = peer_manager.find_by_node_id(c.peer_node_id()).await {
-                peers.push(p);
-            }
+        let node_ids = connections
+            .iter()
+            .map(|c| c.peer_node_id())
+            .cloned()
+            .collect::<Vec<_>>();
+        self.data.connected_peers = peer_manager.get_peers_by_node_ids(&node_ids).await?;
+        if self.data.connected_peers.is_empty() {
+            debug!(target: LOG_TARGET, "Failed to look up {} peers", node_ids.len());
         }
-        self.data.connected_peers = peers;
         self.updated = true;
         Ok(())
     }

applications/minotari_console_wallet/src/ui/state/app_state.rs

@@ -26,7 +26,7 @@ use minotari_wallet::{
     transaction_service::error::TransactionServiceError,
 };
 use tari_common_types::tari_address::TariAddressError;
-use tari_comms::connectivity::ConnectivityError;
+use tari_comms::{connectivity::ConnectivityError, peer_manager::PeerManagerError};
 use tari_contacts::contacts_service::error::ContactsServiceError;
 use tari_utilities::hex::HexError;
 use thiserror::Error;
@@ -47,6 +47,8 @@ pub enum UiError {
     WalletError(#[from] WalletError),
     #[error(transparent)]
     WalletStorageError(#[from] WalletStorageError),
+    #[error(transparent)]
+    PeerManagerError(#[from] PeerManagerError),
     #[error("Could not parse Tari Address: `{0}`")]
     TariAddressParseError(#[from] TariAddressError),
     #[error("Could not convert string into Net Address")]

applications/minotari_console_wallet/src/ui/ui_error.rs

@@ -60,7 +60,7 @@ impl HandleCommand<ArgsAddPeer> for CommandContext {
             String::new(),
         );
         // If the peer exists, this will merge the given address
-        peer_manager.add_peer(peer).await?;
+        peer_manager.add_or_update_peer(peer).await?;
         println!("Peer with node id '{}' was added to the base node.", node_id);
         self.dial_peer(node_id).await?;
         Ok(())

applications/minotari_node/src/commands/command/add_peer.rs

@@ -54,12 +54,26 @@ impl CommandContext {
             "Info",
         ]);
         let peer_manager = self.comms.peer_manager();
-        for conn in conns {
-            let peer = peer_manager
-                .find_by_node_id(conn.peer_node_id())
-                .await
-                .expect("Unexpected peer database error")
-                .expect("Peer not found");
+        let node_ids = conns
+            .iter()
+            .map(|conn| conn.peer_node_id())
+            .cloned()
+            .collect::<Vec<_>>();
+        let peers = match peer_manager.get_peers_by_node_ids(&node_ids).await {
+            Ok(val) => val,
+            Err(e) => {
+                println!("Error: Unexpected peer database error: {}", e);
+                return;
+            },
+        };
+        if peers.len() != node_ids.len() {
+            println!("\nError: Peer manager returned fewer peers than requested\n");
+        }
+        for peer in peers {
+            let conn = match conns.iter().find(|conn| conn.peer_node_id() == &peer.node_id) {
+                None => continue,
+                Some(val) => val,
+            };
 
             let chain_height = peer
                 .get_metadata(1)

applications/minotari_node/src/commands/command/list_connections.rs

@@ -24,7 +24,7 @@ use anyhow::Error;
 use async_trait::async_trait;
 use chrono::Utc;
 use clap::Parser;
-use tari_comms::peer_manager::PeerQuery;
+use tari_comms::peer_manager::PeerFeatures;
 use tari_core::base_node::state_machine_service::states::PeerMetadata;
 
 use super::{CommandContext, HandleCommand};
@@ -45,16 +45,17 @@ impl HandleCommand<Args> for CommandContext {
 
 impl CommandContext {
     pub async fn list_peers(&self, filter: Option<String>) -> Result<(), Error> {
-        let mut query = PeerQuery::new();
-        if let Some(f) = filter {
-            let filter = f.to_lowercase();
-            query = query.select_where(move |p| match filter.as_str() {
-                "basenode" | "basenodes" | "base_node" | "base-node" | "bn" => p.features.is_node(),
-                "wallet" | "wallets" | "w" => p.features.is_client(),
-                _ => false,
-            })
-        }
-        let mut peers = self.comms.peer_manager().perform_query(query).await?;
+        let features = filter.as_ref().and_then(|value| match value.to_lowercase().as_str() {
+            "basenode" | "basenodes" | "base_node" | "base-node" | "bn" => Some(PeerFeatures::COMMUNICATION_NODE),
+            "wallet" | "wallets" | "w" => Some(PeerFeatures::COMMUNICATION_CLIENT),
+            _ => {
+                println!("Unknown filter '{:?}'; list-peers", filter);
+                println!("  Try 'basenode', 'basenodes', 'base-node', 'base_node', 'wallet', 'wallets', 'w'");
+                None
+            },
+        });
+
+        let mut peers = self.comms.peer_manager().all(features).await?;
         let num_peers = peers.len();
         println!();
         let mut table = Table::new();