// Copyright 2024 MaidSafe.net limited.
//
// This SAFE Network Software is licensed to you under The General Public License (GPL), version 3.
// Unless required by applicable law or agreed to in writing, the SAFE Network Software distributed
// under the GPL Licence is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. Please review the Licences for the specific language governing
// permissions and limitations relating to use of the SAFE Network Software.

use std::collections::BTreeSet;

use crate::error::PutValidationError;
use crate::{Marker, Result, node::Node};
use ant_evm::ProofOfPayment;
use ant_evm::merkle_payment_vault::get_merkle_payment_info;
use ant_evm::merkle_payments::MerklePaymentProof;
use ant_evm::payment_vault::verify_data_payment;
use ant_protocol::storage::GraphEntry;
use ant_protocol::{
    NetworkAddress, PrettyPrintRecordKey,
    storage::{
        Chunk, ChunkAddress, DataTypes, GraphEntryAddress, Pointer, PointerAddress, RecordHeader,
        RecordKind, Scratchpad, ValidationType, try_deserialize_record, try_serialize_record,
    },
};
use libp2p::PeerId;
use libp2p::kad::{KBucketDistance as Distance, Record, RecordKey, U256};
use std::collections::HashSet;
use xor_name::XorName;

// We retry the payment verification once after waiting this many seconds to rule out the possibility of an EVM node state desync
const RETRY_PAYMENT_VERIFICATION_WAIT_TIME_SECS: u64 = 5;

impl Node {
    /// Validate a record and its payment, and store the record to the RecordStore
    pub(crate) async fn validate_and_store_record(
        &self,
        record: Record,
    ) -> Result<(), PutValidationError> {
        let record_header = RecordHeader::from_record(&record)
            .map_err(|_| PutValidationError::InvalidRecordHeader)?;

        match record_header.kind {
            RecordKind::DataWithPayment(DataTypes::Chunk) => {
                let record_key = record.key.clone();
                let (payment, chunk) = try_deserialize_record::<(ProofOfPayment, Chunk)>(&record)
                    .map_err(|_| {
                    PutValidationError::InvalidRecord(
                        PrettyPrintRecordKey::from(&record.key).into_owned(),
                    )
                })?;
                let already_exists = self
                    .validate_key_and_existence(&chunk.network_address(), &record_key)
                    .await?;

                // Validate the payment and that we received what we asked.
                // This stores any payments to disk
                let payment_res = self
                    .payment_for_us_exists_and_is_still_valid(
                        &chunk.network_address(),
                        DataTypes::Chunk,
                        payment.clone(),
                    )
                    .await;

                // Now that we've taken any money passed to us, regardless of the payment's validity,
                // if we already have the data we can return early
                if already_exists {
                    // Client changed to upload to ALL payees, hence no longer need this.
                    // May need again once client change back to upload to just one to save traffic.
                    // self.replicate_valid_fresh_record(
                    //     record_key,
                    //     DataTypes::Chunk,
                    //     ValidationType::Chunk,
                    //     Some(payment),
                    // );

                    // Notify replication_fetcher to mark the attempt as completed.
                    // Send the notification earlier to avoid it got skipped due to:
                    // the record becomes stored during the fetch because of other interleaved process.
                    self.network()
                        .notify_fetch_completed(record.key.clone(), ValidationType::Chunk);

                    debug!(
                        "Chunk with addr {:?} already exists: {already_exists}, payment extracted.",
                        chunk.network_address()
                    );
                    return Ok(());
                }

                // Finally before we store, lets bail for any payment issues
                payment_res?;

                // Writing chunk to disk takes time, hence try to execute it first.
                // So that when the replicate target asking for the copy,
                // the node can have a higher chance to respond.
                let store_chunk_result = self.store_chunk(&chunk, true);

                if store_chunk_result.is_ok() {
                    Marker::ValidPaidChunkPutFromClient(&PrettyPrintRecordKey::from(&record.key))
                        .log();
                    // Client changed to upload to ALL payees, hence no longer need this.
                    // May need again once client change back to upload to just one to save traffic.
                    // self.replicate_valid_fresh_record(
                    //     record_key,
                    //     DataTypes::Chunk,
                    //     ValidationType::Chunk,
                    //     Some(payment),
                    // );

                    // Notify replication_fetcher to mark the attempt as completed.
                    // Send the notification earlier to avoid it got skipped due to:
                    // the record becomes stored during the fetch because of other interleaved process.
                    self.network()
                        .notify_fetch_completed(record.key.clone(), ValidationType::Chunk);
                }

                store_chunk_result
            }

            RecordKind::DataOnly(DataTypes::Chunk) => {
                error!("Chunk should not be validated at this point. Got a PUT without payment.");
                Err(PutValidationError::NoPayment(
                    PrettyPrintRecordKey::from(&record.key).into_owned(),
                ))
            }
            RecordKind::DataWithPayment(DataTypes::Scratchpad) => {
                let record_key = record.key.clone();
                let (payment, scratchpad) = try_deserialize_record::<(ProofOfPayment, Scratchpad)>(
                    &record,
                )
                .map_err(|_| {
                    PutValidationError::InvalidRecord(
                        PrettyPrintRecordKey::from(&record.key).into_owned(),
                    )
                })?;
                let _already_exists = self
                    .validate_key_and_existence(&scratchpad.network_address(), &record_key)
                    .await?;

                // Validate the payment and that we received what we asked.
                // This stores any payments to disk
                let payment_res = self
                    .payment_for_us_exists_and_is_still_valid(
                        &scratchpad.network_address(),
                        DataTypes::Scratchpad,
                        payment.clone(),
                    )
                    .await;

                // Finally before we store, lets bail for any payment issues
                payment_res?;

                // Writing records to disk takes time, hence try to execute it first.
                // So that when the replicate target asking for the copy,
                // the node can have a higher chance to respond.
                let store_scratchpad_result = self
                    .validate_and_store_scratchpad_record(scratchpad, record_key.clone(), true)
                    .await;

                match store_scratchpad_result {
                    // if we're receiving this scratchpad PUT again, and we have been paid,
                    // we eagerly retry replicaiton as it seems like other nodes are having trouble
                    // did not manage to get this scratchpad as yet.
                    Ok(_) | Err(PutValidationError::OutdatedRecordCounter { .. }) => {
                        let content_hash = XorName::from_content(&record.value);
                        Marker::ValidScratchpadRecordPutFromClient(&PrettyPrintRecordKey::from(
                            &record_key,
                        ))
                        .log();

                        // Notify replication_fetcher to mark the attempt as completed.
                        // Send the notification earlier to avoid it got skipped due to:
                        // the record becomes stored during the fetch because of other interleaved process.
                        self.network().notify_fetch_completed(
                            record_key,
                            ValidationType::NonChunk(content_hash),
                        );
                    }
                    Err(_) => {}
                }

                store_scratchpad_result
            }
            RecordKind::DataOnly(DataTypes::Scratchpad) => {
                // make sure we already have this scratchpad locally, else reject it as first time upload needs payment
                let key = record.key.clone();
                let scratchpad = try_deserialize_record::<Scratchpad>(&record).map_err(|_| {
                    PutValidationError::InvalidRecord(
                        PrettyPrintRecordKey::from(&record.key).into_owned(),
                    )
                })?;
                let net_addr = NetworkAddress::ScratchpadAddress(*scratchpad.address());
                let pretty_key = PrettyPrintRecordKey::from(&key);
                trace!("Got record to store without payment for scratchpad at {pretty_key:?}");
                if !self.validate_key_and_existence(&net_addr, &key).await? {
                    warn!("Ignore store without payment for scratchpad at {pretty_key:?}");
                    return Err(PutValidationError::NoPayment(
                        PrettyPrintRecordKey::from(&record.key).into_owned(),
                    ));
                }

                // store the scratchpad
                self.validate_and_store_scratchpad_record(scratchpad, key, true)
                    .await
            }
            RecordKind::DataOnly(DataTypes::GraphEntry) => {
                // Transactions should always be paid for
                error!("Transaction should not be validated at this point");
                Err(PutValidationError::NoPayment(
                    PrettyPrintRecordKey::from(&record.key).into_owned(),
                ))
            }
            RecordKind::DataWithPayment(DataTypes::GraphEntry) => {
                let (payment, graph_entry) =
                    try_deserialize_record::<(ProofOfPayment, GraphEntry)>(&record).map_err(
                        |_| {
                            PutValidationError::InvalidRecord(
                                PrettyPrintRecordKey::from(&record.key).into_owned(),
                            )
                        },
                    )?;

                // check if the deserialized value's GraphEntryAddress matches the record's key
                let net_addr = NetworkAddress::from(graph_entry.address());
                let key = net_addr.to_record_key();
                let pretty_key = PrettyPrintRecordKey::from(&key);
                if record.key != key {
                    warn!(
                        "Record's key {pretty_key:?} does not match with the value's GraphEntryAddress, ignoring PUT."
                    );
                    return Err(PutValidationError::RecordKeyMismatch);
                }

                let already_exists = self.validate_key_and_existence(&net_addr, &key).await?;

                // The GraphEntry may already exist during the replication.
                // The payment shall get deposit to self even the GraphEntry already presents.
                // However, if the GraphEntry is already present, the incoming one shall be
                // appended with the existing one, if content is different.
                if let Err(err) = self
                    .payment_for_us_exists_and_is_still_valid(
                        &net_addr,
                        DataTypes::GraphEntry,
                        payment.clone(),
                    )
                    .await
                {
                    if already_exists {
                        debug!(
                            "Payment of the incoming existing GraphEntry {pretty_key:?} having error {err:?}"
                        );
                    } else {
                        error!(
                            "Payment of the incoming new GraphEntry {pretty_key:?} having error {err:?}"
                        );
                        return Err(err);
                    }
                }

                let res = self
                    .validate_merge_and_store_graphentries(vec![graph_entry], &key, true)
                    .await;
                if res.is_ok() {
                    let content_hash = XorName::from_content(&record.value);
                    Marker::ValidGraphEntryPutFromClient(&PrettyPrintRecordKey::from(&record.key))
                        .log();
                    // Client changed to upload to ALL payees, hence no longer need this.
                    // May need again once client change back to upload to just one to save traffic.
                    // self.replicate_valid_fresh_record(
                    //     record.key.clone(),
                    //     DataTypes::GraphEntry,
                    //     ValidationType::NonChunk(content_hash),
                    //     Some(payment),
                    // );

                    // Notify replication_fetcher to mark the attempt as completed.
                    // Send the notification earlier to avoid it got skipped due to:
                    // the record becomes stored during the fetch because of other interleaved process.
                    self.network().notify_fetch_completed(
                        record.key.clone(),
                        ValidationType::NonChunk(content_hash),
                    );
                }
                res
            }
            RecordKind::DataOnly(DataTypes::Pointer) => {
                let pointer = try_deserialize_record::<Pointer>(&record).map_err(|_| {
                    PutValidationError::InvalidRecord(
                        PrettyPrintRecordKey::from(&record.key).into_owned(),
                    )
                })?;
                let net_addr = NetworkAddress::from(pointer.address());
                let pretty_key = PrettyPrintRecordKey::from(&record.key);
                let already_exists = self
                    .validate_key_and_existence(&net_addr, &record.key)
                    .await?;

                if !already_exists {
                    warn!(
                        "Pointer at address: {:?}, key: {:?} does not exist locally, rejecting PUT without payment",
                        pointer.address(),
                        pretty_key
                    );
                    return Err(PutValidationError::NoPayment(
                        PrettyPrintRecordKey::from(&record.key).into_owned(),
                    ));
                }

                let res = self
                    .validate_and_store_pointer_record(pointer, record.key.clone(), true)
                    .await;
                if res.is_ok() {
                    let content_hash = XorName::from_content(&record.value);
                    Marker::ValidPointerPutFromClient(&pretty_key).log();

                    // Notify replication_fetcher to mark the attempt as completed.
                    self.network().notify_fetch_completed(
                        record.key.clone(),
                        ValidationType::NonChunk(content_hash),
                    );
                }
                res
            }
            RecordKind::DataWithPayment(DataTypes::Pointer) => {
                let (payment, pointer) =
                    try_deserialize_record::<(ProofOfPayment, Pointer)>(&record).map_err(|_| {
                        PutValidationError::InvalidRecord(
                            PrettyPrintRecordKey::from(&record.key).into_owned(),
                        )
                    })?;

                let net_addr = NetworkAddress::from(pointer.address());
                let pretty_key = PrettyPrintRecordKey::from(&record.key);
                let already_exists = self
                    .validate_key_and_existence(&net_addr, &record.key)
                    .await?;

                // The pointer may already exist during the replication.
                // The payment shall get deposit to self even if the pointer already exists.
                if let Err(err) = self
                    .payment_for_us_exists_and_is_still_valid(
                        &net_addr,
                        DataTypes::Pointer,
                        payment.clone(),
                    )
                    .await
                {
                    if already_exists {
                        debug!(
                            "Payment of the incoming exists pointer {pretty_key:?} having error {err:?}"
                        );
                    } else {
                        error!(
                            "Payment of the incoming non-exist pointer {pretty_key:?} having error {err:?}"
                        );
                        return Err(err);
                    }
                }

                let res = self
                    .validate_and_store_pointer_record(pointer, record.key.clone(), true)
                    .await;
                if res.is_ok() {
                    let content_hash = XorName::from_content(&record.value);
                    Marker::ValidPointerPutFromClient(&pretty_key).log();

                    // Notify replication_fetcher to mark the attempt as completed.
                    self.network().notify_fetch_completed(
                        record.key.clone(),
                        ValidationType::NonChunk(content_hash),
                    );
                }
                res
            }
            // Merkle payment records
            RecordKind::DataWithMerklePayment(data_type) => {
                self.validate_and_store_merkle_payment_record(record, data_type)
                    .await
            }
        }
    }

    /// Store a pre-validated, and already paid record to the RecordStore
    pub(crate) async fn store_replicated_in_record(
        &self,
        record: Record,
    ) -> Result<(), PutValidationError> {
        debug!(
            "Storing record which was replicated to us {:?}",
            PrettyPrintRecordKey::from(&record.key)
        );
        let record_header = RecordHeader::from_record(&record)
            .map_err(|_| PutValidationError::InvalidRecordHeader)?;
        match record_header.kind {
            // A separate flow handles record with payment
            RecordKind::DataWithPayment(_) => {
                warn!("Prepaid record came with Payment, which should be handled in another flow");
                Err(PutValidationError::UnexpectedRecordWithPayment(
                    PrettyPrintRecordKey::from(&record.key).into_owned(),
                ))
            }
            // Merkle payment records should not be replicated with payment proof
            RecordKind::DataWithMerklePayment(_) => {
                warn!("Record came with Merkle Payment, which should be handled in another flow");
                Err(PutValidationError::UnexpectedRecordWithPayment(
                    PrettyPrintRecordKey::from(&record.key).into_owned(),
                ))
            }
            RecordKind::DataOnly(DataTypes::Chunk) => {
                let chunk = try_deserialize_record::<Chunk>(&record).map_err(|_| {
                    PutValidationError::InvalidRecord(
                        PrettyPrintRecordKey::from(&record.key).into_owned(),
                    )
                })?;

                let record_key = record.key.clone();
                let already_exists = self
                    .validate_key_and_existence(&chunk.network_address(), &record_key)
                    .await?;
                if already_exists {
                    debug!(
                        "Chunk with addr {:?} already exists?: {already_exists}, do nothing",
                        chunk.network_address()
                    );
                    return Ok(());
                }

                self.store_chunk(&chunk, false)
            }
            RecordKind::DataOnly(DataTypes::Scratchpad) => {
                let key = record.key.clone();
                let scratchpad = try_deserialize_record::<Scratchpad>(&record).map_err(|_| {
                    PutValidationError::InvalidRecord(
                        PrettyPrintRecordKey::from(&record.key).into_owned(),
                    )
                })?;
                self.validate_and_store_scratchpad_record(scratchpad, key, false)
                    .await
            }
            RecordKind::DataOnly(DataTypes::GraphEntry) => {
                let record_key = record.key.clone();
                let graph_entries =
                    try_deserialize_record::<Vec<GraphEntry>>(&record).map_err(|_| {
                        PutValidationError::InvalidRecord(
                            PrettyPrintRecordKey::from(&record.key).into_owned(),
                        )
                    })?;
                self.validate_merge_and_store_graphentries(graph_entries, &record_key, false)
                    .await
            }
            RecordKind::DataOnly(DataTypes::Pointer) => {
                let pointer = try_deserialize_record::<Pointer>(&record).map_err(|_| {
                    PutValidationError::InvalidRecord(
                        PrettyPrintRecordKey::from(&record.key).into_owned(),
                    )
                })?;
                let key = record.key.clone();
                self.validate_and_store_pointer_record(pointer, key, false)
                    .await
            }
        }
    }

    /// Check key is valid compared to the network name, and if we already have this data or not.
    /// returns true if data already exists locally
    pub(crate) async fn validate_key_and_existence(
        &self,
        address: &NetworkAddress,
        expected_record_key: &RecordKey,
    ) -> Result<bool, PutValidationError> {
        let data_key = address.to_record_key();
        let pretty_key = PrettyPrintRecordKey::from(&data_key);

        if expected_record_key != &data_key {
            warn!(
                "record key: {:?}, key: {:?}",
                PrettyPrintRecordKey::from(expected_record_key),
                pretty_key
            );
            warn!("Record's key does not match with the value's address, ignoring PUT.");
            return Err(PutValidationError::RecordKeyMismatch);
        }

        let present_locally = self
            .network()
            .is_record_key_present_locally(&data_key)
            .await
            .map_err(|_| PutValidationError::LocalSwarmError)?;

        if present_locally {
            // We may short circuit if the Record::key is present locally;
            debug!(
                "Record with addr {:?} already exists, not overwriting",
                address
            );
            return Ok(true);
        }

        Ok(false)
    }

    /// Store a `Chunk` to the RecordStore
    pub(crate) fn store_chunk(
        &self,
        chunk: &Chunk,
        is_client_put: bool,
    ) -> Result<(), PutValidationError> {
        let key = NetworkAddress::from(*chunk.address()).to_record_key();
        let pretty_key = PrettyPrintRecordKey::from(&key).into_owned();

        // reject if chunk is too large
        if chunk.size() > Chunk::MAX_SIZE {
            warn!(
                "Chunk at {pretty_key:?} is too large: {} bytes, when max size is {} bytes",
                chunk.size(),
                Chunk::MAX_SIZE
            );
            return Err(PutValidationError::OversizedChunk(
                chunk.size(),
                Chunk::MAX_SIZE,
            ));
        }

        let record = Record {
            key,
            value: try_serialize_record(&chunk, RecordKind::DataOnly(DataTypes::Chunk))
                .map_err(|_| PutValidationError::RecordSerializationFailed(pretty_key.clone()))?
                .to_vec(),
            publisher: None,
            expires: None,
        };

        // finally store the Record directly into the local storage
        self.network().put_local_record(record, is_client_put);

        self.record_metrics(Marker::ValidChunkRecordPutFromNetwork(&pretty_key));

        // TODO: currently ignored, re-enable once start to handle
        // self.events_channel()
        //     .broadcast(crate::NodeEvent::ChunkStored(chunk_addr));

        Ok(())
    }

    /// Validate and store a `Scratchpad` to the RecordStore
    ///
    /// When a node receives an update packet:
    /// Verify Name: It MUST hash the provided public key and confirm it matches the name in the packet.
    /// Check Counter: It MUST ensure that the new counter value is strictly greater than the currently stored value to prevent replay attacks.
    /// Verify Signature: It MUST use the public key to verify the BLS12-381 signature against the content hash and the counter.
    /// Accept or Reject: If all verifications succeed, the node MUST accept the packet and replace any previous version. Otherwise, it MUST reject the update.
    pub(crate) async fn validate_and_store_scratchpad_record(
        &self,
        scratchpad: Scratchpad,
        record_key: RecordKey,
        is_client_put: bool,
    ) -> Result<(), PutValidationError> {
        // owner PK is defined herein, so as long as record key and this match, we're good
        let addr = scratchpad.address();
        let count = scratchpad.counter();
        debug!("Validating and storing scratchpad {addr:?} with count {count}");

        // check if the deserialized value's ScratchpadAddress matches the record's key
        let scratchpad_key = NetworkAddress::ScratchpadAddress(*addr).to_record_key();
        if scratchpad_key != record_key {
            warn!("Record's key does not match with the value's ScratchpadAddress, ignoring PUT.");
            return Err(PutValidationError::RecordKeyMismatch);
        }

        // check if the Scratchpad is present locally that we don't have a newer version
        if let Some(local_pad) = self
            .network()
            .get_local_record(&scratchpad_key)
            .await
            .map_err(|_| PutValidationError::LocalSwarmError)?
        {
            let local_pad = try_deserialize_record::<Scratchpad>(&local_pad).map_err(|_| {
                PutValidationError::InvalidRecord(
                    PrettyPrintRecordKey::from(&scratchpad_key).into_owned(),
                )
            })?;
            if local_pad == scratchpad {
                debug!("Scratchpad at {addr:?} already exists locally, skip to success");
                return Ok(());
            }
            if local_pad.counter() >= scratchpad.counter() {
                warn!(
                    "Rejecting Scratchpad PUT with counter less than or equal to the current counter"
                );
                return Err(PutValidationError::OutdatedRecordCounter {
                    counter: scratchpad.counter(),
                    expected: local_pad.counter(),
                });
            }
        }

        // ensure data integrity
        if !scratchpad.verify_signature() {
            warn!("Rejecting Scratchpad PUT with invalid signature");
            return Err(PutValidationError::InvalidScratchpadSignature);
        }

        // ensure the scratchpad is not too big
        if scratchpad.is_too_big() {
            warn!("Rejecting Scratchpad PUT with too big size");
            return Err(PutValidationError::ScratchpadTooBig(scratchpad.size()));
        }

        info!(
            "Storing sratchpad {addr:?} with content of {:?} as Record locally",
            scratchpad.encrypted_data_hash()
        );

        let record = Record {
            key: scratchpad_key.clone(),
            value: try_serialize_record(&scratchpad, RecordKind::DataOnly(DataTypes::Scratchpad))
                .map_err(|_| {
                    PutValidationError::RecordSerializationFailed(
                        PrettyPrintRecordKey::from(&scratchpad_key).into_owned(),
                    )
                })?
                .to_vec(),
            publisher: None,
            expires: None,
        };
        self.network()
            .put_local_record(record.clone(), is_client_put);

        let pretty_key = PrettyPrintRecordKey::from(&scratchpad_key);

        self.record_metrics(Marker::ValidScratchpadRecordPutFromNetwork(&pretty_key));

        // Client changed to upload to ALL payees, hence no longer need this.
        // May need again once client change back to upload to just one to save traffic.
        // if is_client_put {
        //     let content_hash = XorName::from_content(&record.value);
        //     // ScratchPad update is a special upload that without payment,
        //     // but must have an existing copy to update.
        //     self.replicate_valid_fresh_record(
        //         scratchpad_key,
        //         DataTypes::Scratchpad,
        //         ValidationType::NonChunk(content_hash),
        //         payment,
        //     );
        // }

        Ok(())
    }

    /// Validate and store `Vec<GraphEntry>` to the RecordStore
    /// If we already have a GraphEntry at this address, the Vec is extended and stored.
    pub(crate) async fn validate_merge_and_store_graphentries(
        &self,
        entries: Vec<GraphEntry>,
        record_key: &RecordKey,
        is_client_put: bool,
    ) -> Result<(), PutValidationError> {
        let pretty_key = PrettyPrintRecordKey::from(record_key);
        debug!("Validating GraphEntries before storage at {pretty_key:?}");

        // only keep GraphEntries that match the record key
        let entries_for_key: Vec<GraphEntry> = entries
            .into_iter()
            .filter(|s| {
                // get the record key for the GraphEntry
                let graph_entry_address = s.address();
                let network_address = NetworkAddress::from(graph_entry_address);
                let graph_entry_record_key = network_address.to_record_key();
                let graph_entry_pretty = PrettyPrintRecordKey::from(&graph_entry_record_key);
                if &graph_entry_record_key != record_key {
                    warn!("Ignoring GraphEntry for another record key {graph_entry_pretty:?} when verifying: {pretty_key:?}");
                    return false;
                }
                true
            })
            .collect();

        // if we have no GraphEntries to verify, return early
        if entries_for_key.is_empty() {
            warn!("Found no valid GraphEntries to verify upon validation for {pretty_key:?}");
            return Err(PutValidationError::EmptyGraphEntry(
                pretty_key.clone().into_owned(),
            ));
        }

        // verify the GraphEntries
        let mut validated_entries: BTreeSet<GraphEntry> = entries_for_key
            .into_iter()
            .filter(|t| t.verify_signature())
            .collect();

        // skip if none are valid
        let addr = match validated_entries.first() {
            None => {
                warn!("Found no validated GraphEntries to store at {pretty_key:?}");
                return Ok(());
            }
            Some(t) => t.address(),
        };

        // add local GraphEntries to the validated GraphEntries, turn to Vec
        let local_entries = self.get_local_graphentries(addr).await?;
        let existing_entry = local_entries.len();
        validated_entries.extend(local_entries.into_iter());
        let validated_entries: Vec<GraphEntry> = validated_entries.into_iter().collect();

        // No need to write to disk if nothing new.
        if existing_entry == validated_entries.len() {
            debug!("No new entry of the GraphEntry {pretty_key:?}");
            return Ok(());
        }

        // store the record into the local storage
        let record = Record {
            key: record_key.clone(),
            value: try_serialize_record(
                &validated_entries,
                RecordKind::DataOnly(DataTypes::GraphEntry),
            )
            .map_err(|_| {
                PutValidationError::RecordSerializationFailed(pretty_key.clone().into_owned())
            })?
            .to_vec(),
            publisher: None,
            expires: None,
        };
        self.network().put_local_record(record, is_client_put);
        debug!("Successfully stored validated GraphEntries at {pretty_key:?}");

        // Just log the multiple GraphEntries
        if validated_entries.len() > 1 {
            debug!(
                "Got multiple GraphEntry(s) of len {} at {pretty_key:?}",
                validated_entries.len()
            );
        }

        self.record_metrics(Marker::ValidGraphEntryRecordPutFromNetwork(&pretty_key));
        Ok(())
    }

    /// Perform validations on the provided `Record`.
    pub(crate) async fn payment_for_us_exists_and_is_still_valid(
        &self,
        address: &NetworkAddress,
        data_type: DataTypes,
        payment: ProofOfPayment,
    ) -> Result<(), PutValidationError> {
        let key = address.to_record_key();
        let pretty_key = PrettyPrintRecordKey::from(&key).into_owned();

        // check if the quote is valid
        let self_peer_id = self.network().peer_id();
        if !payment.verify_for(self_peer_id) {
            warn!("Payment is not valid for record {pretty_key}");
            return Err(PutValidationError::PaymentNotMadeToOurNode(
                pretty_key.clone(),
            ));
        }

        // verify data type matches
        let own_quotes: Vec<_> = payment.quotes_by_peer(&self_peer_id);
        if !payment.verify_data_type(data_type.get_index()) {
            warn!("Payment quote has wrong data type for record {pretty_key}");
            return Err(PutValidationError::PaymentMadeToIncorrectDataType(
                pretty_key.clone(),
            ));
        }

        // verify the claimed payees are all known to us within the certain range.
        // note: self is already included in the returned list
        let closest_k_peers = self
            .network()
            .get_k_closest_local_peers_to_the_target(Some(address.clone()))
            .await
            .map_err(|_| PutValidationError::LocalSwarmError)?;

        let mut payees = payment.payees();
        payees.retain(|peer_id| !closest_k_peers.iter().any(|(p, _)| p == peer_id));
        if !payees.is_empty() {
            // There might be payee got blocked by us or churned out from our perspective.
            // We shall still consider the payment is valid whenever payees are close enough.
            // In case we don't have enough knowledge of the network, we shall trust the payment.
            if let Some(network_density) = self
                .network()
                .get_network_density()
                .await
                .map_err(|_| PutValidationError::LocalSwarmError)?
            {
                payees.retain(|peer_id| {
                    NetworkAddress::from(*peer_id).distance(address) > network_density
                });

                if !payees.is_empty() {
                    warn!(
                        "Payment quote has out-of-range payees for record {pretty_key}. Payees: {payees:?}"
                    );
                    return Err(PutValidationError::PaymentQuoteOutOfRange {
                        record_key: pretty_key.clone(),
                        payees: payees.clone(),
                    });
                }
            }
        }

        // check if payment is valid on chain
        let payments_to_verify = payment.digest();
        let owned_payment_quotes: Vec<_> = own_quotes.iter().map(|quote| quote.hash()).collect();
        let reward_amount = match verify_data_payment(
            self.evm_network(),
            owned_payment_quotes.clone(),
            payments_to_verify.clone(),
        )
        .await
        {
            Ok(amount) => amount,
            Err(e) => {
                warn!("Failed to verify record payment on the first attempt: {e}");
                // Try again, because there could be a possible EVM node desync
                tokio::time::sleep(std::time::Duration::from_secs(
                    RETRY_PAYMENT_VERIFICATION_WAIT_TIME_SECS,
                ))
                .await;
                verify_data_payment(self.evm_network(), owned_payment_quotes, payments_to_verify)
                    .await
                    .inspect_err(|e| {
                        warn!("Failed to verify record payment on the second attempt: {e}");
                    })
                    .map_err(|e| PutValidationError::PaymentVerificationFailed {
                        record_key: pretty_key.clone(),
                        error: e,
                    })?
            }
        };

        debug!("Payment of {reward_amount:?} is valid for record {pretty_key}");

        if !reward_amount.is_zero() {
            // Notify `record_store` that the node received a payment.
            self.network().notify_payment_received();

            #[cfg(feature = "open-metrics")]
            if let Some(metrics_recorder) = self.metrics_recorder() {
                // FIXME: We would reach the MAX if the storecost is scaled up.
                let current_value = metrics_recorder.current_reward_wallet_balance.get();
                let new_value =
                    current_value.saturating_add(reward_amount.try_into().unwrap_or(i64::MAX));
                let _ = metrics_recorder
                    .current_reward_wallet_balance
                    .set(new_value);
            }

            // TODO: currently ignored, re-enable once going to handle this.
            // self.events_channel()
            //     .broadcast(crate::NodeEvent::RewardReceived(
            //         AttoTokens::from(reward_amount),
            //         address.clone(),
            //     ));

            // vdash metric (if modified please notify at https://github.com/happybeing/vdash/issues):
            info!(
                "Total payment of {reward_amount:?} atto tokens accepted for record {pretty_key}"
            );

            // loud mode: print a celebratory message to console
            #[cfg(feature = "loud")]
            {
                println!("🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟   RECEIVED REWARD   🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟");
                println!(
                    "Total payment of {reward_amount:?} atto tokens accepted for record {pretty_key}"
                );
                println!("🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟🌟");
            }
        }

        Ok(())
    }

    /// Get the local GraphEntries for the provided `GraphEntryAddress`
    /// This only fetches the GraphEntries from the local store and does not perform any network operations.
    async fn get_local_graphentries(
        &self,
        addr: GraphEntryAddress,
    ) -> Result<Vec<GraphEntry>, PutValidationError> {
        // get the local GraphEntries
        let record_key = NetworkAddress::from(addr).to_record_key();
        debug!("Checking for local GraphEntries with key: {record_key:?}");
        let local_record = match self
            .network()
            .get_local_record(&record_key)
            .await
            .map_err(|_| PutValidationError::LocalSwarmError)?
        {
            Some(r) => r,
            None => {
                debug!("GraphEntry is not present locally: {record_key:?}");
                return Ok(vec![]);
            }
        };

        let local_entries: Vec<GraphEntry> =
            try_deserialize_record(&local_record).map_err(|_| {
                PutValidationError::InvalidRecord(
                    PrettyPrintRecordKey::from(&record_key).into_owned(),
                )
            })?;
        Ok(local_entries)
    }

    /// Get the local Pointer for the provided `PointerAddress`
    /// This only fetches the Pointer from the local store and does not perform any network operations.
    /// If the local Pointer is not present or corrupted, returns `None`.
    async fn get_local_pointer(&self, addr: PointerAddress) -> Option<Pointer> {
        // get the local Pointer
        let record_key = NetworkAddress::from(addr).to_record_key();
        debug!("Checking for local Pointer with key: {record_key:?}");
        let local_record = match self.network().get_local_record(&record_key).await {
            Ok(Some(r)) => r,
            Ok(None) => {
                debug!("Pointer is not present locally: {record_key:?}");
                return None;
            }
            Err(e) => {
                error!("Failed to get Pointer record at {addr:?}: {e}");
                return None;
            }
        };

        // deserialize the record and get the Pointer
        let local_header = match RecordHeader::from_record(&local_record) {
            Ok(h) => h,
            Err(_) => {
                error!("Failed to deserialize Pointer record at {addr:?}");
                return None;
            }
        };
        let record_kind = local_header.kind;
        if !matches!(record_kind, RecordKind::DataOnly(DataTypes::Pointer)) {
            error!("Found a {record_kind} when expecting to find Pointer at {addr:?}");
            return None;
        }
        let local_pointer: Pointer = match try_deserialize_record(&local_record) {
            Ok(p) => p,
            Err(_) => {
                error!("Failed to deserialize Pointer record at {addr:?}");
                return None;
            }
        };
        Some(local_pointer)
    }

    /// Validate and store a pointer record
    pub(crate) async fn validate_and_store_pointer_record(
        &self,
        pointer: Pointer,
        key: RecordKey,
        is_client_put: bool,
    ) -> Result<(), PutValidationError> {
        // Verify the pointer's signature
        if !pointer.verify_signature() {
            warn!("Pointer signature verification failed");
            return Err(PutValidationError::InvalidPointerSignature);
        }

        // Check if the pointer's address matches the record key
        let net_addr = NetworkAddress::from(pointer.address());
        if key != net_addr.to_record_key() {
            warn!("Pointer address does not match record key");
            return Err(PutValidationError::RecordKeyMismatch);
        }

        // Keep the pointer with the highest counter
        if let Some(local_pointer) = self.get_local_pointer(pointer.address()).await {
            if pointer == local_pointer {
                debug!("Pointer at {key:?} already exists locally, skipping validation");
                return Ok(());
            }
            if pointer.counter() <= local_pointer.counter() {
                warn!(
                    "Ignoring Pointer PUT at {key:?} with counter less than or equal to the current counter ({} <= {})",
                    pointer.counter(),
                    local_pointer.counter()
                );
                return Err(PutValidationError::OutdatedRecordCounter {
                    counter: pointer.counter(),
                    expected: local_pointer.counter(),
                });
            }
        }

        // Store the pointer
        let record = Record {
            key: key.clone(),
            value: try_serialize_record(&pointer, RecordKind::DataOnly(DataTypes::Pointer))
                .map_err(|_| {
                    PutValidationError::RecordSerializationFailed(
                        PrettyPrintRecordKey::from(&key).into_owned(),
                    )
                })?
                .to_vec(),
            publisher: None,
            expires: None,
        };
        self.network()
            .put_local_record(record.clone(), is_client_put);

        // Client changed to upload to ALL payees, hence no longer need this.
        // May need again once client change back to upload to just one to save traffic.
        // if is_client_put {
        //     let content_hash = XorName::from_content(&record.value);
        //     self.replicate_valid_fresh_record(
        //         key.clone(),
        //         DataTypes::Pointer,
        //         ValidationType::NonChunk(content_hash),
        //         payment,
        //     );
        // }

        info!("Successfully stored Pointer record at {key:?}");
        Ok(())
    }

    /// Validate and store a record with Merkle batch payment
    async fn validate_and_store_merkle_payment_record(
        &self,
        record: Record,
        data_type: DataTypes,
    ) -> Result<(), PutValidationError> {
        use ant_evm::merkle_payments::MerklePaymentProof;

        let record_key = record.key.clone();
        let pretty_key = PrettyPrintRecordKey::from(&record_key);

        match data_type {
            DataTypes::Chunk => {
                let (proof, chunk) = try_deserialize_record::<(MerklePaymentProof, Chunk)>(&record)
                    .map_err(|_| {
                        PutValidationError::InvalidRecord(pretty_key.clone().into_owned())
                    })?;
                let net_addr = chunk.network_address();

                // Check if already exists
                let already_exists = self
                    .validate_key_and_existence(&net_addr, &record_key)
                    .await?;
                if already_exists {
                    debug!(
                        "Chunk at {net_addr:?} already exists, skipping Merkle payment verification"
                    );
                    self.network()
                        .notify_fetch_completed(record_key, ValidationType::Chunk);
                    return Ok(());
                }

                // Verify Merkle payment
                self.verify_merkle_payment(&proof, &net_addr).await?;

                // Store chunk
                let result = self.store_chunk(&chunk, true);
                if result.is_ok() {
                    Marker::ValidPaidChunkPutFromClient(&pretty_key).log();
                    self.network()
                        .notify_fetch_completed(record_key, ValidationType::Chunk);
                }
                result
            }
            DataTypes::Scratchpad => {
                let (proof, scratchpad) =
                    try_deserialize_record::<(MerklePaymentProof, Scratchpad)>(&record).map_err(
                        |_| PutValidationError::InvalidRecord(pretty_key.clone().into_owned()),
                    )?;
                let net_addr = scratchpad.network_address();

                // Verify Merkle payment
                self.verify_merkle_payment(&proof, &net_addr).await?;

                // Store scratchpad
                let result = self
                    .validate_and_store_scratchpad_record(scratchpad, record_key.clone(), true)
                    .await;
                if result.is_ok() {
                    let content_hash = XorName::from_content(&record.value);
                    Marker::ValidScratchpadRecordPutFromClient(&pretty_key).log();
                    self.network()
                        .notify_fetch_completed(record_key, ValidationType::NonChunk(content_hash));
                }
                result
            }
            DataTypes::GraphEntry => {
                let (proof, graph_entry) =
                    try_deserialize_record::<(MerklePaymentProof, GraphEntry)>(&record).map_err(
                        |_| PutValidationError::InvalidRecord(pretty_key.clone().into_owned()),
                    )?;
                let net_addr = NetworkAddress::from(graph_entry.address());

                // Verify Merkle payment
                self.verify_merkle_payment(&proof, &net_addr).await?;

                // Store graph entry
                let result = self
                    .validate_merge_and_store_graphentries(vec![graph_entry], &record_key, true)
                    .await;
                if result.is_ok() {
                    let content_hash = XorName::from_content(&record.value);
                    Marker::ValidGraphEntryPutFromClient(&pretty_key).log();
                    self.network()
                        .notify_fetch_completed(record_key, ValidationType::NonChunk(content_hash));
                }
                result
            }
            DataTypes::Pointer => {
                let (proof, pointer) = try_deserialize_record::<(MerklePaymentProof, Pointer)>(
                    &record,
                )
                .map_err(|_| PutValidationError::InvalidRecord(pretty_key.clone().into_owned()))?;
                let net_addr = NetworkAddress::from(pointer.address());

                // Verify Merkle payment
                self.verify_merkle_payment(&proof, &net_addr).await?;

                // Store pointer
                let result = self
                    .validate_and_store_pointer_record(pointer, record_key.clone(), true)
                    .await;
                if result.is_ok() {
                    let content_hash = XorName::from_content(&record.value);
                    Marker::ValidPointerPutFromClient(&pretty_key).log();
                    self.network()
                        .notify_fetch_completed(record_key, ValidationType::NonChunk(content_hash));
                }
                result
            }
        }
    }

    /// Verify Merkle batch payment for a data address
    ///
    /// This performs:
    /// 0. Verify proof address matches target address
    /// 1. Query smart contract to get payment info
    /// 2. Verify Merkle proof structure and signatures
    /// 3. Verify network topology (paid nodes among closest 20)
    async fn verify_merkle_payment(
        &self,
        proof: &MerklePaymentProof,
        target_address: &NetworkAddress,
    ) -> Result<(), PutValidationError> {
        let record_key = target_address.to_record_key();
        let pretty_key = PrettyPrintRecordKey::from(&record_key);

        // Verify proof address matches target address
        if proof.address.0.to_vec() != target_address.as_bytes() {
            let error_msg = format!(
                "Merkle proof address mismatch: proof is for {:?} but data is at {:?}. \
                 This prevents reusing a proof paid for one address to store data at another address.",
                hex::encode(proof.address.0),
                hex::encode(target_address.as_bytes())
            );
            warn!("{error_msg}");
            return Err(PutValidationError::MerklePaymentVerificationFailed {
                record_key: pretty_key.clone().into_owned(),
                error: error_msg,
            });
        }

        // Query smart contract to get payment info
        let winner_pool_hash = proof.winner_pool_hash();

        let contract_payment_info =
            if let Ok(info) = get_merkle_payment_info(self.evm_network(), winner_pool_hash).await {
                info
            } else {
                warn!("Failed to get Merkle payment info on the first attempt, retrying...");
                // Try again, because there could be a possible EVM node desync
                tokio::time::sleep(std::time::Duration::from_secs(
                    RETRY_PAYMENT_VERIFICATION_WAIT_TIME_SECS,
                ))
                .await;

                get_merkle_payment_info(self.evm_network(), winner_pool_hash)
                    .await
                    .inspect_err(|e| {
                        warn!("Failed to get Merkle payment info on the second attempt: {e}");
                    })
                    .map_err(|e| {
                        let error_msg = format!(
                            "Failed to get payment info for winner pool hash {}: {e:?}",
                            hex::encode(winner_pool_hash)
                        );
                        PutValidationError::MerklePaymentVerificationFailed {
                            record_key: pretty_key.clone().into_owned(),
                            error: error_msg,
                        }
                    })?
            };

        // Convert contract payment info to the format expected by the verification code
        let payment_info = ant_evm::merkle_batch_payment::OnChainPaymentInfo {
            depth: contract_payment_info.depth,
            merkle_payment_timestamp: contract_payment_info.merklePaymentTimestamp,
            paid_node_addresses: contract_payment_info
                .paidNodeAddresses
                .into_iter()
                .map(|paid_node| {
                    let index = paid_node.poolIndex as usize;
                    (paid_node.rewardsAddress, index)
                })
                .collect(),
        };

        debug!(
            "merkle payment: got payment info: depth={}, timestamp={}, paid_nodes={}",
            payment_info.depth,
            payment_info.merkle_payment_timestamp,
            payment_info.paid_node_addresses.len()
        );

        // Verify Merkle proof structure using smart contract data
        debug!("merkle payment: verifying Merkle proof structure");
        proof
            .verify(
                payment_info.depth,
                payment_info.merkle_payment_timestamp,
                &winner_pool_hash,
                &payment_info.paid_node_addresses,
            )
            .map_err(|e| {
                let error_msg = format!("Merkle proof verification failed: {e:?}");
                warn!("{error_msg} for {pretty_key:?}");
                PutValidationError::MerklePaymentVerificationFailed {
                    record_key: pretty_key.clone().into_owned(),
                    error: error_msg,
                }
            })?;

        debug!("Merkle proof structure verified successfully for {pretty_key:?}");

        // Verify network topology (>50% of paid nodes in closest)
        // Get closest peers to the midpoint address (same address the client used to collect candidates)
        // Use majority knowledge method for accuracy in this critical payment verification
        let midpoint_address = proof.winner_pool.midpoint_proof.address();
        let reward_pool_address = NetworkAddress::ChunkAddress(ChunkAddress::new(midpoint_address));
        let all_closest_peers = match self
            .network()
            .get_closest_peers_with_majority_knowledge(&reward_pool_address)
            .await
        {
            Ok(peers) => peers,
            Err(e) => {
                warn!(
                    "Failed to get closest peers with majority knowledge for topology verification: {e:?}"
                );
                return Err(PutValidationError::MerklePaymentVerificationFailed {
                    record_key: pretty_key.clone().into_owned(),
                    error: format!(
                        "Failed to get closest peers with majority knowledge to {midpoint_address:?}: {e:?}"
                    ),
                });
            }
        };

        // Take all the closest nodes
        let closest_peers: Vec<_> = all_closest_peers.into_iter().collect();
        let closest_peer_ids: HashSet<_> =
            closest_peers.iter().map(|(peer_id, _)| *peer_id).collect();

        // Get the peer ids of the PAID nodes using their indices from the smart contract.
        // The indices allow us to identify exact nodes even if multiple nodes share the same reward address.
        let paid_peer_ids = proof
            .corresponding_peer_ids(&payment_info.paid_node_addresses)
            .map_err(|e| {
                let error_msg = format!("Failed to get paid node peer ids: {e:?}");
                warn!("{error_msg}");
                PutValidationError::MerklePaymentVerificationFailed {
                    record_key: pretty_key.clone().into_owned(),
                    error: error_msg,
                }
            })?;

        // Distance validation: verify ALL paid nodes are within 10x responsible distance range
        self.verify_paid_nodes_distance(&paid_peer_ids, &reward_pool_address)
            .await
            .map_err(|error| {
                warn!("{error} for {pretty_key:?}");
                PutValidationError::MerklePaymentVerificationFailed {
                    record_key: pretty_key.clone().into_owned(),
                    error,
                }
            })?;

        // Compare the two sets of peer ids: how many PAID nodes are in the closest?
        let legitimate_paid_nodes = paid_peer_ids
            .iter()
            .filter(|peer_id| closest_peer_ids.contains(peer_id))
            .count();
        let closest_len = closest_peer_ids.len();
        let paid_nodes_count = payment_info.paid_node_addresses.len();
        let validity_threshold = paid_nodes_count / 2;
        let reached_validity_threshold = legitimate_paid_nodes > validity_threshold;

        if !reached_validity_threshold {
            warn!(
                "Network topology verification failed for {pretty_key:?}: only {legitimate_paid_nodes}/{paid_nodes_count} paid nodes in closest {closest_len} (need >{validity_threshold})"
            );

            return Err(PutValidationError::TopologyVerificationFailed {
                target_address: reward_pool_address.clone(),
                valid_count: legitimate_paid_nodes,
                total_paid: paid_nodes_count,
                closest_count: closest_len,
                node_peers: closest_peer_ids.into_iter().collect(),
                paid_peers: paid_peer_ids.to_vec(),
            });
        }

        debug!(
            "Network topology verified: {legitimate_paid_nodes}/{paid_nodes_count} paid nodes in closest {closest_len} for {pretty_key:?}",
        );

        info!("Merkle payment verified successfully for {pretty_key:?}");
        Ok(())
    }

    /// Verify that all paid nodes are within a reasonable distance to the target address.
    /// Uses 10x the responsible distance range as the effective tolerance.
    ///
    /// Returns Ok(()) if all nodes are within range, or an error message if any are outside.
    async fn verify_paid_nodes_distance(
        &self,
        paid_peer_ids: &[PeerId],
        target_address: &NetworkAddress,
    ) -> Result<(), String> {
        // Get the node's responsible distance range
        let distance_range = match self.network().get_network_density().await.ok().flatten() {
            Some(range) => range,
            None => {
                // If we can't get the distance range, skip this check
                debug!("No responsible distance range available, skipping distance validation");
                return Ok(());
            }
        };

        // Use 10x tolerance for the effective range
        let effective_range = Distance(distance_range.0.saturating_mul(U256::from(10)));

        // Check each paid node's distance to the target address
        let nodes_outside_range: Vec<_> = paid_peer_ids
            .iter()
            .filter_map(|peer_id| {
                let peer_addr = NetworkAddress::from(*peer_id);
                let distance = peer_addr.distance(target_address);
                if distance > effective_range {
                    Some((*peer_id, distance))
                } else {
                    None
                }
            })
            .collect();

        // Log distance check results
        debug!(
            "Distance check: {}/{} paid nodes within 10x range (base={:?}, effective={:?})",
            paid_peer_ids.len() - nodes_outside_range.len(),
            paid_peer_ids.len(),
            distance_range.ilog2(),
            effective_range.ilog2()
        );

        // Reject if ANY paid node is outside the 10x range
        if !nodes_outside_range.is_empty() {
            return Err(format!(
                "Distance validation failed: {} paid nodes outside 10x responsible range \
                 (effective ilog2={:?}). Nodes outside: {:?}",
                nodes_outside_range.len(),
                effective_range.ilog2(),
                nodes_outside_range
                    .iter()
                    .map(|(p, d)| format!("{p:?}@{:?}", d.ilog2()))
                    .collect::<Vec<_>>()
            ));
        }

        Ok(())
    }
}
