// Copyright 2024 Moonbeam foundation

// This file is part of Moonbeam.

// Moonbeam 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.

// Moonbeam 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 Moonbeam.  If not, see <http://www.gnu.org/licenses/>.

use crate::{

	lazy_loading, open_frontier_backend, rpc, set_prometheus_registry, BlockImportPipeline,

	ClientCustomizations, FrontierBlockImport, HostFunctions, PartialComponentsResult,

	PendingConsensusDataProvider, RuntimeApiCollection, SOFT_DEADLINE_PERCENT,

};

use cumulus_client_parachain_inherent::{MockValidationDataInherentDataProvider, MockXcmConfig};

use cumulus_primitives_core::{relay_chain, BlockT, ParaId};

use cumulus_primitives_parachain_inherent::ParachainInherentData;

use cumulus_test_relay_sproof_builder::RelayStateSproofBuilder;

use fc_rpc::StorageOverrideHandler;

use fc_rpc_core::types::{FeeHistoryCache, FilterPool};

use futures::{FutureExt, StreamExt};

use moonbeam_cli_opt::{EthApi as EthApiCmd, LazyLoadingConfig, RpcConfig};

use moonbeam_core_primitives::{Block, Hash};

use nimbus_consensus::NimbusManualSealConsensusDataProvider;

use nimbus_primitives::NimbusId;

use parity_scale_codec::Encode;

use polkadot_primitives::{

	AbridgedHostConfiguration, AsyncBackingParams, PersistedValidationData, Slot, UpgradeGoAhead,

};

use sc_chain_spec::{get_extension, BuildGenesisBlock, GenesisBlockBuilder};

use sc_client_api::{Backend, BadBlocks, ExecutorProvider, ForkBlocks, StorageProvider};

use sc_executor::{HeapAllocStrategy, RuntimeVersionOf, WasmExecutor, DEFAULT_HEAP_ALLOC_STRATEGY};

use sc_network::config::FullNetworkConfiguration;

use sc_network::NetworkBackend;

use sc_network_common::sync::SyncMode;

use sc_service::{

	error::Error as ServiceError, ClientConfig, Configuration, Error, KeystoreContainer,

	LocalCallExecutor, PartialComponents, TaskManager,

};

use sc_telemetry::{TelemetryHandle, TelemetryWorker};

use sc_transaction_pool_api::OffchainTransactionPoolFactory;

use sp_api::ConstructRuntimeApi;

use sp_blockchain::HeaderBackend;

use sp_core::traits::CodeExecutor;

use sp_core::{twox_128, H256};

use sp_runtime::traits::NumberFor;

use sp_storage::StorageKey;

use std::collections::BTreeMap;

use std::str::FromStr;

use std::sync::{Arc, Mutex};

use std::time::Duration;

pub mod backend;

pub mod call_executor;

mod client;

mod helpers;

mod lock;

mod state_overrides;

pub const LAZY_LOADING_LOG_TARGET: &'static str = "lazy-loading";

/// Lazy loading client type.

pub type TLazyLoadingClient<TBl, TRtApi, TExec> = sc_service::client::Client<

	TLazyLoadingBackend<TBl>,

	TLazyLoadingCallExecutor<TBl, TExec>,

	TBl,

	TRtApi,

>;

/// Lazy loading client backend type.

pub type TLazyLoadingBackend<TBl> = backend::Backend<TBl>;

/// Lazy loading client call executor type.

pub type TLazyLoadingCallExecutor<TBl, TExec> = call_executor::LazyLoadingCallExecutor<

	TBl,

	LocalCallExecutor<TBl, TLazyLoadingBackend<TBl>, TExec>,

>;

/// Lazy loading parts type.

pub type TLazyLoadingParts<TBl, TRtApi, TExec> = (

	TLazyLoadingClient<TBl, TRtApi, TExec>,

	Arc<TLazyLoadingBackend<TBl>>,

	KeystoreContainer,

	TaskManager,

);

type LazyLoadingClient<RuntimeApi> =

	TLazyLoadingClient<Block, RuntimeApi, WasmExecutor<HostFunctions>>;

type LazyLoadingBackend = TLazyLoadingBackend<Block>;

/// Create the initial parts of a lazy loading node.

/// Create the initial parts of a lazy loading node.

	};

					SyncMode::LightState { .. } | SyncMode::Warp { .. }

				),

				enable_import_proof_recording: false,

			},

/// Builds the PartialComponents for a lazy loading node.

#[allow(clippy::type_complexity)]

	// Use ethereum style for subscription ids

/// Builds a new lazy loading service. This service uses manual seal, and mocks

/// the parachain inherent.

pub async fn new_lazy_loading_service<RuntimeApi, Customizations, Net>(

	mut config: Configuration,

	_author_id: Option<NimbusId>,

	sealing: moonbeam_cli_opt::Sealing,

	rpc_config: RpcConfig,

	lazy_loading_config: LazyLoadingConfig,

	hwbench: Option<sc_sysinfo::HwBench>,

) -> Result<TaskManager, ServiceError>

where

	RuntimeApi: ConstructRuntimeApi<Block, LazyLoadingClient<RuntimeApi>> + Send + Sync + 'static,

	RuntimeApi::RuntimeApi: RuntimeApiCollection,

	Customizations: ClientCustomizations + 'static,

	Net: NetworkBackend<Block, Hash>,

{

	use async_io::Timer;

	use futures::Stream;

	use sc_consensus_manual_seal::{run_manual_seal, EngineCommand, ManualSealParams};

	let sc_service::PartialComponents {

		client,

		backend,

		mut task_manager,

		import_queue,

		keystore_container,

		select_chain: maybe_select_chain,

		transaction_pool,

		other:

			(

				block_import_pipeline,

				filter_pool,

				mut telemetry,

				_telemetry_worker_handle,

				frontier_backend,

				fee_history_cache,

			),

	} = lazy_loading::new_lazy_loading_partial::<RuntimeApi, Customizations>(

		&mut config,

		&rpc_config,

		&lazy_loading_config,

	)?;

	let start_delay = 10;

	let lazy_loading_startup_disclaimer = format!(

		r#"

		You are now running the Moonbeam client in lazy loading mode, where data is retrieved

		from a live RPC node on demand.

		Using remote state from: {rpc}

		Forking from block: {fork_block}

		To ensure the client works properly, please note the following:

		    1. *Avoid Throttling*: Ensure that the backing RPC node is not limiting the number of

		    requests, as this can prevent the lazy loading client from functioning correctly;

		    2. *Be Patient*: As the client may take approximately 20 times longer than normal to

		    retrieve and process the necessary data for the requested operation.

		The service will start in {start_delay} seconds...

		"#,

		rpc = lazy_loading_config.state_rpc,

		fork_block = backend.fork_checkpoint.number

	);

	log::warn!(

		"{}",

		ansi_term::Colour::Yellow.paint(lazy_loading_startup_disclaimer)

	);

	tokio::time::sleep(Duration::from_secs(start_delay)).await;

	let block_import = if let BlockImportPipeline::Dev(block_import) = block_import_pipeline {

		block_import

	} else {

		return Err(ServiceError::Other(

			"Block import pipeline is not dev".to_string(),

		));

	};

	let prometheus_registry = config.prometheus_registry().cloned();

	let net_config =

		FullNetworkConfiguration::<_, _, Net>::new(&config.network, prometheus_registry.clone());

	let metrics = Net::register_notification_metrics(

	);

	let (network, system_rpc_tx, tx_handler_controller, network_starter, sync_service) =

		sc_service::build_network(sc_service::BuildNetworkParams {

			config: &config,

			client: client.clone(),

			transaction_pool: transaction_pool.clone(),

			spawn_handle: task_manager.spawn_handle(),

			import_queue,

			block_announce_validator_builder: None,

			warp_sync_config: None,

			net_config,

			block_relay: None,

			metrics,

		})?;

	if config.offchain_worker.enabled {

		task_manager.spawn_handle().spawn(

			"offchain-workers-runner",

			"offchain-work",

			sc_offchain::OffchainWorkers::new(sc_offchain::OffchainWorkerOptions {

				runtime_api_provider: client.clone(),

				keystore: Some(keystore_container.keystore()),

				offchain_db: backend.offchain_storage(),

				transaction_pool: Some(OffchainTransactionPoolFactory::new(

					transaction_pool.clone(),

				)),

				network_provider: Arc::new(network.clone()),

				is_validator: config.role.is_authority(),

				enable_http_requests: true,

			})

			.run(client.clone(), task_manager.spawn_handle())

			.boxed(),

		);

	}

	let prometheus_registry = config.prometheus_registry().cloned();

	let overrides = Arc::new(StorageOverrideHandler::new(client.clone()));

	let fee_history_limit = rpc_config.fee_history_limit;

	let mut command_sink = None;

	let mut dev_rpc_data = None;

	let collator = config.role.is_authority();

	if collator {

		let mut env = sc_basic_authorship::ProposerFactory::with_proof_recording(

			task_manager.spawn_handle(),

			client.clone(),

			transaction_pool.clone(),

			prometheus_registry.as_ref(),

		);

		env.set_soft_deadline(SOFT_DEADLINE_PERCENT);

		let commands_stream: Box<dyn Stream<Item = EngineCommand<H256>> + Send + Sync + Unpin> =

			match sealing {

				moonbeam_cli_opt::Sealing::Instant => {

					Box::new(

						// This bit cribbed from the implementation of instant seal.

						transaction_pool

							.pool()

							.validated_pool()

							.import_notification_stream()

					)

				}

				moonbeam_cli_opt::Sealing::Manual => {

					let (sink, stream) = futures::channel::mpsc::channel(1000);

					// Keep a reference to the other end of the channel. It goes to the RPC.

					command_sink = Some(sink);

					Box::new(stream)

				}

				moonbeam_cli_opt::Sealing::Interval(millis) => Box::new(StreamExt::map(

					Timer::interval(Duration::from_millis(millis)),

				)),

			};

		let select_chain = maybe_select_chain.expect(

			"`new_lazy_loading_partial` builds a `LongestChainRule` when building dev service.\

				We specified the dev service when calling `new_partial`.\

				Therefore, a `LongestChainRule` is present. qed.",

		);

		let client_set_aside_for_cidp = client.clone();

		// Create channels for mocked XCM messages.

		let (downward_xcm_sender, downward_xcm_receiver) = flume::bounded::<Vec<u8>>(100);

		let (hrmp_xcm_sender, hrmp_xcm_receiver) = flume::bounded::<(ParaId, Vec<u8>)>(100);

		let additional_relay_offset = Arc::new(std::sync::atomic::AtomicU32::new(0));

		dev_rpc_data = Some((

			downward_xcm_sender,

			hrmp_xcm_sender,

			additional_relay_offset,

		));

		let client_clone = client.clone();

		let keystore_clone = keystore_container.keystore().clone();

		let maybe_provide_vrf_digest =

		let parachain_id = helpers::get_parachain_id(backend.rpc_client.clone())

		task_manager.spawn_essential_handle().spawn_blocking(

			"authorship_task",

			Some("block-authoring"),

			run_manual_seal(ManualSealParams {

				block_import,

				env,

				client: client.clone(),

				pool: transaction_pool.clone(),

				commands_stream,

				select_chain,

				consensus_data_provider: Some(Box::new(NimbusManualSealConsensusDataProvider {

					keystore: keystore_container.keystore(),

					client: client.clone(),

					additional_digests_provider: maybe_provide_vrf_digest,

					_phantom: Default::default(),

				})),

			}),

		);

	}

	// Sinks for pubsub notifications.

	// Everytime a new subscription is created, a new mpsc channel is added to the sink pool.

	// The MappingSyncWorker sends through the channel on block import and the subscription emits a

	// notification to the subscriber on receiving a message through this channel.

	// This way we avoid race conditions when using native substrate block import notification

	// stream.

	let pubsub_notification_sinks: fc_mapping_sync::EthereumBlockNotificationSinks<

		fc_mapping_sync::EthereumBlockNotification<Block>,

	> = Default::default();

	let pubsub_notification_sinks = Arc::new(pubsub_notification_sinks);

	rpc::spawn_essential_tasks(

		rpc::SpawnTasksParams {

			task_manager: &task_manager,

			client: client.clone(),

			substrate_backend: backend.clone(),

			frontier_backend: frontier_backend.clone(),

			filter_pool: filter_pool.clone(),

			overrides: overrides.clone(),

			fee_history_limit,

			fee_history_cache: fee_history_cache.clone(),

		},

		sync_service.clone(),

		pubsub_notification_sinks.clone(),

	);

	let ethapi_cmd = rpc_config.ethapi.clone();

	let tracing_requesters =

		if ethapi_cmd.contains(&EthApiCmd::Debug) || ethapi_cmd.contains(&EthApiCmd::Trace) {

			rpc::tracing::spawn_tracing_tasks(

				&rpc_config,

				prometheus_registry.clone(),

				rpc::SpawnTasksParams {

					task_manager: &task_manager,

					client: client.clone(),

					substrate_backend: backend.clone(),

					frontier_backend: frontier_backend.clone(),

					filter_pool: filter_pool.clone(),

					overrides: overrides.clone(),

					fee_history_limit,

					fee_history_cache: fee_history_cache.clone(),

				},

			)

		} else {

			rpc::tracing::RpcRequesters {

				debug: None,

				trace: None,

			}

		};

	let block_data_cache = Arc::new(fc_rpc::EthBlockDataCacheTask::new(

		task_manager.spawn_handle(),

		overrides.clone(),

		rpc_config.eth_log_block_cache,

		rpc_config.eth_statuses_cache,

		prometheus_registry,

	));

	let rpc_builder = {

		let client = client.clone();

		let pool = transaction_pool.clone();

		let backend = backend.clone();

		let network = network.clone();

		let sync = sync_service.clone();

		let ethapi_cmd = ethapi_cmd.clone();

		let max_past_logs = rpc_config.max_past_logs;

		let overrides = overrides.clone();

		let fee_history_cache = fee_history_cache.clone();

		let block_data_cache = block_data_cache.clone();

		let pubsub_notification_sinks = pubsub_notification_sinks.clone();

		let keystore = keystore_container.keystore();

		let command_sink_for_task = command_sink.clone();

				},

			} else {

			}

	};

	let _rpc_handlers = sc_service::spawn_tasks(sc_service::SpawnTasksParams {

		network,

		client,

		keystore: keystore_container.keystore(),

		task_manager: &mut task_manager,

		transaction_pool,

		rpc_builder: Box::new(rpc_builder),

		backend,

		system_rpc_tx,

		sync_service: sync_service.clone(),

		config,

		tx_handler_controller,

		telemetry: None,

	})?;

	if let Some(hwbench) = hwbench {

		sc_sysinfo::print_hwbench(&hwbench);

		if let Some(ref mut telemetry) = telemetry {

			let telemetry_handle = telemetry.handle();

			task_manager.spawn_handle().spawn(

				"telemetry_hwbench",

				None,

				sc_sysinfo::initialize_hwbench_telemetry(telemetry_handle, hwbench),

			);

		}

	}

	network_starter.start_network();

	log::info!("Service Ready");

	Ok(task_manager)

}