// Copyright (C) Moondance Labs Ltd.

// This file is part of Tanssi.

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

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

//! Service and ServiceFactory implementation. Specialized wrapper over substrate service.

use {

    crate::command::solochain::{

        build_solochain_config_dir, copy_zombienet_keystore, dummy_config, keystore_config,

    },

    core::marker::PhantomData,

    cumulus_client_cli::CollatorOptions,

    cumulus_client_collator::service::CollatorService,

    cumulus_client_consensus_proposer::Proposer,

    cumulus_client_parachain_inherent::{MockValidationDataInherentDataProvider, MockXcmConfig},

    cumulus_client_service::{

        prepare_node_config, start_relay_chain_tasks, DARecoveryProfile, StartRelayChainTasksParams,

    },

    cumulus_primitives_core::{

        relay_chain::{well_known_keys as RelayWellKnownKeys, CollatorPair},

        ParaId,

    },

    cumulus_relay_chain_interface::{

        call_remote_runtime_function, OverseerHandle, RelayChainInterface,

    },

    dancebox_runtime::{

        opaque::{Block, Hash},

        AccountId, RuntimeApi,

    },

    dc_orchestrator_chain_interface::{

        BlockNumber, ContainerChainGenesisData, DataPreserverAssignment, DataPreserverProfileId,

        OrchestratorChainError, OrchestratorChainInterface, OrchestratorChainResult, PHash,

        PHeader,

    },

    futures::{Stream, StreamExt},

    nimbus_primitives::{NimbusId, NimbusPair},

    node_common::service::{ManualSealConfiguration, NodeBuilder, NodeBuilderConfig, Sealing},

    pallet_author_noting_runtime_api::AuthorNotingApi,

    pallet_data_preservers_runtime_api::DataPreserversApi,

    pallet_registrar_runtime_api::RegistrarApi,

    parity_scale_codec::Encode,

    polkadot_cli::ProvideRuntimeApi,

    polkadot_parachain_primitives::primitives::HeadData,

    polkadot_service::Handle,

    sc_cli::CliConfiguration,

    sc_client_api::{

        AuxStore, Backend as BackendT, BlockchainEvents, HeaderBackend, UsageProvider,

    },

    sc_consensus::BasicQueue,

    sc_network::NetworkBlock,

    sc_network_common::role::Role,

    sc_network_sync::SyncingService,

    sc_service::{Configuration, KeystoreContainer, SpawnTaskHandle, TFullBackend, TaskManager},

    sc_telemetry::TelemetryHandle,

    sc_transaction_pool::FullPool,

    sp_api::StorageProof,

    sp_consensus::SyncOracle,

    sp_consensus_slots::Slot,

    sp_core::{traits::SpawnEssentialNamed, H256},

    sp_keystore::KeystorePtr,

    sp_state_machine::{Backend as StateBackend, StorageValue},

    std::{pin::Pin, sync::Arc, time::Duration},

    tc_consensus::{

        collators::lookahead::{

            self as lookahead_tanssi_aura, BuyCoreParams, Params as LookaheadTanssiAuraParams,

        },

        OnDemandBlockProductionApi, OrchestratorAuraWorkerAuxData, TanssiAuthorityAssignmentApi,

    },

    tc_service_container_chain::{

        cli::ContainerChainCli,

        monitor,

        service::{

            DevParachainBlockImport, ParachainBlockImport, ParachainClient, ParachainExecutor,

            ParachainProposerFactory,

        },

        spawner::{self, CcSpawnMsg, ContainerChainSpawnParams, ContainerChainSpawner},

    },

    tokio::sync::mpsc::{unbounded_channel, UnboundedSender},

    tokio_util::sync::CancellationToken,

};

mod mocked_relay_keys;

type FullBackend = TFullBackend<Block>;

pub struct NodeConfig;

impl NodeBuilderConfig for NodeConfig {

    type Block = Block;

    type RuntimeApi = RuntimeApi;

    type ParachainExecutor = ParachainExecutor;

}

thread_local!(static TIMESTAMP: std::cell::RefCell<u64> = const { std::cell::RefCell::new(0) });

/// Provide a mock duration starting at 0 in millisecond for timestamp inherent.

/// Each call will increment timestamp by slot_duration making Aura think time has passed.

struct MockTimestampInherentDataProvider;

#[async_trait::async_trait]

impl sp_inherents::InherentDataProvider for MockTimestampInherentDataProvider {

    async fn provide_inherent_data(

        &self,

        inherent_data: &mut sp_inherents::InherentData,

    async fn try_handle_error(

        &self,

        _identifier: &sp_inherents::InherentIdentifier,

        _error: &[u8],

        // The pallet never reports error.

}

/// Background task used to detect changes to container chain assignment,

/// and start/stop container chains on demand. The check runs on every new block.

        // Subscribe to new blocks in order to react to para id assignment

        // This must be the stream of finalized blocks, otherwise the collators may rotate to a

        // different chain before the block is finalized, and that could lead to a stalled chain

        }

/// Check the parachain assignment using the orchestrator chain client, and send a `CcSpawnMsg` to

/// start or stop the required container chains.

///

/// Checks the assignment for the next block, so if there is a session change on block 15, this will

/// detect the assignment change after importing block 14.

    // Check current assignment

    // Check assignment in the next session

/// Start a node with the given parachain `Configuration` and relay chain `Configuration`.

///

/// This is the actual implementation that is abstract over the executor and the runtime api.

#[sc_tracing::logging::prefix_logs_with("Orchestrator")]

async fn start_node_impl(

    orchestrator_config: Configuration,

    polkadot_config: Configuration,

    container_chain_config: Option<(ContainerChainCli, tokio::runtime::Handle)>,

    collator_options: CollatorOptions,

    para_id: ParaId,

    hwbench: Option<sc_sysinfo::HwBench>,

) -> sc_service::error::Result<(TaskManager, Arc<ParachainClient>)> {

    let parachain_config = prepare_node_config(orchestrator_config);

    let chain_type: sc_chain_spec::ChainType = parachain_config.chain_spec.chain_type();

    let relay_chain = crate::chain_spec::Extensions::try_get(&*parachain_config.chain_spec)

        .ok_or("Could not find relay_chain extension in chain-spec.")?;

    // Channel to send messages to start/stop container chains

    let (cc_spawn_tx, cc_spawn_rx) = unbounded_channel();

    // Create a `NodeBuilder` which helps setup parachain nodes common systems.

    let mut node_builder = NodeConfig::new_builder(&parachain_config, hwbench.clone())?;

    let (block_import, import_queue) = import_queue(&parachain_config, &node_builder);

    let (relay_chain_interface, collator_key) = node_builder

        .build_relay_chain_interface(&parachain_config, polkadot_config, collator_options.clone())

        .await?;

    let validator = parachain_config.role.is_authority();

    let force_authoring = parachain_config.force_authoring;

    let node_builder = node_builder

        .build_cumulus_network::<_, sc_network::NetworkWorker<_, _>>(

            &parachain_config,

            para_id,

            import_queue,

            relay_chain_interface.clone(),

        )

        .await?;

    let rpc_builder = {

        let client = node_builder.client.clone();

        let transaction_pool = node_builder.transaction_pool.clone();

    };

    let node_builder = node_builder.spawn_common_tasks(parachain_config, rpc_builder)?;

    let relay_chain_slot_duration = Duration::from_secs(6);

    let overseer_handle = relay_chain_interface

        .overseer_handle()

    let sync_keystore = node_builder.keystore_container.keystore();

    let mut collate_on_tanssi: Arc<

        dyn Fn() -> (CancellationToken, futures::channel::oneshot::Receiver<()>) + Send + Sync,

        } else {

        }

    });

    let announce_block = {

        let sync_service = node_builder.network.sync_service.clone();

    };

    let (mut node_builder, import_queue_service) = node_builder.extract_import_queue_service();

    start_relay_chain_tasks(StartRelayChainTasksParams {

        client: node_builder.client.clone(),

        announce_block: announce_block.clone(),

        para_id,

        relay_chain_interface: relay_chain_interface.clone(),

        task_manager: &mut node_builder.task_manager,

        da_recovery_profile: if validator {

            DARecoveryProfile::Collator

        } else {

            DARecoveryProfile::FullNode

        },

        import_queue: import_queue_service,

        relay_chain_slot_duration,

        recovery_handle: Box::new(overseer_handle.clone()),

        sync_service: node_builder.network.sync_service.clone(),

    })?;

    let orchestrator_chain_interface_builder = OrchestratorChainInProcessInterfaceBuilder {

        client: node_builder.client.clone(),

        backend: node_builder.backend.clone(),

        sync_oracle: node_builder.network.sync_service.clone(),

        overseer_handle: overseer_handle.clone(),

    };

    let orchestrator_chain_interface = orchestrator_chain_interface_builder.build();

    if validator {

        let collator_key = collator_key

            .clone()

            .expect("Command line arguments do not allow this. qed");

        // Start task which detects para id assignment, and starts/stops container chains.

        // Note that if this node was started without a `container_chain_config`, we don't

        // support collation on container chains, so there is no need to detect changes to assignment

        if container_chain_config.is_some() {

            build_check_assigned_para_id(

                orchestrator_chain_interface.clone(),

                sync_keystore.clone(),

                cc_spawn_tx.clone(),

                node_builder.task_manager.spawn_essential_handle(),

            );

        }

        let start_collation = {

            // Params for collate_on_tanssi closure

            let node_spawn_handle = node_builder.task_manager.spawn_handle().clone();

            let node_keystore = node_builder.keystore_container.keystore().clone();

            let node_client = node_builder.client.clone();

            let node_backend = node_builder.backend.clone();

            let relay_interface = relay_chain_interface.clone();

            let node_sync_service = node_builder.network.sync_service.clone();

            let orchestrator_tx_pool = node_builder.transaction_pool.clone();

            let overseer = overseer_handle.clone();

            let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(

                node_spawn_handle.clone(),

                node_client.clone(),

                node_builder.transaction_pool.clone(),

                node_builder.prometheus_registry.as_ref(),

                node_telemetry_handle.clone(),

            );

        };

        // Save callback for later, used when collator rotates from container chain back to orchestrator chain

        collate_on_tanssi = Arc::new(start_collation);

    }

    node_builder.network.start_network.start_network();

    let sync_keystore = node_builder.keystore_container.keystore();

    if let Some((container_chain_cli, tokio_handle)) = container_chain_config {

        // If the orchestrator chain is running as a full-node, we start a full node for the

        // container chain immediately, because only collator nodes detect their container chain

        // assignment so otherwise it will never start.

        if !validator {

            if let Some(container_chain_para_id) = container_chain_cli.base.para_id {

                // Spawn new container chain node

                cc_spawn_tx

                    .send(CcSpawnMsg::UpdateAssignment {

                        current: Some(container_chain_para_id.into()),

                        next: Some(container_chain_para_id.into()),

                    })

            }

        }

        // Start container chain spawner task. This will start and stop container chains on demand.

        let orchestrator_client = node_builder.client.clone();

        let orchestrator_tx_pool = node_builder.transaction_pool.clone();

        let spawn_handle = node_builder.task_manager.spawn_handle();

        // This considers that the container chains have the same APIs as dancebox, which

        // is not the case. However the spawner don't call APIs that are not part of the expected

        // common APIs for a container chain.

        // TODO: Depend on the simple container chain runtime which should be the minimal api?

        let container_chain_spawner = ContainerChainSpawner {

            params: ContainerChainSpawnParams {

                orchestrator_chain_interface,

                container_chain_cli,

                tokio_handle,

                chain_type,

                relay_chain,

                relay_chain_interface,

                sync_keystore,

                orchestrator_para_id: para_id,

                data_preserver: false,

                collation_params: if validator {

                    Some(spawner::CollationParams {

                        orchestrator_client: Some(orchestrator_client.clone()),

                        orchestrator_tx_pool: Some(orchestrator_tx_pool),

                        orchestrator_para_id: para_id,

                        collator_key: collator_key

                            .expect("there should be a collator key if we're a validator"),

                        solochain: false,

                    })

                } else {

                    None

                },

                spawn_handle,

                generate_rpc_builder: tc_service_container_chain::rpc::GenerateSubstrateRpcBuilder::<

                    dancebox_runtime::RuntimeApi,

                >::new(),

                phantom: PhantomData,

            },

            state: Default::default(),

            db_folder_cleanup_done: false,

            collate_on_tanssi,

            collation_cancellation_constructs: None,

        };

        let state = container_chain_spawner.state.clone();

        node_builder.task_manager.spawn_essential_handle().spawn(

            "container-chain-spawner-rx-loop",

            None,

            container_chain_spawner.rx_loop(cc_spawn_rx, validator, false),

        );

        node_builder.task_manager.spawn_essential_handle().spawn(

            "container-chain-spawner-debug-state",

            None,

            monitor::monitor_task(state),

        )

    }

    Ok((node_builder.task_manager, node_builder.client))

}

/// Build the import queue for the parachain runtime (manual seal).

/// Start collator task for orchestrator chain.

/// Returns a `CancellationToken` that can be used to cancel the collator task,

/// and a `oneshot::Receiver<()>` that can be used to wait until the task has ended.

                // Fetch duration every block to avoid downtime when passing from 12 to 6s

                    authorities,

                    block_hash

                );

/// Start a parachain node.

/// Start a solochain node.

    // Instead of putting keystore in

    // Collator1000-01/data/chains/simple_container_2000/keystore

    // We put it in

    // Collator1000-01/data/config/keystore

    // And same for "network" folder

    // But zombienet will put the keys in the old path, so we need to manually copy it if we

    // are running under zombienet

    // No metrics so no prometheus registry

    // Each container chain will spawn its own telemetry

    } else {

    };

    // If the orchestrator chain is running as a full-node, we start a full node for the

    // container chain immediately, because only collator nodes detect their container chain

    // assignment so otherwise it will never start.

            // Spawn new container chain node

    // Start container chain spawner task. This will start and stop container chains on demand.

        params: ContainerChainSpawnParams {

            } else {

            },

pub const SOFT_DEADLINE_PERCENT: sp_runtime::Percent = sp_runtime::Percent::from_percent(100);

/// Start a node with the given parachain `Configuration` and relay chain `Configuration`.

///

/// This is the actual implementation that is abstract over the executor and the runtime api.

pub fn start_dev_node(

    orchestrator_config: Configuration,

    sealing: Sealing,

    hwbench: Option<sc_sysinfo::HwBench>,

    para_id: ParaId,

) -> sc_service::error::Result<TaskManager> {

    let parachain_config = prepare_node_config(orchestrator_config);

    // Create a `NodeBuilder` which helps setup parachain nodes common systems.

    let node_builder = NodeConfig::new_builder(&parachain_config, hwbench)?;

    // This node block import.

    let block_import = DevParachainBlockImport::new(node_builder.client.clone());

    let import_queue = build_manual_seal_import_queue(

        node_builder.client.clone(),

        block_import.clone(),

        &parachain_config,

        node_builder

            .telemetry

            .as_ref()

        &node_builder.task_manager,

    )?;

    // Build a Substrate Network. (not cumulus since it is a dev node, it mocks

    // the relaychain)

    let mut node_builder = node_builder

        .build_substrate_network::<sc_network::NetworkWorker<_, _>>(

            &parachain_config,

            import_queue,

        )?;

    // If we're running a collator dev node we must install manual seal block

    // production.

    let mut command_sink = None;

    let mut xcm_senders = None;

    if parachain_config.role.is_authority() {

        let client = node_builder.client.clone();

        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);

        xcm_senders = Some((downward_xcm_sender, hrmp_xcm_sender));

        command_sink = node_builder.install_manual_seal(ManualSealConfiguration {

            block_import,

            sealing,

            soft_deadline: Some(SOFT_DEADLINE_PERCENT),

            select_chain: sc_consensus::LongestChain::new(node_builder.backend.clone()),

            consensus_data_provider: Some(Box::new(

                tc_consensus::OrchestratorManualSealAuraConsensusDataProvider::new(

                    node_builder.client.clone(),

                    node_builder.keystore_container.keystore(),

                    para_id,

                ),

            )),

        })?;

    }

    // This node RPC builder.

    let rpc_builder = {

        let client = node_builder.client.clone();

        let transaction_pool = node_builder.transaction_pool.clone();

    };

    // We spawn all the common substrate tasks to properly run a node.

    let node_builder = node_builder.spawn_common_tasks(parachain_config, rpc_builder)?;

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

    // We start the networking part.

    node_builder.network.start_network.start_network();

    Ok(node_builder.task_manager)

}

/// Can be called for a `Configuration` to check if it is a configuration for

/// the orchestrator network.

pub trait IdentifyVariant {

    /// Returns `true` if this is a configuration for a dev network.

    fn is_dev(&self) -> bool;

}

impl IdentifyVariant for Box<dyn sc_service::ChainSpec> {

}

/// Builder for a concrete relay chain interface, created from a full node. Builds

/// a [`RelayChainInProcessInterface`] to access relay chain data necessary for parachain operation.

///

/// The builder takes a [`polkadot_client::Client`]

/// that wraps a concrete instance. By using [`polkadot_client::ExecuteWithClient`]

/// the builder gets access to this concrete instance and instantiates a [`RelayChainInProcessInterface`] with it.

struct OrchestratorChainInProcessInterfaceBuilder {

    client: Arc<ParachainClient>,

    backend: Arc<FullBackend>,

    sync_oracle: Arc<dyn SyncOracle + Send + Sync>,

    overseer_handle: Handle,

}

impl OrchestratorChainInProcessInterfaceBuilder {

}

/// Builder for a concrete relay chain interface, created from a full node. Builds

/// a [`RelayChainInProcessInterface`] to access relay chain data necessary for parachain operation.

///

/// The builder takes a [`polkadot_client::Client`]

/// that wraps a concrete instance. By using [`polkadot_client::ExecuteWithClient`]

/// the builder gets access to this concrete instance and instantiates a [`RelayChainInProcessInterface`] with it.

struct OrchestratorChainSolochainInterfaceBuilder {

    overseer_handle: Handle,

    relay_chain_interface: Arc<dyn RelayChainInterface>,

}

impl OrchestratorChainSolochainInterfaceBuilder {

}

/// Provides an implementation of the [`RelayChainInterface`] using a local in-process relay chain node.

pub struct OrchestratorChainInProcessInterface<Client> {

    pub full_client: Arc<Client>,

    pub backend: Arc<FullBackend>,

    pub sync_oracle: Arc<dyn SyncOracle + Send + Sync>,

    pub overseer_handle: Handle,

}

impl<Client> OrchestratorChainInProcessInterface<Client> {

    /// Create a new instance of [`RelayChainInProcessInterface`]

}

impl<T> Clone for OrchestratorChainInProcessInterface<T> {

}

#[async_trait::async_trait]

impl<Client> OrchestratorChainInterface for OrchestratorChainInProcessInterface<Client>

where

    Client: ProvideRuntimeApi<Block>

        + BlockchainEvents<Block>

        + AuxStore

        + UsageProvider<Block>

        + Sync

        + Send,

    Client::Api: TanssiAuthorityAssignmentApi<Block, NimbusId>

        + OnDemandBlockProductionApi<Block, ParaId, Slot>

        + RegistrarApi<Block, ParaId>

        + AuthorNotingApi<Block, AccountId, BlockNumber, ParaId>

        + DataPreserversApi<Block, DataPreserverProfileId, ParaId>,

{

    async fn get_storage_by_key(

        &self,

        orchestrator_parent: PHash,

        key: &[u8],

    async fn prove_read(

        &self,

        orchestrator_parent: PHash,

        relevant_keys: &Vec<Vec<u8>>,

    /// Get a stream of import block notifications.

    async fn import_notification_stream(

        &self,

    /// Get a stream of new best block notifications.

    async fn new_best_notification_stream(

        &self,

    /// Get a stream of finality notifications.

    async fn finality_notification_stream(

        &self,

    async fn genesis_data(

        &self,

        orchestrator_parent: PHash,

        para_id: ParaId,

    async fn boot_nodes(

        &self,

        orchestrator_parent: PHash,

        para_id: ParaId,

    async fn latest_block_number(

        &self,

        orchestrator_parent: PHash,

        para_id: ParaId,

    async fn data_preserver_active_assignment(

        &self,

        orchestrator_parent: PHash,

        profile_id: DataPreserverProfileId,

        use {

            dc_orchestrator_chain_interface::DataPreserverAssignment as InterfaceAssignment,

            pallet_data_preservers_runtime_api::Assignment as RuntimeAssignment,

        };

        Ok(

            },

        )

    async fn check_para_id_assignment(

        &self,

        orchestrator_parent: PHash,

        authority: NimbusId,

    async fn check_para_id_assignment_next_session(

        &self,

        orchestrator_parent: PHash,

        authority: NimbusId,

}

/// Provides an implementation of the [`RelayChainInterface`] using a local in-process relay chain node.

pub struct OrchestratorChainSolochainInterface {

    pub overseer_handle: Handle,

    pub relay_chain_interface: Arc<dyn RelayChainInterface>,

}

impl OrchestratorChainSolochainInterface {

    /// Create a new instance of [`RelayChainInProcessInterface`]

}

#[async_trait::async_trait]

impl OrchestratorChainInterface for OrchestratorChainSolochainInterface {

    async fn get_storage_by_key(

        &self,

        relay_parent: PHash,

        key: &[u8],

    async fn prove_read(

        &self,

        relay_parent: PHash,

        relevant_keys: &Vec<Vec<u8>>,

    /// Get a stream of import block notifications.

    async fn import_notification_stream(

        &self,