pact_plugin_driver/

plugin_manager.rs

//! Manages interactions with Pact plugins
use std::collections::HashMap;
use std::env;
use std::fs;
use std::fs::File;
use std::io::{BufReader, Write};
use std::path::PathBuf;
use std::str::FromStr;
use std::str::from_utf8;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::thread;

use anyhow::{Context, anyhow, bail};
use bytes::Bytes;
use itertools::Either;
use lazy_static::lazy_static;
use maplit::hashmap;
use pact_models::PactSpecification;
use pact_models::bodies::OptionalBody;
use pact_models::json_utils::json_to_string;
use pact_models::prelude::v4::V4Pact;
use pact_models::prelude::{ContentType, Pact};
use pact_models::v4::interaction::V4Interaction;
use reqwest::Client;
use semver::Version;
use serde_json::Value;
use tracing::{debug, info, trace, warn};

use crate::catalogue_manager::{
  CatalogueEntry, all_entries, core_entries, register_plugin_entries, remove_plugin_entries,
};
use crate::content::ContentMismatch;
use crate::download::{download_json_from_github, download_plugin_executable, fetch_json_from_url};
use crate::grpc_plugin::{GrpcPactPlugin, start_plugin_process};
use crate::metrics::send_metrics;
use crate::mock_server::{MockServerConfig, MockServerDetails, MockServerResults};
use crate::plugin_models::{
  PactPlugin, PactPluginManifest, PactPluginRpc, PluginDependency, PluginInitRequest,
  PluginInstance, PluginInterfaceVersion,
};
use crate::proto::*;
use crate::proto_v2;
use crate::repository::{USER_AGENT, fetch_repository_index};
use crate::utils::{
  optional_string, proto_value_to_json, to_proto_struct, to_proto_value, versions_compatible,
};
use crate::verification::{InteractionVerificationData, InteractionVerificationResult};

#[derive(Debug, Clone)]
struct RegisteredPlugin {
  /// The running plugin instance, accessed via the PluginInstance trait.
  instance: Arc<dyn PluginInstance + Send + Sync>,
  /// Kept so load_plugin can return PactPlugin without breaking external callers.
  plugin: PactPlugin,
  access_count: Arc<AtomicUsize>,
}

impl RegisteredPlugin {
  fn new(instance: Arc<dyn PluginInstance + Send + Sync>, plugin: PactPlugin) -> Self {
    RegisteredPlugin {
      instance,
      plugin,
      access_count: Arc::new(AtomicUsize::new(1)),
    }
  }

  fn update_access(&self) {
    let count = self.access_count.fetch_add(1, Ordering::SeqCst);
    trace!(
      "update_access: Plugin {}/{} access is now {}",
      self.plugin.manifest.name,
      self.plugin.manifest.version,
      count + 1
    );
  }

  fn drop_access(&self) -> usize {
    let check = self
      .access_count
      .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |count| {
        if count > 0 { Some(count - 1) } else { None }
      });
    let count = if let Ok(v) = check {
      if v > 0 { v - 1 } else { v }
    } else {
      0
    };
    trace!(
      "drop_access: Plugin {}/{} access is now {}",
      self.plugin.manifest.name, self.plugin.manifest.version, count
    );
    count
  }
}

lazy_static! {
  static ref PLUGIN_MANIFEST_REGISTER: Mutex<HashMap<String, PactPluginManifest>> =
    Mutex::new(HashMap::new());
  static ref PLUGIN_REGISTER: Mutex<HashMap<String, RegisteredPlugin>> = Mutex::new(HashMap::new());
  /// Maps plugin_instance_id → plugin_name so the PluginHost Log RPC handler can
  /// attach the plugin name to forwarded log entries without a proto field for it.
  static ref INSTANCE_NAMES: Mutex<HashMap<String, String>> = Mutex::new(HashMap::new());
}

pub(crate) fn plugin_name_for_instance(instance_id: &str) -> Option<String> {
  INSTANCE_NAMES.lock().unwrap().get(instance_id).cloned()
}

pub(crate) fn register_plugin_instance(instance_id: &str, plugin_name: &str) {
  INSTANCE_NAMES.lock().unwrap().insert(instance_id.to_string(), plugin_name.to_string());
}

pub(crate) fn deregister_plugin_instance(instance_id: &str) {
  INSTANCE_NAMES.lock().unwrap().remove(instance_id);
}

fn host_capabilities() -> Vec<String> {
  core_entries()
    .into_iter()
    .map(|entry| format!("{}/{}", entry.entry_type, entry.key))
    .collect()
}

/// Load the plugin defined by the dependency information. Will first look in the global
/// plugin registry.
pub async fn load_plugin(plugin: &PluginDependency) -> anyhow::Result<PactPlugin> {
  let thread_id = thread::current().id();
  debug!("Loading plugin {:?}", plugin);
  trace!(
    "Rust plugin driver version {}",
    option_env!("CARGO_PKG_VERSION").unwrap_or_default()
  );
  trace!(
    "load_plugin {:?}: Waiting on PLUGIN_REGISTER lock",
    thread_id
  );
  let mut inner = PLUGIN_REGISTER.lock().unwrap();
  trace!("load_plugin {:?}: Got PLUGIN_REGISTER lock", thread_id);
  let result = match lookup_plugin_inner(plugin, &inner) {
    Some(entry) => {
      debug!("Found running plugin {:?}", entry.plugin.manifest);
      entry.update_access();
      Ok(entry.plugin.clone())
    }
    None => {
      debug!("Did not find plugin, will attempt to start it");
      let manifest = match load_plugin_manifest(plugin) {
        Ok(manifest) => manifest,
        Err(err) => {
          warn!(
            "Could not load plugin manifest from disk, will try auto install it: {}",
            err
          );
          let http_client = reqwest::ClientBuilder::new()
            .user_agent(USER_AGENT)
            .build()?;
          let index = fetch_repository_index(&http_client, None).await?;
          match index.lookup_plugin_version(&plugin.name, &plugin.version) {
            Some(entry) => {
              info!("Found an entry for the plugin in the plugin index, will try install that");
              install_plugin_from_url(&http_client, entry.source.value().as_str()).await?
            }
            None => Err(err)?,
          }
        }
      };
      send_metrics(&manifest);
      initialise_plugin(&manifest, &mut inner).await
    }
  };
  trace!(
    "load_plugin {:?}: Releasing PLUGIN_REGISTER lock",
    thread_id
  );
  result
}

fn lookup_plugin_inner<'a>(
  plugin: &PluginDependency,
  plugin_register: &'a HashMap<String, RegisteredPlugin>,
) -> Option<&'a RegisteredPlugin> {
  if let Some(version) = &plugin.version {
    plugin_register.get(format!("{}/{}", plugin.name, version).as_str())
  } else {
    plugin_register
      .iter()
      .filter(|(_, value)| value.plugin.manifest.name == plugin.name)
      .max_by(|(_, v1), (_, v2)| v1.plugin.manifest.version.cmp(&v2.plugin.manifest.version))
      .map(|(_, entry)| entry)
  }
}

/// Look up the plugin in the global plugin register
pub fn lookup_plugin(plugin: &PluginDependency) -> Option<Arc<dyn PluginInstance + Send + Sync>> {
  let thread_id = thread::current().id();
  trace!(
    "lookup_plugin {:?}: Waiting on PLUGIN_REGISTER lock",
    thread_id
  );
  let inner = PLUGIN_REGISTER.lock().unwrap();
  trace!("lookup_plugin {:?}: Got PLUGIN_REGISTER lock", thread_id);
  let entry = lookup_plugin_inner(plugin, &inner);
  trace!(
    "lookup_plugin {:?}: Releasing PLUGIN_REGISTER lock",
    thread_id
  );
  entry.map(|e| e.instance.clone())
}

/// Return the plugin manifest for the given plugin. Will first look in the global plugin manifest
/// registry.
pub fn load_plugin_manifest(plugin_dep: &PluginDependency) -> anyhow::Result<PactPluginManifest> {
  debug!("Loading plugin manifest for plugin {:?}", plugin_dep);
  match lookup_plugin_manifest(plugin_dep) {
    Some(manifest) => Ok(manifest),
    None => load_manifest_from_disk(plugin_dep),
  }
}

fn load_manifest_from_disk(plugin_dep: &PluginDependency) -> anyhow::Result<PactPluginManifest> {
  let plugin_dir = pact_plugin_dir()?;
  debug!("Looking for plugin in {:?}", plugin_dir);

  if plugin_dir.exists() {
    load_manifest_from_dir(plugin_dep, &plugin_dir)
  } else {
    Err(anyhow!("Plugin directory {:?} does not exist", plugin_dir))
  }
}

fn load_manifest_from_dir(
  plugin_dep: &PluginDependency,
  plugin_dir: &PathBuf,
) -> anyhow::Result<PactPluginManifest> {
  let mut manifests = vec![];
  for entry in fs::read_dir(plugin_dir)? {
    let path = entry?.path();
    trace!("Found: {:?}", path);

    if path.is_dir() {
      let manifest_file = path.join("pact-plugin.json");
      if manifest_file.exists() && manifest_file.is_file() {
        debug!("Found plugin manifest: {:?}", manifest_file);
        let file = File::open(manifest_file)?;
        let reader = BufReader::new(file);
        let manifest: PactPluginManifest = serde_json::from_reader(reader)?;
        trace!("Parsed plugin manifest: {:?}", manifest);
        let version = manifest.version.clone();
        if manifest.name == plugin_dep.name
          && versions_compatible(version.as_str(), &plugin_dep.version)
        {
          let manifest = PactPluginManifest {
            plugin_dir: path.to_string_lossy().to_string(),
            ..manifest
          };
          manifests.push(manifest);
        }
      }
    }
  }

  let manifest = manifests.iter().max_by(|a, b| {
    let a = Version::parse(&a.version).unwrap_or_else(|_| Version::new(0, 0, 0));
    let b = Version::parse(&b.version).unwrap_or_else(|_| Version::new(0, 0, 0));
    a.cmp(&b)
  });
  if let Some(manifest) = manifest {
    let key = format!("{}/{}", manifest.name, manifest.version);
    {
      let mut guard = PLUGIN_MANIFEST_REGISTER.lock().unwrap();
      guard.insert(key.clone(), manifest.clone());
    }
    Ok(manifest.clone())
  } else {
    Err(anyhow!(
      "Plugin {} was not found (in $HOME/.pact/plugins or $PACT_PLUGIN_DIR)",
      plugin_dep
    ))
  }
}

pub(crate) fn pact_plugin_dir() -> anyhow::Result<PathBuf> {
  let env_var = env::var_os("PACT_PLUGIN_DIR");
  let plugin_dir = env_var.unwrap_or_default();
  let plugin_dir = plugin_dir.to_string_lossy();
  if plugin_dir.is_empty() {
    home::home_dir().map(|dir| dir.join(".pact").join("plugins"))
  } else {
    PathBuf::from_str(plugin_dir.as_ref()).ok()
  }
  .ok_or_else(|| {
    anyhow!("No Pact plugin directory was found (in $HOME/.pact/plugins or $PACT_PLUGIN_DIR)")
  })
}

/// Lookup the plugin manifest in the global plugin manifest registry.
pub fn lookup_plugin_manifest(plugin: &PluginDependency) -> Option<PactPluginManifest> {
  let guard = PLUGIN_MANIFEST_REGISTER.lock().unwrap();
  if let Some(version) = &plugin.version {
    let key = format!("{}/{}", plugin.name, version);
    guard.get(&key).cloned()
  } else {
    guard
      .iter()
      .filter(|(_, value)| value.name == plugin.name)
      .max_by(|(_, v1), (_, v2)| v1.version.cmp(&v2.version))
      .map(|(_, p)| p.clone())
  }
}

async fn initialise_plugin(
  manifest: &PactPluginManifest,
  plugin_register: &mut HashMap<String, RegisteredPlugin>,
) -> anyhow::Result<PactPlugin> {
  let interface_version = PluginInterfaceVersion::try_from(manifest.plugin_interface_version)
    .with_context(|| {
      format!(
        "Plugin {}:{} declared an invalid interface version",
        manifest.name, manifest.version
      )
    })?;

  match interface_version {
    PluginInterfaceVersion::V1 | PluginInterfaceVersion::V2 => {
      match manifest.executable_type.as_str() {
        "exec" => {
          let plugin = start_plugin_process(manifest).await?;
          #[allow(deprecated)]
          let port = plugin.port();
          debug!(
            "Plugin process started OK (port = {}), sending init message",
            port
          );

          let instance_id = plugin.instance_id.clone();
          let mut grpc_plugin = GrpcPactPlugin::new(plugin);
          let response = init_handshake(manifest, &mut grpc_plugin, &instance_id).await.map_err(|err| {
            deregister_plugin_instance(&instance_id);
            grpc_plugin.kill();
            anyhow!("Failed to send init request to the plugin - {}", err)
          })?;
          grpc_plugin.plugin.plugin_capabilities = response.plugin_capabilities;
          let pact_plugin = grpc_plugin.plugin.clone();

          let key = format!("{}/{}", manifest.name, manifest.version);
          let instance: Arc<dyn PluginInstance + Send + Sync> = Arc::new(grpc_plugin);
          plugin_register.insert(key, RegisteredPlugin::new(instance, pact_plugin.clone()));

          Ok(pact_plugin)
        }
        "lua" => {
          #[cfg(feature = "lua")]
          {
            let instance_id = uuid::Uuid::new_v4().to_string();
            let mut lua_plugin = crate::lua_plugin::start_lua_plugin(manifest, instance_id.clone())?;
            let response = init_handshake(manifest, &mut lua_plugin, &instance_id).await.map_err(|err| {
              anyhow!("Failed to send init request to the Lua plugin - {}", err)
            })?;
            lua_plugin.set_plugin_capabilities(response.plugin_capabilities.clone());

            #[allow(deprecated)]
            let child = crate::child_process::ChildPluginProcess {
              child_pid: 0,
              plugin_info: crate::child_process::RunningPluginInfo {
                port: 0,
                server_key: String::new(),
              },
              instance_id: instance_id.clone(),
            };
            let mut pact_plugin = PactPlugin::new(manifest, child)?;
            pact_plugin.plugin_capabilities = response.plugin_capabilities.clone();

            let key = format!("{}/{}", manifest.name, manifest.version);
            let instance: Arc<dyn PluginInstance + Send + Sync> = Arc::new(lua_plugin);
            plugin_register.insert(key, RegisteredPlugin::new(instance, pact_plugin.clone()));

            Ok(pact_plugin)
          }
          #[cfg(not(feature = "lua"))]
          {
            Err(anyhow!(
              "Lua plugins are not supported (the 'lua' feature of pact-plugin-driver is not enabled)"
            ))
          }
        }
        _ => Err(anyhow!(
          "Plugin executable type of {} is not supported",
          manifest.executable_type
        )),
      }
    }
  }
}

/// Internal function: public for testing
pub async fn init_handshake(
  manifest: &PactPluginManifest,
  plugin: &mut (dyn PactPluginRpc + Send + Sync),
  instance_id: &str,
) -> anyhow::Result<crate::plugin_models::PluginInitResponse> {
  let request = PluginInitRequest {
    implementation: "plugin-driver-rust".to_string(),
    version: option_env!("CARGO_PKG_VERSION").unwrap_or("0").to_string(),
    host_capabilities: host_capabilities(),
    plugin_instance_id: instance_id.to_string(),
  };
  let response = plugin.init_plugin(request).await?;
  debug!(
    "Got init response {:?} from plugin {}",
    response, manifest.name
  );
  register_plugin_entries(manifest, &response.catalogue);
  tokio::task::spawn(publish_updated_catalogue());
  Ok(response)
}

/// Shut down all plugin processes
pub fn shutdown_plugins() {
  let thread_id = thread::current().id();
  debug!("Shutting down all plugins");
  trace!(
    "shutdown_plugins {:?}: Waiting on PLUGIN_REGISTER lock",
    thread_id
  );
  let mut guard = PLUGIN_REGISTER.lock().unwrap();
  trace!("shutdown_plugins {:?}: Got PLUGIN_REGISTER lock", thread_id);
  for entry in guard.values() {
    debug!("Shutting down plugin {:?}", entry.plugin.manifest);
    deregister_plugin_instance(&entry.plugin.instance_id);
    entry.instance.kill();
    remove_plugin_entries(&entry.plugin.manifest.name);
  }
  guard.clear();
  trace!(
    "shutdown_plugins {:?}: Releasing PLUGIN_REGISTER lock",
    thread_id
  );
}

/// Shutdown the given plugin
pub fn shutdown_plugin(plugin: &dyn PluginInstance) {
  debug!(
    "Shutting down plugin {}:{}",
    plugin.manifest().name, plugin.manifest().version
  );
  deregister_plugin_instance(plugin.instance_id());
  plugin.kill();
  remove_plugin_entries(&plugin.manifest().name);
}

/// Publish the current catalogue to all plugins
pub async fn publish_updated_catalogue() {
  let thread_id = thread::current().id();

  let request = Catalogue {
    catalogue: all_entries()
      .iter()
      .map(|entry| crate::proto::CatalogueEntry {
        r#type: entry.entry_type.to_proto_type() as i32,
        key: entry.key.clone(),
        values: entry.values.clone(),
      })
      .collect(),
  };

  let plugins = {
    trace!(
      "publish_updated_catalogue {:?}: Waiting on PLUGIN_REGISTER lock",
      thread_id
    );
    let inner = PLUGIN_REGISTER.lock().unwrap();
    trace!(
      "publish_updated_catalogue {:?}: Got PLUGIN_REGISTER lock",
      thread_id
    );
    let plugins = inner.values().map(|e| e.instance.clone()).collect::<Vec<_>>();
    trace!(
      "publish_updated_catalogue {:?}: Releasing PLUGIN_REGISTER lock",
      thread_id
    );
    plugins
  };

  for plugin in plugins {
    if let Err(err) = plugin.update_catalogue(request.clone()).await {
      warn!(
        "Failed to send updated catalogue to plugin '{}' - {}",
        plugin.manifest().name, err
      );
    }
  }
}

/// Increment access to the plugin.
#[tracing::instrument]
pub fn increment_plugin_access(plugin: &PluginDependency) {
  let thread_id = thread::current().id();

  trace!(
    "increment_plugin_access {:?}: Waiting on PLUGIN_REGISTER lock",
    thread_id
  );
  let inner = PLUGIN_REGISTER.lock().unwrap();
  trace!(
    "increment_plugin_access {:?}: Got PLUGIN_REGISTER lock",
    thread_id
  );

  if let Some(entry) = lookup_plugin_inner(plugin, &inner) {
    entry.update_access();
  }

  trace!(
    "increment_plugin_access {:?}: Releasing PLUGIN_REGISTER lock",
    thread_id
  );
}

/// Decrement access to the plugin. If the current access count is zero, shut down the plugin
#[tracing::instrument]
pub fn drop_plugin_access(plugin: &PluginDependency) {
  let thread_id = thread::current().id();

  trace!(
    "drop_plugin_access {:?}: Waiting on PLUGIN_REGISTER lock",
    thread_id
  );
  let mut inner = PLUGIN_REGISTER.lock().unwrap();
  trace!(
    "drop_plugin_access {:?}: Got PLUGIN_REGISTER lock",
    thread_id
  );

  let shutdown_info = lookup_plugin_inner(plugin, &inner).and_then(|entry| {
    let key = format!("{}/{}", entry.plugin.manifest.name, entry.plugin.manifest.version);
    let instance_ref = entry.instance.clone();
    if entry.drop_access() == 0 {
      Some((key, instance_ref))
    } else {
      None
    }
  });
  if let Some((key, instance_ref)) = shutdown_info {
    shutdown_plugin(instance_ref.as_ref());
    inner.remove(key.as_str());
  }

  trace!(
    "drop_plugin_access {:?}: Releasing PLUGIN_REGISTER lock",
    thread_id
  );
}

/// Starts a mock server given the catalog entry for it and a Pact
#[deprecated(
  note = "Use start_mock_server_v2 which takes a test context map",
  since = "0.2.2"
)]
pub async fn start_mock_server(
  catalogue_entry: &CatalogueEntry,
  pact: Box<dyn Pact + Send + Sync>,
  config: MockServerConfig,
) -> anyhow::Result<MockServerDetails> {
  start_mock_server_v2(catalogue_entry, pact, config, hashmap! {}).await
}

/// Starts a mock server given the catalog entry for it and a Pact.
/// Any transport specific configuration must be passed in the `test_context` field under
/// the `transport_config` key.
pub async fn start_mock_server_v2(
  catalogue_entry: &CatalogueEntry,
  pact: Box<dyn Pact + Send + Sync>,
  config: MockServerConfig,
  test_context: HashMap<String, Value>,
) -> anyhow::Result<MockServerDetails> {
  let manifest = catalogue_entry
    .plugin
    .as_ref()
    .ok_or_else(|| anyhow!("Catalogue entry did not have an associated plugin manifest"))?;
  let plugin = lookup_plugin(&manifest.as_dependency())
    .ok_or_else(|| anyhow!("Did not find a running plugin for manifest {:?}", manifest))?;

  debug!(
    plugin_name = manifest.name.as_str(),
    plugin_version = manifest.version.as_str(),
    ?test_context,
    "Sending startMockServer request to plugin"
  );

  let response = if manifest.plugin_interface_version >= 2 {
    let v4_pact = pact.as_v4_pact().map_err(|_| anyhow!("Pact must be a V4 pact for V2 plugin interface"))?;
    let interactions = build_v2_interaction_contents(manifest, &v4_pact);
    let request = proto_v2::StartMockServerRequest {
      host_interface: config.host_interface.clone().unwrap_or_default(),
      port: config.port,
      tls: config.tls,
      interactions,
      test_context: Some(to_proto_struct(&test_context)),
    };
    plugin.start_mock_server_v2(request).await?
  } else {
    let request = StartMockServerRequest {
      host_interface: config.host_interface.unwrap_or_default(),
      port: config.port,
      tls: config.tls,
      pact: pact.to_json(PactSpecification::V4)?.to_string(),
      test_context: Some(to_proto_struct(&test_context)),
    };
    plugin.start_mock_server(request).await?
  };

  debug!("Got response ${response:?}");

  let mock_server_response = response
    .response
    .ok_or_else(|| anyhow!("Did not get a valid response from the start mock server call"))?;
  match mock_server_response {
    start_mock_server_response::Response::Error(err) => {
      Err(anyhow!("Mock server failed to start: {}", err))
    }
    start_mock_server_response::Response::Details(details) => Ok(MockServerDetails {
      key: details.key.clone(),
      base_url: details.address.clone(),
      port: details.port,
      plugin,
    }),
  }
}

/// Convert a V1 InteractionData to a V2 InteractionData (structurally identical; binary conversion is safe).
fn to_proto_v2_interaction_data(data: InteractionData) -> proto_v2::InteractionData {
  use prost::Message;
  proto_v2::InteractionData::decode(data.encode_to_vec().as_slice())
    .expect("V1 and V2 InteractionData have identical wire format")
}

fn value_to_proto_struct(v: Value) -> prost_types::Struct {
  match v {
    Value::Object(map) => {
      let hmap: HashMap<String, Value> = map.into_iter().collect();
      to_proto_struct(&hmap)
    }
    _ => prost_types::Struct::default(),
  }
}

/// Build a list of V2 InteractionContents from a V4 pact and plugin manifest.
fn build_v2_interaction_contents(
  manifest: &PactPluginManifest,
  pact: &V4Pact,
) -> Vec<proto_v2::InteractionContents> {
  let plugin_name = &manifest.name;
  let consumer = pact.consumer.name.clone();
  let provider = pact.provider.name.clone();
  let pact_configuration = pact.plugin_data()
    .into_iter()
    .find(|p| p.name == *plugin_name)
    .and_then(|p| p.configuration.get("pactConfiguration").cloned());
  pact.interactions.iter().map(|interaction| {
    build_interaction_contents_inner(
      plugin_name,
      &consumer,
      &provider,
      pact_configuration.clone(),
      interaction.as_ref(),
    )
  }).collect()
}

/// Build V2 InteractionContents for a single interaction.
fn build_v2_single_interaction_contents(
  manifest: &PactPluginManifest,
  pact: &V4Pact,
  interaction: &(dyn V4Interaction + Send + Sync),
) -> proto_v2::InteractionContents {
  let plugin_name = &manifest.name;
  let pact_configuration = pact.plugin_data()
    .into_iter()
    .find(|p| p.name == *plugin_name)
    .and_then(|p| p.configuration.get("pactConfiguration").cloned());
  build_interaction_contents_inner(
    plugin_name,
    &pact.consumer.name,
    &pact.provider.name,
    pact_configuration,
    interaction,
  )
}

fn build_interaction_contents_inner(
  plugin_name: &str,
  consumer: &str,
  provider: &str,
  pact_configuration: Option<Value>,
  interaction: &(dyn V4Interaction + Send + Sync),
) -> proto_v2::InteractionContents {
  let plugin_config = interaction.plugin_config();
  let interaction_configuration = plugin_config
    .get(plugin_name)
    .map(|config| Value::Object(config.iter().map(|(k, v)| (k.clone(), v.clone())).collect()));
  proto_v2::InteractionContents {
    interaction_type: interaction.v4_type().to_string(),
    plugin_configuration: Some(proto_v2::PluginConfiguration {
      interaction_configuration: interaction_configuration.map(value_to_proto_struct),
      pact_configuration: pact_configuration.map(value_to_proto_struct),
    }),
    consumer: consumer.to_string(),
    provider: provider.to_string(),
  }
}

/// Shutdowns a running mock server. Will return any errors from the mock server.
pub async fn shutdown_mock_server(
  mock_server: &MockServerDetails,
) -> anyhow::Result<Vec<MockServerResults>> {
  let request = ShutdownMockServerRequest {
    server_key: mock_server.key.to_string(),
  };

  let manifest = mock_server.plugin.manifest();
  debug!(
    plugin_name = manifest.name.as_str(),
    plugin_version = manifest.version.as_str(),
    server_key = mock_server.key.as_str(),
    "Sending shutdownMockServer request to plugin"
  );
  let response = mock_server.plugin.shutdown_mock_server(request).await?;
  debug!("Got response: {response:?}");

  if response.ok {
    Ok(vec![])
  } else {
    Ok(
      response
        .results
        .iter()
        .map(|result| MockServerResults {
          path: result.path.clone(),
          error: result.error.clone(),
          mismatches: result
            .mismatches
            .iter()
            .map(|mismatch| ContentMismatch {
              expected: mismatch
                .expected
                .as_ref()
                .map(|e| from_utf8(&e).unwrap_or_default().to_string())
                .unwrap_or_default(),
              actual: mismatch
                .actual
                .as_ref()
                .map(|a| from_utf8(&a).unwrap_or_default().to_string())
                .unwrap_or_default(),
              mismatch: mismatch.mismatch.clone(),
              path: mismatch.path.clone(),
              diff: optional_string(&mismatch.diff),
              mismatch_type: optional_string(&mismatch.mismatch_type),
            })
            .collect(),
        })
        .collect(),
    )
  }
}

/// Gets the results from a running mock server.
pub async fn get_mock_server_results(
  mock_server: &MockServerDetails,
) -> anyhow::Result<Vec<MockServerResults>> {
  let request = MockServerRequest {
    server_key: mock_server.key.to_string(),
  };

  let manifest = mock_server.plugin.manifest();
  debug!(
    plugin_name = manifest.name.as_str(),
    plugin_version = manifest.version.as_str(),
    server_key = mock_server.key.as_str(),
    "Sending getMockServerResults request to plugin"
  );
  let response = mock_server.plugin.get_mock_server_results(request).await?;
  debug!("Got response: {response:?}");

  if response.ok {
    Ok(vec![])
  } else {
    Ok(
      response
        .results
        .iter()
        .map(|result| MockServerResults {
          path: result.path.clone(),
          error: result.error.clone(),
          mismatches: result
            .mismatches
            .iter()
            .map(|mismatch| ContentMismatch {
              expected: mismatch
                .expected
                .as_ref()
                .map(|e| from_utf8(&e).unwrap_or_default().to_string())
                .unwrap_or_default(),
              actual: mismatch
                .actual
                .as_ref()
                .map(|a| from_utf8(&a).unwrap_or_default().to_string())
                .unwrap_or_default(),
              mismatch: mismatch.mismatch.clone(),
              path: mismatch.path.clone(),
              diff: optional_string(&mismatch.diff),
              mismatch_type: optional_string(&mismatch.mismatch_type),
            })
            .collect(),
        })
        .collect(),
    )
  }
}

/// Sets up a transport request to be made. This is the first phase when verifying, and it allows the
/// users to add additional values to any requests that are made.
pub async fn prepare_validation_for_interaction(
  transport_entry: &CatalogueEntry,
  pact: &V4Pact,
  interaction: &(dyn V4Interaction + Send + Sync),
  context: &HashMap<String, Value>,
) -> anyhow::Result<InteractionVerificationData> {
  let manifest = transport_entry.plugin.as_ref().ok_or_else(|| {
    anyhow!("Transport catalogue entry did not have an associated plugin manifest")
  })?;
  let plugin = lookup_plugin(&manifest.as_dependency())
    .ok_or_else(|| anyhow!("Did not find a running plugin for manifest {:?}", manifest))?;

  prepare_validation_for_interaction_inner(plugin.as_ref(), pact, interaction, context).await
}

pub(crate) async fn prepare_validation_for_interaction_inner(
  plugin: &dyn PluginInstance,
  pact: &V4Pact,
  interaction: &(dyn V4Interaction + Send + Sync),
  context: &HashMap<String, Value>,
) -> anyhow::Result<InteractionVerificationData> {
  let manifest = plugin.manifest();
  debug!(
    plugin_name = manifest.name.as_str(),
    plugin_version = manifest.version.as_str(),
    "Sending prepareValidationForInteraction request to plugin"
  );

  let response = if manifest.plugin_interface_version >= 2 {
    let interaction_contents = build_v2_single_interaction_contents(manifest, pact, interaction);
    let request = proto_v2::VerificationPreparationRequest {
      interaction_contents: Some(interaction_contents),
      config: Some(to_proto_struct(context)),
      test_context: None,
    };
    plugin.prepare_interaction_for_verification_v2(request).await?
  } else {
    let mut pact = pact.clone();
    pact.interactions = pact
      .interactions
      .iter()
      .map(|i| {
        if i.key().is_none() {
          i.with_unique_key()
        } else {
          i.boxed_v4()
        }
      })
      .collect();
    let request = VerificationPreparationRequest {
      pact: pact.to_json(PactSpecification::V4)?.to_string(),
      interaction_key: interaction.unique_key(),
      config: Some(to_proto_struct(context)),
    };
    plugin.prepare_interaction_for_verification(request).await?
  };
  debug!("Got response: {response:?}");

  let validation_response = response.response.ok_or_else(|| {
    anyhow!("Did not get a valid response from the prepare interaction for verification call")
  })?;
  match &validation_response {
    verification_preparation_response::Response::Error(err) => {
      Err(anyhow!("Failed to prepare the request: {}", err))
    }
    verification_preparation_response::Response::InteractionData(data) => {
      let content_type = data
        .body
        .as_ref()
        .and_then(|body| ContentType::parse(body.content_type.as_str()).ok());
      Ok(InteractionVerificationData {
        request_data: data
          .body
          .as_ref()
          .and_then(|body| body.content.as_ref())
          .map(|body| OptionalBody::Present(Bytes::from(body.clone()), content_type, None))
          .unwrap_or_default(),
        metadata: data
          .metadata
          .iter()
          .map(|(k, v)| {
            let value = match &v.value {
              Some(v) => match &v {
                metadata_value::Value::NonBinaryValue(v) => Either::Left(proto_value_to_json(v)),
                metadata_value::Value::BinaryValue(b) => Either::Right(Bytes::from(b.clone())),
              },
              None => Either::Left(Value::Null),
            };
            (k.clone(), value)
          })
          .collect(),
      })
    }
  }
}

/// Executes the verification of the interaction that was configured with the prepare_validation_for_interaction call
pub async fn verify_interaction(
  transport_entry: &CatalogueEntry,
  verification_data: &InteractionVerificationData,
  config: &HashMap<String, Value>,
  pact: &V4Pact,
  interaction: &(dyn V4Interaction + Send + Sync),
) -> anyhow::Result<InteractionVerificationResult> {
  let manifest = transport_entry.plugin.as_ref().ok_or_else(|| {
    anyhow!("Transport catalogue entry did not have an associated plugin manifest")
  })?;
  let plugin = lookup_plugin(&manifest.as_dependency())
    .ok_or_else(|| anyhow!("Did not find a running plugin for manifest {:?}", manifest))?;

  verify_interaction_inner(
    plugin.as_ref(),
    verification_data,
    config,
    pact,
    interaction,
  )
  .await
}

pub(crate) async fn verify_interaction_inner(
  plugin: &dyn PluginInstance,
  verification_data: &InteractionVerificationData,
  config: &HashMap<String, Value>,
  pact: &V4Pact,
  interaction: &(dyn V4Interaction + Send + Sync),
) -> anyhow::Result<InteractionVerificationResult> {
  let manifest = plugin.manifest();
  debug!(
    plugin_name = manifest.name.as_str(),
    plugin_version = manifest.version.as_str(),
    "Sending verifyInteraction request to plugin"
  );

  let interaction_data = InteractionData {
    body: Some((&verification_data.request_data).into()),
    metadata: verification_data
      .metadata
      .iter()
      .map(|(k, v)| {
        (
          k.clone(),
          MetadataValue {
            value: Some(match v {
              Either::Left(value) => metadata_value::Value::NonBinaryValue(to_proto_value(value)),
              Either::Right(b) => metadata_value::Value::BinaryValue(b.to_vec()),
            }),
          },
        )
      })
      .collect(),
  };

  let response = if manifest.plugin_interface_version >= 2 {
    let interaction_contents = build_v2_single_interaction_contents(manifest, pact, interaction);
    let request = proto_v2::VerifyInteractionRequest {
      interaction_data: Some(to_proto_v2_interaction_data(interaction_data)),
      config: Some(to_proto_struct(config)),
      interaction_contents: Some(interaction_contents),
      test_context: None,
    };
    plugin.verify_interaction_v2(request).await?
  } else {
    let mut pact = pact.clone();
    pact.interactions = pact
      .interactions
      .iter()
      .map(|i| {
        if i.key().is_none() {
          i.with_unique_key()
        } else {
          i.boxed_v4()
        }
      })
      .collect();
    let request = VerifyInteractionRequest {
      pact: pact.to_json(PactSpecification::V4)?.to_string(),
      interaction_key: interaction.unique_key(),
      config: Some(to_proto_struct(config)),
      interaction_data: Some(interaction_data),
    };
    plugin.verify_interaction(request).await?
  };
  debug!("Got response: {response:?}");

  let validation_response = response
    .response
    .ok_or_else(|| anyhow!("Did not get a valid response from the verification call"))?;
  match &validation_response {
    verify_interaction_response::Response::Error(err) => {
      Err(anyhow!("Failed to verify the request: {}", err))
    }
    verify_interaction_response::Response::Result(data) => Ok(data.into()),
  }
}

/// Tries to download and install the plugin from the given URL, returning the manifest for the
/// plugin if successful.
pub async fn install_plugin_from_url(
  http_client: &Client,
  source_url: &str,
) -> anyhow::Result<PactPluginManifest> {
  let response = fetch_json_from_url(source_url, http_client).await?;
  if let Some(map) = response.as_object() {
    if let Some(tag) = map.get("tag_name") {
      let tag = json_to_string(tag);
      debug!(%tag, "Found tag");
      let url = if source_url.ends_with("/latest") {
        source_url.strip_suffix("/latest").unwrap_or(source_url)
      } else {
        let suffix = format!("/tag/{}", tag);
        source_url
          .strip_suffix(suffix.as_str())
          .unwrap_or(source_url)
      };
      let manifest_json = download_json_from_github(&http_client, url, &tag, "pact-plugin.json")
        .await
        .context("Downloading manifest file from GitHub")?;
      let manifest: PactPluginManifest = serde_json::from_value(manifest_json)
        .context("Failed to parsing JSON manifest file from GitHub")?;
      debug!(?manifest, "Loaded manifest from GitHub");

      debug!(
        "Installing plugin {} version {}",
        manifest.name, manifest.version
      );
      let plugin_dir =
        create_plugin_dir(&manifest).context("Failed to creating plugins directory")?;
      download_plugin_executable(&manifest, &plugin_dir, &http_client, url, &tag, false).await?;

      Ok(PactPluginManifest {
        plugin_dir: plugin_dir.to_string_lossy().to_string(),
        ..manifest
      })
    } else {
      bail!("GitHub release page does not have a valid tag_name attribute");
    }
  } else {
    bail!("Response from source is not a valid JSON from a GitHub release page")
  }
}

fn create_plugin_dir(manifest: &PactPluginManifest) -> anyhow::Result<PathBuf> {
  let plugins_dir = pact_plugin_dir()?;
  if !plugins_dir.exists() {
    info!(plugins_dir = %plugins_dir.display(), "Creating plugins directory");
    fs::create_dir_all(plugins_dir.clone())?;
  }

  let plugin_dir = plugins_dir.join(format!("{}-{}", manifest.name, manifest.version));
  info!(plugin_dir = %plugin_dir.display(), "Creating plugin directory");
  fs::create_dir(plugin_dir.clone())?;

  info!("Writing plugin manifest file");
  let file_name = plugin_dir.join("pact-plugin.json");
  let mut f = File::create(file_name)?;
  let json = serde_json::to_string(manifest)?;
  f.write_all(json.as_bytes())?;

  Ok(plugin_dir.clone())
}

#[cfg(test)]
mod tests {
  use std::collections::HashMap;
  use std::fs::{self, File};

  use maplit::hashmap;
  use pact_models::prelude::v4::V4Pact;
  use pact_models::v4::interaction::V4Interaction;
  use pact_models::v4::sync_message::SynchronousMessage;

  use expectest::prelude::*;
  use tempdir::TempDir;

  use crate::plugin_manager::prepare_validation_for_interaction_inner;
  use crate::plugin_manager::verify_interaction_inner;
  use crate::plugin_models::PluginDependency;
  use crate::plugin_models::tests::{FailingInitPlugin, InitRecordingPlugin, MockPlugin};
  use crate::verification::InteractionVerificationData;

  use crate::catalogue_manager::{
    CatalogueEntry, CatalogueEntryProviderType, CatalogueEntryType, register_core_entries,
  };

  use super::{PactPluginManifest, init_handshake, initialise_plugin, load_manifest_from_dir};

  #[test]
  fn load_manifest_from_dir_test() {
    let tmp_dir = TempDir::new("load_manifest_from_dir").unwrap();

    let manifest_1 = PactPluginManifest {
      name: "test-plugin".to_string(),
      version: "0.1.5".to_string(),
      ..PactPluginManifest::default()
    };
    let path_1 = tmp_dir.path().join("1");
    fs::create_dir_all(&path_1).unwrap();
    let file_1 = File::create(path_1.join("pact-plugin.json")).unwrap();
    serde_json::to_writer(file_1, &manifest_1).unwrap();

    let manifest_2 = PactPluginManifest {
      name: "test-plugin".to_string(),
      version: "0.1.20".to_string(),
      ..PactPluginManifest::default()
    };
    let path_2 = tmp_dir.path().join("2");
    fs::create_dir_all(&path_2).unwrap();
    let file_2 = File::create(path_2.join("pact-plugin.json")).unwrap();
    serde_json::to_writer(file_2, &manifest_2).unwrap();

    let manifest_3 = PactPluginManifest {
      name: "test-plugin".to_string(),
      version: "0.1.7".to_string(),
      ..PactPluginManifest::default()
    };
    let path_3 = tmp_dir.path().join("3");
    fs::create_dir_all(&path_3).unwrap();
    let file_3 = File::create(path_3.join("pact-plugin.json")).unwrap();
    serde_json::to_writer(file_3, &manifest_3).unwrap();

    let manifest_4 = PactPluginManifest {
      name: "test-plugin".to_string(),
      version: "0.1.14".to_string(),
      ..PactPluginManifest::default()
    };
    let path_4 = tmp_dir.path().join("4");
    fs::create_dir_all(&path_4).unwrap();
    let file_4 = File::create(path_4.join("pact-plugin.json")).unwrap();
    serde_json::to_writer(file_4, &manifest_4).unwrap();

    let manifest_5 = PactPluginManifest {
      name: "test-plugin".to_string(),
      version: "0.1.12".to_string(),
      ..PactPluginManifest::default()
    };
    let path_5 = tmp_dir.path().join("5");
    fs::create_dir_all(&path_5).unwrap();
    let file_5 = File::create(path_5.join("pact-plugin.json")).unwrap();
    serde_json::to_writer(file_5, &manifest_5).unwrap();

    let dep = PluginDependency {
      name: "test-plugin".to_string(),
      version: None,
      dependency_type: Default::default(),
    };

    let result = load_manifest_from_dir(&dep, &tmp_dir.path().to_path_buf()).unwrap();
    expect!(result.version).to(be_equal_to("0.1.20"));
  }

  #[test_log::test(tokio::test)]
  async fn initialise_plugin_rejects_unsupported_interface_versions() {
    let manifest = PactPluginManifest {
      name: "test-plugin".to_string(),
      version: "0.0.0".to_string(),
      executable_type: "exec".to_string(),
      plugin_interface_version: 3,
      ..PactPluginManifest::default()
    };

    let mut plugin_register = HashMap::new();
    let err = initialise_plugin(&manifest, &mut plugin_register)
      .await
      .unwrap_err();

    expect!(err.to_string()).to(be_equal_to(
      "Plugin test-plugin:0.0.0 declared an invalid interface version".to_string(),
    ));
  }

  #[test_log::test(tokio::test)]
  async fn init_handshake_sends_host_capabilities_and_returns_plugin_capabilities() {
    register_core_entries(&vec![CatalogueEntry {
      entry_type: CatalogueEntryType::CONTENT_MATCHER,
      provider_type: CatalogueEntryProviderType::CORE,
      plugin: None,
      key: "test-content-type".to_string(),
      values: Default::default(),
    }]);

    let manifest = PactPluginManifest {
      name: "test-plugin".to_string(),
      version: "0.0.0".to_string(),
      ..PactPluginManifest::default()
    };
    let mut plugin = InitRecordingPlugin::default();

    let response = init_handshake(&manifest, &mut plugin, "test-instance-id").await.unwrap();
    let request = plugin.request.read().unwrap().clone().unwrap();

    assert!(
      request.host_capabilities.contains(&"content-matcher/test-content-type".to_string()),
      "expected host_capabilities to contain 'content-matcher/test-content-type', got: {:?}",
      request.host_capabilities
    );
    expect!(response.plugin_capabilities).to(be_equal_to(vec![
      "interaction/request-response".to_string(),
    ]));
  }

  #[test_log::test(tokio::test)]
  async fn init_handshake_propagates_plugin_init_failure() {
    let manifest = PactPluginManifest {
      name: "test-plugin".to_string(),
      version: "0.0.0".to_string(),
      ..PactPluginManifest::default()
    };
    let expected_error = "CSV plugin requires request/response-scoped interaction support \
      (missing host capabilities: interaction/request-response)";
    let mut plugin = FailingInitPlugin { error: expected_error.to_string() };

    let err = init_handshake(&manifest, &mut plugin, "test-instance-id").await.unwrap_err();

    expect!(err.to_string()).to(be_equal_to(expected_error.to_string()));
  }

  #[test_log::test(tokio::test)]
  async fn prepare_validation_for_interaction_passes_in_pact_with_interaction_keys_set() {
    let mock_plugin = MockPlugin {
      manifest: PactPluginManifest {
        name: "test-plugin".to_string(),
        version: "0.0.0".to_string(),
        ..PactPluginManifest::default()
      },
      ..MockPlugin::default()
    };

    let interaction = SynchronousMessage {
      ..SynchronousMessage::default()
    };
    let pact = V4Pact {
      interactions: vec![interaction.boxed_v4()],
      ..V4Pact::default()
    };
    let context = hashmap! {};

    let result = prepare_validation_for_interaction_inner(
      &mock_plugin,
      &pact,
      &interaction,
      &context,
    )
    .await;

    expect!(result).to(be_ok());
    let request = {
      let r = mock_plugin.prepare_request.read().unwrap();
      r.clone()
    };
    let pact_in =
      V4Pact::pact_from_json(&serde_json::from_str(request.pact.as_str()).unwrap(), "").unwrap();
    expect!(pact_in.interactions[0].key().unwrap()).to(be_equal_to(request.interaction_key));
  }

  #[test_log::test(tokio::test)]
  async fn prepare_validation_for_interaction_handles_pact_with_keys_already_set() {
    let mock_plugin = MockPlugin {
      manifest: PactPluginManifest {
        name: "test-plugin".to_string(),
        version: "0.0.0".to_string(),
        ..PactPluginManifest::default()
      },
      ..MockPlugin::default()
    };

    let interaction = SynchronousMessage {
      key: Some("1234567890".to_string()),
      ..SynchronousMessage::default()
    };
    let pact = V4Pact {
      interactions: vec![interaction.boxed_v4()],
      ..V4Pact::default()
    };
    let context = hashmap! {};

    let result = prepare_validation_for_interaction_inner(
      &mock_plugin,
      &pact,
      &interaction,
      &context,
    )
    .await;

    expect!(result).to(be_ok());
    let request = {
      let r = mock_plugin.prepare_request.read().unwrap();
      r.clone()
    };
    let pact_in =
      V4Pact::pact_from_json(&serde_json::from_str(request.pact.as_str()).unwrap(), "").unwrap();
    expect!(request.interaction_key.as_str()).to(be_equal_to("1234567890"));
    expect!(pact_in.interactions[0].key().unwrap()).to(be_equal_to(request.interaction_key));
  }

  #[test_log::test(tokio::test)]
  async fn verify_interaction_passes_in_pact_with_interaction_keys_set() {
    let mock_plugin = MockPlugin {
      manifest: PactPluginManifest {
        name: "test-plugin".to_string(),
        version: "0.0.0".to_string(),
        ..PactPluginManifest::default()
      },
      ..MockPlugin::default()
    };

    let interaction = SynchronousMessage {
      ..SynchronousMessage::default()
    };
    let pact = V4Pact {
      interactions: vec![interaction.boxed_v4()],
      ..V4Pact::default()
    };
    let context = hashmap! {};
    let data = InteractionVerificationData::default();

    let result = verify_interaction_inner(
      &mock_plugin,
      &data,
      &context,
      &pact,
      &interaction,
    )
    .await;

    expect!(result).to(be_ok());
    let request = {
      let r = mock_plugin.verify_request.read().unwrap();
      r.clone()
    };
    let pact_in =
      V4Pact::pact_from_json(&serde_json::from_str(request.pact.as_str()).unwrap(), "").unwrap();
    expect!(pact_in.interactions[0].key().unwrap()).to(be_equal_to(request.interaction_key));
  }

  #[test_log::test(tokio::test)]
  async fn verify_interaction_handles_interaction_with_key_already_set() {
    let mock_plugin = MockPlugin {
      manifest: PactPluginManifest {
        name: "test-plugin".to_string(),
        version: "0.0.0".to_string(),
        ..PactPluginManifest::default()
      },
      ..MockPlugin::default()
    };

    let interaction = SynchronousMessage {
      key: Some("1234567890".to_string()),
      ..SynchronousMessage::default()
    };
    let pact = V4Pact {
      interactions: vec![interaction.boxed_v4()],
      ..V4Pact::default()
    };
    let context = hashmap! {};
    let data = InteractionVerificationData::default();

    let result = verify_interaction_inner(
      &mock_plugin,
      &data,
      &context,
      &pact,
      &interaction,
    )
    .await;

    expect!(result).to(be_ok());
    let request = {
      let r = mock_plugin.verify_request.read().unwrap();
      r.clone()
    };
    let pact_in =
      V4Pact::pact_from_json(&serde_json::from_str(request.pact.as_str()).unwrap(), "").unwrap();
    expect!(request.interaction_key.as_str()).to(be_equal_to("1234567890"));
    expect!(pact_in.interactions[0].key().unwrap()).to(be_equal_to(request.interaction_key));
  }
}