use crate::common::binary_conversions::{binary_to_ip, binary_to_string, ip_to_binary}; use crate::common::network_startup::get_ip_and_port; use crate::encode; use crate::records::memory::network_mapping::structs::{ AddAddressParams, DeleteAddressParams, SignedNodeEdit, NODE_ADDED_BY_OFFSET, NODE_ADDED_SIGNATURE_OFFSET, NODE_ADDED_TIMESTAMP_OFFSET, NODE_BLOCKS_MINED_OFFSET, NODE_DELETED_BLOCK_OFFSET, NODE_DELETED_BY_OFFSET, NODE_DELETED_SIGNATURE_OFFSET, NODE_DELETED_TIMESTAMP_OFFSET, NODE_IP_OFFSET, NODE_RECORD_BYTES, }; use crate::records::memory::network_mapping::NodeInfo; use crate::records::memory::response_channels::{reserve_entry, Command}; use crate::rpc::command_maps::{RPC_ADD_NETWORK_NODE, RPC_REQUEST_NODE_LIST}; use crate::rpc::responses::RpcResponse; use crate::sled::Db; use crate::wallets::structures::Wallet; use crate::Arc; use crate::Mutex; use crate::TcpStream; use crate::Utc; fn decode_optional_signature(bytes: &[u8]) -> String { // Network mapping records use zero-filled signature fields when no signature exists yet. if bytes.iter().all(|&byte| byte == 0) { String::new() } else { encode(bytes) } } pub async fn announce_self_to_network( unlocked_stream: Arc>, address: &str, command_map: Arc>, db: &Db, wallet: Arc, connections_key: &str, ) { // announce the local node to the connected peer, then // request its current network mapping on success let message_type = RPC_ADD_NETWORK_NODE; let (ip, _, _) = get_ip_and_port().await; // Reserve a reply key so the peer's acknowledgement returns to this request. let (hashmap_key, _hashmap_tx, hashmap_rx) = reserve_entry(command_map.clone()).await; // Encode the local node identity into the same binary shape used by network commands. let ip_bytes = ip_to_binary(&ip); let address_bytes = match Wallet::short_address_to_bytes(address) { Some(bytes) => bytes, None => return, }; let modified_by_bytes = vec![0u8; Wallet::ADDRESS_BYTES_LENGTH]; let time = Utc::now().timestamp_millis() as u64; let modified_timestamp_bytes = time.to_le_bytes(); // Self-announcement is intentionally unsigned. The receiving node // adopts it by re-signing the membership edit with its own wallet. let modified_signature_bytes = vec![0u8; Wallet::SIGNATURE_LENGTH]; let mut message: Vec = Vec::with_capacity( 1 + 3 + Wallet::SHORT_ADDRESS_BYTES_LENGTH + 16 + Wallet::ADDRESS_BYTES_LENGTH + 8 + Wallet::SIGNATURE_LENGTH, ); message.push(message_type); message.extend_from_slice(&hashmap_key); message.extend_from_slice(&address_bytes); message.extend_from_slice(&ip_bytes); message.extend_from_slice(&modified_by_bytes); message.extend_from_slice(&modified_timestamp_bytes); message.extend_from_slice(&modified_signature_bytes); RpcResponse::send_raw(&unlocked_stream, Some(connections_key), &message).await; // Only pull the peer's network mapping after it accepts our self-announcement. let mut rx = hashmap_rx.lock().await; // Handle the received data if let Some(buffer) = rx.recv().await { if binary_to_string(buffer.clone()) == "Success" { get_network_mapping( unlocked_stream, command_map.clone(), db, wallet, connections_key, ) .await } } } pub async fn get_network_mapping( unlocked_stream: Arc>, command_map: Arc>, db: &Db, wallet: Arc, connections_key: &str, ) { // request the remote peer's serialized node list and // import each advertised add/delete record locally let message_type = RPC_REQUEST_NODE_LIST; let (download_hashmap_key, _hashmap_tx, download_hashmap_rx) = reserve_entry(command_map.clone()).await; let mut message: Vec = Vec::new(); message.push(message_type); message.extend_from_slice(&download_hashmap_key); // The remote reply is a concatenated list of fixed-width node records. RpcResponse::send_raw(&unlocked_stream, Some(connections_key), &message).await; let mut rx = download_hashmap_rx.lock().await; // each node record is serialized into one fixed-width payload so // startup import can safely drain the response in-place. if let Some(mut buffer) = rx.recv().await { while buffer.len() >= NODE_RECORD_BYTES { let chunk: Vec = buffer.drain(0..NODE_RECORD_BYTES).collect(); // The first part of each record describes the advertised node address and IP. let Some(address) = Wallet::bytes_to_short_address(&chunk[0..NODE_IP_OFFSET]) else { continue; }; let ip = binary_to_ip(chunk[NODE_IP_OFFSET..NODE_BLOCKS_MINED_OFFSET].to_vec()); let _advertised_blocks_mined = chunk[NODE_BLOCKS_MINED_OFFSET]; let blocks_mined = 0_u8; let added_by_bytes = &chunk[NODE_ADDED_BY_OFFSET..NODE_ADDED_TIMESTAMP_OFFSET]; let added_by = if added_by_bytes.iter().all(|&byte| byte == 0) { String::new() } else { Wallet::bytes_to_long_address(added_by_bytes.to_vec()) }; let added_timestamp = u64::from_le_bytes( chunk[NODE_ADDED_TIMESTAMP_OFFSET..NODE_ADDED_SIGNATURE_OFFSET] .try_into() .unwrap(), ); let added_signature = decode_optional_signature( &chunk[NODE_ADDED_SIGNATURE_OFFSET..NODE_DELETED_BY_OFFSET], ); let remote_ip = ""; // Add records are imported through NodeInfo so local validation/signing rules stay central. NodeInfo::add_address(AddAddressParams { map: command_map.clone(), edit: SignedNodeEdit { address: address.clone(), ip: ip.clone(), modified_by: added_by, modified_timestamp: added_timestamp, modified_signature: added_signature, }, blocks_mined, remote_ip: remote_ip.to_string(), db: db.clone(), wallet: wallet.clone(), connections_key: connections_key.to_string(), }) .await; if !chunk[NODE_DELETED_BY_OFFSET..NODE_DELETED_TIMESTAMP_OFFSET] .iter() .all(|&byte| byte == 0) { // Deleted fields are optional; when present, replay the delete record locally too. let deleted_by = Wallet::bytes_to_long_address( chunk[NODE_DELETED_BY_OFFSET..NODE_DELETED_TIMESTAMP_OFFSET].to_vec(), ); let deleted_timestamp = u64::from_le_bytes( chunk[NODE_DELETED_TIMESTAMP_OFFSET..NODE_DELETED_BLOCK_OFFSET] .try_into() .unwrap(), ); let deleted_block = u32::from_le_bytes( chunk[NODE_DELETED_BLOCK_OFFSET..NODE_DELETED_SIGNATURE_OFFSET] .try_into() .unwrap(), ); let deleted_signature = decode_optional_signature( &chunk[NODE_DELETED_SIGNATURE_OFFSET..NODE_RECORD_BYTES], ); let deleted_address = address.clone(); let _ = NodeInfo::delete_address(DeleteAddressParams { map: command_map.clone(), edit: SignedNodeEdit { address, ip, modified_by: deleted_by, modified_timestamp: deleted_timestamp, modified_signature: deleted_signature, }, remote_ip: remote_ip.to_string(), db: db.clone(), wallet: wallet.clone(), connections_key: connections_key.to_string(), }) .await; NodeInfo::set_deleted_block_from_mapping(&deleted_address, deleted_block).await; } } } }