use crate::BuilderTask;
use bytes::Bytes;
use codec::Encode;
use corevm_host::{ServiceMessage, StorageKey};
use jam_std_common::{Node, NodeError, NodeExt as _};
use jam_types::HeaderHash;
use std::collections::VecDeque;

/// This is artificial limit to cover niche cases when the guest _either_ systematically fails to
/// refine all inter-service messages that it receives _or_ accumulation fails somehow.
const MAX_NOT_YET_ACCUMULATED_MESSAGES: usize = 64 * 1024;

/// Inter-service messages state.
///
/// Maintains builder-side refined messages queue and synchronizes it with the queue in the service
/// storage after each guest invocation.
pub struct ServiceMessages {
	/// Inter-service messages that were refined in one of the previous guest invocations but not
	/// yet accumulated.
	// TODO @ivan We should probably save this queue on-disk and read it from there when the
	// builder is restarted.
	refined: VecDeque<ServiceMessage>,
	/// Last known copy of the inter-service message queue (usually from the previous slot).
	///
	/// Initially it is fetched from the service storage, and then it's updated after each guest
	/// invocation.
	queue: VecDeque<ServiceMessage>,
}

impl ServiceMessages {
	pub fn new() -> Self {
		Self { refined: VecDeque::new(), queue: VecDeque::new() }
	}

	/// Appends refine messsages.
	///
	/// This should be done after each guest invocation.
	pub fn push_refined(
		&mut self,
		refined: impl IntoIterator<Item = ServiceMessage>,
	) -> Result<(), NodeError> {
		self.refined.extend(refined);
		if self.refined.len() > MAX_NOT_YET_ACCUMULATED_MESSAGES {
			return Err(NodeError::Other(format!(
                "Max. number of not yet accumulated inter-service messages reached ({MAX_NOT_YET_ACCUMULATED_MESSAGES})"
            )));
		}
		Ok(())
	}

	/// Synchronizes builder-side and service-side message queues.
	///
	/// `new_queue` is the new message queue from the service storage.
	///
	/// After the method returns `new_queue` will only contain not yet refined messages. This is
	/// builder's perspective, from JAM node's perspective some of the refined messages haven't been
	/// accumulated yet. We, however, can't supply such messages to the next guest invocation.
	fn sync(&mut self, new_queue: &mut VecDeque<ServiceMessage>) -> Result<(), NodeError> {
		while let Some(accumulated_message) =
			self.queue.pop_front_if(|message| Some(message) != new_queue.front_mut())
		{
			// Remove accumulated message from the queue of refined messages.
			self.refined
				.pop_front_if(|message| message == &accumulated_message)
				.ok_or_else(|| {
					NodeError::Other(format!(
						"Bad accumulated inter-service message: builder has {:?}, service storage contains {:?}",
						self.refined.front(),
						accumulated_message
					))
				})?;
		}
		// Remove messages that were refined but not yet accumulated.
		for refined_message in self.refined.iter() {
			new_queue.pop_front_if(|message| message == refined_message).ok_or_else(|| {
				NodeError::Other(format!(
                    "Bad refined inter-service message: builder has {:?}, service storage contains {:?}",
                    refined_message,
                    new_queue.front(),
                ))
			})?;
		}
		Ok(())
	}
}

impl BuilderTask {
	/// Returns not yet refined inter-service messages.
	///
	/// Synchronizes refined messages with the queue in the service storage.
	pub async fn unrefined_inter_service_messages(
		&mut self,
		node: &dyn Node,
		head: HeaderHash,
	) -> Result<VecDeque<(ServiceMessage, Bytes)>, NodeError> {
		let mut queue: VecDeque<ServiceMessage> = node
			.typed_service_value(head, self.service_id, &StorageKey::IncomingServiceMessages)
			.await?
			.unwrap_or_default();
		self.service_messages.sync(&mut queue)?;
		// Retrieve the contents of the messages.
		let mut imported_messages = VecDeque::with_capacity(queue.len());
		for message in queue.into_iter() {
			let key = StorageKey::IncomingMessage(message).encode();
			let contents = match node.service_value(head, self.service_id, &key).await? {
				Some(contents) => contents,
				None => {
					log::debug!(
						"Failed to retrieve message {} from the storage of {:x}",
						message.index,
						message.source
					);
					continue;
				},
			};
			imported_messages.push_back((message, contents));
		}
		Ok(imported_messages)
	}
}
