// This file is part of Substrate.

// Copyright (C) Parity Technologies (UK) Ltd.

// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0

// This program 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.

// This program 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 this program. If not, see <https://www.gnu.org/licenses/>.

//! Client backend that is backed by a database.

//!

//! # Canonicality vs. Finality

//!

//! Finality indicates that a block will not be reverted, according to the consensus algorithm,

//! while canonicality indicates that the block may be reverted, but we will be unable to do so,

//! having discarded heavy state that will allow a chain reorganization.

//!

//! Finality implies canonicality but not vice-versa.

#![warn(missing_docs)]

pub mod offchain;

pub mod bench;

mod children;

mod parity_db;

mod pinned_blocks_cache;

mod record_stats_state;

mod stats;

#[cfg(any(feature = "rocksdb", test))]

mod upgrade;

mod utils;

use linked_hash_map::LinkedHashMap;

use log::{debug, trace, warn};

use parking_lot::{Mutex, RwLock};

use std::{

	collections::{HashMap, HashSet},

	io,

	path::{Path, PathBuf},

	sync::Arc,

};

use crate::{

	pinned_blocks_cache::PinnedBlocksCache,

	record_stats_state::RecordStatsState,

	stats::StateUsageStats,

	utils::{meta_keys, read_db, read_meta, DatabaseType, Meta},

};

use codec::{Decode, Encode};

use hash_db::Prefix;

use sc_client_api::{

	backend::NewBlockState,

	leaves::{FinalizationOutcome, LeafSet},

	utils::is_descendent_of,

	IoInfo, MemoryInfo, MemorySize, UsageInfo,

};

use sc_state_db::{IsPruned, LastCanonicalized, StateDb};

use sp_arithmetic::traits::Saturating;

use sp_blockchain::{

	Backend as _, CachedHeaderMetadata, DisplacedLeavesAfterFinalization, Error as ClientError,

	HeaderBackend, HeaderMetadata, HeaderMetadataCache, Result as ClientResult,

};

use sp_core::{

	offchain::OffchainOverlayedChange,

	storage::{well_known_keys, ChildInfo},

};

use sp_database::Transaction;

use sp_runtime::{

	generic::BlockId,

	traits::{

		Block as BlockT, Hash, HashingFor, Header as HeaderT, NumberFor, One, SaturatedConversion,

		Zero,

	},

	Justification, Justifications, StateVersion, Storage,

};

use sp_state_machine::{

	backend::{AsTrieBackend, Backend as StateBackend},

	BackendTransaction, ChildStorageCollection, DBValue, IndexOperation, IterArgs,

	OffchainChangesCollection, StateMachineStats, StorageCollection, StorageIterator, StorageKey,

	StorageValue, UsageInfo as StateUsageInfo,

};

use sp_trie::{cache::SharedTrieCache, prefixed_key, MemoryDB, MerkleValue, PrefixedMemoryDB};

// Re-export the Database trait so that one can pass an implementation of it.

pub use sc_state_db::PruningMode;

pub use sp_database::Database;

pub use bench::BenchmarkingState;

const CACHE_HEADERS: usize = 8;

/// DB-backed patricia trie state, transaction type is an overlay of changes to commit.

pub type DbState<H> = sp_state_machine::TrieBackend<Arc<dyn sp_state_machine::Storage<H>>, H>;

/// Builder for [`DbState`].

pub type DbStateBuilder<Hasher> =

	sp_state_machine::TrieBackendBuilder<Arc<dyn sp_state_machine::Storage<Hasher>>, Hasher>;

/// Length of a [`DbHash`].

const DB_HASH_LEN: usize = 32;

/// Hash type that this backend uses for the database.

pub type DbHash = sp_core::H256;

/// An extrinsic entry in the database.

#[derive(Debug, Encode, Decode)]

enum DbExtrinsic<B: BlockT> {

	/// Extrinsic that contains indexed data.

	Indexed {

		/// Hash of the indexed part.

		hash: DbHash,

		/// Extrinsic header.

		header: Vec<u8>,

	},

	/// Complete extrinsic data.

	Full(B::Extrinsic),

}

/// A reference tracking state.

///

/// It makes sure that the hash we are using stays pinned in storage

/// until this structure is dropped.

pub struct RefTrackingState<Block: BlockT> {

	state: DbState<HashingFor<Block>>,

	storage: Arc<StorageDb<Block>>,

	parent_hash: Option<Block::Hash>,

}

impl<B: BlockT> RefTrackingState<B> {

	fn new(

		state: DbState<HashingFor<B>>,

		storage: Arc<StorageDb<B>>,

		parent_hash: Option<B::Hash>,

	) -> Self {

		RefTrackingState { state, parent_hash, storage }

	}

}

impl<B: BlockT> Drop for RefTrackingState<B> {

	fn drop(&mut self) {

		if let Some(hash) = &self.parent_hash {

			self.storage.state_db.unpin(hash);

		}

	}

}

impl<Block: BlockT> std::fmt::Debug for RefTrackingState<Block> {

	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {

		write!(f, "Block {:?}", self.parent_hash)

	}

}

/// A raw iterator over the `RefTrackingState`.

pub struct RawIter<B: BlockT> {

	inner: <DbState<HashingFor<B>> as StateBackend<HashingFor<B>>>::RawIter,

}

impl<B: BlockT> StorageIterator<HashingFor<B>> for RawIter<B> {

	type Backend = RefTrackingState<B>;

	type Error = <DbState<HashingFor<B>> as StateBackend<HashingFor<B>>>::Error;

	fn next_key(&mut self, backend: &Self::Backend) -> Option<Result<StorageKey, Self::Error>> {

		self.inner.next_key(&backend.state)

	}

	fn next_pair(

		&mut self,

		backend: &Self::Backend,

	) -> Option<Result<(StorageKey, StorageValue), Self::Error>> {

		self.inner.next_pair(&backend.state)

	}

	fn was_complete(&self) -> bool {

		self.inner.was_complete()

	}

}

impl<B: BlockT> StateBackend<HashingFor<B>> for RefTrackingState<B> {

	type Error = <DbState<HashingFor<B>> as StateBackend<HashingFor<B>>>::Error;

	type TrieBackendStorage =

		<DbState<HashingFor<B>> as StateBackend<HashingFor<B>>>::TrieBackendStorage;

	type RawIter = RawIter<B>;

	fn storage(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {

		self.state.storage(key)

	}

	fn storage_hash(&self, key: &[u8]) -> Result<Option<B::Hash>, Self::Error> {

		self.state.storage_hash(key)

	}

	fn child_storage(

		&self,

		child_info: &ChildInfo,

		key: &[u8],

	) -> Result<Option<Vec<u8>>, Self::Error> {

		self.state.child_storage(child_info, key)

	}

	fn child_storage_hash(

		&self,

		child_info: &ChildInfo,

		key: &[u8],

	) -> Result<Option<B::Hash>, Self::Error> {

		self.state.child_storage_hash(child_info, key)

	}

	fn closest_merkle_value(

		&self,

		key: &[u8],

	) -> Result<Option<MerkleValue<B::Hash>>, Self::Error> {

		self.state.closest_merkle_value(key)

	}

	fn child_closest_merkle_value(

		&self,

		child_info: &ChildInfo,

		key: &[u8],

	) -> Result<Option<MerkleValue<B::Hash>>, Self::Error> {

		self.state.child_closest_merkle_value(child_info, key)

	}

	fn exists_storage(&self, key: &[u8]) -> Result<bool, Self::Error> {

		self.state.exists_storage(key)

	}

	fn exists_child_storage(

		&self,

		child_info: &ChildInfo,

		key: &[u8],

	) -> Result<bool, Self::Error> {

		self.state.exists_child_storage(child_info, key)

	}

	fn next_storage_key(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {

		self.state.next_storage_key(key)

	}

	fn next_child_storage_key(

		&self,

		child_info: &ChildInfo,

		key: &[u8],

	) -> Result<Option<Vec<u8>>, Self::Error> {

		self.state.next_child_storage_key(child_info, key)

	}

	fn storage_root<'a>(

		&self,

		delta: impl Iterator<Item = (&'a [u8], Option<&'a [u8]>)>,

		state_version: StateVersion,

	) -> (B::Hash, BackendTransaction<HashingFor<B>>) {

		self.state.storage_root(delta, state_version)

	}

	fn child_storage_root<'a>(

		&self,

		child_info: &ChildInfo,

		delta: impl Iterator<Item = (&'a [u8], Option<&'a [u8]>)>,

		state_version: StateVersion,

	) -> (B::Hash, bool, BackendTransaction<HashingFor<B>>) {

		self.state.child_storage_root(child_info, delta, state_version)

	}

	fn raw_iter(&self, args: IterArgs) -> Result<Self::RawIter, Self::Error> {

		self.state.raw_iter(args).map(|inner| RawIter { inner })

	}

	fn register_overlay_stats(&self, stats: &StateMachineStats) {

		self.state.register_overlay_stats(stats);

	}

	fn usage_info(&self) -> StateUsageInfo {

		self.state.usage_info()

	}

}

impl<B: BlockT> AsTrieBackend<HashingFor<B>> for RefTrackingState<B> {

	type TrieBackendStorage =

		<DbState<HashingFor<B>> as StateBackend<HashingFor<B>>>::TrieBackendStorage;

	fn as_trie_backend(

		&self,

	) -> &sp_state_machine::TrieBackend<Self::TrieBackendStorage, HashingFor<B>> {

		&self.state.as_trie_backend()

	}

}

/// Database settings.

pub struct DatabaseSettings {

	/// The maximum trie cache size in bytes.

	///

	/// If `None` is given, the cache is disabled.

	pub trie_cache_maximum_size: Option<usize>,

	/// Requested state pruning mode.

	pub state_pruning: Option<PruningMode>,

	/// Where to find the database.

	pub source: DatabaseSource,

	/// Block pruning mode.

	///

	/// NOTE: only finalized blocks are subject for removal!

	pub blocks_pruning: BlocksPruning,

}

/// Block pruning settings.

#[derive(Debug, Clone, Copy, PartialEq)]

pub enum BlocksPruning {

	/// Keep full block history, of every block that was ever imported.

	KeepAll,

	/// Keep full finalized block history.

	KeepFinalized,

	/// Keep N recent finalized blocks.

	Some(u32),

}

impl BlocksPruning {

	/// True if this is an archive pruning mode (either KeepAll or KeepFinalized).

	pub fn is_archive(&self) -> bool {

		match *self {

			BlocksPruning::KeepAll | BlocksPruning::KeepFinalized => true,

			BlocksPruning::Some(_) => false,

		}

	}

}

/// Where to find the database..

#[derive(Debug, Clone)]

pub enum DatabaseSource {

	/// Check given path, and see if there is an existing database there. If it's either `RocksDb`

	/// or `ParityDb`, use it. If there is none, create a new instance of `ParityDb`.

	Auto {

		/// Path to the paritydb database.

		paritydb_path: PathBuf,

		/// Path to the rocksdb database.

		rocksdb_path: PathBuf,

		/// Cache size in MiB. Used only by `RocksDb` variant of `DatabaseSource`.

		cache_size: usize,

	},

	/// Load a RocksDB database from a given path. Recommended for most uses.

	#[cfg(feature = "rocksdb")]

	RocksDb {

		/// Path to the database.

		path: PathBuf,

		/// Cache size in MiB.

		cache_size: usize,

	},

	/// Load a ParityDb database from a given path.

	ParityDb {

		/// Path to the database.

		path: PathBuf,

	},

	/// Use a custom already-open database.

	Custom {

		/// the handle to the custom storage

		db: Arc<dyn Database<DbHash>>,

		/// if set, the `create` flag will be required to open such datasource

		require_create_flag: bool,

	},

}

impl DatabaseSource {

	/// Return path for databases that are stored on disk.

			// as per https://github.com/paritytech/substrate/pull/9500#discussion_r684312550

			//

			// IIUC this is needed for polkadot to create its own dbs, so until it can use parity db

			// I would think rocksdb, but later parity-db.

			#[cfg(feature = "rocksdb")]

		}

	/// Set path for databases that are stored on disk.

	pub fn set_path(&mut self, p: &Path) -> bool {

		match self {

			DatabaseSource::Auto { ref mut paritydb_path, .. } => {

				*paritydb_path = p.into();

				true

			},

			#[cfg(feature = "rocksdb")]

			DatabaseSource::RocksDb { ref mut path, .. } => {

				*path = p.into();

				true

			},

			DatabaseSource::ParityDb { ref mut path } => {

				*path = p.into();

				true

			},

			DatabaseSource::Custom { .. } => false,

		}

	}

}

impl std::fmt::Display for DatabaseSource {

	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {

		let name = match self {

			DatabaseSource::Auto { .. } => "Auto",

			#[cfg(feature = "rocksdb")]

			DatabaseSource::RocksDb { .. } => "RocksDb",

			DatabaseSource::ParityDb { .. } => "ParityDb",

			DatabaseSource::Custom { .. } => "Custom",

		};

		write!(f, "{}", name)

	}

}

pub(crate) mod columns {

	pub const META: u32 = crate::utils::COLUMN_META;

	pub const STATE: u32 = 1;

	pub const STATE_META: u32 = 2;

	/// maps hashes to lookup keys and numbers to canon hashes.

	pub const KEY_LOOKUP: u32 = 3;

	pub const HEADER: u32 = 4;

	pub const BODY: u32 = 5;

	pub const JUSTIFICATIONS: u32 = 6;

	pub const AUX: u32 = 8;

	/// Offchain workers local storage

	pub const OFFCHAIN: u32 = 9;

	/// Transactions

	pub const TRANSACTION: u32 = 11;

	pub const BODY_INDEX: u32 = 12;

}

struct PendingBlock<Block: BlockT> {

	header: Block::Header,

	justifications: Option<Justifications>,

	body: Option<Vec<Block::Extrinsic>>,

	indexed_body: Option<Vec<Vec<u8>>>,

	leaf_state: NewBlockState,

}

// wrapper that implements trait required for state_db

#[derive(Clone)]

struct StateMetaDb(Arc<dyn Database<DbHash>>);

impl sc_state_db::MetaDb for StateMetaDb {

	type Error = sp_database::error::DatabaseError;

	fn get_meta(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {

		Ok(self.0.get(columns::STATE_META, key))

	}

}

struct MetaUpdate<Block: BlockT> {

	pub hash: Block::Hash,

	pub number: NumberFor<Block>,

	pub is_best: bool,

	pub is_finalized: bool,

	pub with_state: bool,

}

fn cache_header<Hash: std::cmp::Eq + std::hash::Hash, Header>(

	cache: &mut LinkedHashMap<Hash, Option<Header>>,

	hash: Hash,

	header: Option<Header>,

) {

	cache.insert(hash, header);

	while cache.len() > CACHE_HEADERS {

		cache.pop_front();

	}

}

/// Block database

pub struct BlockchainDb<Block: BlockT> {

	db: Arc<dyn Database<DbHash>>,

	meta: Arc<RwLock<Meta<NumberFor<Block>, Block::Hash>>>,

	leaves: RwLock<LeafSet<Block::Hash, NumberFor<Block>>>,

	header_metadata_cache: Arc<HeaderMetadataCache<Block>>,

	header_cache: Mutex<LinkedHashMap<Block::Hash, Option<Block::Header>>>,

	pinned_blocks_cache: Arc<RwLock<PinnedBlocksCache<Block>>>,

}

impl<Block: BlockT> BlockchainDb<Block> {

	fn new(db: Arc<dyn Database<DbHash>>) -> ClientResult<Self> {

		let meta = read_meta::<Block>(&*db, columns::HEADER)?;

		let leaves = LeafSet::read_from_db(&*db, columns::META, meta_keys::LEAF_PREFIX)?;

		Ok(BlockchainDb {

			db,

			leaves: RwLock::new(leaves),

			meta: Arc::new(RwLock::new(meta)),

			header_metadata_cache: Arc::new(HeaderMetadataCache::default()),

			header_cache: Default::default(),

			pinned_blocks_cache: Arc::new(RwLock::new(PinnedBlocksCache::new())),

		})

	}

	fn update_meta(&self, update: MetaUpdate<Block>) {

		let MetaUpdate { hash, number, is_best, is_finalized, with_state } = update;

		let mut meta = self.meta.write();

		if number.is_zero() {

			meta.genesis_hash = hash;

		}

		if is_best {

			meta.best_number = number;

			meta.best_hash = hash;

		}

		if is_finalized {

			if with_state {

				meta.finalized_state = Some((hash, number));

			}

			meta.finalized_number = number;

			meta.finalized_hash = hash;

		}

	}

	fn update_block_gap(&self, gap: Option<(NumberFor<Block>, NumberFor<Block>)>) {

		let mut meta = self.meta.write();

		meta.block_gap = gap;

	}

	/// Empty the cache of pinned items.

	fn clear_pinning_cache(&self) {

		self.pinned_blocks_cache.write().clear();

	}

	/// Load a justification into the cache of pinned items.

	/// Reference count of the item will not be increased. Use this

	/// to load values for items into the cache which have already been pinned.

	fn insert_justifications_if_pinned(&self, hash: Block::Hash, justification: Justification) {

		let mut cache = self.pinned_blocks_cache.write();

		if !cache.contains(hash) {

			return

		}

		let justifications = Justifications::from(justification);

		cache.insert_justifications(hash, Some(justifications));

	}

	/// Load a justification from the db into the cache of pinned items.

	/// Reference count of the item will not be increased. Use this

	/// to load values for items into the cache which have already been pinned.

	fn insert_persisted_justifications_if_pinned(&self, hash: Block::Hash) -> ClientResult<()> {

		let mut cache = self.pinned_blocks_cache.write();

		if !cache.contains(hash) {

			return Ok(())

		}

		let justifications = self.justifications_uncached(hash)?;

		cache.insert_justifications(hash, justifications);

		Ok(())

	}

	/// Load a block body from the db into the cache of pinned items.

	/// Reference count of the item will not be increased. Use this

	/// to load values for items items into the cache which have already been pinned.

	fn insert_persisted_body_if_pinned(&self, hash: Block::Hash) -> ClientResult<()> {

		let mut cache = self.pinned_blocks_cache.write();

		if !cache.contains(hash) {

			return Ok(())

		}

		let body = self.body_uncached(hash)?;

		cache.insert_body(hash, body);

		Ok(())

	}

	/// Bump reference count for pinned item.

	fn bump_ref(&self, hash: Block::Hash) {

		self.pinned_blocks_cache.write().pin(hash);

	}

	/// Decrease reference count for pinned item and remove if reference count is 0.

	fn unpin(&self, hash: Block::Hash) {

		self.pinned_blocks_cache.write().unpin(hash);

	}

	fn justifications_uncached(&self, hash: Block::Hash) -> ClientResult<Option<Justifications>> {

		match read_db(

			&*self.db,

			columns::KEY_LOOKUP,

			columns::JUSTIFICATIONS,

			BlockId::<Block>::Hash(hash),

		)? {

			Some(justifications) => match Decode::decode(&mut &justifications[..]) {

				Ok(justifications) => Ok(Some(justifications)),

				Err(err) =>

					return Err(sp_blockchain::Error::Backend(format!(

						"Error decoding justifications: {}",

						err

					))),

			},

			None => Ok(None),

		}

	}

	fn body_uncached(&self, hash: Block::Hash) -> ClientResult<Option<Vec<Block::Extrinsic>>> {

		if let Some(body) =

			read_db(&*self.db, columns::KEY_LOOKUP, columns::BODY, BlockId::Hash::<Block>(hash))?

		{

			// Plain body

			match Decode::decode(&mut &body[..]) {

				Ok(body) => return Ok(Some(body)),

				Err(err) =>

					return Err(sp_blockchain::Error::Backend(format!(

						"Error decoding body: {}",

						err

					))),

			}

		}

		if let Some(index) = read_db(

			&*self.db,

			columns::KEY_LOOKUP,

			columns::BODY_INDEX,

			BlockId::Hash::<Block>(hash),

		)? {

			match Vec::<DbExtrinsic<Block>>::decode(&mut &index[..]) {

				Ok(index) => {

					let mut body = Vec::new();

					for ex in index {

						match ex {

							DbExtrinsic::Indexed { hash, header } => {

								match self.db.get(columns::TRANSACTION, hash.as_ref()) {

									Some(t) => {

										let mut input =

											utils::join_input(header.as_ref(), t.as_ref());

										let ex = Block::Extrinsic::decode(&mut input).map_err(

											|err| {

												sp_blockchain::Error::Backend(format!(

													"Error decoding indexed extrinsic: {}",

													err

												))

											},

										)?;

										body.push(ex);

									},

									None =>

										return Err(sp_blockchain::Error::Backend(format!(

											"Missing indexed transaction {:?}",

											hash

										))),

								};

							},

							DbExtrinsic::Full(ex) => {

								body.push(ex);

							},

						}

					}

					return Ok(Some(body))

				},

				Err(err) =>

					return Err(sp_blockchain::Error::Backend(format!(

						"Error decoding body list: {}",

						err

					))),

			}

		}

		Ok(None)

	}

}

impl<Block: BlockT> sc_client_api::blockchain::HeaderBackend<Block> for BlockchainDb<Block> {

	fn header(&self, hash: Block::Hash) -> ClientResult<Option<Block::Header>> {

		let mut cache = self.header_cache.lock();

		if let Some(result) = cache.get_refresh(&hash) {

			return Ok(result.clone())

		}

		let header = utils::read_header(

			&*self.db,

			columns::KEY_LOOKUP,

			columns::HEADER,

			BlockId::<Block>::Hash(hash),

		)?;

		cache_header(&mut cache, hash, header.clone());

		Ok(header)

	}

	fn info(&self) -> sc_client_api::blockchain::Info<Block> {

		let meta = self.meta.read();

		sc_client_api::blockchain::Info {

			best_hash: meta.best_hash,

			best_number: meta.best_number,

			genesis_hash: meta.genesis_hash,

			finalized_hash: meta.finalized_hash,

			finalized_number: meta.finalized_number,

			finalized_state: meta.finalized_state,

			number_leaves: self.leaves.read().count(),

			block_gap: meta.block_gap,

		}

	}

	fn status(&self, hash: Block::Hash) -> ClientResult<sc_client_api::blockchain::BlockStatus> {

		match self.header(hash)?.is_some() {

			true => Ok(sc_client_api::blockchain::BlockStatus::InChain),

			false => Ok(sc_client_api::blockchain::BlockStatus::Unknown),

		}

	}

	fn number(&self, hash: Block::Hash) -> ClientResult<Option<NumberFor<Block>>> {

		Ok(self.header_metadata(hash).ok().map(|header_metadata| header_metadata.number))

	}

	fn hash(&self, number: NumberFor<Block>) -> ClientResult<Option<Block::Hash>> {

		Ok(utils::read_header::<Block>(

			&*self.db,

			columns::KEY_LOOKUP,

			columns::HEADER,

			BlockId::Number(number),

		)?

		.map(|header| header.hash()))

	}

}

impl<Block: BlockT> sc_client_api::blockchain::Backend<Block> for BlockchainDb<Block> {

	fn body(&self, hash: Block::Hash) -> ClientResult<Option<Vec<Block::Extrinsic>>> {

		let cache = self.pinned_blocks_cache.read();

		if let Some(result) = cache.body(&hash) {

			return Ok(result.clone())

		}

		self.body_uncached(hash)

	}

	fn justifications(&self, hash: Block::Hash) -> ClientResult<Option<Justifications>> {

		let cache = self.pinned_blocks_cache.read();

		if let Some(result) = cache.justifications(&hash) {

			return Ok(result.clone())

		}

		self.justifications_uncached(hash)

	}

	fn last_finalized(&self) -> ClientResult<Block::Hash> {

		Ok(self.meta.read().finalized_hash)

	}

	fn leaves(&self) -> ClientResult<Vec<Block::Hash>> {

		Ok(self.leaves.read().hashes())

	}

	fn children(&self, parent_hash: Block::Hash) -> ClientResult<Vec<Block::Hash>> {

		children::read_children(&*self.db, columns::META, meta_keys::CHILDREN_PREFIX, parent_hash)

	}

	fn indexed_transaction(&self, hash: Block::Hash) -> ClientResult<Option<Vec<u8>>> {

		Ok(self.db.get(columns::TRANSACTION, hash.as_ref()))

	}

	fn has_indexed_transaction(&self, hash: Block::Hash) -> ClientResult<bool> {

		Ok(self.db.contains(columns::TRANSACTION, hash.as_ref()))

	}

	fn block_indexed_body(&self, hash: Block::Hash) -> ClientResult<Option<Vec<Vec<u8>>>> {

		let body = match read_db(

			&*self.db,

			columns::KEY_LOOKUP,

			columns::BODY_INDEX,

			BlockId::<Block>::Hash(hash),

		)? {

			Some(body) => body,

			None => return Ok(None),

		};

		match Vec::<DbExtrinsic<Block>>::decode(&mut &body[..]) {

			Ok(index) => {

				let mut transactions = Vec::new();

				for ex in index.into_iter() {

					if let DbExtrinsic::Indexed { hash, .. } = ex {

						match self.db.get(columns::TRANSACTION, hash.as_ref()) {

							Some(t) => transactions.push(t),

							None =>

								return Err(sp_blockchain::Error::Backend(format!(

									"Missing indexed transaction {:?}",

									hash

								))),

						}

					}

				}

				Ok(Some(transactions))

			},

			Err(err) =>

				Err(sp_blockchain::Error::Backend(format!("Error decoding body list: {}", err))),

		}

	}

}

impl<Block: BlockT> HeaderMetadata<Block> for BlockchainDb<Block> {

	type Error = sp_blockchain::Error;

	fn header_metadata(

		&self,

		hash: Block::Hash,

	) -> Result<CachedHeaderMetadata<Block>, Self::Error> {

		self.header_metadata_cache.header_metadata(hash).map_or_else(

			|| {

				self.header(hash)?

					.map(|header| {

						let header_metadata = CachedHeaderMetadata::from(&header);

						self.header_metadata_cache

							.insert_header_metadata(header_metadata.hash, header_metadata.clone());

						header_metadata

					})

					.ok_or_else(|| {

						ClientError::UnknownBlock(format!(

							"Header was not found in the database: {:?}",

							hash

						))

					})

			},

			Ok,

		)

	}

	fn insert_header_metadata(&self, hash: Block::Hash, metadata: CachedHeaderMetadata<Block>) {

		self.header_metadata_cache.insert_header_metadata(hash, metadata)

	}

	fn remove_header_metadata(&self, hash: Block::Hash) {

		self.header_cache.lock().remove(&hash);

		self.header_metadata_cache.remove_header_metadata(hash);

	}

}

/// Database transaction

pub struct BlockImportOperation<Block: BlockT> {

	old_state: RecordStatsState<RefTrackingState<Block>, Block>,

	db_updates: PrefixedMemoryDB<HashingFor<Block>>,

	storage_updates: StorageCollection,

	child_storage_updates: ChildStorageCollection,

	offchain_storage_updates: OffchainChangesCollection,

	pending_block: Option<PendingBlock<Block>>,

	aux_ops: Vec<(Vec<u8>, Option<Vec<u8>>)>,

	finalized_blocks: Vec<(Block::Hash, Option<Justification>)>,

	set_head: Option<Block::Hash>,

	commit_state: bool,

	index_ops: Vec<IndexOperation>,

}

impl<Block: BlockT> BlockImportOperation<Block> {

	fn apply_offchain(&mut self, transaction: &mut Transaction<DbHash>) {

		let mut count = 0;

		for ((prefix, key), value_operation) in self.offchain_storage_updates.drain(..) {

			count += 1;

			let key = crate::offchain::concatenate_prefix_and_key(&prefix, &key);

			match value_operation {

				OffchainOverlayedChange::SetValue(val) =>

					transaction.set_from_vec(columns::OFFCHAIN, &key, val),

				OffchainOverlayedChange::Remove => transaction.remove(columns::OFFCHAIN, &key),

			}

		}

		if count > 0 {

			log::debug!(target: "sc_offchain", "Applied {} offchain indexing changes.", count);

		}

	}

	fn apply_aux(&mut self, transaction: &mut Transaction<DbHash>) {

		for (key, maybe_val) in self.aux_ops.drain(..) {

			match maybe_val {

				Some(val) => transaction.set_from_vec(columns::AUX, &key, val),

				None => transaction.remove(columns::AUX, &key),

			}

		}

	}

	fn apply_new_state(

		&mut self,

		storage: Storage,

		state_version: StateVersion,

	) -> ClientResult<Block::Hash> {

		if storage.top.keys().any(|k| well_known_keys::is_child_storage_key(k)) {

			return Err(sp_blockchain::Error::InvalidState)

		}

		let child_delta = storage.children_default.values().map(|child_content| {

			(

				&child_content.child_info,

				child_content.data.iter().map(|(k, v)| (&k[..], Some(&v[..]))),

			)

		});

		let (root, transaction) = self.old_state.full_storage_root(

			storage.top.iter().map(|(k, v)| (&k[..], Some(&v[..]))),

			child_delta,

			state_version,

		);

		self.db_updates = transaction;

		Ok(root)

	}

}

impl<Block: BlockT> sc_client_api::backend::BlockImportOperation<Block>

	for BlockImportOperation<Block>

{

	type State = RecordStatsState<RefTrackingState<Block>, Block>;

	fn state(&self) -> ClientResult<Option<&Self::State>> {

		Ok(Some(&self.old_state))

	}

	fn set_block_data(

		&mut self,

		header: Block::Header,

		body: Option<Vec<Block::Extrinsic>>,

		indexed_body: Option<Vec<Vec<u8>>>,

		justifications: Option<Justifications>,

		leaf_state: NewBlockState,

	) -> ClientResult<()> {

		assert!(self.pending_block.is_none(), "Only one block per operation is allowed");

		self.pending_block =

			Some(PendingBlock { header, body, indexed_body, justifications, leaf_state });

		Ok(())

	}

	fn update_db_storage(

		&mut self,

		update: PrefixedMemoryDB<HashingFor<Block>>,

	) -> ClientResult<()> {

		self.db_updates = update;

		Ok(())

	}

	fn reset_storage(

		&mut self,

		storage: Storage,

		state_version: StateVersion,

	) -> ClientResult<Block::Hash> {

		let root = self.apply_new_state(storage, state_version)?;

		self.commit_state = true;

		Ok(root)

	}

	fn set_genesis_state(

		&mut self,

		storage: Storage,

		commit: bool,

		state_version: StateVersion,

	) -> ClientResult<Block::Hash> {

		let root = self.apply_new_state(storage, state_version)?;

		self.commit_state = commit;

		Ok(root)

	}

	fn insert_aux<I>(&mut self, ops: I) -> ClientResult<()>

	where

		I: IntoIterator<Item = (Vec<u8>, Option<Vec<u8>>)>,

	{

		self.aux_ops.append(&mut ops.into_iter().collect());

		Ok(())

	}

	fn update_storage(

		&mut self,

		update: StorageCollection,

		child_update: ChildStorageCollection,

	) -> ClientResult<()> {

		self.storage_updates = update;

		self.child_storage_updates = child_update;

		Ok(())

	}

	fn update_offchain_storage(

		&mut self,

		offchain_update: OffchainChangesCollection,

	) -> ClientResult<()> {

		self.offchain_storage_updates = offchain_update;

		Ok(())

	}

	fn mark_finalized(

		&mut self,

		block: Block::Hash,

		justification: Option<Justification>,

	) -> ClientResult<()> {

		self.finalized_blocks.push((block, justification));

		Ok(())

	}

	fn mark_head(&mut self, hash: Block::Hash) -> ClientResult<()> {

		assert!(self.set_head.is_none(), "Only one set head per operation is allowed");

		self.set_head = Some(hash);

		Ok(())

	}

	fn update_transaction_index(&mut self, index_ops: Vec<IndexOperation>) -> ClientResult<()> {

		self.index_ops = index_ops;

		Ok(())

	}

}

struct StorageDb<Block: BlockT> {

	pub db: Arc<dyn Database<DbHash>>,

	pub state_db: StateDb<Block::Hash, Vec<u8>, StateMetaDb>,

	prefix_keys: bool,

}

impl<Block: BlockT> sp_state_machine::Storage<HashingFor<Block>> for StorageDb<Block> {

	fn get(&self, key: &Block::Hash, prefix: Prefix) -> Result<Option<DBValue>, String> {

		if self.prefix_keys {

			let key = prefixed_key::<HashingFor<Block>>(key, prefix);

			self.state_db.get(&key, self)

		} else {

			self.state_db.get(key.as_ref(), self)

		}

		.map_err(|e| format!("Database backend error: {:?}", e))

	}

}

impl<Block: BlockT> sc_state_db::NodeDb for StorageDb<Block> {

	type Error = io::Error;

	type Key = [u8];

	fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {

		Ok(self.db.get(columns::STATE, key))

	}

}

struct DbGenesisStorage<Block: BlockT> {

	root: Block::Hash,

	storage: PrefixedMemoryDB<HashingFor<Block>>,

}

impl<Block: BlockT> DbGenesisStorage<Block> {

	pub fn new(root: Block::Hash, storage: PrefixedMemoryDB<HashingFor<Block>>) -> Self {

		DbGenesisStorage { root, storage }

	}

}

impl<Block: BlockT> sp_state_machine::Storage<HashingFor<Block>> for DbGenesisStorage<Block> {

	fn get(&self, key: &Block::Hash, prefix: Prefix) -> Result<Option<DBValue>, String> {

		use hash_db::HashDB;

		Ok(self.storage.get(key, prefix))

	}

}

struct EmptyStorage<Block: BlockT>(pub Block::Hash);

impl<Block: BlockT> EmptyStorage<Block> {

	pub fn new() -> Self {

		let mut root = Block::Hash::default();

		let mut mdb = MemoryDB::<HashingFor<Block>>::default();

		// both triedbmut are the same on empty storage.

		sp_trie::trie_types::TrieDBMutBuilderV1::<HashingFor<Block>>::new(&mut mdb, &mut root)

			.build();

		EmptyStorage(root)

	}

}

impl<Block: BlockT> sp_state_machine::Storage<HashingFor<Block>> for EmptyStorage<Block> {

	fn get(&self, _key: &Block::Hash, _prefix: Prefix) -> Result<Option<DBValue>, String> {

		Ok(None)

	}

}

/// Frozen `value` at time `at`.

///

/// Used as inner structure under lock in `FrozenForDuration`.

struct Frozen<T: Clone> {

	at: std::time::Instant,

	value: Option<T>,

}

/// Some value frozen for period of time.

///

/// If time `duration` not passed since the value was instantiated,

/// current frozen value is returned. Otherwise, you have to provide

/// a new value which will be again frozen for `duration`.

pub(crate) struct FrozenForDuration<T: Clone> {

	duration: std::time::Duration,

	value: parking_lot::Mutex<Frozen<T>>,

}

impl<T: Clone> FrozenForDuration<T> {

	fn new(duration: std::time::Duration) -> Self {

		Self { duration, value: Frozen { at: std::time::Instant::now(), value: None }.into() }

	}

	fn take_or_else<F>(&self, f: F) -> T

	where

		F: FnOnce() -> T,

	{

		let mut lock = self.value.lock();

		let now = std::time::Instant::now();

		if now.saturating_duration_since(lock.at) > self.duration || lock.value.is_none() {

			let new_value = f();

			lock.at = now;

			lock.value = Some(new_value.clone());

			new_value

		} else {

			lock.value.as_ref().expect("Checked with in branch above; qed").clone()

		}

	}

}

/// Disk backend.

///

/// Disk backend keeps data in a key-value store. In archive mode, trie nodes are kept from all

/// blocks. Otherwise, trie nodes are kept only from some recent blocks.

pub struct Backend<Block: BlockT> {

	storage: Arc<StorageDb<Block>>,

	offchain_storage: offchain::LocalStorage,

	blockchain: BlockchainDb<Block>,

	canonicalization_delay: u64,

	import_lock: Arc<RwLock<()>>,

	is_archive: bool,

	blocks_pruning: BlocksPruning,

	io_stats: FrozenForDuration<(kvdb::IoStats, StateUsageInfo)>,

	state_usage: Arc<StateUsageStats>,

	genesis_state: RwLock<Option<Arc<DbGenesisStorage<Block>>>>,

	shared_trie_cache: Option<sp_trie::cache::SharedTrieCache<HashingFor<Block>>>,

}

impl<Block: BlockT> Backend<Block> {

	/// Create a new instance of database backend.

	///

	/// The pruning window is how old a block must be before the state is pruned.

	pub fn new(db_config: DatabaseSettings, canonicalization_delay: u64) -> ClientResult<Self> {

		use utils::OpenDbError;

		let db_source = &db_config.source;

		let (needs_init, db) =

			match crate::utils::open_database::<Block>(db_source, DatabaseType::Full, false) {

				Ok(db) => (false, db),

				Err(OpenDbError::DoesNotExist) => {

					let db =

						crate::utils::open_database::<Block>(db_source, DatabaseType::Full, true)?;

					(true, db)

				},

				Err(as_is) => return Err(as_is.into()),

			};

		Self::from_database(db as Arc<_>, canonicalization_delay, &db_config, needs_init)

	}

	/// Reset the shared trie cache.

	pub fn reset_trie_cache(&self) {

		if let Some(cache) = &self.shared_trie_cache {

			cache.reset();

		}

	}

	/// Create new memory-backed client backend for tests.

	#[cfg(any(test, feature = "test-helpers"))]

	pub fn new_test(blocks_pruning: u32, canonicalization_delay: u64) -> Self {

		Self::new_test_with_tx_storage(BlocksPruning::Some(blocks_pruning), canonicalization_delay)

	}

	/// Create new memory-backed client backend for tests.

	#[cfg(any(test, feature = "test-helpers"))]

	pub fn new_test_with_tx_storage(

		blocks_pruning: BlocksPruning,

		canonicalization_delay: u64,

	) -> Self {

		let db = kvdb_memorydb::create(crate::utils::NUM_COLUMNS);

		let db = sp_database::as_database(db);

		let state_pruning = match blocks_pruning {

			BlocksPruning::KeepAll => PruningMode::ArchiveAll,

			BlocksPruning::KeepFinalized => PruningMode::ArchiveCanonical,

			BlocksPruning::Some(n) => PruningMode::blocks_pruning(n),

		};

		let db_setting = DatabaseSettings {

			trie_cache_maximum_size: Some(16 * 1024 * 1024),

			state_pruning: Some(state_pruning),

			source: DatabaseSource::Custom { db, require_create_flag: true },

			blocks_pruning,

		};

		Self::new(db_setting, canonicalization_delay).expect("failed to create test-db")

	}

	/// Expose the Database that is used by this backend.

	/// The second argument is the Column that stores the State.

	///

	/// Should only be needed for benchmarking.

	#[cfg(any(feature = "runtime-benchmarks"))]

	pub fn expose_db(&self) -> (Arc<dyn sp_database::Database<DbHash>>, sp_database::ColumnId) {

		(self.storage.db.clone(), columns::STATE)

	}

	/// Expose the Storage that is used by this backend.

	///

	/// Should only be needed for benchmarking.

	#[cfg(any(feature = "runtime-benchmarks"))]

	pub fn expose_storage(&self) -> Arc<dyn sp_state_machine::Storage<HashingFor<Block>>> {

		self.storage.clone()

	}

	fn from_database(

		db: Arc<dyn Database<DbHash>>,

		canonicalization_delay: u64,

		config: &DatabaseSettings,

		should_init: bool,

	) -> ClientResult<Self> {

		let mut db_init_transaction = Transaction::new();

		let requested_state_pruning = config.state_pruning.clone();

		let state_meta_db = StateMetaDb(db.clone());

		let map_e = sp_blockchain::Error::from_state_db;

		let (state_db_init_commit_set, state_db) = StateDb::open(

			state_meta_db,

			requested_state_pruning,

			!db.supports_ref_counting(),

			should_init,

		)

		.map_err(map_e)?;

		apply_state_commit(&mut db_init_transaction, state_db_init_commit_set);

		let state_pruning_used = state_db.pruning_mode();

		let is_archive_pruning = state_pruning_used.is_archive();

		let blockchain = BlockchainDb::new(db.clone())?;

		let storage_db =

			StorageDb { db: db.clone(), state_db, prefix_keys: !db.supports_ref_counting() };

		let offchain_storage = offchain::LocalStorage::new(db.clone());

		let backend = Backend {

			storage: Arc::new(storage_db),

			offchain_storage,

			blockchain,

			canonicalization_delay,

			import_lock: Default::default(),

			is_archive: is_archive_pruning,

			io_stats: FrozenForDuration::new(std::time::Duration::from_secs(1)),

			state_usage: Arc::new(StateUsageStats::new()),

			blocks_pruning: config.blocks_pruning,

			genesis_state: RwLock::new(None),

			shared_trie_cache: config.trie_cache_maximum_size.map(|maximum_size| {

				SharedTrieCache::new(sp_trie::cache::CacheSize::new(maximum_size))

			}),

		};

		// Older DB versions have no last state key. Check if the state is available and set it.

		let info = backend.blockchain.info();

		if info.finalized_state.is_none() &&

			info.finalized_hash != Default::default() &&

			sc_client_api::Backend::have_state_at(

				&backend,

				info.finalized_hash,

				info.finalized_number,

			) {

			backend.blockchain.update_meta(MetaUpdate {

				hash: info.finalized_hash,

				number: info.finalized_number,

				is_best: info.finalized_hash == info.best_hash,

				is_finalized: true,

				with_state: true,

			});

		}

		db.commit(db_init_transaction)?;

		Ok(backend)

	}

	/// Handle setting head within a transaction. `route_to` should be the last

	/// block that existed in the database. `best_to` should be the best block

	/// to be set.

	///

	/// In the case where the new best block is a block to be imported, `route_to`

	/// should be the parent of `best_to`. In the case where we set an existing block

	/// to be best, `route_to` should equal to `best_to`.

	fn set_head_with_transaction(

		&self,

		transaction: &mut Transaction<DbHash>,

		route_to: Block::Hash,

		best_to: (NumberFor<Block>, Block::Hash),

	) -> ClientResult<(Vec<Block::Hash>, Vec<Block::Hash>)> {

		let mut enacted = Vec::default();

		let mut retracted = Vec::default();

		let (best_number, best_hash) = best_to;

		let meta = self.blockchain.meta.read();

		if meta.best_number.saturating_sub(best_number).saturated_into::<u64>() >

			self.canonicalization_delay

		{

			return Err(sp_blockchain::Error::SetHeadTooOld)

		}

		let parent_exists =

			self.blockchain.status(route_to)? == sp_blockchain::BlockStatus::InChain;

		// Cannot find tree route with empty DB or when imported a detached block.

		if meta.best_hash != Default::default() && parent_exists {

			let tree_route = sp_blockchain::tree_route(&self.blockchain, meta.best_hash, route_to)?;

			// uncanonicalize: check safety violations and ensure the numbers no longer

			// point to these block hashes in the key mapping.

			for r in tree_route.retracted() {

				if r.hash == meta.finalized_hash {

					warn!(

						"Potential safety failure: reverting finalized block {:?}",

						(&r.number, &r.hash)

					);

					return Err(sp_blockchain::Error::NotInFinalizedChain)

				}

				retracted.push(r.hash);

				utils::remove_number_to_key_mapping(transaction, columns::KEY_LOOKUP, r.number)?;

			}

			// canonicalize: set the number lookup to map to this block's hash.

			for e in tree_route.enacted() {

				enacted.push(e.hash);

				utils::insert_number_to_key_mapping(

					transaction,

					columns::KEY_LOOKUP,

					e.number,

					e.hash,

				)?;

			}

		}

		let lookup_key = utils::number_and_hash_to_lookup_key(best_number, &best_hash)?;

		transaction.set_from_vec(columns::META, meta_keys::BEST_BLOCK, lookup_key);

		utils::insert_number_to_key_mapping(

			transaction,