// Copyright 2020 The finn Developers
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Adapters connecting new block, new transaction, and accepted transaction
//! events to consumers of those events.

use crate::util::RwLock;
use std::fs::File;
use std::path::PathBuf;
use std::sync::{Arc, Weak};
use std::thread;
use std::time::Instant;

use crate::chain::{
	self, BlockStatus, ChainAdapter, Options, SyncState, SyncStatus, TxHashsetDownloadStats,
};
use std::collections::HashMap;
use std::sync::Mutex;

use crate::common::hooks::{ChainEvents, NetEvents};
use crate::common::types::{ChainValidationMode, DandelionEpoch, ServerConfig};
use crate::core::core::hash::{Hash, Hashed};
use crate::core::core::transaction::Transaction;
use crate::core::core::verifier_cache::VerifierCache;
use crate::core::core::{BlockHeader, BlockSums, CompactBlock, Inputs, OutputIdentifier};
use crate::core::pow::Difficulty;
use crate::core::ser::ProtocolVersion;
use crate::core::{core, global};
use crate::p2p;
use crate::p2p::types::PeerInfo;
use crate::pool::{self, BlockChain, PoolAdapter};
use crate::util::OneTime;
use chrono::prelude::*;
use chrono::Duration;
use rand::prelude::*;
use std::sync::atomic::AtomicI64;
use std::sync::atomic::Ordering;

// NetToChainAdapter need a memory cache to prevent data overloading for network core nodes (non leaf nodes)
// This cache will drop sequense of the events during the second
struct EventCache {
	event: RwLock<Hash>,
	time: AtomicI64,
}

impl EventCache {
	fn new() -> Self {
		EventCache {
			event: RwLock::new(Hash::default()),
			time: AtomicI64::new(0),
		}
	}

	// Check if it is contain the hash value
	pub fn contains(&self, hash: &Hash, update: bool) -> bool {
		let now = Utc::now().timestamp_millis();
		let time_limit = now - 1000; // lock for a 1 second, should be enough to reduce the load.
		if self.time.load(Ordering::Relaxed) < time_limit {
			if update {
				*(self.event.write()) = *hash;
				self.time.store(now, Ordering::Relaxed);
			}
			return false;
		}

		if *self.event.read() == *hash {
			true
		} else {
			if update {
				*(self.event.write()) = *hash;
				self.time.store(now, Ordering::Relaxed);
			}
			false
		}
	}
}

/// Implementation of the NetAdapter for the . Gets notified when new
/// blocks and transactions are received and forwards to the chain and pool
/// implementations.
pub struct NetToChainAdapter<B, P, V>
where
	B: BlockChain,
	P: PoolAdapter,
	V: VerifierCache + 'static,
{
	sync_state: Arc<SyncState>,
	chain: Weak<chain::Chain>,
	tx_pool: Arc<RwLock<pool::TransactionPool<B, P, V>>>,
	verifier_cache: Arc<RwLock<V>>,
	peers: OneTime<Weak<p2p::Peers>>,
	config: ServerConfig,
	hooks: Vec<Box<dyn NetEvents + Send + Sync>>,

	// local in mem cache
	processed_headers: EventCache,
	processed_blocks: EventCache,
	processed_transactions: EventCache,

	header_cache: Arc<Mutex<HashMap<u64, core::BlockHeader>>>,
	tip_processed: Arc<Mutex<u64>>,
	reset_tip: Arc<Mutex<u64>>,
}

impl<B, P, V> p2p::ChainAdapter for NetToChainAdapter<B, P, V>
where
	B: BlockChain,
	P: PoolAdapter,
	V: VerifierCache + 'static,
{
	fn total_difficulty(&self) -> Result<Difficulty, chain::Error> {
		Ok(self.chain().head()?.total_difficulty)
	}

	fn total_height(&self) -> Result<u64, chain::Error> {
		Ok(self.chain().head()?.height)
	}

	fn get_transaction(&self, kernel_hash: Hash) -> Option<core::Transaction> {
		self.tx_pool.read().retrieve_tx_by_kernel_hash(kernel_hash)
	}

	fn tx_kernel_received(
		&self,
		kernel_hash: Hash,
		peer_info: &PeerInfo,
	) -> Result<bool, chain::Error> {
		// nothing much we can do with a new transaction while syncing
		if self.sync_state.is_syncing() {
			return Ok(true);
		}

		let tx = self.tx_pool.read().retrieve_tx_by_kernel_hash(kernel_hash);

		if tx.is_none() {
			self.request_transaction(kernel_hash, peer_info);
		}
		Ok(true)
	}

	fn transaction_received(
		&self,
		tx: core::Transaction,
		stem: bool,
	) -> Result<bool, chain::Error> {
		if self.sync_state.is_syncing() {
			return Ok(true);
		}

		// nothing much we can do with a new transaction while syncing
		let tx_hash = tx.hash();
		// For transaction we allow double processing, we want to be sure that TX will be stored in the pool
		// because there is no recovery plan for transactions. So we want to use natural retry to help us handle failures
		if self.processed_transactions.contains(&tx_hash, false) {
			debug!("transaction_received, cache for {} Rejected", tx_hash);
			return Ok(true);
		} else {
			debug!("transaction_received, cache for {} OK", tx_hash);
		}

		let source = pool::TxSource::Broadcast;

		let header = self.chain().head_header()?;

		for hook in &self.hooks {
			hook.on_transaction_received(&tx);
		}

		let mut tx_pool = self.tx_pool.write();
		match tx_pool.add_to_pool(source, tx, stem, &header) {
			Ok(_) => {
				self.processed_transactions.contains(&tx_hash, true);
				Ok(true)
			}
			Err(e) => {
				debug!("Transaction {} rejected: {:?}", tx_hash, e);
				Ok(false)
			}
		}
	}

	fn block_received(
		&self,
		b: core::Block,
		peer_info: &PeerInfo,
		opts: chain::Options,
	) -> Result<bool, chain::Error> {
		let b_hash = b.hash();
		if self.processed_blocks.contains(&b_hash, true) {
			debug!("block_received, cache for {} Rejected", b_hash);
			return Ok(true);
		} else {
			debug!("block_received, cache for {} OK", b_hash);
		}

		if self.chain().block_exists(b.hash())? {
			return Ok(true);
		}

		info!(
			"Received block {} at {} from {} [in/out/kern: {}/{}/{}] going to process.",
			b.hash(),
			b.header.height,
			peer_info.addr,
			b.inputs().len(),
			b.outputs().len(),
			b.kernels().len(),
		);

		self.process_block(b, peer_info, opts)
	}

	fn compact_block_received(
		&self,
		cb: core::CompactBlock,
		peer_info: &PeerInfo,
	) -> Result<bool, chain::Error> {
		// No need to process this compact block if we have previously accepted the _full block_.
		if self.chain().block_exists(cb.hash())? {
			return Ok(true);
		}
		let bhash = cb.hash();
		debug!(
			"Received compact_block {} at {} from {} [out/kern/kern_ids: {}/{}/{}] going to process.",
			bhash,
			cb.header.height,
			peer_info.addr,
			cb.out_full().len(),
			cb.kern_full().len(),
			cb.kern_ids().len(),
		);

		let cb_hash = cb.hash();
		if cb.kern_ids().is_empty() {
			// push the freshly hydrated block through the chain pipeline
			match core::Block::hydrate_from(cb, &[]) {
				Ok(block) => {
					debug!(
						"successfully hydrated (empty) block: {} at {} ({})",
						block.header.hash(),
						block.header.height,
						block.inputs().version_str(),
					);
					if !self.sync_state.is_syncing() {
						for hook in &self.hooks {
							hook.on_block_received(&block, &peer_info.addr);
						}
					}
					self.process_block(block, peer_info, chain::Options::NONE)
				}
				Err(e) => {
					debug!("Invalid hydrated block {}: {:?}", cb_hash, e);
					return Ok(false);
				}
			}
		} else {
			// check at least the header is valid before hydrating
			if let Err(e) = self
				.chain()
				.process_block_header(&cb.header, chain::Options::NONE)
			{
				debug!("Invalid compact block header {}: {:?}", cb_hash, e.kind());
				return Ok(!e.is_bad_data());
			}

			let (txs, missing_short_ids) = {
				self.tx_pool
					.read()
					.retrieve_transactions(cb.hash(), cb.nonce, cb.kern_ids())
			};

			debug!(
				"compact_block_received: txs from tx pool - {}, (unknown kern_ids: {})",
				txs.len(),
				missing_short_ids.len(),
			);

			// If we have missing kernels then we know we cannot hydrate this compact block.
			if !missing_short_ids.is_empty() {
				self.request_block(&cb.header, peer_info, chain::Options::NONE);
				return Ok(true);
			}

			let block = match core::Block::hydrate_from(cb.clone(), &txs) {
				Ok(block) => {
					if !self.sync_state.is_syncing() {
						for hook in &self.hooks {
							hook.on_block_received(&block, &peer_info.addr);
						}
					}
					block
				}
				Err(e) => {
					debug!("Invalid hydrated block {}: {:?}", cb.hash(), e);
					return Ok(false);
				}
			};

			if let Ok(prev) = self.chain().get_previous_header(&cb.header) {
				if block
					.validate(&prev.total_kernel_offset, self.verifier_cache.clone())
					.is_ok()
				{
					debug!(
						"successfully hydrated block: {} at {} ({})",
						block.header.hash(),
						block.header.height,
						block.inputs().version_str(),
					);
					self.process_block(block, peer_info, chain::Options::NONE)
				} else if self.sync_state.status() == SyncStatus::NoSync {
					debug!("adapter: block invalid after hydration, requesting full block");
					self.request_block(&cb.header, peer_info, chain::Options::NONE);
					Ok(true)
				} else {
					debug!("block invalid after hydration, ignoring it, cause still syncing");
					Ok(true)
				}
			} else {
				debug!("failed to retrieve previous block header (still syncing?)");
				Ok(true)
			}
		}
	}

	fn header_received(
		&self,
		bh: core::BlockHeader,
		peer_info: &PeerInfo,
	) -> Result<bool, chain::Error> {
		// No need to process this header if we have previously accepted the _full block_.
		let bh_hash = bh.hash();

		// A shortcut to refuse the known bad block header.
		if bh_hash == Hash::from_hex(crate::chain::BLOCK_TO_BAN)? {
			debug!(
				"header_received: known bad header {} at {} refused by chain",
				bh_hash, bh.height,
			);
			// serious enough to need to ban the peer
			return Ok(false);
		}

		if self.processed_headers.contains(&bh_hash, true) {
			debug!("header_received, cache for {} Rejected", bh_hash);
			return Ok(true);
		} else {
			debug!("header_received, cache for {} OK", bh_hash);
		}

		if self.chain().block_exists(bh.hash())? {
			return Ok(true);
		}
		if !self.sync_state.is_syncing() {
			for hook in &self.hooks {
				hook.on_header_received(&bh, &peer_info.addr);
			}
		}

		// pushing the new block header through the header chain pipeline
		// we will go ask for the block if this is a new header
		let res = self.chain().process_block_header(&bh, chain::Options::NONE);

		if let Err(e) = res {
			debug!(
				"Block header {} refused by chain: {:?}",
				bh.hash(),
				e.kind()
			);
			if e.is_bad_data() {
				return Ok(false);
			} else {
				// we got an error when trying to process the block header
				// but nothing serious enough to need to ban the peer upstream
				return Err(e);
			}
		}

		// we have successfully processed a block header
		// so we can go request the block itself
		self.request_compact_block(&bh, peer_info);

		// done receiving the header
		Ok(true)
	}

	fn headers_received(
		&self,
		bhs: &[core::BlockHeader],
		peer_info: &PeerInfo,
		header_cache_size: u64,
	) -> Result<bool, chain::Error> {
		let tip_processed = {
			let mut tip_processed = self.tip_processed.lock().unwrap();
			let sync_head_height = self.chain().get_sync_head()?.height;

			let mut reset_tip = self.reset_tip.lock().unwrap();
			if *reset_tip != 0 {
				warn!(
					"reset of tip to {} from {} due to differing headers.",
					*reset_tip, *tip_processed
				);
				*tip_processed = *reset_tip;
				*reset_tip = 0;
			} else if *tip_processed < sync_head_height {
				*tip_processed = sync_head_height;
			}

			*tip_processed
		};

		if bhs.is_empty() {
			return Ok(false);
		}
		let bad_block = Hash::from_hex(crate::chain::BLOCK_TO_BAN)?;
		if bhs.iter().find(|h| h.hash() == bad_block).is_some() {
			debug!("headers_received: found known bad header, all data is rejected");
			return Ok(false);
		}

		info!(
			"Received {} block headers from {}, height {}, hash = {}, tip_processed = {}",
			bhs.len(),
			peer_info.addr,
			bhs[0].height,
			bhs[0].hash(),
			tip_processed,
		);

		if bhs[0].height > tip_processed + 1 {
			// we can't process this yet.
			// try to process anything in the cache that we can

			if header_cache_size > 0 {
				for bh in bhs {
					let mut hashmap = self.header_cache.lock().unwrap();
					hashmap.insert(bh.height, bh.clone());
					if bh.height > header_cache_size {
						hashmap.remove(&(bh.height - header_cache_size));
					}
				}
			}
			return Ok(true);
		}
		if header_cache_size > 0 {
			let mut itt = tip_processed + 1;
			let mut bh_backlog: Vec<core::BlockHeader> = Vec::new();
			let mut backlog_processed = false;
			loop {
				{
					let hashmap = self.header_cache.lock().unwrap();
					let next = hashmap.get(&itt);
					if !next.is_some() {
						break;
					}
					let next = next.unwrap();
					//info!("adding headers to the backlog: {}", next.height);
					bh_backlog.push(next.clone());
				}

				if bh_backlog.len() >= 256 {
					// getting too big, process and continue
					self.process_add_headers_sync(&bh_backlog.as_slice(), header_cache_size)?;
					bh_backlog = Vec::new();
					backlog_processed = true;
				}

				itt = itt + 1;
			}

			if bh_backlog.len() > 0 {
				self.process_add_headers_sync(&bh_backlog.as_slice(), header_cache_size)?;
				return Ok(true);
			}
			if backlog_processed {
				return Ok(true);
			}
		}

		let first_height = bhs[0].height;
		for bh in bhs {
			if header_cache_size > 0 {
				// set highest processed block
				let mut hashmap = self.header_cache.lock().unwrap();
				let value = hashmap.get(&bh.height);
				if value.is_some() {
					// we already have something here.
					// does it match? If so return.
					let cache_value = value.unwrap();
					if bh.prev_hash == cache_value.prev_hash {
						if first_height <= tip_processed {
							return Ok(true);
						}
					} else {
						// it doesn't match! there must have
						// been a reorg or someone gave us bad headers.
						// clear the entire hashmap to be safe.
						// go back to previous logic at this point hashmap.clear();
						warn!(
							"different header value at height = {}. clearing cache.",
							bh.height
						);
						hashmap.clear();
						*(self.reset_tip.lock().unwrap()) = first_height - 1;
						break;
					}
				}
			}
		}
		self.process_add_headers_sync(bhs, header_cache_size)
	}

	fn process_add_headers_sync(
		&self,
		bhs: &[core::BlockHeader],
		header_cache_size: u64,
	) -> Result<bool, chain::Error> {
		let mut hashmap = self.header_cache.lock().unwrap();
		// try to add headers to our header chain
		match self.chain().sync_block_headers(bhs, chain::Options::SYNC) {
			Ok(_) => {
				for bh in bhs {
					let mut tip_processed = self.tip_processed.lock().unwrap();
					if *tip_processed < bh.height {
						*tip_processed = bh.height;
					}
					if header_cache_size > 0 {
						hashmap.insert(bh.height, bh.clone());
						if bh.height > header_cache_size {
							hashmap.remove(&(bh.height - header_cache_size));
						}
					}
				}
				Ok(true)
			}
			Err(e) => {
				debug!("Block headers refused by chain: {:?}", e);
				if e.is_bad_data() {
					Ok(false)
				} else {
					Err(e)
				}
			}
		}
	}

	fn locate_headers(&self, locator: &[Hash]) -> Result<Vec<core::BlockHeader>, chain::Error> {
		debug!("locator: {:?}", locator);

		let header = match self.find_common_header(locator) {
			Some(header) => header,
			None => return Ok(vec![]),
		};

		let max_height = self.chain().header_head()?.height;

		let header_pmmr = self.chain().header_pmmr();
		let header_pmmr = header_pmmr.read();

		// looks like we know one, getting as many following headers as allowed
		let hh = header.height;
		let mut headers = vec![];
		for h in (hh + 1)..=(hh + (p2p::MAX_BLOCK_HEADERS as u64)) {
			if h > max_height {
				break;
			}

			if let Ok(hash) = header_pmmr.get_header_hash_by_height(h) {
				let header = self.chain().get_block_header(&hash)?;
				headers.push(header);
			} else {
				error!("Failed to locate headers successfully.");
				break;
			}
		}

		debug!("returning headers: {}", headers.len());

		Ok(headers)
	}

	/// Gets a full block by its hash.
	/// Will convert to v2 compatibility based on peer protocol version.
	fn get_block(&self, h: Hash, peer_info: &PeerInfo) -> Option<core::Block> {
		self.chain()
			.get_block(&h)
			.map(|b| match peer_info.version.value() {
				0..=2 => self.chain().convert_block_v2(b).ok(),
				3..=ProtocolVersion::MAX => Some(b),
			})
			.unwrap_or(None)
	}

	/// Provides a reading view into the current txhashset state as well as
	/// the required indexes for a consumer to rewind to a consistent state
	/// at the provided block hash.
	fn txhashset_read(&self, h: Hash) -> Option<p2p::TxHashSetRead> {
		match self.chain().txhashset_read(h.clone()) {
			Ok((out_index, kernel_index, read)) => Some(p2p::TxHashSetRead {
				output_index: out_index,
				kernel_index: kernel_index,
				reader: read,
			}),
			Err(e) => {
				warn!("Couldn't produce txhashset data for block {}: {:?}", h, e);
				None
			}
		}
	}

	fn txhashset_archive_header(&self) -> Result<core::BlockHeader, chain::Error> {
		self.chain().txhashset_archive_header()
	}

	fn txhashset_receive_ready(&self) -> bool {
		match self.sync_state.status() {
			SyncStatus::TxHashsetDownload { .. } => true,
			_ => false,
		}
	}

	fn txhashset_download_update(
		&self,
		start_time: DateTime<Utc>,
		downloaded_size: u64,
		total_size: u64,
	) -> bool {
		match self.sync_state.status() {
			SyncStatus::TxHashsetDownload(prev) => {
				self.sync_state
					.update_txhashset_download(TxHashsetDownloadStats {
						start_time,
						prev_update_time: prev.update_time,
						update_time: Utc::now(),
						prev_downloaded_size: prev.downloaded_size,
						downloaded_size,
						total_size,
					});
				true
			}
			_ => false,
		}
	}

	/// Writes a reading view on a txhashset state that's been provided to us.
	/// If we're willing to accept that new state, the data stream will be
	/// read as a zip file, unzipped and the resulting state files should be
	/// rewound to the provided indexes.
	fn txhashset_write(
		&self,
		h: Hash,
		txhashset_data: File,
		_peer_info: &PeerInfo,
	) -> Result<bool, chain::Error> {
		// check status again after download, in case 2 txhashsets made it somehow
		if let SyncStatus::TxHashsetDownload { .. } = self.sync_state.status() {
		} else {
			return Ok(false);
		}

		match self
			.chain()
			.txhashset_write(h, txhashset_data, self.sync_state.as_ref())
		{
			Ok(is_bad_data) => {
				if is_bad_data {
					self.chain().clean_txhashset_sandbox();
					error!("Failed to save txhashset archive: bad data");
					self.sync_state.set_sync_error(
						chain::ErrorKind::TxHashSetErr("bad txhashset data".to_string()).into(),
					);
				} else {
					info!("Received valid txhashset data for {}.", h);
				}
				Ok(is_bad_data)
			}
			Err(e) => {
				self.chain().clean_txhashset_sandbox();
				error!("Failed to save txhashset archive: {}", e);
				self.sync_state.set_sync_error(e);
				Ok(false)
			}
		}
	}

	fn get_tmp_dir(&self) -> PathBuf {
		self.chain().get_tmp_dir()
	}

	fn get_tmpfile_pathname(&self, tmpfile_name: String) -> PathBuf {
		self.chain().get_tmpfile_pathname(tmpfile_name)
	}
}

impl<B, P, V> NetToChainAdapter<B, P, V>
where
	B: BlockChain,
	P: PoolAdapter,
	V: VerifierCache + 'static,
{
	/// Construct a new NetToChainAdapter instance
	pub fn new(
		sync_state: Arc<SyncState>,
		chain: Arc<chain::Chain>,
		tx_pool: Arc<RwLock<pool::TransactionPool<B, P, V>>>,
		verifier_cache: Arc<RwLock<V>>,
		config: ServerConfig,
		hooks: Vec<Box<dyn NetEvents + Send + Sync>>,
	) -> Self {
		NetToChainAdapter {
			sync_state,
			chain: Arc::downgrade(&chain),
			tx_pool,
			verifier_cache,
			peers: OneTime::new(),
			config,
			hooks,
			processed_headers: EventCache::new(),
			processed_blocks: EventCache::new(),
			processed_transactions: EventCache::new(),
			header_cache: Arc::new(Mutex::new(HashMap::new())),
			tip_processed: Arc::new(Mutex::new(0)),
			reset_tip: Arc::new(Mutex::new(0)),
		}
	}

	/// Initialize a NetToChainAdaptor with reference to a Peers object.
	/// Should only be called once.
	pub fn init(&self, peers: Arc<p2p::Peers>) {
		self.peers.init(Arc::downgrade(&peers));
	}

	fn peers(&self) -> Arc<p2p::Peers> {
		self.peers
			.borrow()
			.upgrade()
			.expect("Failed to upgrade weak ref to our peers.")
	}

	fn chain(&self) -> Arc<chain::Chain> {
		self.chain
			.upgrade()
			.expect("Failed to upgrade weak ref to our chain.")
	}

	// Find the first locator hash that refers to a known header on our main chain.
	fn find_common_header(&self, locator: &[Hash]) -> Option<BlockHeader> {
		let header_pmmr = self.chain().header_pmmr();
		let header_pmmr = header_pmmr.read();

		for hash in locator {
			if let Ok(header) = self.chain().get_block_header(&hash) {
				if let Ok(hash_at_height) = header_pmmr.get_header_hash_by_height(header.height) {
					if let Ok(header_at_height) = self.chain().get_block_header(&hash_at_height) {
						if header.hash() == header_at_height.hash() {
							return Some(header);
						}
					}
				}
			}
		}
		None
	}

	// pushing the new block through the chain pipeline
	// remembering to reset the head if we have a bad block
	fn process_block(
		&self,
		b: core::Block,
		peer_info: &PeerInfo,
		opts: chain::Options,
	) -> Result<bool, chain::Error> {
		// We cannot process blocks earlier than the horizon so check for this here.
		{
			let head = self.chain().head()?;
			let horizon = head
				.height
				.saturating_sub(global::cut_through_horizon() as u64);
			if b.header.height < horizon {
				return Ok(true);
			}
		}

		let bhash = b.hash();
		let previous = self.chain().get_previous_header(&b.header);

		match self.chain().process_block(b, opts) {
			Ok(_) => {
				self.validate_chain(bhash);
				self.check_compact();
				Ok(true)
			}
			Err(ref e) if e.is_bad_data() => {
				self.validate_chain(bhash);
				Ok(false)
			}
			Err(e) => {
				match e.kind() {
					chain::ErrorKind::Orphan(orph_msg) => {
						if let Ok(previous) = previous {
							// make sure we did not miss the parent block
							if !self.chain().is_orphan(&previous.hash())
								&& !self.sync_state.is_syncing()
							{
								debug!("process_block: received an orphan block: {}, checking the parent: {:}", orph_msg, previous.hash());
								self.request_block(&previous, peer_info, chain::Options::NONE)
							}
						}
						Ok(true)
					}
					_ => {
						debug!(
							"process_block: block {} refused by chain: {}",
							bhash,
							e.kind()
						);
						Ok(true)
					}
				}
			}
		}
	}

	fn validate_chain(&self, bhash: Hash) {
		// If we are running in "validate the full chain every block" then
		// panic here if validation fails for any reason.
		// We are out of consensus at this point and want to track the problem
		// down as soon as possible.
		// Skip this if we are currently syncing (too slow).
		if self.config.chain_validation_mode == ChainValidationMode::EveryBlock
			&& self.chain().head().unwrap().height > 0
			&& !self.sync_state.is_syncing()
		{
			let now = Instant::now();

			debug!(
				"process_block: ***** validating full chain state at {}",
				bhash,
			);

			self.chain()
				.validate(true)
				.expect("chain validation failed, hard stop");

			debug!(
				"process_block: ***** done validating full chain state, took {}s",
				now.elapsed().as_secs(),
			);
		}
	}

	fn check_compact(&self) {
		// Skip compaction if we are syncing.
		if self.sync_state.is_syncing() {
			return;
		}

		// Roll the dice to trigger compaction at 1/COMPACTION_CHECK chance per block,
		// uses a different thread to avoid blocking the caller thread (likely a peer)
		let mut rng = thread_rng();
		if 0 == rng.gen_range(0, global::COMPACTION_CHECK) {
			let chain = self.chain();
			let _ = thread::Builder::new()
				.name("compactor".to_string())
				.spawn(move || {
					if let Err(e) = chain.compact() {
						error!("Could not compact chain: {:?}", e);
					}
				});
		}
	}

	fn request_transaction(&self, h: Hash, peer_info: &PeerInfo) {
		self.send_tx_request_to_peer(h, peer_info, |peer, h| peer.send_tx_request(h))
	}

	// After receiving a compact block if we cannot successfully hydrate
	// it into a full block then fallback to requesting the full block
	// from the same peer that gave us the compact block
	// consider additional peers for redundancy?
	fn request_block(&self, bh: &BlockHeader, peer_info: &PeerInfo, opts: Options) {
		self.send_block_request_to_peer(bh.hash(), peer_info, |peer, h| {
			peer.send_block_request(h, opts)
		})
	}

	// After we have received a block header in "header first" propagation
	// we need to go request the block (compact representation) from the
	// same peer that gave us the header (unless we have already accepted the block)
	fn request_compact_block(&self, bh: &BlockHeader, peer_info: &PeerInfo) {
		self.send_block_request_to_peer(bh.hash(), peer_info, |peer, h| {
			peer.send_compact_block_request(h)
		})
	}

	fn send_tx_request_to_peer<F>(&self, h: Hash, peer_info: &PeerInfo, f: F)
	where
		F: Fn(&p2p::Peer, Hash) -> Result<(), p2p::Error>,
	{
		match self.peers().get_connected_peer(peer_info.addr.clone()) {
			None => debug!(
				"send_tx_request_to_peer: can't send request to peer {:?}, not connected",
				peer_info.addr
			),
			Some(peer) => {
				if let Err(e) = f(&peer, h) {
					error!("send_tx_request_to_peer: failed: {:?}", e)
				}
			}
		}
	}

	fn send_block_request_to_peer<F>(&self, h: Hash, peer_info: &PeerInfo, f: F)
	where
		F: Fn(&p2p::Peer, Hash) -> Result<(), p2p::Error>,
	{
		match self.chain().block_exists(h) {
			Ok(false) => match self.peers().get_connected_peer(peer_info.addr.clone()) {
				None => debug!(
					"send_block_request_to_peer: can't send request to peer {:?}, not connected",
					peer_info.addr
				),
				Some(peer) => {
					if let Err(e) = f(&peer, h) {
						error!("send_block_request_to_peer: failed: {:?}", e)
					}
				}
			},
			Ok(true) => debug!("send_block_request_to_peer: block {} already known", h),
			Err(e) => error!(
				"send_block_request_to_peer: failed to check block exists: {:?}",
				e
			),
		}
	}
}

/// Implementation of the ChainAdapter for the network. Gets notified when the
///  accepted a new block, asking the pool to update its state and
/// the network to broadcast the block
pub struct ChainToPoolAndNetAdapter<B, P, V>
where
	B: BlockChain,
	P: PoolAdapter,
	V: VerifierCache + 'static,
{
	tx_pool: Arc<RwLock<pool::TransactionPool<B, P, V>>>,
	peers: OneTime<Weak<p2p::Peers>>,
	hooks: Vec<Box<dyn ChainEvents + Send + Sync>>,
}

impl<B, P, V> ChainAdapter for ChainToPoolAndNetAdapter<B, P, V>
where
	B: BlockChain,
	P: PoolAdapter,
	V: VerifierCache + 'static,
{
	fn block_accepted(&self, b: &core::Block, status: BlockStatus, opts: Options) {
		// not broadcasting blocks received through sync
		if !opts.contains(chain::Options::SYNC) {
			for hook in &self.hooks {
				hook.on_block_accepted(b, status);
			}
			// If we mined the block then we want to broadcast the compact block.
			// If we received the block from another node then broadcast "header first"
			// to minimize network traffic.
			if opts.contains(Options::MINE) {
				// propagate compact block out if we mined the block
				let cb: CompactBlock = b.clone().into();
				self.peers().broadcast_compact_block(&cb);
			} else {
				// "header first" propagation if we are not the originator of this block
				self.peers().broadcast_header(&b.header);
			}
		}

		// Reconcile the txpool against the new block *after* we have broadcast it too our peers.
		// This may be slow and we do not want to delay block propagation.
		// We only want to reconcile the txpool against the new block *if* total work has increased.

		if status.is_next() || status.is_reorg() {
			let mut tx_pool = self.tx_pool.write();

			let _ = tx_pool.reconcile_block(b);

			// First "age out" any old txs in the reorg_cache.
			let cutoff = Utc::now() - Duration::minutes(tx_pool.config.reorg_cache_timeout);
			tx_pool.truncate_reorg_cache(cutoff);
		}

		if status.is_reorg() {
			let _ = self.tx_pool.write().reconcile_reorg_cache(&b.header);
		}
	}
}

impl<B, P, V> ChainToPoolAndNetAdapter<B, P, V>
where
	B: BlockChain,
	P: PoolAdapter,
	V: VerifierCache + 'static,
{
	/// Construct a ChainToPoolAndNetAdapter instance.
	pub fn new(
		tx_pool: Arc<RwLock<pool::TransactionPool<B, P, V>>>,
		hooks: Vec<Box<dyn ChainEvents + Send + Sync>>,
	) -> Self {
		ChainToPoolAndNetAdapter {
			tx_pool,
			peers: OneTime::new(),
			hooks: hooks,
		}
	}

	/// Initialize a ChainToPoolAndNetAdapter instance with handle to a Peers
	/// object. Should only be called once.
	pub fn init(&self, peers: Arc<p2p::Peers>) {
		self.peers.init(Arc::downgrade(&peers));
	}

	fn peers(&self) -> Arc<p2p::Peers> {
		self.peers
			.borrow()
			.upgrade()
			.expect("Failed to upgrade weak ref to our peers.")
	}
}

/// Adapter between the transaction pool and the network, to relay
/// transactions that have been accepted.
pub struct PoolToNetAdapter {
	peers: OneTime<Weak<p2p::Peers>>,
	dandelion_epoch: Arc<RwLock<DandelionEpoch>>,
}

/// Adapter between the Dandelion monitor and the current Dandelion "epoch".
pub trait DandelionAdapter: Send + Sync {
	/// Is the node stemming (or fluffing) transactions in the current epoch?
	fn is_stem(&self) -> bool;

	/// Is the current Dandelion epoch expired?
	fn is_expired(&self) -> bool;

	/// Transition to the next Dandelion epoch (new stem/fluff state, select new relay peer).
	fn next_epoch(&self);
}

impl DandelionAdapter for PoolToNetAdapter {
	fn is_stem(&self) -> bool {
		self.dandelion_epoch.read().is_stem()
	}

	fn is_expired(&self) -> bool {
		self.dandelion_epoch.read().is_expired()
	}

	fn next_epoch(&self) {
		self.dandelion_epoch.write().next_epoch(&self.peers());
	}
}

impl pool::PoolAdapter for PoolToNetAdapter {
	fn tx_accepted(&self, entry: &pool::PoolEntry) {
		self.peers().broadcast_transaction(&entry.tx);
	}

	fn stem_tx_accepted(&self, entry: &pool::PoolEntry) -> Result<(), pool::PoolError> {
		// Take write lock on the current epoch.
		// We need to be able to update the current relay peer if not currently connected.
		let mut epoch = self.dandelion_epoch.write();

		// If "stem" epoch attempt to relay the tx to the next Dandelion relay.
		// Fallback to immediately fluffing the tx if we cannot stem for any reason.
		// If "fluff" epoch then nothing to do right now (fluff via Dandelion monitor).
		// If node is configured to always stem our (pushed via api) txs then do so.
		if epoch.is_stem() || (entry.src.is_pushed() && epoch.always_stem_our_txs()) {
			if let Some(peer) = epoch.relay_peer(&self.peers()) {
				match peer.send_stem_transaction(&entry.tx) {
					Ok(_) => {
						info!("Stemming this epoch, relaying to next peer.");
						Ok(())
					}
					Err(e) => {
						error!("Stemming tx failed. Fluffing. {:?}", e);
						Err(pool::PoolError::DandelionError)
					}
				}
			} else {
				error!("No relay peer. Fluffing.");
				Err(pool::PoolError::DandelionError)
			}
		} else {
			info!("Fluff epoch. Aggregating stem tx(s). Will fluff via Dandelion monitor.");
			Ok(())
		}
	}
}

impl PoolToNetAdapter {
	/// Create a new pool to net adapter
	pub fn new(config: pool::DandelionConfig) -> PoolToNetAdapter {
		PoolToNetAdapter {
			peers: OneTime::new(),
			dandelion_epoch: Arc::new(RwLock::new(DandelionEpoch::new(config))),
		}
	}

	/// Setup the p2p server on the adapter
	pub fn init(&self, peers: Arc<p2p::Peers>) {
		self.peers.init(Arc::downgrade(&peers));
	}

	fn peers(&self) -> Arc<p2p::Peers> {
		self.peers
			.borrow()
			.upgrade()
			.expect("Failed to upgrade weak ref to our peers.")
	}
}

/// Implements the view of the  required by the TransactionPool to
/// operate. Mostly needed to break any direct lifecycle or implementation
/// dependency between the pool and the chain.
#[derive(Clone)]
pub struct PoolToChainAdapter {
	chain: OneTime<Weak<chain::Chain>>,
}

impl PoolToChainAdapter {
	/// Create a new pool adapter
	pub fn new() -> PoolToChainAdapter {
		PoolToChainAdapter {
			chain: OneTime::new(),
		}
	}

	/// Set the pool adapter's chain. Should only be called once.
	pub fn set_chain(&self, chain_ref: Arc<chain::Chain>) {
		self.chain.init(Arc::downgrade(&chain_ref));
	}

	fn chain(&self) -> Arc<chain::Chain> {
		self.chain
			.borrow()
			.upgrade()
			.expect("Failed to upgrade the weak ref to our chain.")
	}
}

impl pool::BlockChain for PoolToChainAdapter {
	fn chain_head(&self) -> Result<BlockHeader, pool::PoolError> {
		self.chain()
			.head_header()
			.map_err(|e| pool::PoolError::Other(format!("failed to get head_header, {}", e)))
	}

	fn get_block_header(&self, hash: &Hash) -> Result<BlockHeader, pool::PoolError> {
		self.chain()
			.get_block_header(hash)
			.map_err(|e| pool::PoolError::Other(format!("failed to get block_header, {}", e)))
	}

	fn get_block_sums(&self, hash: &Hash) -> Result<BlockSums, pool::PoolError> {
		self.chain()
			.get_block_sums(hash)
			.map_err(|e| pool::PoolError::Other(format!("failed to get block_sums, {}", e)))
	}

	fn validate_tx(&self, tx: &Transaction) -> Result<(), pool::PoolError> {
		self.chain()
			.validate_tx(tx)
			.map_err(|e| pool::PoolError::Other(format!("failed to validate tx, {}", e)))
	}

	fn validate_inputs(&self, inputs: &Inputs) -> Result<Vec<OutputIdentifier>, pool::PoolError> {
		self.chain()
			.validate_inputs(inputs)
			.map(|outputs| outputs.into_iter().map(|(out, _)| out).collect::<Vec<_>>())
			.map_err(|_| pool::PoolError::Other("failed to validate tx".to_string()))
	}

	fn verify_coinbase_maturity(&self, inputs: &Inputs) -> Result<(), pool::PoolError> {
		self.chain()
			.verify_coinbase_maturity(inputs)
			.map_err(|_| pool::PoolError::ImmatureCoinbase)
	}

	fn verify_tx_lock_height(&self, tx: &Transaction) -> Result<(), pool::PoolError> {
		self.chain()
			.verify_tx_lock_height(tx)
			.map_err(|_| pool::PoolError::ImmatureTransaction)
	}

	fn replay_attack_check(&self, tx: &Transaction) -> Result<(), pool::PoolError> {
		self.chain()
			.replay_attack_check(tx)
			.map_err(|_| pool::PoolError::DuplicateKernelOrDuplicateSpent)
	}
}

#[cfg(test)]
mod test {
	use super::*;
	use std::time::Duration;

	#[test]
	fn test_event_cache() {
		let cache = EventCache::new();
		let def_hash1 = Hash::default();
		let def_hash2 = Hash::default();
		let hash2_1 =
			Hash::from_hex("735cf2a4492b437e292a295549c31df5f1e8e6d09e58ed20abdd808c2261d1f1")
				.unwrap();
		let hash2_2 =
			Hash::from_hex("735cf2a4492b437e292a295549c31df5f1e8e6d09e58ed20abdd808c2261d1f1")
				.unwrap();
		let hash3 =
			Hash::from_hex("9686b69cab945146fd431ec4459a0eef6efbcc5553480b7454edd32f9c3b4d52")
				.unwrap();

		assert_eq!(cache.contains(&def_hash1, true), false);
		assert_eq!(cache.contains(&def_hash2, true), true);
		thread::sleep(Duration::from_millis(1050));
		assert_eq!(cache.contains(&def_hash1, false), false);
		assert_eq!(cache.contains(&def_hash1, true), false);
		assert_eq!(cache.contains(&def_hash2, true), true);
		assert_eq!(cache.contains(&hash2_2, false), false);
		assert_eq!(cache.contains(&hash2_1, true), false);
		assert_eq!(cache.contains(&hash2_2, true), true);
		assert_eq!(cache.contains(&hash3, true), false);
	}
}
