// Copyright 2019-2025 PureStake Inc.

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

//! `trace_filter` RPC handler and its associated service task.

//! The RPC handler rely on `CacheTask` which provides a future that must be run inside a tokio

//! executor.

//!

//! The implementation is composed of multiple tasks :

//! - Many calls the RPC handler `Trace::filter`, communicating with the main task.

//! - A main `CacheTask` managing the cache and the communication between tasks.

//! - For each traced block an async task responsible to wait for a permit, spawn a blocking

//!   task and waiting for the result, then send it to the main `CacheTask`.

use futures::{select, stream::FuturesUnordered, FutureExt, StreamExt};

use std::{collections::BTreeMap, future::Future, marker::PhantomData, sync::Arc, time::Duration};

use tokio::{

	sync::{mpsc, oneshot, Semaphore},

	time::sleep,

};

use tracing::{instrument, Instrument};

use sc_client_api::backend::{Backend, StateBackend, StorageProvider};

use sc_utils::mpsc::TracingUnboundedSender;

use sp_api::{ApiExt, Core, ProvideRuntimeApi};

use sp_block_builder::BlockBuilder;

use sp_blockchain::{

	Backend as BlockchainBackend, Error as BlockChainError, HeaderBackend, HeaderMetadata,

};

use sp_runtime::traits::{BlakeTwo256, Block as BlockT, Header as HeaderT};

use substrate_prometheus_endpoint::{

	register, Counter, PrometheusError, Registry as PrometheusRegistry, U64,

};

use ethereum_types::H256;

use fc_storage::StorageOverride;

use fp_rpc::EthereumRuntimeRPCApi;

use moonbeam_client_evm_tracing::{

	formatters::ResponseFormatter,

	types::block::{self, TransactionTrace},

};

pub use moonbeam_rpc_core_trace::{FilterRequest, TraceServer};

use moonbeam_rpc_core_types::{RequestBlockId, RequestBlockTag};

use moonbeam_rpc_primitives_debug::DebugRuntimeApi;

type TxsTraceRes = Result<Vec<TransactionTrace>, String>;

/// RPC handler. Will communicate with a `CacheTask` through a `CacheRequester`.

pub struct Trace<B, C> {

	_phantom: PhantomData<B>,

	client: Arc<C>,

	requester: CacheRequester,

	max_count: u32,

}

impl<B, C> Clone for Trace<B, C> {

}

impl<B, C> Trace<B, C>

where

	B: BlockT<Hash = H256> + Send + Sync + 'static,

	B::Header: HeaderT<Number = u32>,

	C: HeaderMetadata<B, Error = BlockChainError> + HeaderBackend<B>,

	C: Send + Sync + 'static,

{

	/// Create a new RPC handler.

	/// Convert an optional block ID (number or tag) to a block height.

			None | Some(RequestBlockId::Tag(RequestBlockTag::Latest)) => {

			}

			Some(RequestBlockId::Tag(RequestBlockTag::Finalized)) => {

			}

			Some(RequestBlockId::Tag(RequestBlockTag::Pending)) => {

			}

		}

	/// `trace_filter` endpoint (wrapped in the trait implementation with futures compatibility)

		}

		// Start a batch with these blocks.

		// Fetch all the traces. It is done in another function to simplify error handling and allow

		// to call the following `stop_batch` regardless of the result. This is important for the

		// cache cleanup to work properly.

		// Stop the batch, allowing the cache task to remove useless non-started block traces and

		// start the expiration delay.

			// Request the traces of this block to the cache service.

			// This will resolve quickly if the block is already cached, or wait until the block

			// has finished tracing.

			// Filter addresses.

					}

					}

					}

		}

}

#[jsonrpsee::core::async_trait]

impl<B, C> TraceServer for Trace<B, C>

where

	B: BlockT<Hash = H256> + Send + Sync + 'static,

	B::Header: HeaderT<Number = u32>,

	C: HeaderMetadata<B, Error = BlockChainError> + HeaderBackend<B>,

	C: Send + Sync + 'static,

{

	async fn filter(

		&self,

		filter: FilterRequest,

}

/// An opaque batch ID.

#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]

pub struct CacheBatchId(u64);

/// Requests the cache task can accept.

enum CacheRequest {

	/// Request to start caching the provided range of blocks.

	/// The task will add to blocks to its pool and immediately return a new batch ID.

	StartBatch {

		/// Returns the ID of the batch for cancellation.

		sender: oneshot::Sender<CacheBatchId>,

		/// List of block hash to trace.

		blocks: Vec<H256>,

	},

	/// Fetch the traces for given block hash.

	/// The task will answer only when it has processed this block.

	GetTraces {

		/// Returns the array of traces or an error.

		sender: oneshot::Sender<TxsTraceRes>,

		/// Hash of the block.

		block: H256,

	},

	/// Notify the cache that it can stop the batch with that ID. Any block contained only in

	/// this batch and still not started will be discarded.

	StopBatch { batch_id: CacheBatchId },

}

/// Allows to interact with the cache task.

#[derive(Clone)]

pub struct CacheRequester(TracingUnboundedSender<CacheRequest>);

impl CacheRequester {

	/// Request to start caching the provided range of blocks.

	/// The task will add to blocks to its pool and immediately return the batch ID.

	#[instrument(skip(self))]

	pub async fn start_batch(&self, blocks: Vec<H256>) -> Result<CacheBatchId, String> {

		let (response_tx, response_rx) = oneshot::channel();

		let sender = self.0.clone();

		sender

			.unbounded_send(CacheRequest::StartBatch {

				sender: response_tx,

				blocks,

			})

	}

	/// Fetch the traces for given block hash.

	/// The task will answer only when it has processed this block.

	/// The block should be part of a batch first. If no batch has requested the block it will

	/// return an error.

	#[instrument(skip(self))]

	pub async fn get_traces(&self, block: H256) -> TxsTraceRes {

		let (response_tx, response_rx) = oneshot::channel();

		let sender = self.0.clone();

		sender

			.unbounded_send(CacheRequest::GetTraces {

				sender: response_tx,

				block,

			})

		response_rx

			.await

	}

	/// Notify the cache that it can stop the batch with that ID. Any block contained only in

	/// this batch and still in the waiting pool will be discarded.

	#[instrument(skip(self))]

	pub async fn stop_batch(&self, batch_id: CacheBatchId) {

		let sender = self.0.clone();

		// Here we don't care if the request has been accepted or refused, the caller can't

		// do anything with it.

		let _ = sender

			.unbounded_send(CacheRequest::StopBatch { batch_id })

	}

}

/// Data stored for each block in the cache.

/// `active_batch_count` represents the number of batches using this

/// block. It will increase immediately when a batch is created, but will be

/// decrease only after the batch ends and its expiration delay passes.

/// It allows to keep the data in the cache for following requests that would use

/// this block, which is important to handle pagination efficiently.

struct CacheBlock {

	active_batch_count: usize,

	state: CacheBlockState,

}

/// State of a cached block. It can either be polled to be traced or cached.

enum CacheBlockState {

	/// Block has been added to the pool blocks to be replayed.

	/// It may be currently waiting to be replayed or being replayed.

	Pooled {

		started: bool,

		/// Multiple requests might query the same block while it is pooled to be

		/// traced. They response channel is stored here, and the result will be

		/// sent in all of them when the tracing is finished.

		waiting_requests: Vec<oneshot::Sender<TxsTraceRes>>,

		/// Channel used to unqueue a tracing that has not yet started.

		/// A tracing will be unqueued if it has not yet been started and the last batch

		/// needing this block is ended (ignoring the expiration delay).

		/// It is not used directly, but dropping will wake up the receiver.

		#[allow(dead_code)]

		unqueue_sender: oneshot::Sender<()>,

	},

	/// Tracing has been completed and the result is available. No Runtime API call

	/// will be needed until this block cache is removed.

	Cached { traces: TxsTraceRes },

}

/// Tracing a block is done in a separate tokio blocking task to avoid clogging the async threads.

/// For this reason a channel using this type is used by the blocking task to communicate with the

/// main cache task.

enum BlockingTaskMessage {

	/// Notify the tracing for this block has started as the blocking task got a permit from

	/// the semaphore. This is used to prevent the deletion of a cache entry for a block that has

	/// started being traced.

	Started { block_hash: H256 },

	/// The tracing is finished and the result is sent to the main task.

	Finished {

		block_hash: H256,

		result: TxsTraceRes,

	},

}

/// Type wrapper for the cache task, generic over the Client, Block and Backend types.

pub struct CacheTask<B, C, BE> {

	client: Arc<C>,

	backend: Arc<BE>,

	blocking_permits: Arc<Semaphore>,

	cached_blocks: BTreeMap<H256, CacheBlock>,

	batches: BTreeMap<u64, Vec<H256>>,

	next_batch_id: u64,

	metrics: Option<Metrics>,

	_phantom: PhantomData<B>,

}

impl<B, C, BE> CacheTask<B, C, BE>

where

	BE: Backend<B> + 'static,

	BE::State: StateBackend<BlakeTwo256>,

	C: ProvideRuntimeApi<B>,

	C: StorageProvider<B, BE>,

	C: HeaderMetadata<B, Error = BlockChainError> + HeaderBackend<B>,

	C: Send + Sync + 'static,

	B: BlockT<Hash = H256> + Send + Sync + 'static,

	B::Header: HeaderT<Number = u32>,

	C::Api: BlockBuilder<B>,

	C::Api: DebugRuntimeApi<B>,

	C::Api: EthereumRuntimeRPCApi<B>,

	C::Api: ApiExt<B>,

{

	/// Create a new cache task.

	///

	/// Returns a Future that needs to be added to a tokio executor, and a handle allowing to

	/// send requests to the task.

		// Task running in the service.

					}

				}

			} else {

			};

			// Contains the inner state of the cache task, excluding the pooled futures/channels.

			// Having this object allows to refactor each event into its own function, simplifying

			// the main loop.

			// Main event loop. This loop must not contain any direct .await, as we want to

			// react to events as fast as possible.

			loop {

						}

					},

						}

					},

						}

					}

				}

			}

	/// Handle the creation of a batch.

	/// Will start the tracing process for blocks that are not already in the cache.

	#[instrument(skip(self, blocking_tx, sender, blocks, overrides))]

	fn request_start_batch(

		&mut self,

		blocking_tx: &mpsc::Sender<BlockingTaskMessage>,

		sender: oneshot::Sender<CacheBatchId>,

		blocks: Vec<H256>,

		overrides: Arc<dyn StorageOverride<B>>,

	) {

		tracing::trace!("Starting batch {}", self.next_batch_id);

		self.batches.insert(self.next_batch_id, blocks.clone());

		for block in blocks {

			// The block is already in the cache, awesome!

			if let Some(block_cache) = self.cached_blocks.get_mut(&block) {

				block_cache.active_batch_count += 1;

				tracing::trace!(

					"Cache hit for block {}, now used by {} batches.",

					block,

					block_cache.active_batch_count

				);

			}

			// Otherwise we need to queue this block for tracing.

			else {

				tracing::trace!("Cache miss for block {}, pooling it for tracing.", block);

				let blocking_permits = Arc::clone(&self.blocking_permits);

				let (unqueue_sender, unqueue_receiver) = oneshot::channel();

				let client = Arc::clone(&self.client);

				let backend = Arc::clone(&self.backend);

				let blocking_tx = blocking_tx.clone();

				let overrides = overrides.clone();

				// Spawn all block caching asynchronously.

				// It will wait to obtain a permit, then spawn a blocking task.

				// When the blocking task returns its result, it is sent

				// thought a channel to the main task loop.

				tokio::spawn(

							},

						// Perform block tracing in a tokio blocking task.

						// Send a response to the main task.

					.instrument(tracing::trace_span!("Block tracing", block = %block)),

				);

				// Insert the block in the cache.

				self.cached_blocks.insert(

					block,

					CacheBlock {

						active_batch_count: 1,

						state: CacheBlockState::Pooled {

							started: false,

							waiting_requests: vec![],

							unqueue_sender,

						},

					},

				);

			}

		}

		// Respond with the batch ID.

		let _ = sender.send(CacheBatchId(self.next_batch_id));

		// Increase batch ID for the next request.

		self.next_batch_id = self.next_batch_id.overflowing_add(1).0;

	}

	/// Handle a request to get the traces of the provided block.

	/// - If the result is stored in the cache, it sends it immediately.

	/// - If the block is currently being pooled, it is added to this block cache waiting list,

	///   and all requests concerning this block will be satisfied when the tracing for this block

	///   is finished.

	/// - If this block is missing from the cache, it means no batch asked for it. All requested

	///   blocks should be contained in a batch beforehand, and thus an error is returned.

	#[instrument(skip(self))]

	fn request_get_traces(&mut self, sender: oneshot::Sender<TxsTraceRes>, block: H256) {

		if let Some(block_cache) = self.cached_blocks.get_mut(&block) {

			match &mut block_cache.state {

				CacheBlockState::Pooled {

					ref mut waiting_requests,

					..

				} => {

					tracing::warn!(

						"A request asked a pooled block ({}), adding it to the list of \

						waiting requests.",

						block

					);

					waiting_requests.push(sender);

					if let Some(metrics) = &self.metrics {

						metrics.tracing_cache_misses.inc();

					}

				}

				CacheBlockState::Cached { traces, .. } => {

					tracing::warn!(

						"A request asked a cached block ({}), sending the traces directly.",

						block

					);

					let _ = sender.send(traces.clone());

					if let Some(metrics) = &self.metrics {

						metrics.tracing_cache_hits.inc();

					}

				}

			}

		} else {

			tracing::warn!(

				"An RPC request asked to get a block ({}) which was not batched.",

				block

			);

			let _ = sender.send(Err(format!(

				"RPC request asked a block ({}) that was not batched",

				block

			)));

		}

	}

	/// Handle a request to stop a batch.

	/// For all blocks that needed to be traced, are only in this batch and not yet started, their

	/// tracing is cancelled to save CPU-time and avoid attacks requesting large amount of blocks.

	/// This batch data is not yet removed however. Instead a expiration delay timer is started

	/// after which the data will indeed be cleared. (the code for that is in the main loop code

	/// as it involved an unnamable type :C)

	#[instrument(skip(self))]

	fn request_stop_batch(&mut self, batch_id: CacheBatchId) {

		tracing::trace!("Stopping batch {}", batch_id.0);

		if let Some(blocks) = self.batches.get(&batch_id.0) {

			for block in blocks {

				let mut remove = false;

				// We remove early the block cache if this batch is the last

				// pooling this block.

				if let Some(block_cache) = self.cached_blocks.get_mut(block) {

					if block_cache.active_batch_count == 1

						&& matches!(

							block_cache.state,

							CacheBlockState::Pooled { started: false, .. }

						) {

						remove = true;

					}

				}

				if remove {

					tracing::trace!("Pooled block {} is no longer requested.", block);

					// Remove block from the cache. Drops the value,

					// closing all the channels contained in it.

					let _ = self.cached_blocks.remove(block);

				}

			}

		}

	}

	/// A tracing blocking task notifies it got a permit and is starting the tracing.

	/// This started status is stored to avoid removing this block entry.

	#[instrument(skip(self))]

	fn blocking_started(&mut self, block_hash: H256) {

		if let Some(block_cache) = self.cached_blocks.get_mut(&block_hash) {

			if let CacheBlockState::Pooled {

				ref mut started, ..

			} = block_cache.state

			{

				*started = true;

			}

		}

	}

	/// A tracing blocking task notifies it has finished the tracing and provide the result.

	#[instrument(skip(self, result))]

	fn blocking_finished(&mut self, block_hash: H256, result: TxsTraceRes) {

		// In some cases it might be possible to receive traces of a block

		// that has no entry in the cache because it was removed of the pool

		// and received a permit concurrently. We just ignore it.

		//

		// TODO : Should we add it back ? Should it have an active_batch_count

		// of 1 then ?

		if let Some(block_cache) = self.cached_blocks.get_mut(&block_hash) {

			if let CacheBlockState::Pooled {

				ref mut waiting_requests,

				..

			} = block_cache.state

			{

				tracing::trace!(

					"A new block ({}) has been traced, adding it to the cache and responding to \

					{} waiting requests.",

					block_hash,

					waiting_requests.len()

				);

				// Send result in waiting channels

				while let Some(channel) = waiting_requests.pop() {

					let _ = channel.send(result.clone());

				}

				// Update cache entry

				block_cache.state = CacheBlockState::Cached { traces: result };

			}

		}

	}

	/// A batch expiration delay timer has completed. It performs the cache cleaning for blocks

	/// not longer used by other batches.

	#[instrument(skip(self))]

	fn expired_batch(&mut self, batch_id: CacheBatchId) {

		if let Some(batch) = self.batches.remove(&batch_id.0) {

			for block in batch {

				// For each block of the batch, we remove it if it was the

				// last batch containing it.

				let mut remove = false;

				if let Some(block_cache) = self.cached_blocks.get_mut(&block) {

					block_cache.active_batch_count -= 1;

					if block_cache.active_batch_count == 0 {

						remove = true;

					}

				}

				if remove {

					let _ = self.cached_blocks.remove(&block);

				}

			}

		}

	}

	/// (In blocking task) Use the Runtime API to trace the block.

	#[instrument(skip(client, backend, overrides))]

	fn cache_block(

		client: Arc<C>,

		backend: Arc<BE>,

		substrate_hash: H256,

		overrides: Arc<dyn StorageOverride<B>>,

	) -> TxsTraceRes {

		// Get Substrate block data.

		let api = client.runtime_api();

		let block_header = client

			.header(substrate_hash)

		let height = *block_header.number();

		let substrate_parent_hash = *block_header.parent_hash();

		// Get Ethereum block data.

		let (eth_block, eth_transactions) = match (

			overrides.current_block(substrate_hash),

			overrides.current_transaction_statuses(substrate_hash),

		) {

			(Some(a), Some(b)) => (a, b),

			_ => {

				return Err(format!(

					"Failed to get Ethereum block data for Substrate block {}",

					substrate_hash

				))

			}

		};

		let eth_block_hash = eth_block.header.hash();

		let eth_tx_hashes = eth_transactions

			.iter()

			.collect();

		// Get extrinsics (containing Ethereum ones)

		let extrinsics = backend

			.blockchain()

			.body(substrate_hash)

		// Get DebugRuntimeApi version

		let trace_api_version = if let Ok(Some(api_version)) =

			api.api_version::<dyn DebugRuntimeApi<B>>(substrate_parent_hash)

		{

			api_version

		} else {

			return Err("Runtime api version call failed (trace)".to_string());

		};

		// Trace the block.

			} else {

				// Get core runtime api version

				{

				} else {

				};

				// Initialize block: calls the "on_initialize" hook on every pallet

				// in AllPalletsWithSystem

				// This was fine before pallet-message-queue because the XCM messages

				// were processed by the "setValidationData" inherent call and not on an

				// "on_initialize" hook, which runs before enabling XCM tracing

				} else {

					#[allow(deprecated)]

				}

				#[allow(deprecated)]

			};

						target: "tracing",

						height,

						e

					);

		let eth_transactions_by_index: BTreeMap<u32, H256> = eth_transactions

			.iter()

			.collect();

		let mut proxy = moonbeam_client_evm_tracing::listeners::CallList::default();

		proxy.using(f)?;

		let traces: Vec<TransactionTrace> =

			moonbeam_client_evm_tracing::formatters::TraceFilter::format(proxy)

				.ok_or("Fail to format proxy")?

				.into_iter()

							// Reformat error messages.

							{

						}

						None => {

								height,

								trace,

							);

						}

					}

				.collect();

		Ok(traces)

	}

}

/// Prometheus metrics for tracing.

#[derive(Clone)]

pub(crate) struct Metrics {

	tracing_cache_hits: Counter<U64>,

	tracing_cache_misses: Counter<U64>,

}

impl Metrics {

			tracing_cache_misses: register(

		})

}