// Copyright 2015-2017 Parity Technologies (UK) Ltd.
// This file is part of Parity.

// Parity is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.

// Parity is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.

// You should have received a copy of the GNU General Public License
// along with Parity.  If not, see <http://www.gnu.org/licenses/>.

///
/// Blockchain downloader
///

use std::collections::{HashSet, VecDeque};
use std::cmp;
use heapsize::HeapSizeOf;
use ethereum_types::H256;
use rlp::Rlp;
use ethcore::views::BlockView;
use ethcore::header::{BlockNumber, Header as BlockHeader};
use ethcore::client::{BlockStatus, BlockId, BlockImportError};
use ethcore::block::Block;
use ethcore::error::{ImportError, BlockError};
use sync_io::SyncIo;
use blocks::BlockCollection;

const MAX_HEADERS_TO_REQUEST: usize = 128;
const MAX_BODIES_TO_REQUEST: usize = 32;
const MAX_RECEPITS_TO_REQUEST: usize = 128;
const SUBCHAIN_SIZE: u64 = 256;
const MAX_ROUND_PARENTS: usize = 16;
const MAX_PARALLEL_SUBCHAIN_DOWNLOAD: usize = 5;

#[derive(Copy, Clone, Eq, PartialEq, Debug)]
/// Downloader state
pub enum State {
	/// No active downloads.
	Idle,
	/// Downloading subchain heads
	ChainHead,
	/// Downloading blocks
	Blocks,
	/// Download is complete
	Complete,
}

/// Data that needs to be requested from a peer.
pub enum BlockRequest {
	Headers {
		start: H256,
		count: u64,
		skip: u64,
	},
	Bodies {
		hashes: Vec<H256>,
	},
	Receipts {
		hashes: Vec<H256>,
	},
}

/// Indicates sync action
pub enum DownloadAction {
	/// Do nothing
	None,
	/// Reset downloads for all peers
	Reset
}

#[derive(Eq, PartialEq, Debug)]
pub enum BlockDownloaderImportError {
	/// Imported data is rejected as invalid.
	Invalid,
	/// Imported data is valid but rejected cause the downloader does not need it.
	Useless,
}

/// Block downloader strategy.
/// Manages state and block data for a block download process.
pub struct BlockDownloader {
	/// Downloader state
	state: State,
	/// Highest block number seen
	highest_block: Option<BlockNumber>,
	/// Downloaded blocks, holds `H`, `B` and `S`
	blocks: BlockCollection,
	/// Last impoted block number
	last_imported_block: BlockNumber,
	/// Last impoted block hash
	last_imported_hash: H256,
	/// Number of blocks imported this round
	imported_this_round: Option<usize>,
	/// Block number the last round started with.
	last_round_start: BlockNumber,
	last_round_start_hash: H256,
	/// Block parents imported this round (hash, parent)
	round_parents: VecDeque<(H256, H256)>,
	/// Do we need to download block recetips.
	download_receipts: bool,
	/// Sync up to the block with this hash.
	target_hash: Option<H256>,
	/// Probing range for seeking common best block.
	retract_step: u64,
	/// Whether reorg should be limited.
	limit_reorg: bool,
}

impl BlockDownloader {
	/// Create a new instance of syncing strategy. This won't reorganize to before the
	/// last kept state.
	pub fn new(sync_receipts: bool, start_hash: &H256, start_number: BlockNumber) -> Self {
		BlockDownloader {
			state: State::Idle,
			highest_block: None,
			last_imported_block: start_number,
			last_imported_hash: start_hash.clone(),
			last_round_start: start_number,
			last_round_start_hash: start_hash.clone(),
			blocks: BlockCollection::new(sync_receipts),
			imported_this_round: None,
			round_parents: VecDeque::new(),
			download_receipts: sync_receipts,
			target_hash: None,
			retract_step: 1,
			limit_reorg: true,
		}
	}

	/// Create a new instance of sync with unlimited reorg allowed.
	pub fn with_unlimited_reorg(sync_receipts: bool, start_hash: &H256, start_number: BlockNumber) -> Self {
		BlockDownloader {
			state: State::Idle,
			highest_block: None,
			last_imported_block: start_number,
			last_imported_hash: start_hash.clone(),
			last_round_start: start_number,
			last_round_start_hash: start_hash.clone(),
			blocks: BlockCollection::new(sync_receipts),
			imported_this_round: None,
			round_parents: VecDeque::new(),
			download_receipts: sync_receipts,
			target_hash: None,
			retract_step: 1,
			limit_reorg: false,
		}
	}

	/// Reset sync. Clear all local downloaded data.
	pub fn reset(&mut self) {
		self.blocks.clear();
		self.state = State::Idle;
	}

	/// Mark a block as known in the chain
	pub fn mark_as_known(&mut self, hash: &H256, number: BlockNumber) {
		if number >= self.last_imported_block + 1 {
			self.last_imported_block = number;
			self.last_imported_hash = hash.clone();
			self.imported_this_round = Some(self.imported_this_round.unwrap_or(0) + 1);
			self.last_round_start = number;
			self.last_round_start_hash = hash.clone();
		}
	}

	/// Check if download is complete
	pub fn is_complete(&self) -> bool {
		self.state == State::Complete
	}

	/// Check if particular block hash is being downloaded
	pub fn is_downloading(&self, hash: &H256) -> bool {
		self.blocks.is_downloading(hash)
	}

	/// Set starting sync block
	pub fn set_target(&mut self, hash: &H256) {
		self.target_hash = Some(hash.clone());
	}

	/// Unmark header as being downloaded.
	pub fn clear_header_download(&mut self, hash: &H256) {
		self.blocks.clear_header_download(hash)
	}

	/// Unmark block body as being downloaded.
	pub fn clear_body_download(&mut self, hashes: &[H256]) {
		self.blocks.clear_body_download(hashes)
	}

	/// Unmark block receipt as being downloaded.
	pub fn clear_receipt_download(&mut self, hashes: &[H256]) {
		self.blocks.clear_receipt_download(hashes)
	}
	/// Reset collection for a new sync round with given subchain block hashes.
	pub fn reset_to(&mut self, hashes: Vec<H256>) {
		self.reset();
		self.blocks.reset_to(hashes);
		self.state = State::Blocks;
	}

	/// Returns used heap memory size.
	pub fn heap_size(&self) -> usize {
		self.blocks.heap_size() + self.round_parents.heap_size_of_children()
	}

	/// Returns best imported block number.
	pub fn last_imported_block_number(&self) -> BlockNumber {
		self.last_imported_block
	}

	/// Add new block headers.
	pub fn import_headers(&mut self, io: &mut SyncIo, r: &Rlp, expected_hash: Option<H256>) -> Result<DownloadAction, BlockDownloaderImportError> {
		let item_count = r.item_count().unwrap_or(0);
		if self.state == State::Idle {
			trace!(target: "sync", "Ignored unexpected block headers");
			return Ok(DownloadAction::None)
		}
		if item_count == 0 && (self.state == State::Blocks) {
			return Err(BlockDownloaderImportError::Invalid);
		}

		let mut headers = Vec::new();
		let mut hashes = Vec::new();
		let mut valid_response = item_count == 0; //empty response is valid
		let mut any_known = false;
		for i in 0..item_count {
			let info: BlockHeader = r.val_at(i).map_err(|e| {
				trace!(target: "sync", "Error decoding block header RLP: {:?}", e);
				BlockDownloaderImportError::Invalid
			})?;
			let number = BlockNumber::from(info.number());
			// Check if any of the headers matches the hash we requested
			if !valid_response {
				if let Some(expected) = expected_hash {
					valid_response = expected == info.hash()
				}
			}
			any_known = any_known || self.blocks.contains_head(&info.hash());
			if self.blocks.contains(&info.hash()) {
				trace!(target: "sync", "Skipping existing block header {} ({:?})", number, info.hash());
				continue;
			}

			if self.highest_block.as_ref().map_or(true, |n| number > *n) {
				self.highest_block = Some(number);
			}
			let hash = info.hash();
			let hdr = r.at(i).map_err(|e| {
				trace!(target: "sync", "Error decoding block header RLP: {:?}", e);
				BlockDownloaderImportError::Invalid
			})?;
			match io.chain().block_status(BlockId::Hash(hash.clone())) {
				BlockStatus::InChain | BlockStatus::Queued => {
					match self.state {
						State::Blocks => trace!(target: "sync", "Header already in chain {} ({})", number, hash),
						_ => trace!(target: "sync", "Header already in chain {} ({}), state = {:?}", number, hash, self.state),
					}
					headers.push(hdr.as_raw().to_vec());
					hashes.push(hash);
				},
				BlockStatus::Bad => {
					return Err(BlockDownloaderImportError::Invalid);
				},
				BlockStatus::Unknown | BlockStatus::Pending => {
					headers.push(hdr.as_raw().to_vec());
					hashes.push(hash);
				}
			}
		}

		// Disable the peer for this syncing round if it gives invalid chain
		if !valid_response {
			trace!(target: "sync", "Invalid headers response");
			return Err(BlockDownloaderImportError::Invalid);
		}

		match self.state {
			State::ChainHead => {
				if !headers.is_empty() {
					// TODO: validate heads better. E.g. check that there is enough distance between blocks.
					trace!(target: "sync", "Received {} subchain heads, proceeding to download", headers.len());
					self.blocks.reset_to(hashes);
					self.state = State::Blocks;
					return Ok(DownloadAction::Reset);
				} else {
					let best = io.chain().chain_info().best_block_number;
					let oldest_reorg = io.chain().pruning_info().earliest_state;
					let last = self.last_imported_block;
					if self.limit_reorg && best > last && (last == 0 || last < oldest_reorg) {
						trace!(target: "sync", "No common block, disabling peer");
						return Err(BlockDownloaderImportError::Invalid);
					}
				}
			},
			State::Blocks => {
				let count = headers.len();
				// At least one of the heades must advance the subchain. Otherwise they are all useless.
				if count == 0 || !any_known {
					trace!(target: "sync", "No useful headers");
					return Err(BlockDownloaderImportError::Useless);
				}
				self.blocks.insert_headers(headers);
				trace!(target: "sync", "Inserted {} headers", count);
			},
			_ => trace!(target: "sync", "Unexpected headers({})", headers.len()),
		}

		Ok(DownloadAction::None)
	}

	/// Called by peer once it has new block bodies
	pub fn import_bodies(&mut self, _io: &mut SyncIo, r: &Rlp) -> Result<(), BlockDownloaderImportError> {
		let item_count = r.item_count().unwrap_or(0);
		if item_count == 0 {
			return Err(BlockDownloaderImportError::Useless);
		}
		else if self.state != State::Blocks {
			trace!(target: "sync", "Ignored unexpected block bodies");
		}
		else {
			let mut bodies = Vec::with_capacity(item_count);
			for i in 0..item_count {
				let body = r.at(i).map_err(|e| {
					trace!(target: "sync", "Error decoding block boides RLP: {:?}", e);
					BlockDownloaderImportError::Invalid
				})?;
				bodies.push(body.as_raw().to_vec());
			}
			if self.blocks.insert_bodies(bodies) != item_count {
				trace!(target: "sync", "Deactivating peer for giving invalid block bodies");
				return Err(BlockDownloaderImportError::Invalid);
			}
		}
		Ok(())
	}

	/// Called by peer once it has new block bodies
	pub fn import_receipts(&mut self, _io: &mut SyncIo, r: &Rlp) -> Result<(), BlockDownloaderImportError> {
		let item_count = r.item_count().unwrap_or(0);
		if item_count == 0 {
			return Err(BlockDownloaderImportError::Useless);
		}
		else if self.state != State::Blocks {
			trace!(target: "sync", "Ignored unexpected block receipts");
		}
		else {
			let mut receipts = Vec::with_capacity(item_count);
			for i in 0..item_count {
				let receipt = r.at(i).map_err(|e| {
					trace!(target: "sync", "Error decoding block receipts RLP: {:?}", e);
					BlockDownloaderImportError::Invalid
				})?;
				receipts.push(receipt.as_raw().to_vec());
			}
			if self.blocks.insert_receipts(receipts) != item_count {
				trace!(target: "sync", "Deactivating peer for giving invalid block receipts");
				return Err(BlockDownloaderImportError::Invalid);
			}
		}
		Ok(())
	}

	fn start_sync_round(&mut self, io: &mut SyncIo) {
		self.state = State::ChainHead;
		trace!(target: "sync", "Starting round (last imported count = {:?}, last started = {}, block = {:?}", self.imported_this_round, self.last_round_start, self.last_imported_block);
		// Check if need to retract to find the common block. The problem is that the peers still return headers by hash even
		// from the non-canonical part of the tree. So we also retract if nothing has been imported last round.
		let start = self.last_round_start;
		let start_hash = self.last_round_start_hash;
		match self.imported_this_round {
			Some(n) if n == 0 && start > 0 => {
				// nothing was imported last round, step back to a previous block
				// search parent in last round known parents first
				if let Some(&(_, p)) = self.round_parents.iter().find(|&&(h, _)| h == start_hash) {
					self.last_imported_block = start - 1;
					self.last_imported_hash = p.clone();
					trace!(target: "sync", "Searching common header from the last round {} ({})", self.last_imported_block, self.last_imported_hash);
				} else {
					let best = io.chain().chain_info().best_block_number;
					let oldest_reorg = io.chain().pruning_info().earliest_state;
					if self.limit_reorg && best > start && start < oldest_reorg {
						debug!(target: "sync", "Could not revert to previous ancient block, last: {} ({})", start, start_hash);
						self.reset();
					} else {
						let n = start - cmp::min(self.retract_step, start);
						self.retract_step *= 2;
						match io.chain().block_hash(BlockId::Number(n)) {
							Some(h) => {
								self.last_imported_block = n;
								self.last_imported_hash = h;
								trace!(target: "sync", "Searching common header in the blockchain {} ({})", start, self.last_imported_hash);
							}
							None => {
								debug!(target: "sync", "Could not revert to previous block, last: {} ({})", start, self.last_imported_hash);
								self.reset();
							}
						}
					}
				}
			},
			_ => {
				self.retract_step = 1;
			},
		}
		self.last_round_start = self.last_imported_block;
		self.last_round_start_hash = self.last_imported_hash;
		self.imported_this_round = None;
	}

	/// Find some headers or blocks to download for a peer.
	pub fn request_blocks(&mut self, io: &mut SyncIo, num_active_peers: usize) -> Option<BlockRequest> {
		match self.state {
			State::Idle => {
				self.start_sync_round(io);
				if self.state == State::ChainHead {
					return self.request_blocks(io, num_active_peers);
				}
			},
			State::ChainHead => {
				if num_active_peers < MAX_PARALLEL_SUBCHAIN_DOWNLOAD {
					// Request subchain headers
					trace!(target: "sync", "Starting sync with better chain");
					// Request MAX_HEADERS_TO_REQUEST - 2 headers apart so that
					// MAX_HEADERS_TO_REQUEST would include headers for neighbouring subchains
					return Some(BlockRequest::Headers {
						start: self.last_imported_hash.clone(),
						count: SUBCHAIN_SIZE,
						skip: (MAX_HEADERS_TO_REQUEST - 2) as u64,
					});
				}
			},
			State::Blocks => {
				// check to see if we need to download any block bodies first
				let needed_bodies = self.blocks.needed_bodies(MAX_BODIES_TO_REQUEST, false);
				if !needed_bodies.is_empty() {
					return Some(BlockRequest::Bodies {
						hashes: needed_bodies,
					});
				}

				if self.download_receipts {
					let needed_receipts = self.blocks.needed_receipts(MAX_RECEPITS_TO_REQUEST, false);
					if !needed_receipts.is_empty() {
						return Some(BlockRequest::Receipts {
							hashes: needed_receipts,
						});
					}
				}

				// find subchain to download
				if let Some((h, count)) = self.blocks.needed_headers(MAX_HEADERS_TO_REQUEST, false) {
					return Some(BlockRequest::Headers {
						start: h,
						count: count as u64,
						skip: 0,
					});
				}
			},
			State::Complete => (),
		}
		None
	}

	/// Checks if there are blocks fully downloaded that can be imported into the blockchain and does the import.
	pub fn collect_blocks(&mut self, io: &mut SyncIo, allow_out_of_order: bool) -> Result<(), BlockDownloaderImportError> {
		let mut bad = false;
		let mut imported = HashSet::new();
		let blocks = self.blocks.drain();
		let count = blocks.len();
		for block_and_receipts in blocks {
			let block = block_and_receipts.block;
			let receipts = block_and_receipts.receipts;
			let (h, number, parent) = {
				let header = view!(BlockView, &block).header_view();
				(header.hash(), header.number(), header.parent_hash())
			};

			// Perform basic block verification
			if !Block::is_good(&block) {
				debug!(target: "sync", "Bad block rlp {:?} : {:?}", h, block);
				bad = true;
				break;
			}

			if self.target_hash.as_ref().map_or(false, |t| t == &h) {
				self.state = State::Complete;
				trace!(target: "sync", "Sync target reached");
				return Ok(());
			}

			let result = if let Some(receipts) = receipts {
				io.chain().import_block_with_receipts(block, receipts)
			} else {
				io.chain().import_block(block)
			};

			match result {
				Err(BlockImportError::Import(ImportError::AlreadyInChain)) => {
					trace!(target: "sync", "Block already in chain {:?}", h);
					self.block_imported(&h, number, &parent);
				},
				Err(BlockImportError::Import(ImportError::AlreadyQueued)) => {
					trace!(target: "sync", "Block already queued {:?}", h);
					self.block_imported(&h, number, &parent);
				},
				Ok(_) => {
					trace!(target: "sync", "Block queued {:?}", h);
					imported.insert(h.clone());
					self.block_imported(&h, number, &parent);
				},
				Err(BlockImportError::Block(BlockError::UnknownParent(_))) if allow_out_of_order => {
					break;
				},
				Err(BlockImportError::Block(BlockError::UnknownParent(_))) => {
					trace!(target: "sync", "Unknown new block parent, restarting sync");
					break;
				},
				Err(BlockImportError::Block(BlockError::TemporarilyInvalid(_))) => {
					debug!(target: "sync", "Block temporarily invalid, restarting sync");
					break;
				},
				Err(e) => {
					debug!(target: "sync", "Bad block {:?} : {:?}", h, e);
					bad = true;
					break;
				}
			}
		}
		trace!(target: "sync", "Imported {} of {}", imported.len(), count);
		self.imported_this_round = Some(self.imported_this_round.unwrap_or(0) + imported.len());

		if bad {
			return Err(BlockDownloaderImportError::Invalid);
		}

		if self.blocks.is_empty() {
			// complete sync round
			trace!(target: "sync", "Sync round complete");
			self.reset();
		}
		Ok(())
	}

	fn block_imported(&mut self, hash: &H256, number: BlockNumber, parent: &H256) {
		self.last_imported_block = number;
		self.last_imported_hash = hash.clone();
		self.round_parents.push_back((hash.clone(), parent.clone()));
		if self.round_parents.len() > MAX_ROUND_PARENTS {
			self.round_parents.pop_front();
		}
	}
}

//TODO: module tests