jamt 0.1.28

use anyhow::anyhow;
use clap::{Parser, ValueEnum};
use codec::{Compact, Decode, Encode};
use corevm_host::{fs as corevm_fs, AudioMode, CoreVmInstruction, VideoMode};
use corevm_tooling::{parsing::CoreVmFsNode, ArcNodeExt, CoreVmCodeInfo};
use futures::StreamExt;
use jam_bootstrap_service_common::Instruction;
use jam_program_blob_common::ConventionalMetadata as Metadata;
use jam_std_common::{hash_raw, BlockDesc, ChainSubUpdate, Node, VersionedParameters};
use jam_tooling::{
	format,
	parsing::{self, Blob, DataId, DataIdLen},
	CodeInfo, CommonArgs, NodeExt,
};
use jam_types::{
	core_count, min_turnaround_period, slot_period_sec, AnyBytes, AnyHash, Balance, CoreIndex,
	Hash, HeaderHash, Memo, ServiceId, SignedGas, Slot, ToAny, UnsignedGas, WorkPackageHash,
};
use jsonrpsee::ws_client::WsClient;
use std::{
	collections::btree_map::BTreeMap,
	ffi::CString,
	path::{PathBuf, MAIN_SEPARATOR},
	sync::Arc,
	time::Duration,
};
use tokio::task::spawn_blocking;

mod formatting;

mod cached_desc;
use self::cached_desc::*;

mod bootstrap;
use self::{bootstrap::*, package::PackageBuilder};

mod package;
mod queue;
mod service;
mod vm;

#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, clap::ValueEnum)]
pub enum BlockId {
	/// The head of the chain.
	Best,
	/// The most recently finalized block in the chain.
	Final,
}

#[derive(clap::Parser, Debug, Clone)]
pub enum Inspections {
	/// The slot of the block.
	Slot,
	/// A service.
	Service {
		/// The ID of the service.
		#[arg(value_parser = parsing::service_id)]
		id: ServiceId,
	},
	/// An item in a service's storage.
	Storage {
		/// The ID of the service.
		#[arg(value_parser = parsing::service_id)]
		id: ServiceId,
		/// The key into the service's storage.
		#[arg(value_parser = parsing::blob)]
		key: Blob,
		/// Print storage values as raw hex bytes instead of using pretty-printing.
		#[arg(long, default_value_t = false)]
		raw: bool,
	},
	/// A preimage stored in a service's storage.
	Preimage {
		/// The ID of the service.
		#[arg(value_parser = parsing::service_id)]
		id: ServiceId,
		/// The hash of the target preimage.
		#[arg(value_parser = parsing::data_id)]
		key: DataId,
	},
	/// A preimage request in a service's storage.
	Request {
		/// The ID of the service.
		#[arg(value_parser = parsing::service_id)]
		id: ServiceId,
		/// The hash and length of the target preimage.
		#[arg(value_parser = parsing::data_id_len)]
		key: DataIdLen,
	},
}

/// Common VM arguments.
#[derive(clap::Args, Clone, Debug)]
pub struct VmArgs {
	/// Root file system directory (service_id:hash or local path).
	#[arg(
        short = 'r',
        long = "root",
        value_name = "SERVICE_ID:HASH|DIR",
        value_parser = corevm_tooling::parsing::corevm_fs_dir_or_local_dir
    )]
	root_dir: Option<CoreVmFsNode>,
	/// Environment variables.
	#[arg(short = 'e', long = "env", value_name = "KEY=VALUE")]
	env: Vec<CString>,
	/// Command line arguments.
	#[arg(trailing_var_arg = true, allow_hyphen_values = true, value_name = "ARGS...")]
	args: Vec<CString>,
	/// Inner VM gas limit.
	#[clap(
        long = "gas",
        value_name = "GAS",
        default_value_t = SignedGas::MAX as UnsignedGas
	)]
	gas: UnsignedGas,
	/// Video input mode.
	///
	/// Mandatory parameters:
	///
	/// - `WIDTH`: Video stream width.
	/// - `HEIGHT`: Video stream height.
	/// - `REFRESH_RATE`: Frame rate in frames per second.
	///
	/// Optional parameters:
	///
	/// - `quantization-level`: Video codec quantization level (0-14). Typical value is 3-4.
	/// - `chroma-subsampling`: Video codec chroma subsampling (`true` means YUV420, `false` means
	///   YUV444).
	/// - `raw`: If `true`, then video frames are transmitted without encoding. This is useful if
	///   you're using PolkaVM interpreter.
	#[arg(
        long = "video-input",
        value_name = "WIDTH x HEIGHT @ REFRESH_RATE, param1=value1, ...",
        value_parser = corevm_tooling::parsing::corevm_video_mode,
        verbatim_doc_comment
	)]
	video_input: Option<VideoMode>,
	/// Audio input mode.
	///
	/// Parameters:
	///
	/// - `SAMPLE_RATE`: Sample rate in Hz.
	/// - `CHANNELS`: The number of audio channels.
	/// - `SAMPLE_FORMAT`: Audio sample format. Possible values: `s16le`.
	#[arg(
        long = "audio-input",
        value_name = "SAMPLE_RATE:CHANNELS:SAMPLE_FORMAT",
        value_parser = corevm_tooling::parsing::corevm_audio_mode,
        verbatim_doc_comment
	)]
	audio_input: Option<AudioMode>,
	/// Input public key.
	///
	/// This is used to verify input chunks sent by CoreVM monitor.
	#[arg(long = "input-key", value_name = "HEX", value_parser = parsing::ed25519_public_key)]
	input_key: Option<[u8; 32]>,
}

#[derive(clap::Subcommand, Debug, Clone)]
pub enum VmCommands {
	/// Create a new CoreVM service and load code.
	New {
		/// The code to use for the service.
		#[arg(value_parser = parsing::blob)]
		code: Blob,
		/// The initial endowment for the service.
		#[arg(default_value_t = 0)]
		amount: Balance,
		/// The minimum accumulation gas each work-item must have.
		#[arg(short = 'G', long, default_value_t = 1_000_000)]
		min_item_gas: UnsignedGas,
		/// The minimum on-transfer gas each memo must have.
		#[arg(short = 'g', long, default_value_t = 1_000_000)]
		min_memo_gas: UnsignedGas,
		#[command(flatten)]
		args: VmArgs,
	},
	/// Update code; this will only work if we know how to transmit from the owner service.
	Upgrade {
		/// The service to update.
		#[arg(value_parser = parsing::service_id)]
		service_id: ServiceId,
		/// The code to use for the service.
		#[arg(value_parser = parsing::blob)]
		code: Blob,
		#[command(flatten)]
		args: VmArgs,
	},
	/// Destroy CoreVM service.
	///
	/// This command zombifies the service and forgets all associated preimages (program and
	/// root file system).
	Destroy {
		/// The service to destroy.
		#[arg(value_parser = parsing::service_id)]
		target: ServiceId,
	},
	/// Print all preimages that the service uses.
	///
	/// Prints service id, preimage hash and preimage length.
	Preimages {
		#[arg(value_parser = parsing::service_id)]
		target: ServiceId,
	},
}

#[derive(clap::Subcommand, Debug, Clone)]
pub enum FsCommands {
	/// Upload the specified local files/directories to the preimage store.
	Upload {
		/// File system block size to use.
		#[arg(
            long = "block-size",
            value_name = "BYTES",
            value_parser = parsing::exact_bytes,
            default_value = format::exact_bytes(corevm_fs::MAX_BLOCK_SIZE as u64),
		)]
		block_size: u64,
		/// Input path in the local file system.
		#[arg(value_name = "PATH")]
		paths: Vec<PathBuf>,
	},
	/// Download the specified file/directory from the preimage store.
	Download {
		/// The service ID and hash of the input file/directory in the preimage store.
		#[arg(
            value_name = "SERVICE_ID:HASH",
            value_parser = corevm_tooling::parsing::corevm_fs_block_ref
		)]
		block_ref: corevm_fs::BlockRef,
		/// Output path in the local file system.
		#[arg(value_name = "PATH")]
		output_path: PathBuf,
	},
}

#[derive(clap::Subcommand, Debug, Clone)]
pub enum Commands {
	/// Create a new service.
	CreateService {
		/// The code to use for the service.
		#[arg(value_parser = parsing::data_id_len)]
		code: DataIdLen,
		/// The initial endowment for the service.
		#[arg(default_value_t = 0)]
		amount: Balance,
		/// The memo given to the service for the endowment.
		#[arg(value_parser = parsing::memo, default_value = "")]
		memo: Memo,
		/// The minimum accumulation gas each work-item must have.
		#[arg(short = 'G', long, default_value_t = 1_000_000)]
		min_item_gas: UnsignedGas,
		/// The minimum on-transfer gas each memo must have.
		#[arg(short = 'g', long, default_value_t = 1_000_000)]
		min_memo_gas: UnsignedGas,
		/// Register the service with the Bootstrap service.
		#[arg(short = 'r', long)]
		register: Option<String>,
		/// Print only the new service ID to stdout; progress messages go to stderr.
		#[arg(long, default_value_t = false)]
		raw: bool,
	},
	/// Build and submit a work package with a single work-item.
	Item {
		/// The service to submit the work-item to.
		#[arg(value_parser = parsing::service_id)]
		service_id: ServiceId,
		/// The payload to submit.
		#[arg(value_parser = parsing::blob, default_value = "")]
		payload: Blob,
		/// The amount of gas to provide for refinement of the work-item. If none provided, then
		/// the full work-package amount will be used.
		#[arg(short = 'G', long)]
		refine_gas: Option<UnsignedGas>,
		/// The amount of gas to provide for accumulation of the work-item. If none provided, then
		/// the full work-package amount will be used.
		#[arg(short = 'g', long)]
		accumulate_gas: Option<UnsignedGas>,
		/// The number of exports that the item is expected to produce.
		#[arg(short = 'e', long, default_value_t = 0)]
		exports: u16,
		/// Import definition, either "wp:<hex_encoded_work_package_hash>:<import_index>"
		/// for access by work package hash or "sr:<hex_encoded_segment_root>:<import_index>"
		/// for access by segment root.
		#[arg(short = 'i', long, value_parser = parsing::import_spec)]
		import: Vec<jam_types::ImportSpec>,
	},
	/// Make a transfer from the Bootstrap service.
	Transfer {
		/// The service to which the transfer is made.
		#[arg(value_parser = parsing::service_id)]
		service_id: ServiceId,
		/// The amount to transfer to the service.
		#[arg(default_value_t = 0)]
		amount: Balance,
		/// The memo to transfer to the service.
		#[arg(value_parser = parsing::memo, default_value = "")]
		memo: Memo,
		/// The amount of gas to use for the transfer, defaulting to the minimum transfer amount of
		/// the destination service.
		#[arg(short = 'g', long, value_parser = parsing::gas)]
		gas: Option<UnsignedGas>,
	},
	/// Get information concerning a block.
	Inspect {
		/// The block to query.
		#[arg(default_value = "best")]
		id: BlockId,
		/// The field to extract and view.
		#[command(subcommand)]
		field: Inspections,
	},
	/// List all service IDs at a block.
	List {
		/// The block to query.
		#[arg(default_value = "best")]
		id: BlockId,
	},
	/// Submit a solicited preimage to the chain.
	Provide {
		/// The service that has requested the preimage.
		#[arg(value_parser = parsing::service_id)]
		requester: ServiceId,
		/// The preimage blob.
		#[arg(value_parser = parsing::blob)]
		preimage: Blob,
	},
	/// Inspect the queue.
	Queue,
	/// Submit all enqueued work-items.
	Pack,
	/// Build and submit multiple work packages. One work package with a single work-item is
	/// created per payload, assigned to cores sequentially starting from `--force-core` (or 0).
	PackList {
		/// The service to submit the work-items to.
		#[arg(value_parser = parsing::service_id)]
		service_id: ServiceId,
		/// The payloads to submit. One work-item is created per payload.
		#[arg(value_parser = parsing::blob)]
		payloads: Vec<Blob>,
		/// The amount of gas to provide for refinement of each work-item.
		#[arg(short = 'G', long)]
		refine_gas: Option<UnsignedGas>,
		/// The amount of gas to provide for accumulation of each work-item.
		#[arg(short = 'g', long)]
		accumulate_gas: Option<UnsignedGas>,
	},
	/// A bunch of VM commands.
	Vm {
		/// The command to run.
		#[command(subcommand)]
		command: VmCommands,
	},
	/// File system operations.
	Fs {
		/// The command to run.
		#[command(subcommand)]
		command: FsCommands,
	},
}

#[derive(ValueEnum, Debug, Default, Clone, Copy, Eq, PartialEq)]
pub enum Provision {
	/// No provision; any data requirements must be satisfied manually.
	None,
	/// Provide direct to the service, not via Bootstrap.
	Direct,
	/// Provide via the Bootstrap service.
	#[default]
	Bootstrap,
}

/// Options controlling work-package submission and data provision.
#[derive(Parser, Debug, Clone)]
pub struct PackageOptions {
	/// The strategy for providing data to services.
	#[arg(short = 'p', long, value_enum, default_value_t = Provision::Bootstrap)]
	pub provision: Provision,
	/// Force the operation to happen on a particular core if relevant.
	#[arg(short = 'c', long)]
	pub force_core: Option<CoreIndex>,
	/// When creating a work-item, do not package and submit. Instead queue up the item for a later
	/// submission, potentially with other items.
	#[arg(short = 'q', long, default_value_t = false)]
	pub queue: bool,
}

impl PackageOptions {
	pub fn core(&self) -> CoreIndex {
		self.force_core.unwrap_or(0)
	}
}

/// JAM CLI tool
#[derive(Parser, Debug)]
#[command(name = "jamt", version, long_version = format!("{}\nGP {}", env!("CARGO_PKG_VERSION"), jam_types::GP_VERSION))]
pub struct BootArgs {
	#[command(flatten)]
	common: CommonArgs,
	/// The service ID of the Bootstrap service.
	#[arg(short = 'b', long, value_parser = parsing::service_id, default_value_t = 0)]
	bootstrap_service_id: ServiceId,
	#[command(flatten)]
	package: PackageOptions,
	/// The command to run.
	#[command(subcommand)]
	command: Commands,
}

pub struct BootstrapProvision {
	node: Arc<WsClient>,
	preimages: Vec<(Hash, Vec<u8>)>,
	service_id: ServiceId,
}

impl BootstrapProvision {
	pub async fn ensure(
		&mut self,
		header_hash: HeaderHash,
		data: Vec<u8>,
	) -> anyhow::Result<(bool, Hash)> {
		let hash = hash_raw(&data);
		Ok((
			match self
				.node
				.service_request(header_hash, self.service_id, hash, data.len() as u32)
				.await?
				.map(|v| v.len())
			{
				Some(1) => false,
				_ => {
					self.preimages.push((hash, data));
					true
				},
			},
			hash,
		))
	}

	async fn provide(self, context: &mut Context) -> anyhow::Result<BTreeMap<Hash, Slot>> {
		// Solicit everything.
		context.boot_solicit(self.preimages.iter().map(|(h, d)| (*h, d.len()))).await?;

		// Provide those which are needed.
		let mut items = Vec::new();
		for (hash, data) in self.preimages.into_iter() {
			let len = data.len() as u32;
			let mut sub =
				self.node.subscribe_service_request(self.service_id, hash, len, false).await?;
			loop {
				let ChainSubUpdate { value, .. } = sub
					.next()
					.await
					.ok_or(anyhow!("Subscription to monitor creation terminated unexpectedly"))??;
				match value.map(|v| v.len()) {
					Some(0) => {
						// Requested and provision needed.
						items.push((hash, data.len() as u32));
						self.node.submit_preimage(self.service_id, data.into()).await?;
						break
					},
					Some(1) | Some(3) => {
						// Requested and no provision needed!
						break
					},
					_ => {
						// Not yet requested.
					},
				}
			}
		}

		// Wait for the solicitations to be satisfied.
		let mut map = BTreeMap::new();
		for (hash, len) in items.into_iter() {
			let mut sub =
				self.node.subscribe_service_request(self.service_id, hash, len, false).await?;
			let slot = loop {
				let ChainSubUpdate { value, .. } = sub
					.next()
					.await
					.ok_or(anyhow!("Subscription to monitor creation terminated unexpectedly"))??;
				match value.map(|v| (v.len(), v)) {
					Some((1, v)) | Some((3, v)) => {
						// Already provided; nothing to do.
						break *v.last().expect("Already matched length 1; qed")
					},
					_ => {
						// Not yet provided.
					},
				}
			};
			map.insert(hash, slot);
		}
		eprintln!("Provided {} item(s)", map.len());
		Ok(map)
	}
}

struct Context {
	args: BootArgs,
	node: Arc<WsClient>,
	bootstrap_service: Option<Arc<BootstrapService>>,
}
impl Context {
	async fn new(args: BootArgs) -> anyhow::Result<Self> {
		let node = Arc::new(args.common.connect_rpc(0).await?);
		let VersionedParameters::V1(params) = node.parameters().await?;
		params.apply().map_err(|s| anyhow!("{}", s))?;
		if let Some(core) = args.package.force_core {
			if core >= core_count() {
				return Err(anyhow!(
					"Invalid core ({core}); the number of cores is {}",
					core_count()
				));
			}
		}

		let s = Self { args, node, bootstrap_service: None };
		Ok(s)
	}

	fn bootstrap_provider(&self) -> BootstrapProvision {
		BootstrapProvision {
			node: self.node.clone(),
			service_id: self.args.bootstrap_service_id,
			preimages: vec![],
		}
	}

	fn package_builder(&self) -> PackageBuilder<WsClient> {
		PackageBuilder::new(
			self.node.clone(),
			self.args.bootstrap_service_id,
			self.args.package.clone(),
		)
	}

	async fn ensure_boot(&mut self) -> anyhow::Result<Arc<BootstrapService>> {
		match self.bootstrap_service {
			Some(ref service) => Ok(service.clone()),
			None => {
				let BlockDesc { header_hash: head, .. } = self.node.best_block().await?;
				let id = self.args.bootstrap_service_id;
				let service = BootstrapService::query(self.node.clone(), head, id).await?;
				Ok(self.bootstrap_service.insert(Arc::new(service)).clone())
			},
		}
	}

	async fn execute(mut self) -> anyhow::Result<()> {
		match self.args.command.clone() {
			Commands::CreateService {
				code,
				amount,
				memo,
				min_item_gas,
				min_memo_gas,
				register,
				raw,
			} => {
				let boot = self.ensure_boot().await?;
				boot.create_service(
					code,
					amount,
					memo,
					min_item_gas,
					min_memo_gas,
					register,
					raw,
					&self.args.package,
				)
				.await?;
			},
			Commands::Transfer { service_id, amount, memo, gas } => {
				let boot = self.ensure_boot().await?;
				self.package_builder()
					.transfer(boot.as_ref(), service_id, amount, memo, gas)
					.await?;
			},
			Commands::Item { service_id, payload, refine_gas, accumulate_gas, exports, import } => {
				self.package_builder()
					.item(service_id, payload, refine_gas, accumulate_gas, exports, import)
					.await?;
			},
			Commands::PackList { service_id, payloads, refine_gas, accumulate_gas } => {
				self.package_builder()
					.pack_list(service_id, payloads, refine_gas, accumulate_gas)
					.await?;
			},
			Commands::Pack => {
				let boot = self.ensure_boot().await?;
				self.package_builder().pack(boot.as_ref()).await?;
			},
			Commands::Queue => queue::queue_status()?,
			Commands::Inspect { id, field } => {
				let desc = match id {
					BlockId::Best => self.node.best_block().await?,
					BlockId::Final => self.node.finalized_block().await?,
				};
				match field {
					Inspections::Slot => println!("{}", desc.slot),
					Inspections::Service { id } => {
						let (_service, maybe_meta, maybe_inner_meta) =
							self.node.query_service_and_corevm_guest(desc.header_hash, id).await?;
						if let CodeInfo::Known(meta) = maybe_meta {
							println!("Service {meta}");
						} else {
							eprintln!("Service code not found");
						}
						match maybe_inner_meta {
							None => {},
							Some(CoreVmCodeInfo::Undefined(block_ref)) => {
								eprintln!(
									"CoreVM guest code {block_ref} is defined but unreadable"
								);
							},
							Some(CoreVmCodeInfo::NotProvided(block_ref)) => {
								eprintln!("File {block_ref} is not provided");
							},
							Some(CoreVmCodeInfo::Known(meta)) => {
								println!("Guest {meta}");
							},
						}
					},
					Inspections::Storage { id, key, raw } => {
						let value = self.node.service_value(desc.header_hash, id, &key).await?;
						if let Some(value) = value {
							if raw {
								println!("0x{}", jam_types::hex::hex(&value));
							} else {
								println!("{}", AnyBytes::from(value));
							}
						} else {
							return Err(anyhow!("No value found for key {:?}", key))
						}
					},
					Inspections::Request { id, key } => {
						let hash = key.hash();
						let len = key.len() as u32;
						let maybe_request =
							self.node.service_request(desc.header_hash, id, hash, len).await?;
						let Some(request) = maybe_request else {
							return Err(anyhow!(
								"No preimage request found for hash {:?}",
								hash.any()
							));
						};
						match request.len() {
							0 => {
								println!("Unprovided request found");
							},
							1 => {
								println!("Request active, provided at slot {}", request[0]);
							},
							2 => {
								println!(
									"Request inactive since {}, provided at slot {}",
									request[1], request[0]
								);
							},
							3 => {
								println!(
									"Request active since {}, unrequested at slot {}, provided at slot {}",
									request[2], request[1], request[0]
								);
							},
							_ => unreachable!(),
						}
					},
					Inspections::Preimage { id, key } => {
						let hash = key.hash();
						let maybe_data =
							self.node.service_preimage(desc.header_hash, id, hash).await?;
						let Some(data) = maybe_data else {
							return Err(anyhow!("No preimage provided for hash {:?}", hash.any()));
						};
						println!("Preimage found of length {}", data.len());
					},
				}
			},
			Commands::List { id } => {
				let desc = match id {
					BlockId::Best => self.node.best_block().await?,
					BlockId::Final => self.node.finalized_block().await?,
				};
				let services = self.node.list_services(desc.header_hash).await?;
				let infos = futures::future::try_join_all(services.iter().map(|&id| {
					let node = self.node.clone();
					async move {
						let (_, meta, inner) =
							node.query_service_and_corevm_guest(desc.header_hash, id).await?;
						anyhow::Ok((id, meta, inner))
					}
				}))
				.await?;
				for (id, meta, inner) in infos {
					let name = match (&meta, &inner) {
						(CodeInfo::Known(m), Some(CoreVmCodeInfo::Known(im))) =>
							if m.name == "corevm" {
								format!("vm: {}", im.name)
							} else {
								format!("{}: {}", m.name, im.name)
							},
						(CodeInfo::Known(m), _) => m.name.clone(),
						(CodeInfo::Undefined(h), _) | (CodeInfo::CodeNotProvided(h), _) =>
							format!("{}", AnyHash(*h)),
					};
					println!("{id:08x}\t{name}");
				}
			},
			Commands::Provide { requester: service_id, preimage: pre_image } => {
				let boot = self.ensure_boot().await?;
				let at = boot.provide(service_id, pre_image).await?;
				eprintln!("Preimage provided at slot {at}");
			},
			Commands::Vm { command } => match command {
				VmCommands::New {
					code,
					amount,
					min_item_gas,
					min_memo_gas,
					args: VmArgs { gas, root_dir, env, args, video_input, audio_input, input_key },
				} => {
					let gas: SignedGas =
						gas.try_into().map_err(|_| anyhow::anyhow!("Too much gas"))?;
					if let Ok((_, Metadata::Info(i))) =
						<(Compact<u32>, Metadata)>::decode(&mut &corevm_bin::BLOB[..])
					{
						eprintln!("Using CoreVM module {i}");
					}

					if let Ok((_, Metadata::Info(i))) =
						<(Compact<u32>, Metadata)>::decode(&mut &code[..])
					{
						eprintln!("Guest identifies as {i}");
					} else {
						eprintln!("Guest code metadata not found");
					}
					if (video_input.is_some() || audio_input.is_some()) && input_key.is_none() {
						return Err(anyhow!(
							"You should specify input key when audio/video input is enabled"
						));
					}

					// TODO: Respect provision strategy for CoreVM code.
					// We'll eventually want to allow a CoreVM service instance to host its guest
					// code   directly, but for now we just provide it to the Bootstrap
					// service.
					assert_eq!(
						self.args.package.provision,
						Provision::Bootstrap,
						"Only provision via bootstrap service is supported"
					);

					self.node.wait_for_sync().await?;
					let (memo, code_ref, root_dir_ref) = self
						.reset_vm_instruction(
							gas,
							code,
							root_dir,
							args,
							env,
							video_input,
							audio_input,
							input_key,
							true,
						)
						.await?;
					let inst = Instruction::CreateService {
						code_hash: corevm_bin::HASH.into(),
						code_len: corevm_bin::BLOB.len() as u64,
						min_item_gas,
						min_memo_gas,
						endowment: amount,
						memo,
						registration: None,
					};
					let boot = self.ensure_boot().await?;
					let core = self.args.package.force_core.unwrap_or(0);
					boot.submit_instructions(vec![inst], core).await?;

					let id = boot.wait_for_the_service_to_be_created().await?;
					println!("Service id: {id:x}");
					println!("Code file: {code_ref}");
					if !root_dir_ref.hash.is_zero() {
						println!("Root directory: {root_dir_ref}");
					}
				},
				VmCommands::Upgrade {
					service_id,
					code,
					args: VmArgs { gas, root_dir, env, args, video_input, audio_input, input_key },
				} => {
					let gas: SignedGas =
						gas.try_into().map_err(|_| anyhow::anyhow!("Too much gas"))?;
					if (video_input.is_some() || audio_input.is_some()) && input_key.is_none() {
						return Err(anyhow!(
							"You should specify input key when audio/video input is enabled"
						));
					}
					let best = self.node.best_block().await?;
					let (service, _corevm_code) =
						service::query(self.node.as_ref(), best.header_hash, service_id, "corevm")
							.await?;

					let (memo, code_ref, root_dir_ref) = self
						.reset_vm_instruction(
							gas,
							code,
							root_dir,
							args,
							env,
							video_input,
							audio_input,
							input_key,
							false,
						)
						.await?;
					let inst = Instruction::Transfer {
						destination: service_id,
						amount: 0,
						gas_limit: service.min_memo_gas,
						memo,
					};
					let boot = self.ensure_boot().await?;
					let core = self.args.package.force_core.unwrap_or(0);
					boot.submit_instructions(vec![inst], core).await?;
					eprintln!("Code file: {code_ref}");
					if !root_dir_ref.hash.is_zero() {
						eprintln!("Root directory: {root_dir_ref}");
					}
				},
				VmCommands::Destroy { target } => {
					vm::destroy(
						self.node,
						target,
						self.args.bootstrap_service_id,
						self.args.package.force_core.unwrap_or(0),
					)
					.await?;
					eprintln!(
						"Success! The service can be ejected in {}",
						format::exact_duration(Duration::from_secs(
							u64::from(min_turnaround_period()) * slot_period_sec()
						)),
					);
				},
				VmCommands::Preimages { target } => {
					let preimages = vm::preimages(self.node, target).await?;
					for (service_id, hash, len) in preimages.into_iter() {
						println!("{service_id:8x}  {}  {len}", format::hash(&hash));
					}
				},
			},
			Commands::Fs { command } => match command {
				FsCommands::Upload { paths, block_size } => {
					let block_size = block_size as usize;
					if paths.is_empty() {
						return Err(anyhow::anyhow!("No files specified."))
					}
					assert_eq!(
						self.args.package.provision,
						Provision::Bootstrap,
						"Only provision via bootstrap service is supported"
					);
					self.node.wait_for_sync().await?;
					let provider = self.bootstrap_provider();
					let (block_refs, writer) = spawn_blocking(move || {
						let mut writer = PreimageBlockWriter::new(provider);
						let mut block_refs = Vec::with_capacity(paths.len());
						for path in paths.into_iter() {
							let block_ref = if path.is_dir() {
								let dir_reader =
									VerboseDirRead(corevm_fs::StdDirReader::new(path.clone())?);
								corevm_fs::copy_dir_in(
									dir_reader,
									self.args.bootstrap_service_id,
									&mut writer,
									block_size,
								)?
							} else {
								use std::io::Seek;
								let mut file = std::fs::File::open(&path)?;
								let file_size = file.seek(std::io::SeekFrom::End(0))?;
								file.rewind()?;
								eprintln!("Copying {}... [{file_size} B]", path.display());
								corevm_fs::copy_file_in(
									&mut file,
									self.args.bootstrap_service_id,
									&mut writer,
									block_size,
								)?
							};
							block_refs.push((path, block_ref));
						}
						Ok::<_, anyhow::Error>((block_refs, writer))
					})
					.await??;
					let m = writer.finish(&mut self).await?;
					eprintln!("Provided {} preimages to Bootstrap service", m.len());
					for (path, block_ref) in block_refs.into_iter() {
						eprintln!("{}: {block_ref}", path.display());
					}
				},
				FsCommands::Download { block_ref, output_path } => {
					assert_eq!(
						self.args.package.provision,
						Provision::Bootstrap,
						"Only provision via bootstrap service is supported"
					);
					self.node.wait_for_sync().await?;
					let best = self.node.best_block().await?;
					spawn_blocking(move || {
						let mut reader = corevm_tooling::PreimageReader::new(
							self.node.clone(),
							best.header_hash,
						);
						let fs_writer = VerboseFsWrite(corevm_fs::StdWriter::new(output_path));
						corevm_fs::copy_out(&block_ref, &mut reader, fs_writer)
					})
					.await??;
				},
			},
		}
		Ok(())
	}

	async fn boot_solicit(
		&mut self,
		items: impl IntoIterator<Item = (Hash, usize)>,
	) -> anyhow::Result<WorkPackageHash> {
		let boot = self.ensure_boot().await?;
		boot.boot_solicit(items, &self.args.package).await
	}

	async fn upload_exec_env(
		&mut self,
		code: Vec<u8>,
		root_dir: Option<CoreVmFsNode>,
		args: Vec<CString>,
		env: Vec<CString>,
		video_input: Option<VideoMode>,
		audio_input: Option<AudioMode>,
		input_key: Option<[u8; 32]>,
		provide_corevm_code: bool,
	) -> anyhow::Result<(corevm_host::ExecEnv, corevm_fs::BlockRef)> {
		let mut provider = self.bootstrap_provider();
		if provide_corevm_code {
			let best = self.node.best_block().await?;
			if provider.ensure(best.header_hash, corevm_bin::BLOB.to_vec()).await?.0 {
				eprintln!("Providing CoreVM module code...");
			} else {
				eprintln!("CoreVM module already provided.");
			}
		}
		let service_id = self.args.bootstrap_service_id;
		let (exec_ref, exec, writer) = spawn_blocking(move || {
			let mut writer = PreimageBlockWriter::new(provider);
			let code_ref = corevm_fs::copy_file_in(
				&mut std::io::Cursor::new(&code[..]),
				service_id,
				&mut writer,
				corevm_fs::MAX_BLOCK_SIZE,
			)?;
			let root_dir_ref = match root_dir {
				None => corevm_fs::BlockRef { service_id: 0, hash: Default::default() },
				Some(CoreVmFsNode::BlockRef(block_ref)) => block_ref,
				Some(CoreVmFsNode::Local(path)) => {
					let dir_reader = VerboseDirRead(corevm_fs::StdDirReader::new(path)?);
					corevm_fs::copy_dir_in(
						dir_reader,
						service_id,
						&mut writer,
						corevm_fs::MAX_BLOCK_SIZE,
					)?
				},
			};
			let exec = corevm_host::ExecEnv {
				program: code_ref,
				root_dir: root_dir_ref,
				args: args.into_iter().map(corevm_host::Arg).collect(),
				env: env.into_iter().map(corevm_host::Arg).collect(),
				video_input,
				audio_input,
				input_key,
			};
			let exec_encoded = exec.encode();
			let exec_ref = corevm_fs::copy_file_in(
				&mut std::io::Cursor::new(&exec_encoded[..]),
				service_id,
				&mut writer,
				corevm_fs::MAX_BLOCK_SIZE,
			)?;
			Ok::<_, anyhow::Error>((exec_ref, exec, writer))
		})
		.await??;
		writer.finish(self).await?;
		Ok((exec, exec_ref))
	}

	async fn reset_vm_instruction(
		&mut self,
		gas: SignedGas,
		code: Vec<u8>,
		root_dir: Option<CoreVmFsNode>,
		args: Vec<CString>,
		env: Vec<CString>,
		video_input: Option<VideoMode>,
		audio_input: Option<AudioMode>,
		input_key: Option<[u8; 32]>,
		provide_corevm_code: bool,
	) -> anyhow::Result<(Memo, corevm_fs::BlockRef, corevm_fs::BlockRef)> {
		let (exec, exec_ref) = self
			.upload_exec_env(
				code,
				root_dir,
				args,
				env,
				video_input,
				audio_input,
				input_key,
				provide_corevm_code,
			)
			.await?;
		let mut memo = Memo::zero();
		CoreVmInstruction::Reset { gas, exec_ref }.encode_to(&mut &mut memo[..]);
		Ok((memo, exec.program, exec.root_dir))
	}
}

pub async fn boot_main() -> anyhow::Result<()> {
	Context::new(BootArgs::parse()).await?.execute().await
}

struct PreimageBlockWriter {
	provider: BootstrapProvision,
	best_block: CachedBestBlock,
	block_counter: usize,
}

impl PreimageBlockWriter {
	fn new(provider: BootstrapProvision) -> Self {
		Self { block_counter: 0, provider, best_block: CachedBestBlock::new() }
	}

	async fn finish(self, context: &mut Context) -> anyhow::Result<BTreeMap<Hash, Slot>> {
		self.provider.provide(context).await
	}
}

impl corevm_fs::WriteBlock for PreimageBlockWriter {
	fn write_block(&mut self, service_id: ServiceId, buf: &[u8]) -> Result<(), corevm_fs::IoError> {
		// We use bootstrap provider, hence service id should always be the same.
		assert_eq!(self.provider.service_id, service_id);
		let preimage_len = buf.len();
		log::debug!("Soliciting file block ({preimage_len} B)");
		let rt = tokio::runtime::Handle::current();
		let best = rt.block_on(self.best_block.get(self.provider.node.as_ref())).map_err(|e| {
			log::debug!("Failed to get best block: {e}");
			corevm_fs::IoError
		})?;
		let (providing, _) =
			rt.block_on(self.provider.ensure(best.header_hash, buf.to_vec())).map_err(|e| {
				log::debug!("Failed to solicit file block ({preimage_len} B): {e}");
				corevm_fs::IoError
			})?;
		if providing {
			self.block_counter += 1;
			log::debug!("Providing file block {}...", self.block_counter);
		}
		Ok(())
	}
}

struct VerboseDirRead(pub corevm_fs::StdDirReader);

impl corevm_fs::HostDirRead<std::fs::File> for VerboseDirRead {
	type Id = (u64, u64);

	fn next_entry(&mut self) -> Option<Result<corevm_fs::HostDirEntry, corevm_fs::IoError>> {
		self.0.next_entry()
	}

	fn open_file(
		&mut self,
		name: &corevm_fs::FileName,
	) -> Result<std::fs::File, corevm_fs::IoError> {
		use std::io::Seek;
		let mut file = self.0.open_file(name)?;
		let file_size = file.seek(std::io::SeekFrom::End(0)).map_err(|_| corevm_fs::IoError)?;
		file.rewind().map_err(|_| corevm_fs::IoError)?;
		let name = name.as_ref().to_string_lossy();
		eprintln!("Copying {}{MAIN_SEPARATOR}{name}... [{file_size} B]", self.0.path().display());
		Ok(file)
	}

	fn open_dir(
		&mut self,
		name: &corevm_fs::FileName,
	) -> Result<(Self, Option<Self::Id>), corevm_fs::IoError> {
		let name_str = name.as_ref().to_string_lossy();
		eprintln!("Copying {}{MAIN_SEPARATOR}{name_str}...", self.0.path().display());
		self.0.open_dir(name).map(|(dir, dir_id)| (Self(dir), dir_id))
	}
}

struct VerboseDirWrite(pub corevm_fs::StdDirWriter);

impl corevm_fs::HostDirWrite for VerboseDirWrite {
	type FileWrite = std::fs::File;

	fn create_file(
		&mut self,
		name: &corevm_fs::FileName,
	) -> Result<std::fs::File, corevm_fs::IoError> {
		let name_str = name.as_ref().to_string_lossy();
		eprintln!("Creating file {}{MAIN_SEPARATOR}{name_str}...", self.0.path().display());
		self.0.create_file(name)
	}

	fn create_dir(&mut self, name: &corevm_fs::FileName) -> Result<Self, corevm_fs::IoError> {
		let name_str = name.as_ref().to_string_lossy();
		eprintln!("Creating directory {}{MAIN_SEPARATOR}{name_str}...", self.0.path().display());
		self.0.create_dir(name).map(Self)
	}
}

struct VerboseFsWrite(pub corevm_fs::StdWriter);

impl corevm_fs::HostWrite for VerboseFsWrite {
	type FileWrite = std::fs::File;
	type DirWrite = VerboseDirWrite;

	fn into_file_writer(self) -> Result<Self::FileWrite, corevm_fs::IoError> {
		eprintln!("Writing {}...", self.0.path().display());
		self.0.into_file_writer()
	}

	fn into_dir_writer(self) -> Result<Self::DirWrite, corevm_fs::IoError> {
		self.0.into_dir_writer().map(VerboseDirWrite)
	}
}

#[cfg(test)]
mod tests {
	use super::*;
	use clap::CommandFactory;

	#[test]
	fn args_work() {
		BootArgs::command().debug_assert();
	}
}