// Copyright (c) 2022 Red Hat
//
// SPDX-License-Identifier: Apache-2.0
//

use super::cmdline_generator::{get_network_device, QemuCmdLine};
use super::qmp::Qmp;
use crate::device::pci_path::PciPath;
use crate::device::topology::PCIePort;
use crate::qemu::cmdline_generator::VfioDeviceConfig;
use crate::qemu::qmp::get_qmp_socket_path;
use crate::{
    device::driver::ProtectionDeviceConfig, hypervisor_persist::HypervisorState, selinux,
    HypervisorConfig, MemoryConfig, VcpuThreadIds, VsockDevice, HYPERVISOR_QEMU, KATA_BLK_DEV_TYPE,
    KATA_CCW_DEV_TYPE, KATA_NVDIMM_DEV_TYPE, KATA_SCSI_DEV_TYPE,
};

use crate::utils::{
    bytes_to_megs, create_dir_all_with_inherit_owner, enter_netns, get_jailer_root, megs_to_bytes,
    set_groups, uses_native_ccw_bus, vm_cleanup,
};

use anyhow::{anyhow, Context, Result};
use async_trait::async_trait;
use kata_sys_util::netns::NetnsGuard;
use kata_types::build_path;
use kata_types::config::hypervisor::{RootlessUser, VIRTIO_BLK_CCW, VIRTIO_BLK_PCI};
use kata_types::rootless::is_rootless;
use kata_types::{
    capabilities::{Capabilities, CapabilityBits},
    config::KATA_PATH,
};
use nix::unistd::{setgid, setuid, Gid, Uid};
use persist::sandbox_persist::Persist;
use qapi_qmp::MigrationStatus;
use std::cmp::Ordering;
use std::convert::{TryFrom, TryInto};
use std::path::Path;
use std::process::Stdio;
use std::time::Duration;

use tokio::time::sleep;
use tokio::time::Instant;
use tokio::{
    io::{AsyncBufReadExt, BufReader},
    process::{Child, ChildStderr, Command},
};
use tokio::{
    net::UnixStream,
    sync::{mpsc, Mutex},
};

const VSOCK_SCHEME: &str = "vsock";

#[derive(Debug)]
pub struct QemuInner {
    /// sandbox id
    id: String,

    qemu_process: Mutex<Option<Child>>,
    qmp: Option<Qmp>,

    config: HypervisorConfig,
    devices: Vec<DeviceType>,
    netns: Option<String>,

    exit_notify: Option<mpsc::Sender<()>>,
}

impl QemuInner {
    pub fn new(exit_notify: mpsc::Sender<()>) -> QemuInner {
        QemuInner {
            id: "".to_string(),
            qemu_process: Mutex::new(None),
            qmp: None,
            config: Default::default(),
            devices: Vec::new(),
            netns: None,

            exit_notify: Some(exit_notify),
        }
    }

    pub(crate) async fn prepare_vm(
        &mut self,
        id: &str,
        netns: Option<String>,
        selinux_label: Option<String>,
    ) -> Result<()> {
        info!(sl!(), "Preparing QEMU VM");
        self.id = id.to_string();
        self.netns = netns;

        if !self.hypervisor_config().disable_selinux {
            if let Some(label) = selinux_label.as_ref() {
                self.config.security_info.selinux_label = Some(label.to_string());
                selinux::set_exec_label(label).context("failed to set SELinux process label")?;
            }
        }

        let vm_path = Path::new(build_path(KATA_PATH).as_str()).join(self.id.as_str());
        create_dir_all_with_inherit_owner(vm_path, 0o750)?;

        Ok(())
    }

    pub(crate) async fn start_vm(&mut self, _timeout: i32) -> Result<()> {
        info!(sl!(), "Starting QEMU VM");
        let netns = self.netns.clone().unwrap_or_default();

        // CAUTION: since 'cmdline' contains file descriptors that have to stay
        // open until spawn() is called to launch qemu later in this function,
        // 'cmdline' has to live at least until spawn() is called
        let mut cmdline = QemuCmdLine::new(&self.id, &self.config)?;

        for device in &mut self.devices {
            match device {
                DeviceType::ShareFs(share_fs_dev) => {
                    if share_fs_dev.config.fs_type == "virtio-fs" {
                        cmdline.add_virtiofs_share(
                            &share_fs_dev.config.sock_path,
                            &share_fs_dev.config.mount_tag,
                            share_fs_dev.config.queue_size,
                        );
                    }
                }
                DeviceType::Vsock(vsock_dev) => {
                    let fd = vsock_dev.init_config().await?;
                    cmdline.add_vsock(fd, vsock_dev.config.guest_cid)?;
                }
                DeviceType::BlockModern(ref block_device) => {
                    info!(sl!(), "Starting QMP hotplug for BlockModern device");

                    // First, snapshot parameters within the lock.
                    // Do not hold the lock across the 'await' point of the hotplug operation to avoid blocking.
                    let (
                        device_id,
                        path_on_host,
                        is_direct,
                        is_readonly,
                        discard_unmap,
                        serial_override,
                        driver_option,
                    ) = {
                        let device_locked = block_device.lock().await;
                        let device_id = device_locked.device_id.clone();
                        let cfg = &device_locked.config;
                        (
                            device_id,
                            cfg.path_on_host.clone(),
                            cfg.is_direct
                                .unwrap_or(self.config.blockdev_info.block_device_cache_direct),
                            cfg.is_readonly,
                            cfg.discard_unmap,
                            cfg.serial_override.clone(),
                            cfg.driver_option.clone(),
                        )
                    };
                    if path_on_host == self.config.boot_info.initrd {
                        // If this block device represents initrd we ignore it here, it
                        // will be handled elsewhere by adding `-initrd` to the qemu
                        // command line.
                        continue;
                    }
                    match driver_option.as_str() {
                        KATA_NVDIMM_DEV_TYPE => cmdline.add_nvdimm(
                            &path_on_host,
                            is_readonly,
                        )?,
                        KATA_CCW_DEV_TYPE | KATA_BLK_DEV_TYPE | KATA_SCSI_DEV_TYPE => {
                            let serial = if serial_override.is_empty() {
                                None
                            } else {
                                Some(serial_override.as_str())
                            };
                            cmdline.add_block_device(
                                device_id.as_str(),
                                &path_on_host,
                                is_direct,
                                driver_option.as_str() == KATA_SCSI_DEV_TYPE,
                                discard_unmap,
                                serial,
                            )?
                        }
                        unsupported => {
                            info!(sl!(), "unsupported block device driver: {}", unsupported)
                        }
                    }
                }
                DeviceType::Network(network) => {
                    // we need ensure add_network_device happens in netns.
                    let _netns_guard = NetnsGuard::new(&netns).context("new netns guard")?;

                    cmdline.add_network_device(
                        &network.config.host_dev_name,
                        network.config.guest_mac.clone().unwrap(),
                        network.config.queue_num.max(1) as u32,
                    )?;
                }
                DeviceType::Protection(prot_dev) => match &prot_dev.config {
                    ProtectionDeviceConfig::SevSnp(sev_snp_cfg) => {
                        if sev_snp_cfg.is_snp {
                            cmdline.add_sev_snp_protection_device(
                                sev_snp_cfg.cbitpos,
                                sev_snp_cfg.phys_addr_reduction,
                                &sev_snp_cfg.firmware,
                                &sev_snp_cfg.host_data,
                            )
                        } else {
                            cmdline.add_sev_protection_device(
                                sev_snp_cfg.cbitpos,
                                sev_snp_cfg.phys_addr_reduction,
                                &sev_snp_cfg.firmware,
                            )
                        }
                    }
                    ProtectionDeviceConfig::Se => cmdline.add_se_protection_device(),
                    ProtectionDeviceConfig::Tdx(tdx_config) => cmdline.add_tdx_protection_device(
                        &tdx_config.id,
                        &tdx_config.firmware,
                        tdx_config.qgs_port,
                        &tdx_config.mrconfigid,
                        tdx_config.debug,
                    ),
                },
                DeviceType::PortDevice(port_device) => {
                    let port_type = port_device.config.port_type;
                    let devices_per_port = port_device.port_devices.clone();

                    match port_type {
                        PCIePort::RootPort => cmdline.add_pcie_root_ports(devices_per_port)?,
                        PCIePort::SwitchPort => cmdline.add_pcie_switch_ports(devices_per_port)?,
                        _ => info!(sl!(), "no need to add {} ports", port_type),
                    }
                }
                DeviceType::VfioModern(vfio_dev) => {
                    // Snapshot parameters under the lock; release before doing cmdline work.
                    let (device_type, ap_sysfs_path, devices, bus_port_id) = {
                        let vfio_device = vfio_dev.lock().await;
                        let device_type = vfio_device.device.device_type.clone();
                        let ap_sysfs_path =
                            vfio_device.device.primary.sysfs_path.display().to_string();
                        // Legacy path: take devices from the IOMMU group.
                        // IOMMUFD path: iommu_group is None; fall back to
                        // VfioDevice::devices which holds the single cdev device.
                        let devices = vfio_device
                            .device
                            .iommu_group
                            .as_ref()
                            .map(|g| g.devices.clone())
                            .unwrap_or_else(|| vfio_device.device.devices.clone());
                        (
                            device_type,
                            ap_sysfs_path,
                            devices,
                            vfio_device.config.bus_port_id.clone(),
                        )
                    };

                    if device_type == VfioDeviceType::MediatedAp {
                        // s390x VFIO-AP: -device vfio-ap,sysfsdev=<path>
                        // No PCIe root port, no guest_pci_path.
                        cmdline.add_vfio_ap_device(&ap_sysfs_path)?;
                        info!(
                            sl!(),
                            "Completed VFIOModern AP coldplug for sysfsdev: {}", ap_sysfs_path
                        );
                    } else {
                        // PCI cold plug devices
                        for dev in devices.iter() {
                            let host_bdf = dev.addr.to_string();

                            let vfio_cfg = VfioDeviceConfig::new(
                                host_bdf,
                                bus_port_id.1 as u16,
                                bus_port_id.1 + 1,
                            )
                            .with_vfio_bus(bus_port_id.0.clone());

                            cmdline.add_pcie_vfio_device(vfio_cfg)?;
                        }

                        // Write back guest PCI path
                        let pci_path =
                            PciPath::try_from(format!("{:02x}/00", bus_port_id.1).as_str())?;
                        {
                            let mut vfio_device = vfio_dev.lock().await;
                            vfio_device.config.guest_pci_path = Some(pci_path.clone());
                        }
                        info!(
                            sl!(),
                            "Completed VFIOModern coldplug with returned guest pci path: {:?}",
                            pci_path
                        );
                    }
                }
                DeviceType::Vfio(vfio_dev) => {
                    // Cold-plug physical-endpoint VFs (non-IOMMUFD VFIO) onto
                    // pre-allocated PCIe root ports.  The bus assignment and
                    // guest PCI path were computed by do_add_pcie_endpoint()
                    // at device-registration time.
                    //
                    // We must emit BOTH:
                    //   1. the pcie-root-port for vfio_dev.bus (e.g. "rp1")
                    //   2. the vfio-pci device on that bus
                    //
                    // add_pcie_root_ports() skips allocated ports assuming
                    // VfioModern already emitted them.  For regular Vfio
                    // (physical endpoints) we have to emit the root port here.
                    let port_index = vfio_dev
                        .bus
                        .strip_prefix("rp")
                        .and_then(|s| s.parse::<u32>().ok())
                        .unwrap_or(0);
                    cmdline.add_physical_endpoint_root_port(&vfio_dev.bus, port_index);

                    for hostdev in &vfio_dev.devices {
                        let host_bdf = format!("{}:{}", hostdev.domain, hostdev.bus_slot_func);
                        let (vendor_id, device_id) =
                            hostdev
                                .device_vendor_class
                                .as_ref()
                                .map_or((None, None), |dvc| {
                                    let (dev, vendor) = dvc.get_device_vendor().unwrap_or((0, 0));
                                    let v = if vendor != 0 {
                                        Some(format!("0x{:04x}", vendor))
                                    } else {
                                        None
                                    };
                                    let d = if dev != 0 {
                                        Some(format!("0x{:04x}", dev))
                                    } else {
                                        None
                                    };
                                    (v, d)
                                });
                        cmdline.add_physical_vfio_device(
                            &host_bdf,
                            &hostdev.hostdev_id,
                            &vfio_dev.bus,
                            vendor_id.as_deref(),
                            device_id.as_deref(),
                        );
                        info!(
                            sl!(),
                            "cold-plug physical VFIO device: host={} id={} bus={} \
                             guest_pci_path={:?}",
                            host_bdf,
                            hostdev.hostdev_id,
                            vfio_dev.bus,
                            hostdev.guest_pci_path,
                        );
                    }
                }
                _ => info!(sl!(), "qemu cmdline: unsupported device: {:?}", device),
            }
        }
        // To get access to the VM console for debugging, enable the following
        // line and replace its argument appropriately (open a terminal, run
        // `tty` in it to get its device file path and use it as the argument).
        //cmdline.add_serial_console("/dev/pts/23");

        // Add a console to the devices of the cmdline
        let console_socket_path = Path::new(&self.get_jailer_root().await?).join("console.sock");
        cmdline.add_console(console_socket_path.to_str().unwrap());

        info!(sl!(), "qemu args: {}", cmdline.build().await?.join(" "));
        let mut command = Command::new(&self.config.path);
        command.args(cmdline.build().await?);

        info!(sl!(), "qemu cmd: {:?}", command);

        let user: Option<RootlessUser> = if is_rootless() {
            Some(
                self.config
                    .security_info
                    .rootless_user
                    .clone()
                    .ok_or_else(|| {
                        std::io::Error::other("rootless user must be specified for rootless qemu")
                    })?,
            )
        } else {
            None
        };

        // we need move the qemu process into Network Namespace and set SELinux label.
        unsafe {
            let selinux_label = self.config.security_info.selinux_label.clone();
            let _pre_exec = command.pre_exec(move || {
                let _ = enter_netns(&netns);
                if let Some(label) = selinux_label.as_ref() {
                    if let Err(e) = selinux::set_exec_label(label) {
                        error!(sl!(), "Failed to set SELinux label in child process: {}", e);
                        // Don't return error here to avoid breaking the process startup
                        // Log the error and continue
                    } else {
                        info!(
                            sl!(),
                            "Successfully set SELinux label in child process: {}", &label
                        );
                    }
                }
                if let Some(user) = &user {
                    let groups = user.groups.clone();
                    let gid = Gid::from_raw(user.gid);
                    let uid = Uid::from_raw(user.uid);

                    let _ = set_groups(&groups);
                    let _ = setgid(gid).context("setgid failed");
                    let _ = setuid(uid).context("setuid failed");
                }

                Ok(())
            });
        }

        let mut qemu_process = command.stderr(Stdio::piped()).spawn()?;
        let stderr = qemu_process.stderr.take().unwrap();
        self.qemu_process = Mutex::new(Some(qemu_process));

        info!(sl!(), "qemu process started");

        let exit_notify: mpsc::Sender<()> = self
            .exit_notify
            .take()
            .ok_or_else(|| anyhow!("no exit notify"))?;

        tokio::spawn(log_qemu_stderr(stderr, exit_notify));

        let qmp_socket_path = get_qmp_socket_path(self.id.as_str());

        match Qmp::new(&qmp_socket_path) {
            Ok(mut qmp) => {
                if let Some(subchannel) = cmdline.take_ccw_subchannel() {
                    qmp.set_ccw_subchannel(subchannel);
                }
                // Setup virtio-mem device if enabled.  It requires a memory
                // hotplug region (maxmem) on the QEMU command line; that region
                // is not reserved for every configuration (e.g. on s390x the
                // shared memory-backend used with a virtio-blk-ccw rootfs zeroes
                // maxmem/slots, and static resource management sizes the VM
                // upfront).  Skip the setup in that case -- like other static
                // sizing arches (e.g. arm64) the guest simply runs with its
                // boot memory -- instead of failing VM start with
                // "the configuration is not prepared for memory devices".
                if self.config.memory_info.enable_virtio_mem {
                    if cmdline.has_memory_hotplug_region() {
                        qmp.setup_virtio_mem(
                            self.config.memory_info.default_memory,
                            self.config.memory_info.default_maxmemory,
                            &self.config.machine_info.machine_type,
                            self.config.shared_fs.shared_fs.as_deref(),
                        )
                        .context("Failed to setup virtio-mem during VM initialization")?;
                    } else {
                        info!(
                            sl!(),
                            "virtio-mem enabled but no memory hotplug region (maxmem) was reserved; skipping virtio-mem setup"
                        );
                    }
                }
                let bridge_count = self.config.device_info.default_bridges;
                if bridge_count > 0 {
                    qmp.init_pci_bridges(bridge_count);
                }
                self.qmp = Some(qmp);
            }
            Err(e) => {
                error!(sl!(), "couldn't initialise QMP: {:?}", e);
                return Err(e);
            }
        }

        // Start the virtual machine by restoring it from a VM template if enabled.
        if self.config.vm_template.boot_from_template {
            self.boot_from_template()
                .await
                .context("boot from template")?;
            self.resume_vm().context("resume vm")?;
        }

        // When hypervisor debug is enabled, output the kernel boot messages for debugging.
        if self.config.debug_info.enable_debug {
            let stream = UnixStream::connect(console_socket_path.as_os_str()).await?;
            tokio::spawn(log_qemu_console(stream));
        }

        Ok(())
    }

    async fn boot_from_template(&mut self) -> Result<()> {
        let qmp = self
            .qmp
            .as_mut()
            .context("failed to get QMP connection for boot from template")?;

        qmp.set_ignore_shared_memory_capability()
            .context("failed to set ignore shared memory capability")?;

        let uri = format!("exec:cat {}", self.config.vm_template.device_state_path);

        qmp.execute_migration_incoming(&uri)
            .context("failed to execute migration incoming")?;

        self.wait_for_migration()
            .await
            .context("failed to wait for migration")?;

        info!(sl!(), "migration complete");

        Ok(())
    }

    pub async fn wait_for_migration(&mut self) -> Result<()> {
        // Ensure QMP is connected.
        if self.qmp.is_none() {
            return Err(anyhow!("QMP is not connected"));
        }

        let qmp = self
            .qmp
            .as_mut()
            .context("failed to get QMP connection for boot from template")?;

        // Helper to migrate_completed migration state from `query-migrate`.
        let migrate_completed = |st: Option<MigrationStatus>| -> Result<bool> {
            match st {
                Some(MigrationStatus::completed) => Ok(true), // done
                Some(MigrationStatus::failed) | Some(MigrationStatus::cancelled) => {
                    Err(anyhow!("migration ended early: {:?}", st))
                }
                _ => Ok(false), // still running / unknown
            }
        };

        // If already finished, just return Ok(()).
        let mi = qmp.execute_query_migrate().await?;
        match migrate_completed(mi.status) {
            Ok(true) => return Ok(()),
            Ok(false) => {
                info!(sl!(), "migration not yet completed, entering wait loop");
            }
            Err(e) => return Err(e),
        }

        // Overall timeout for migration.
        // Regarding why the timeout is set to 280ms and whether it should be adjusted, we need more empirical data.
        // For now, we will keep using the previous configuration.
        let timeout = Duration::from_millis(280);

        // Polling interval: start small, then back off to reduce load.
        let poll_interval = Duration::from_millis(20);

        // Deadline with a timeout.
        let deadline = Instant::now()
            .checked_add(timeout)
            .ok_or_else(|| anyhow!("timeout overflow"))?;

        loop {
            // Query migration status via QMP.
            let mi = qmp.execute_query_migrate().await?;
            match migrate_completed(mi.status) {
                Ok(true) => return Ok(()),
                Ok(false) => {
                    info!(sl!(), "migration still not completed, continuing wait loop");
                }
                Err(e) => return Err(e),
            }

            // Stop waiting once we hit the timeout.
            let now = Instant::now();
            if now >= deadline {
                return Err(anyhow!("wait_for_migration timeout after {:?}", timeout));
            }

            // Sleep until next tick, but never beyond deadline
            sleep(poll_interval.min(deadline - now)).await;
        }
    }

    pub(crate) async fn stop_vm(&mut self) -> Result<()> {
        info!(sl!(), "Stopping QEMU VM");

        let mut qemu_process = self.qemu_process.lock().await;
        if let Some(qemu_process) = qemu_process.as_mut() {
            let is_qemu_running = qemu_process.id().is_some();
            if is_qemu_running {
                info!(sl!(), "QemuInner::stop_vm(): kill()'ing qemu");
                qemu_process.kill().await.map_err(anyhow::Error::from)
            } else {
                info!(
                    sl!(),
                    "QemuInner::stop_vm(): qemu process isn't running (likely stopped already)"
                );
                Ok(())
            }
        } else {
            Err(anyhow!("qemu process has not been started yet"))
        }
    }

    pub(crate) async fn wait_vm(&self) -> Result<i32> {
        let mut qemu_process = self.qemu_process.lock().await;

        if let Some(mut qemu_process) = qemu_process.take() {
            let status = qemu_process.wait().await?;
            Ok(status.code().unwrap_or(0))
        } else {
            Err(anyhow!("the process has been reaped"))
        }
    }

    pub(crate) fn pause_vm(&mut self) -> Result<()> {
        let qmp = self.qmp.as_mut().ok_or(anyhow!("qmp not initialized"))?;
        qmp.qmp_stop().context("pause vm")
    }

    pub(crate) fn resume_vm(&mut self) -> Result<()> {
        let qmp = self.qmp.as_mut().ok_or(anyhow!("qmp not initialized"))?;
        qmp.qmp_cont().context("resume vm")
    }

    pub(crate) async fn save_vm(&mut self) -> Result<()> {
        let qmp = self.qmp.as_mut().ok_or(anyhow!("QMP not initialized"))?;

        if self.config.vm_template.boot_to_be_template {
            qmp.set_ignore_shared_memory_capability()
                .context("failed to set ignore shared memory capability")?;
        }

        let uri = format!("exec:cat >{}", self.config.vm_template.device_state_path);

        qmp.execute_migration(&uri)
            .context("failed to execute migration")?;

        self.wait_for_migration()
            .await
            .context("failed to wait for migration")?;

        info!(sl!(), "migration finished successfully");

        Ok(())
    }

    pub(crate) async fn get_agent_socket(&self) -> Result<String> {
        info!(sl!(), "QemuInner::get_agent_socket()");
        let guest_cid = match &self.get_agent_vsock_dev() {
            Some(device) => device.config.guest_cid,
            None => return Err(anyhow!("uninitialized agent vsock".to_owned())),
        };

        Ok(format!("{VSOCK_SCHEME}://{guest_cid}"))
    }

    pub(crate) async fn disconnect(&mut self) {
        info!(sl!(), "QemuInner::disconnect()");
        todo!()
    }

    pub(crate) async fn get_thread_ids(&mut self) -> Result<VcpuThreadIds> {
        info!(sl!(), "QemuInner::get_thread_ids()");

        Ok(self
            .qmp
            .as_mut()
            .and_then(|qmp| qmp.get_vcpu_thread_ids().ok())
            .unwrap_or_default())
    }

    pub(crate) async fn get_vmm_master_tid(&self) -> Result<u32> {
        info!(sl!(), "QemuInner::get_vmm_master_tid()");
        if let Some(qemu_process) = self.qemu_process.lock().await.as_ref() {
            if let Some(qemu_pid) = qemu_process.id() {
                info!(
                    sl!(),
                    "QemuInner::get_vmm_master_tid(): returning {}", qemu_pid
                );
                Ok(qemu_pid)
            } else {
                Err(anyhow!("QemuInner::get_vmm_master_tid(): qemu process isn't running (likely stopped already)"))
            }
        } else {
            Err(anyhow!("qemu process not running"))
        }
    }

    pub(crate) async fn get_ns_path(&self) -> Result<String> {
        info!(sl!(), "QemuInner::get_ns_path()");
        Ok(format!(
            "/proc/{}/task/{}/ns",
            std::process::id(),
            std::process::id()
        ))
    }

    pub(crate) async fn cleanup(&self) -> Result<()> {
        info!(sl!(), "QemuInner::cleanup()");
        let vm_path = [build_path(KATA_PATH).as_str(), self.id.as_str()].join("/");
        vm_cleanup(&self.config, vm_path.as_str())
    }

    pub(crate) async fn resize_vcpu(
        &mut self,
        old_vcpus: u32,
        mut new_vcpus: u32,
    ) -> Result<(u32, u32)> {
        info!(
            sl!(),
            "QemuInner::resize_vcpu(): {} -> {}", old_vcpus, new_vcpus
        );

        // TODO The following sanity checks apparently have to be performed by
        // any hypervisor - wouldn't it make sense to move them to the caller?
        if new_vcpus == old_vcpus {
            return Ok((old_vcpus, new_vcpus));
        }

        if new_vcpus == 0 {
            return Err(anyhow!("resize to 0 vcpus requested"));
        }

        if new_vcpus > self.config.cpu_info.default_maxvcpus {
            warn!(
                sl!(),
                "Cannot allocate more vcpus than the max allowed number of vcpus. The maximum allowed amount of vcpus will be used instead.");
            new_vcpus = self.config.cpu_info.default_maxvcpus;
        }

        if let Some(ref mut qmp) = self.qmp {
            match new_vcpus.cmp(&old_vcpus) {
                Ordering::Greater => {
                    let hotplugged = qmp.hotplug_vcpus(new_vcpus - old_vcpus)?;
                    new_vcpus = old_vcpus + hotplugged;
                }
                Ordering::Less => {
                    let hotunplugged = qmp.hotunplug_vcpus(old_vcpus - new_vcpus)?;
                    new_vcpus = old_vcpus - hotunplugged;
                }
                Ordering::Equal => {}
            }
        }

        Ok((old_vcpus, new_vcpus))
    }

    pub(crate) async fn get_pids(&self) -> Result<Vec<u32>> {
        info!(sl!(), "QemuInner::get_pids()");
        todo!()
    }

    pub(crate) async fn check(&self) -> Result<()> {
        todo!()
    }

    pub(crate) async fn get_jailer_root(&self) -> Result<String> {
        let root_path = get_jailer_root(self.id.as_str());
        create_dir_all_with_inherit_owner(&root_path, 0o750)?;
        Ok(root_path)
    }

    pub(crate) async fn capabilities(&self) -> Result<Capabilities> {
        let mut caps = Capabilities::default();

        // Confidential Guest doesn't permit virtio-fs.
        let flags = if self.hypervisor_config().security_info.confidential_guest
            || self.hypervisor_config().shared_fs.shared_fs.is_none()
        {
            CapabilityBits::BlockDeviceSupport
                | CapabilityBits::BlockDeviceHotplugSupport
                | CapabilityBits::BlockDeviceDiscardSupport
        } else {
            CapabilityBits::BlockDeviceSupport
                | CapabilityBits::BlockDeviceHotplugSupport
                | CapabilityBits::BlockDeviceDiscardSupport
                | CapabilityBits::FsSharingSupport
        };
        caps.set(flags);

        Ok(caps)
    }

    pub fn set_hypervisor_config(&mut self, config: HypervisorConfig) {
        self.config = config;
    }

    pub fn hypervisor_config(&self) -> HypervisorConfig {
        info!(sl!(), "QemuInner::hypervisor_config()");
        self.config.clone()
    }

    pub(crate) async fn get_hypervisor_metrics(&self) -> Result<String> {
        todo!()
    }

    pub(crate) fn set_capabilities(&mut self, flag: CapabilityBits) {
        let mut caps = Capabilities::default();

        caps.set(flag)
    }

    pub(crate) fn set_guest_memory_block_size(&mut self, size: u32) {
        if let Some(ref mut qmp) = self.qmp {
            info!(
                sl!(),
                "QemuInner::set_guest_memory_block_size(): block size set to {}", size
            );
            qmp.set_guest_memory_block_size(size.into());
        } else {
            warn!(
                sl!(),
                "QemuInner::set_guest_memory_block_size(): QMP not initialized"
            );
        }
    }

    pub(crate) fn guest_memory_block_size(&self) -> u32 {
        if let Some(qmp) = &self.qmp {
            qmp.guest_memory_block_size() as u32
        } else {
            warn!(
                sl!(),
                "QemuInner::guest_memory_block_size(): QMP not initialized"
            );
            0
        }
    }

    pub(crate) fn resize_memory(
        &mut self,
        mut new_total_mem_mb: u32,
    ) -> Result<(u32, MemoryConfig)> {
        info!(
            sl!(),
            "QemuInner::resize_memory(): asked to resize memory to {} MB", new_total_mem_mb
        );

        let qmp = match self.qmp {
            Some(ref mut qmp) => qmp,
            None => {
                warn!(sl!(), "QemuInner::resize_memory(): QMP not initialized");
                return Err(anyhow!("QMP not initialized"));
            }
        };

        let coldplugged_mem_mb = self.config.memory_info.default_memory;
        let coldplugged_mem = megs_to_bytes(coldplugged_mem_mb);
        let new_total_mem = megs_to_bytes(new_total_mem_mb);

        if new_total_mem < coldplugged_mem {
            warn!(sl!(), "asked to resize to {} M but that is less than cold-plugged memory size ({}), nothing to do",new_total_mem_mb,
                bytes_to_megs(coldplugged_mem));

            return Ok((coldplugged_mem_mb, MemoryConfig::default()));
        }

        let guest_mem_block_size = qmp.guest_memory_block_size();

        let mut new_hotplugged_mem = new_total_mem - coldplugged_mem;

        info!(
            sl!(),
            "new hotplugged mem before alignment: {} B ({} MB)",
            new_hotplugged_mem,
            bytes_to_megs(new_hotplugged_mem)
        );

        let is_unaligned = !new_hotplugged_mem.is_multiple_of(guest_mem_block_size);
        if is_unaligned {
            new_hotplugged_mem = ch_config::convert::checked_next_multiple_of(
                new_hotplugged_mem,
                guest_mem_block_size,
            )
            .ok_or(anyhow!(format!(
                "alignment of {} B to the block size of {} B failed",
                new_hotplugged_mem, guest_mem_block_size
            )))?
        }
        let new_hotplugged_mem = new_hotplugged_mem;

        info!(
            sl!(),
            "new hotplugged mem after alignment: {} B ({} MB)",
            new_hotplugged_mem,
            bytes_to_megs(new_hotplugged_mem)
        );

        let max_total_mem = megs_to_bytes(self.config.memory_info.default_maxmemory);
        if coldplugged_mem + new_hotplugged_mem > max_total_mem {
            return Err(anyhow!(
                "requested memory ({} M) is greater than maximum allowed ({} M)",
                bytes_to_megs(coldplugged_mem + new_hotplugged_mem),
                bytes_to_megs(max_total_mem)
            ));
        }

        let cur_hotplugged_memory = qmp.hotplugged_memory_size()?;
        info!(
            sl!(),
            "hotplug memory {} -> {}", cur_hotplugged_memory, new_hotplugged_mem
        );

        match new_hotplugged_mem.cmp(&cur_hotplugged_memory) {
            Ordering::Greater => {
                info!(
                    sl!(),
                    "hotplugging {} B of memory",
                    new_hotplugged_mem - cur_hotplugged_memory
                );
                qmp.hotplug_memory(new_hotplugged_mem - cur_hotplugged_memory)
                    .context("qemu hotplug memory")?;
                info!(
                    sl!(),
                    "hotplugged memory after hotplugging: {}",
                    qmp.hotplugged_memory_size()?
                );

                new_total_mem_mb = bytes_to_megs(coldplugged_mem + new_hotplugged_mem);
            }
            Ordering::Less => {
                info!(
                    sl!(),
                    "hotunplugging {} B of memory",
                    cur_hotplugged_memory - new_hotplugged_mem
                );
                let res =
                    qmp.hotunplug_memory((cur_hotplugged_memory - new_hotplugged_mem).try_into()?);
                if let Err(err) = res {
                    info!(sl!(), "hotunplugging failed: {:?}", err);
                } else {
                    new_total_mem_mb = bytes_to_megs(coldplugged_mem + new_hotplugged_mem);
                }
                info!(
                    sl!(),
                    "hotplugged memory after hotunplugging: {}",
                    qmp.hotplugged_memory_size()?
                );
            }
            Ordering::Equal => info!(
                sl!(),
                "VM already has the requested amount of memory, nothing to do"
            ),
        }

        Ok((new_total_mem_mb, MemoryConfig::default()))
    }
}

async fn log_qemu_console(console: UnixStream) -> Result<()> {
    info!(sl!(), "starting reading qemu console");

    let stderr_reader = BufReader::new(console);
    let mut stderr_lines = stderr_reader.lines();

    while let Some(buffer) = stderr_lines
        .next_line()
        .await
        .context("next_line() failed on qemu console")?
    {
        info!(sl!(), "vm console: {:?}", buffer);
    }

    info!(sl!(), "finished reading qemu console");
    Ok(())
}

async fn log_qemu_stderr(stderr: ChildStderr, exit_notify: mpsc::Sender<()>) -> Result<()> {
    info!(sl!(), "starting reading qemu stderr");

    let stderr_reader = BufReader::new(stderr);
    let mut stderr_lines = stderr_reader.lines();

    while let Some(buffer) = stderr_lines
        .next_line()
        .await
        .context("next_line() failed on qemu stderr")?
    {
        info!(sl!(), "qemu stderr: {:?}", buffer);
    }

    // Notfiy the waiter the process exit.
    let _ = exit_notify.try_send(());

    info!(sl!(), "finished reading qemu stderr");
    Ok(())
}

use crate::device::DeviceType;
use crate::vfio_device::VfioDeviceType;

// device manager part of Hypervisor
impl QemuInner {
    pub(crate) async fn add_device(&mut self, mut device: DeviceType) -> Result<DeviceType> {
        info!(sl!(), "QemuInner::add_device() {}", device);
        let is_qemu_ready_to_hotplug = self.qmp.is_some();
        if is_qemu_ready_to_hotplug {
            // hypervisor is running already
            device = self.hotplug_device(device).await?;
        } else {
            // store the device to coldplug it later, on hypervisor launch
            self.devices.push(device.clone());
        }
        Ok(device)
    }

    pub(crate) async fn remove_device(&mut self, device: DeviceType) -> Result<()> {
        info!(sl!(), "QemuInner::remove_device() {} ", device);
        self.hotunplug_device(&device).await?;

        self.devices.retain(|d| match (d, &device) {
            (DeviceType::BlockModern(a), DeviceType::BlockModern(b)) => {
                !std::sync::Arc::ptr_eq(a, b)
            }
            _ => true,
        });

        Ok(())
    }

    async fn hotunplug_device(&mut self, device: &DeviceType) -> Result<()> {
        let qmp = match self.qmp {
            Some(ref mut qmp) => qmp,
            None => return Err(anyhow!("QMP not initialized")),
        };

        match device {
            DeviceType::BlockModern(ref block_device) => {
                let (index, driver) = {
                    let cfg = &block_device.lock().await.config;
                    (
                        cfg.index,
                        self.config.blockdev_info.block_device_driver.clone(),
                    )
                };
                qmp.hotunplug_block_device(&driver, index)
                    .context("hotunplug block device")?;
            }
            DeviceType::Network(_)
            | DeviceType::Vfio(_)
            | DeviceType::VfioModern(_)
            | DeviceType::VhostUserBlk(_)
            | DeviceType::VhostUserNetwork(_)
            | DeviceType::ShareFs(_)
            | DeviceType::HybridVsock(_)
            | DeviceType::Vsock(_)
            | DeviceType::Protection(_)
            | DeviceType::PortDevice(_) => {
                return Err(anyhow!("hotunplug for {} is currently unsupported", device));
            }
        }
        Ok(())
    }

    async fn hotplug_device(&mut self, device: DeviceType) -> Result<DeviceType> {
        let qmp = match self.qmp {
            Some(ref mut qmp) => qmp,
            None => return Err(anyhow!("QMP not initialized")),
        };

        match device {
            DeviceType::Network(mut network_device) => {
                let (netdev, virtio_net_device) = get_network_device(
                    &self.config,
                    &network_device.config.host_dev_name,
                    network_device.config.guest_mac.clone().unwrap(),
                    &mut None,
                    network_device.config.queue_num.max(1) as u32,
                )?;
                qmp.hotplug_network_device(&netdev, &virtio_net_device)?;

                // On s390x the NIC is plugged onto the virtual channel subsystem
                // (virtio-net-ccw) and has no PCI address, so the PCI-path lookup
                // below does not apply.  The agent matches the CCW interface via
                // its uevent instead; forcing a PCI-path resolution here would
                // fail and abort the whole NIC attach (breaking guest networking).
                if !uses_native_ccw_bus() {
                    let frontend_id = format!("frontend-{}", virtio_net_device.get_netdev_id());
                    let pci_path = qmp
                        .get_device_by_qdev_id(&frontend_id)
                        .context("get network device pci path")?;
                    network_device.config.pci_path = Some(pci_path);
                }

                return Ok(DeviceType::Network(network_device));
            }
            DeviceType::Vfio(mut vfiodev) => {
                // FIXME: the first one might not the true device we want to passthrough.
                // The `multifunction=on` is temporarily unsupported.
                // Tracking issue #11292 has been created to monitor progress towards full multifunction support.
                let primary_device = vfiodev.devices.first_mut().unwrap();
                info!(
                    sl!(),
                    "qmp hotplug vfio primary_device {:?}", &primary_device
                );

                primary_device.guest_pci_path = qmp.hotplug_vfio_device(
                    &primary_device.hostdev_id,
                    &primary_device.sysfs_path,
                    &primary_device.bus_slot_func,
                    &vfiodev.driver_type,
                    &vfiodev.bus,
                )?;

                return Ok(DeviceType::Vfio(vfiodev));
            }
            DeviceType::BlockModern(ref block_device) => {
                info!(sl!(), "Starting QMP hotplug for BlockModern device");

                // First, snapshot parameters within the lock.
                // Do not hold the lock across the 'await' point of the hotplug operation to avoid blocking.
                let (
                    index,
                    path_on_host,
                    aio,
                    is_direct,
                    is_readonly,
                    no_drop,
                    block_format,
                    discard_unmap,
                    driver,
                    logical_sector_size,
                    physical_sector_size,
                ) = {
                    let cfg = &block_device.lock().await.config;
                    (
                        cfg.index,
                        cfg.path_on_host.clone(),
                        cfg.blkdev_aio.to_string(),
                        Some(
                            cfg.is_direct
                                .unwrap_or(self.config.blockdev_info.block_device_cache_direct),
                        ),
                        cfg.is_readonly,
                        cfg.no_drop,
                        cfg.format.clone(),
                        cfg.discard_unmap,
                        self.config.blockdev_info.block_device_driver.clone(),
                        cfg.logical_sector_size,
                        cfg.physical_sector_size,
                    )
                };

                // Determine iothread for hotplugged virtio-blk-pci devices.
                // Only attach iothread when:
                // 1. enable_iothreads is true
                // 2. indep_iothreads > 0
                // 3. block driver is virtio-blk-pci
                // 4. TODO: for more complex cases
                let iothread = if self.config.enable_iothreads
                    && self.config.indep_iothreads > 0
                    && driver == VIRTIO_BLK_PCI
                {
                    // Use the first independent iothread (indep_iothread_0)
                    Some("indep_iothread_0")
                } else {
                    None
                };

                // Second, execute the asynchronous hotplug without holding the lock.
                let (pci_path, addr_str) = qmp
                    .hotplug_block_device(
                        &driver,
                        index,
                        &path_on_host,
                        &aio,
                        is_direct,
                        is_readonly,
                        no_drop,
                        discard_unmap,
                        logical_sector_size,
                        physical_sector_size,
                        &block_format,
                        iothread,
                    )
                    .context("hotplug block device")?;

                // Third, re-acquire the lock to write back results.
                {
                    let mut dev = block_device.lock().await;
                    let cfg = &mut dev.config;
                    if let Some(p) = pci_path {
                        cfg.pci_path = Some(p);
                    }
                    if let Some(addr) = addr_str {
                        if driver == VIRTIO_BLK_CCW {
                            cfg.ccw_addr = Some(addr);
                        } else {
                            cfg.scsi_addr = Some(addr);
                        }
                    }
                    info!(sl!(), "Completed BlockModern hotplug: {:?}", &cfg);
                }
                return Ok(DeviceType::BlockModern(block_device.clone()));
            }
            DeviceType::VfioModern(ref vfiodev) => {
                // Snapshot VFIO parameters inside the lock.
                let (hostdev_id, device_type, sysfs_path, address, driver_type, bus) = {
                    let vfio_device = vfiodev.lock().await;
                    let hostdev_id = vfio_device.device_id.clone();
                    let device = &vfio_device.device;
                    let device_type = device.device_type.clone();

                    let sysfs_path = device.primary.sysfs_path.display().to_string();

                    // For AP devices there is no IOMMU group or BDF; use empty strings.
                    let (address, driver_type, bus) = if device_type == VfioDeviceType::MediatedAp {
                        (String::new(), "vfio-ap".to_string(), String::new())
                    } else {
                        // Legacy path: primary comes from the IOMMU group.
                        // IOMMUFD path: iommu_group is None; use VfioDevice::primary
                        // which is the single device returned by discover_vfio_device.
                        // FIXME: multi-function group support tracked in #11292.
                        let primary_device = device
                            .iommu_group
                            .as_ref()
                            .map(|g| g.primary.clone())
                            .unwrap_or_else(|| device.primary.clone());

                        info!(
                            sl!(),
                            "QMP hotplug VFIO primary_device address: {:?}", &primary_device.addr
                        );

                        let driver_type = primary_device
                            .driver
                            .clone()
                            .ok_or_else(|| anyhow!("Driver type missing for primary device"))?;
                        let address = format!("{}", primary_device.addr);
                        let bus = vfio_device.config.bus_port_id.0.clone();
                        (address, driver_type, bus)
                    };

                    (
                        hostdev_id,
                        device_type,
                        sysfs_path,
                        address,
                        driver_type,
                        bus,
                    )
                };

                // Execute hotplug outside the lock.
                let guest_pci_path = qmp.hotplug_vfio_device(
                    &hostdev_id,
                    &sysfs_path,
                    &address,
                    &driver_type,
                    &bus,
                )?;

                // Write the resulting Guest PCI Path back within the lock (PCI only).
                {
                    let mut vfio_device = vfiodev.lock().await;
                    if device_type != VfioDeviceType::MediatedAp {
                        if let Some(p) = guest_pci_path {
                            vfio_device.config.guest_pci_path = Some(p);
                        }
                    }
                    info!(
                        sl!(),
                        "Completed VFIOModern hotplug for device ID: {}", hostdev_id
                    );
                }
            }
            _ => info!(
                sl!(),
                "Hotplugging for {:#?} is currently unsupported", device
            ),
        }
        Ok(device)
    }
}

// private helpers
impl QemuInner {
    fn get_agent_vsock_dev(&self) -> Option<&VsockDevice> {
        self.devices.iter().find_map(|dev| {
            if let DeviceType::Vsock(vsock_dev) = dev {
                Some(vsock_dev)
            } else {
                None
            }
        })
    }

    pub(crate) async fn update_device(&mut self, device: DeviceType) -> Result<()> {
        info!(sl!(), "QemuInner::update_device() {:?}", &device);

        Ok(())
    }

    /// Resolve the in-guest PCIe path for a cold-plugged physical-endpoint VF
    /// via QMP query-pci. Must be called after the VM has started and QMP is
    /// initialised. This is the runtime-rs pair of the Go runtime's
    /// `ResolveColdPlugVFIOGuestPciPaths` / `qomGetPciPath` call.
    pub(crate) fn resolve_vfio_device_pci_path(&mut self, hostdev_id: &str) -> Result<PciPath> {
        let qmp = self.qmp.as_mut().ok_or_else(|| {
            anyhow!(
                "QMP not initialised; cannot resolve PCI path for {}",
                hostdev_id
            )
        })?;
        qmp.get_device_by_qdev_id(hostdev_id)
    }
}

#[async_trait]
impl Persist for QemuInner {
    type State = HypervisorState;
    type ConstructorArgs = mpsc::Sender<()>;

    /// Save a state of hypervisor
    async fn save(&self) -> Result<Self::State> {
        Ok(HypervisorState {
            hypervisor_type: HYPERVISOR_QEMU.to_string(),
            id: self.id.clone(),
            config: self.hypervisor_config(),
            ..Default::default()
        })
    }

    /// Restore hypervisor
    async fn restore(exit_notify: mpsc::Sender<()>, hypervisor_state: Self::State) -> Result<Self> {
        Ok(QemuInner {
            id: hypervisor_state.id,
            qemu_process: Mutex::new(None),
            qmp: None,
            config: hypervisor_state.config,
            devices: Vec::new(),
            netns: None,

            exit_notify: Some(exit_notify),
        })
    }
}
