yaak-mountain-loop/src-tauri/yaak-plugins/src/manager.rs

use crate::error::Error::{
    self, AuthPluginNotFound, ClientNotInitializedErr, PluginErr, PluginNotFoundErr,
    UnknownEventErr,
};
use crate::error::Result;
use crate::events::{
    BootRequest, CallFolderActionRequest, CallGrpcRequestActionRequest,
    CallHttpAuthenticationActionArgs, CallHttpAuthenticationActionRequest,
    CallHttpAuthenticationRequest, CallHttpAuthenticationResponse, CallHttpRequestActionRequest,
    CallTemplateFunctionArgs, CallTemplateFunctionRequest, CallTemplateFunctionResponse,
    CallWebsocketRequestActionRequest, CallWorkspaceActionRequest, Color, EmptyPayload,
    ErrorResponse, FilterRequest, FilterResponse, GetFolderActionsResponse,
    GetGrpcRequestActionsResponse, GetHttpAuthenticationConfigRequest,
    GetHttpAuthenticationConfigResponse, GetHttpAuthenticationSummaryResponse,
    GetHttpRequestActionsResponse, GetTemplateFunctionConfigRequest,
    GetTemplateFunctionConfigResponse, GetTemplateFunctionSummaryResponse, GetThemesRequest,
    GetThemesResponse, GetWebsocketRequestActionsResponse, GetWorkspaceActionsResponse, Icon,
    ImportRequest, ImportResponse, InternalEvent, InternalEventPayload, JsonPrimitive,
    PluginContext, RenderPurpose, ShowToastRequest,
};
use crate::native_template_functions::{template_function_keyring, template_function_secure};
use crate::nodejs::start_nodejs_plugin_runtime;
use crate::plugin_handle::PluginHandle;
use crate::server_ws::PluginRuntimeServerWebsocket;
use crate::template_callback::PluginTemplateCallback;
use log::{error, info, warn};
use serde_json::json;
use std::collections::HashMap;
use std::env;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use tauri::path::BaseDirectory;
use tauri::{AppHandle, Emitter, Manager, Runtime, WebviewWindow, is_dev};
use tokio::fs::read_dir;
use tokio::net::TcpListener;
use tokio::sync::mpsc::error::TrySendError;
use tokio::sync::{Mutex, mpsc};
use tokio::time::{Instant, timeout};
use yaak_models::models::{Environment, Plugin};
use yaak_models::query_manager::QueryManagerExt;
use yaak_models::render::make_vars_hashmap;
use yaak_models::util::{UpdateSource, generate_id};
use yaak_templates::error::Error::RenderError;
use yaak_templates::error::Result as TemplateResult;
use yaak_templates::{RenderErrorBehavior, RenderOptions, render_json_value_raw};

#[derive(Clone)]
pub struct PluginManager {
    subscribers: Arc<Mutex<HashMap<String, mpsc::Sender<InternalEvent>>>>,
    plugin_handles: Arc<Mutex<Vec<PluginHandle>>>,
    kill_tx: tokio::sync::watch::Sender<bool>,
    ws_service: Arc<PluginRuntimeServerWebsocket>,
    vendored_plugin_dir: PathBuf,
    pub(crate) installed_plugin_dir: PathBuf,
}

impl PluginManager {
    pub fn new<R: Runtime>(app_handle: AppHandle<R>) -> PluginManager {
        let (events_tx, mut events_rx) = mpsc::channel(128);
        let (kill_server_tx, kill_server_rx) = tokio::sync::watch::channel(false);

        let (client_disconnect_tx, mut client_disconnect_rx) = mpsc::channel(128);
        let (client_connect_tx, mut client_connect_rx) = tokio::sync::watch::channel(false);
        let ws_service =
            PluginRuntimeServerWebsocket::new(events_tx, client_disconnect_tx, client_connect_tx);

        let vendored_plugin_dir = app_handle
            .path()
            .resolve("vendored/plugins", BaseDirectory::Resource)
            .expect("failed to resolve plugin directory resource");

        let installed_plugin_dir = app_handle
            .path()
            .app_data_dir()
            .expect("failed to get app data dir")
            .join("installed-plugins");

        let plugin_manager = PluginManager {
            plugin_handles: Default::default(),
            subscribers: Default::default(),
            ws_service: Arc::new(ws_service.clone()),
            kill_tx: kill_server_tx,
            vendored_plugin_dir,
            installed_plugin_dir,
        };

        // Forward events to subscribers
        let subscribers = plugin_manager.subscribers.clone();
        tauri::async_runtime::spawn(async move {
            while let Some(event) = events_rx.recv().await {
                for (tx_id, tx) in subscribers.lock().await.iter_mut() {
                    if let Err(e) = tx.try_send(event.clone()) {
                        match e {
                            TrySendError::Full(e) => {
                                error!("Failed to send event to full subscriber {tx_id} {e:?}");
                            }
                            TrySendError::Closed(_) => {
                                // Subscriber already unsubscribed
                            }
                        }
                    }
                }
            }
        });

        // Handle when client plugin runtime disconnects
        tauri::async_runtime::spawn(async move {
            while (client_disconnect_rx.recv().await).is_some() {
                // Happens when the app is closed
                info!("Plugin runtime client disconnected");
            }
        });

        let listen_addr = match option_env!("YAAK_PLUGIN_SERVER_PORT") {
            Some(port) => format!("127.0.0.1:{port}"),
            None => "127.0.0.1:0".to_string(),
        };
        let listener = tauri::async_runtime::block_on(async move {
            TcpListener::bind(listen_addr).await.expect("Failed to bind TCP listener")
        });
        let addr = listener.local_addr().expect("Failed to get local address");

        // 1. Reload all plugins when the Node.js runtime connects
        let init_plugins_task = {
            let plugin_manager = plugin_manager.clone();
            let app_handle = app_handle.clone();
            tauri::async_runtime::spawn(async move {
                match client_connect_rx.changed().await {
                    Ok(_) => {
                        info!("Plugin runtime client connected!");
                        plugin_manager
                            .initialize_all_plugins(&app_handle, &PluginContext::new_empty())
                            .await
                            .expect("Failed to reload plugins");
                    }
                    Err(e) => {
                        warn!("Failed to receive from client connection rx {e:?}");
                    }
                }
            })
        };

        // 1. Spawn server in the background
        info!("Starting plugin server on {addr}");
        tauri::async_runtime::spawn(async move {
            ws_service.listen(listener).await;
        });

        // 2. Start Node.js runtime and initialize plugins
        tauri::async_runtime::block_on(async move {
            start_nodejs_plugin_runtime(&app_handle, addr, &kill_server_rx).await.unwrap();
            info!("Waiting for plugins to initialize");
            init_plugins_task.await.unwrap();
        });

        // 3. Block waiting for plugins to initialize
        tauri::async_runtime::block_on(async move {});

        plugin_manager
    }

    async fn list_available_plugins<R: Runtime>(
        &self,
        app_handle: &AppHandle<R>,
    ) -> Result<Vec<Plugin>> {
        let plugins_dir = if is_dev() {
            // Use plugins directly for easy development
            env::current_dir()
                .map(|cwd| cwd.join("../plugins").canonicalize().unwrap())
                .unwrap_or_else(|_| self.vendored_plugin_dir.to_path_buf())
        } else {
            self.vendored_plugin_dir.to_path_buf()
        };

        info!("Loading bundled plugins from {plugins_dir:?}");

        // Read bundled plugin directories from disk
        let bundled_plugin_dirs: Vec<String> = read_plugins_dir(&plugins_dir)
            .await
            .expect(&format!("Failed to read plugins dir: {:?}", plugins_dir));

        // Ensure all bundled plugins make it into the database
        for dir in &bundled_plugin_dirs {
            if app_handle.db().get_plugin_by_directory(dir).is_none() {
                app_handle.db().upsert_plugin(
                    &Plugin {
                        directory: dir.clone(),
                        enabled: true,
                        url: None,
                        ..Default::default()
                    },
                    &UpdateSource::Background,
                )?;
            }
        }

        Ok(app_handle.db().list_plugins()?)
    }

    pub async fn uninstall(&self, plugin_context: &PluginContext, dir: &str) -> Result<()> {
        let plugin = self.get_plugin_by_dir(dir).await.ok_or(PluginNotFoundErr(dir.to_string()))?;
        self.remove_plugin(plugin_context, &plugin).await
    }

    async fn remove_plugin(
        &self,
        plugin_context: &PluginContext,
        plugin: &PluginHandle,
    ) -> Result<()> {
        // Terminate the plugin if it's enabled
        if plugin.enabled {
            self.send_to_plugin_and_wait(
                plugin_context,
                plugin,
                &InternalEventPayload::TerminateRequest,
            )
            .await?;
        }

        // Remove the plugin from the list
        let mut plugins = self.plugin_handles.lock().await;
        let pos = plugins.iter().position(|p| p.ref_id == plugin.ref_id);
        if let Some(pos) = pos {
            plugins.remove(pos);
        }

        Ok(())
    }

    pub async fn add_plugin(&self, plugin_context: &PluginContext, plugin: &Plugin) -> Result<()> {
        info!("Adding plugin by dir {}", plugin.directory);

        let maybe_tx = self.ws_service.app_to_plugin_events_tx.lock().await;
        let tx = match &*maybe_tx {
            None => return Err(ClientNotInitializedErr),
            Some(tx) => tx,
        };
        let plugin_handle = PluginHandle::new(&plugin.directory, plugin.enabled, tx.clone())?;
        let dir_path = Path::new(&plugin.directory);
        let is_vendored = dir_path.starts_with(self.vendored_plugin_dir.as_path());
        let is_installed = dir_path.starts_with(self.installed_plugin_dir.as_path());

        // Boot the plugin if it's enabled
        if plugin.enabled {
            let event = self
                .send_to_plugin_and_wait(
                    plugin_context,
                    &plugin_handle,
                    &InternalEventPayload::BootRequest(BootRequest {
                        dir: plugin.directory.clone(),
                        watch: !is_vendored && !is_installed,
                    }),
                )
                .await?;

            if !matches!(event.payload, InternalEventPayload::BootResponse) {
                // Add it to the plugin handles anyway...
                let mut plugin_handles = self.plugin_handles.lock().await;
                plugin_handles.retain(|p| p.dir != plugin.directory);
                plugin_handles.push(plugin_handle.clone());
                return Err(UnknownEventErr);
            }
        }

        let mut plugin_handles = self.plugin_handles.lock().await;
        plugin_handles.retain(|p| p.dir != plugin.directory);
        plugin_handles.push(plugin_handle.clone());

        Ok(())
    }

    pub async fn initialize_all_plugins<R: Runtime>(
        &self,
        app_handle: &AppHandle<R>,
        plugin_context: &PluginContext,
    ) -> Result<()> {
        info!("Initializing all plugins");
        let start = Instant::now();
        for plugin in self.list_available_plugins(app_handle).await?.clone() {
            // First remove the plugin if it exists and is enabled
            if let Some(plugin_handle) = self.get_plugin_by_dir(&plugin.directory).await {
                if let Err(e) = self.remove_plugin(plugin_context, &plugin_handle).await {
                    error!("Failed to remove plugin {} {e:?}", plugin.directory);
                    continue;
                }
            }
            if let Err(e) = self.add_plugin(plugin_context, &plugin).await {
                warn!("Failed to add plugin {} {e:?}", plugin.directory);

                // Extract a user-friendly plugin name from the directory path
                let plugin_name = plugin.directory.split('/').last().unwrap_or(&plugin.directory);

                // Show a toast for all plugin failures
                let toast = ShowToastRequest {
                    message: format!("Failed to start plugin '{}': {}", plugin_name, e),
                    color: Some(Color::Danger),
                    icon: Some(Icon::AlertTriangle),
                    timeout: Some(10000),
                };

                if let Err(emit_err) = app_handle.emit("show_toast", toast) {
                    error!("Failed to emit toast for plugin error: {emit_err:?}");
                }
            }
        }

        let plugins = self.plugin_handles.lock().await;
        let names = plugins.iter().map(|p| p.dir.to_string()).collect::<Vec<String>>();
        info!(
            "Initialized {} plugins in {:?}:\n  - {}",
            plugins.len(),
            start.elapsed(),
            names.join("\n  - "),
        );

        Ok(())
    }

    pub async fn subscribe(&self, label: &str) -> (String, mpsc::Receiver<InternalEvent>) {
        let (tx, rx) = mpsc::channel(128);
        let rx_id = format!("{label}_{}", generate_id());
        self.subscribers.lock().await.insert(rx_id.clone(), tx);
        (rx_id, rx)
    }

    pub async fn unsubscribe(&self, rx_id: &str) {
        self.subscribers.lock().await.remove(rx_id);
    }

    pub async fn terminate(&self) {
        self.kill_tx.send_replace(true);

        // Give it a bit of time to kill
        tokio::time::sleep(Duration::from_millis(500)).await;
    }

    pub async fn reply(
        &self,
        source_event: &InternalEvent,
        payload: &InternalEventPayload,
    ) -> Result<()> {
        let plugin_context = source_event.to_owned().context;
        let reply_id = Some(source_event.to_owned().id);
        let plugin = self
            .get_plugin_by_ref_id(source_event.plugin_ref_id.as_str())
            .await
            .ok_or(PluginNotFoundErr(source_event.plugin_ref_id.to_string()))?;
        let event = plugin.build_event_to_send_raw(&plugin_context, &payload, reply_id);
        plugin.send(&event).await
    }

    pub async fn get_plugin_by_ref_id(&self, ref_id: &str) -> Option<PluginHandle> {
        self.plugin_handles.lock().await.iter().find(|p| p.ref_id == ref_id).cloned()
    }

    pub async fn get_plugin_by_dir(&self, dir: &str) -> Option<PluginHandle> {
        self.plugin_handles.lock().await.iter().find(|p| p.dir == dir).cloned()
    }

    pub async fn get_plugin_by_name(&self, name: &str) -> Option<PluginHandle> {
        for plugin in self.plugin_handles.lock().await.iter().cloned() {
            let info = plugin.info();
            if info.name == name {
                return Some(plugin);
            }
        }
        None
    }

    async fn send_to_plugin_and_wait(
        &self,
        plugin_context: &PluginContext,
        plugin: &PluginHandle,
        payload: &InternalEventPayload,
    ) -> Result<InternalEvent> {
        if !plugin.enabled {
            return Err(Error::PluginErr(format!("Plugin {} is disabled", plugin.metadata.name)));
        }

        let events =
            self.send_to_plugins_and_wait(plugin_context, payload, vec![plugin.to_owned()]).await?;
        Ok(events
            .first()
            .ok_or(Error::PluginErr(format!(
                "No plugin events returned for: {}",
                plugin.metadata.name
            )))?
            .to_owned())
    }

    async fn send_and_wait(
        &self,
        plugin_context: &PluginContext,
        payload: &InternalEventPayload,
    ) -> Result<Vec<InternalEvent>> {
        let plugins = { self.plugin_handles.lock().await.clone() };
        self.send_to_plugins_and_wait(plugin_context, payload, plugins).await
    }

    async fn send_to_plugins_and_wait(
        &self,
        plugin_context: &PluginContext,
        payload: &InternalEventPayload,
        plugins: Vec<PluginHandle>,
    ) -> Result<Vec<InternalEvent>> {
        let label = format!("wait[{}.{}]", plugins.len(), payload.type_name());
        let (rx_id, mut rx) = self.subscribe(label.as_str()).await;

        // 1. Build the events with IDs and everything
        let events_to_send = plugins
            .iter()
            .filter(|p| p.enabled)
            .map(|p| p.build_event_to_send(plugin_context, payload, None))
            .collect::<Vec<InternalEvent>>();

        // 2. Spawn thread to subscribe to incoming events and check reply ids
        let sub_events_fut = {
            let events_to_send = events_to_send.clone();

            tokio::spawn(async move {
                let mut found_events = Vec::new();

                let collect_events = async {
                    while let Some(event) = rx.recv().await {
                        let matched_sent_event =
                            events_to_send.iter().any(|e| Some(e.id.to_owned()) == event.reply_id);
                        if matched_sent_event {
                            found_events.push(event.clone());
                        };

                        let found_them_all = found_events.len() == events_to_send.len();
                        if found_them_all {
                            break;
                        }
                    }
                };

                // Timeout after 10 seconds to prevent hanging forever if plugin doesn't respond
                if timeout(Duration::from_secs(5), collect_events).await.is_err() {
                    warn!(
                        "Timeout waiting for plugin responses. Got {}/{} responses",
                        found_events.len(),
                        events_to_send.len()
                    );
                }

                found_events
            })
        };

        // 3. Send the events
        for event in events_to_send {
            let plugin = plugins
                .iter()
                .find(|p| p.ref_id == event.plugin_ref_id)
                .expect("Didn't find plugin in list");
            plugin.send(&event).await?
        }

        // 4. Join on the spawned thread
        let events = sub_events_fut.await.expect("Thread didn't succeed");

        // 5. Unsubscribe
        self.unsubscribe(&rx_id).await;

        Ok(events)
    }

    pub async fn get_themes<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
    ) -> Result<Vec<GetThemesResponse>> {
        let reply_events = self
            .send_and_wait(
                &PluginContext::new(window),
                &InternalEventPayload::GetThemesRequest(GetThemesRequest {}),
            )
            .await?;

        let mut themes = Vec::new();
        for event in reply_events {
            if let InternalEventPayload::GetThemesResponse(resp) = event.payload {
                themes.push(resp.clone());
            }
        }

        Ok(themes)
    }

    pub async fn get_grpc_request_actions<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
    ) -> Result<Vec<GetGrpcRequestActionsResponse>> {
        let reply_events = self
            .send_and_wait(
                &PluginContext::new(window),
                &InternalEventPayload::GetGrpcRequestActionsRequest(EmptyPayload {}),
            )
            .await?;

        let mut all_actions = Vec::new();
        for event in reply_events {
            if let InternalEventPayload::GetGrpcRequestActionsResponse(resp) = event.payload {
                all_actions.push(resp.clone());
            }
        }

        Ok(all_actions)
    }

    pub async fn get_http_request_actions<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
    ) -> Result<Vec<GetHttpRequestActionsResponse>> {
        let reply_events = self
            .send_and_wait(
                &PluginContext::new(window),
                &InternalEventPayload::GetHttpRequestActionsRequest(EmptyPayload {}),
            )
            .await?;

        let mut all_actions = Vec::new();
        for event in reply_events {
            if let InternalEventPayload::GetHttpRequestActionsResponse(resp) = event.payload {
                all_actions.push(resp.clone());
            }
        }

        Ok(all_actions)
    }

    pub async fn get_websocket_request_actions<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
    ) -> Result<Vec<GetWebsocketRequestActionsResponse>> {
        let reply_events = self
            .send_and_wait(
                &PluginContext::new(window),
                &InternalEventPayload::GetWebsocketRequestActionsRequest(EmptyPayload {}),
            )
            .await?;

        let mut all_actions = Vec::new();
        for event in reply_events {
            if let InternalEventPayload::GetWebsocketRequestActionsResponse(resp) = event.payload {
                all_actions.push(resp.clone());
            }
        }

        Ok(all_actions)
    }

    pub async fn get_workspace_actions<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
    ) -> Result<Vec<GetWorkspaceActionsResponse>> {
        let reply_events = self
            .send_and_wait(
                &PluginContext::new(window),
                &InternalEventPayload::GetWorkspaceActionsRequest(EmptyPayload {}),
            )
            .await?;

        let mut all_actions = Vec::new();
        for event in reply_events {
            if let InternalEventPayload::GetWorkspaceActionsResponse(resp) = event.payload {
                all_actions.push(resp.clone());
            }
        }

        Ok(all_actions)
    }

    pub async fn get_folder_actions<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
    ) -> Result<Vec<GetFolderActionsResponse>> {
        let reply_events = self
            .send_and_wait(
                &PluginContext::new(window),
                &InternalEventPayload::GetFolderActionsRequest(EmptyPayload {}),
            )
            .await?;

        let mut all_actions = Vec::new();
        for event in reply_events {
            if let InternalEventPayload::GetFolderActionsResponse(resp) = event.payload {
                all_actions.push(resp.clone());
            }
        }

        Ok(all_actions)
    }

    pub async fn get_template_function_config<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
        fn_name: &str,
        environment_chain: Vec<Environment>,
        values: HashMap<String, JsonPrimitive>,
        model_id: &str,
    ) -> Result<GetTemplateFunctionConfigResponse> {
        let results = self.get_template_function_summaries(window).await?;
        let r = results
            .iter()
            .find(|r| r.functions.iter().any(|f| f.name == fn_name))
            .ok_or_else(|| PluginNotFoundErr(fn_name.into()))?;

        let plugin = match self.get_plugin_by_ref_id(&r.plugin_ref_id).await {
            None => {
                // It's probably a native function, so just fallback to the summary
                let function = r
                    .functions
                    .iter()
                    .find(|f| f.name == fn_name)
                    .ok_or_else(|| PluginNotFoundErr(fn_name.into()))?;
                return Ok(GetTemplateFunctionConfigResponse {
                    function: function.clone(),
                    plugin_ref_id: r.plugin_ref_id.clone(),
                });
            }
            Some(v) => v,
        };

        let plugin_context = &PluginContext::new(&window);
        let vars = &make_vars_hashmap(environment_chain);
        let cb = PluginTemplateCallback::new(
            window.app_handle(),
            &plugin_context,
            RenderPurpose::Preview,
        );
        // We don't want to fail for this op because the UI will not be able to list any auth types then
        let render_opt = RenderOptions { error_behavior: RenderErrorBehavior::ReturnEmpty };
        let rendered_values = render_json_value_raw(json!(values), vars, &cb, &render_opt).await?;
        let context_id = format!("{:x}", md5::compute(model_id));

        let event = self
            .send_to_plugin_and_wait(
                &PluginContext::new(window),
                &plugin,
                &InternalEventPayload::GetTemplateFunctionConfigRequest(
                    GetTemplateFunctionConfigRequest {
                        values: serde_json::from_value(rendered_values)?,
                        name: fn_name.to_string(),
                        context_id,
                    },
                ),
            )
            .await?;
        match event.payload {
            InternalEventPayload::GetTemplateFunctionConfigResponse(resp) => Ok(resp),
            InternalEventPayload::EmptyResponse(_) => {
                Err(PluginErr("Template function plugin returned empty".to_string()))
            }
            InternalEventPayload::ErrorResponse(e) => Err(PluginErr(e.error)),
            e => Err(PluginErr(format!("Template function plugin returned invalid event {:?}", e))),
        }
    }

    pub async fn call_http_request_action<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
        req: CallHttpRequestActionRequest,
    ) -> Result<()> {
        let ref_id = req.plugin_ref_id.clone();
        let plugin =
            self.get_plugin_by_ref_id(ref_id.as_str()).await.ok_or(PluginNotFoundErr(ref_id))?;
        let event = plugin.build_event_to_send(
            &PluginContext::new(window),
            &InternalEventPayload::CallHttpRequestActionRequest(req),
            None,
        );
        plugin.send(&event).await?;
        Ok(())
    }

    pub async fn call_websocket_request_action<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
        req: CallWebsocketRequestActionRequest,
    ) -> Result<()> {
        let ref_id = req.plugin_ref_id.clone();
        let plugin =
            self.get_plugin_by_ref_id(ref_id.as_str()).await.ok_or(PluginNotFoundErr(ref_id))?;
        let event = plugin.build_event_to_send(
            &PluginContext::new(window),
            &InternalEventPayload::CallWebsocketRequestActionRequest(req),
            None,
        );
        plugin.send(&event).await?;
        Ok(())
    }

    pub async fn call_workspace_action<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
        req: CallWorkspaceActionRequest,
    ) -> Result<()> {
        let ref_id = req.plugin_ref_id.clone();
        let plugin =
            self.get_plugin_by_ref_id(ref_id.as_str()).await.ok_or(PluginNotFoundErr(ref_id))?;
        let event = plugin.build_event_to_send(
            &PluginContext::new(window),
            &InternalEventPayload::CallWorkspaceActionRequest(req),
            None,
        );
        plugin.send(&event).await?;
        Ok(())
    }

    pub async fn call_folder_action<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
        req: CallFolderActionRequest,
    ) -> Result<()> {
        let ref_id = req.plugin_ref_id.clone();
        let plugin =
            self.get_plugin_by_ref_id(ref_id.as_str()).await.ok_or(PluginNotFoundErr(ref_id))?;
        let event = plugin.build_event_to_send(
            &PluginContext::new(window),
            &InternalEventPayload::CallFolderActionRequest(req),
            None,
        );
        plugin.send(&event).await?;
        Ok(())
    }

    pub async fn call_grpc_request_action<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
        req: CallGrpcRequestActionRequest,
    ) -> Result<()> {
        let ref_id = req.plugin_ref_id.clone();
        let plugin =
            self.get_plugin_by_ref_id(ref_id.as_str()).await.ok_or(PluginNotFoundErr(ref_id))?;
        let event = plugin.build_event_to_send(
            &PluginContext::new(window),
            &InternalEventPayload::CallGrpcRequestActionRequest(req),
            None,
        );
        plugin.send(&event).await?;
        Ok(())
    }

    pub async fn get_http_authentication_summaries<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
    ) -> Result<Vec<(PluginHandle, GetHttpAuthenticationSummaryResponse)>> {
        let plugin_context = PluginContext::new(window);
        let reply_events = self
            .send_and_wait(
                &plugin_context,
                &InternalEventPayload::GetHttpAuthenticationSummaryRequest(EmptyPayload {}),
            )
            .await?;

        let mut results = Vec::new();
        for event in reply_events {
            if let InternalEventPayload::GetHttpAuthenticationSummaryResponse(resp) = event.payload
            {
                let plugin = self
                    .get_plugin_by_ref_id(&event.plugin_ref_id)
                    .await
                    .ok_or(PluginNotFoundErr(event.plugin_ref_id))?;
                results.push((plugin, resp.clone()));
            }
        }

        Ok(results)
    }

    pub async fn get_http_authentication_config<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
        environment_chain: Vec<Environment>,
        auth_name: &str,
        values: HashMap<String, JsonPrimitive>,
        model_id: &str,
    ) -> Result<GetHttpAuthenticationConfigResponse> {
        if auth_name == "none" {
            return Ok(GetHttpAuthenticationConfigResponse {
                args: Vec::new(),
                plugin_ref_id: "auth-none".to_string(),
                actions: None,
            });
        }

        let results = self.get_http_authentication_summaries(window).await?;
        let plugin = results
            .iter()
            .find_map(|(p, r)| if r.name == auth_name { Some(p) } else { None })
            .ok_or(PluginNotFoundErr(auth_name.into()))?;

        let vars = &make_vars_hashmap(environment_chain);
        let cb = PluginTemplateCallback::new(
            window.app_handle(),
            &PluginContext::new(&window),
            RenderPurpose::Preview,
        );
        // We don't want to fail for this op because the UI will not be able to list any auth types then
        let render_opt = RenderOptions { error_behavior: RenderErrorBehavior::ReturnEmpty };
        let rendered_values = render_json_value_raw(json!(values), vars, &cb, &render_opt).await?;
        let context_id = format!("{:x}", md5::compute(model_id));
        let event = self
            .send_to_plugin_and_wait(
                &PluginContext::new(window),
                &plugin,
                &InternalEventPayload::GetHttpAuthenticationConfigRequest(
                    GetHttpAuthenticationConfigRequest {
                        values: serde_json::from_value(rendered_values)?,
                        context_id,
                    },
                ),
            )
            .await?;
        match event.payload {
            InternalEventPayload::GetHttpAuthenticationConfigResponse(resp) => Ok(resp),
            InternalEventPayload::EmptyResponse(_) => {
                Err(PluginErr("Auth plugin returned empty".to_string()))
            }
            InternalEventPayload::ErrorResponse(e) => Err(PluginErr(e.error)),
            e => Err(PluginErr(format!("Auth plugin returned invalid event {:?}", e))),
        }
    }

    pub async fn call_http_authentication_action<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
        environment_chain: Vec<Environment>,
        auth_name: &str,
        action_index: i32,
        values: HashMap<String, JsonPrimitive>,
        model_id: &str,
    ) -> Result<()> {
        let vars = &make_vars_hashmap(environment_chain);
        let rendered_values = render_json_value_raw(
            json!(values),
            vars,
            &PluginTemplateCallback::new(
                window.app_handle(),
                &PluginContext::new(&window),
                RenderPurpose::Preview,
            ),
            &RenderOptions { error_behavior: RenderErrorBehavior::Throw },
        )
        .await?;
        let results = self.get_http_authentication_summaries(window).await?;
        let plugin = results
            .iter()
            .find_map(|(p, r)| if r.name == auth_name { Some(p) } else { None })
            .ok_or(PluginNotFoundErr(auth_name.into()))?;

        let context_id = format!("{:x}", md5::compute(model_id));
        self.send_to_plugin_and_wait(
            &PluginContext::new(window),
            &plugin,
            &InternalEventPayload::CallHttpAuthenticationActionRequest(
                CallHttpAuthenticationActionRequest {
                    index: action_index,
                    plugin_ref_id: plugin.clone().ref_id,
                    args: CallHttpAuthenticationActionArgs {
                        context_id,
                        values: serde_json::from_value(rendered_values)?,
                    },
                },
            ),
        )
        .await?;
        Ok(())
    }

    pub async fn call_http_authentication<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
        auth_name: &str,
        req: CallHttpAuthenticationRequest,
        plugin_context: &PluginContext,
    ) -> Result<CallHttpAuthenticationResponse> {
        let disabled = match req.values.get("disabled") {
            Some(JsonPrimitive::Boolean(v)) => *v,
            _ => false,
        };

        // Auth is disabled, so don't do anything
        if disabled {
            info!("Not applying disabled auth {:?}", auth_name);
            return Ok(CallHttpAuthenticationResponse {
                set_headers: None,
                set_query_parameters: None,
            });
        }

        let handlers = self.get_http_authentication_summaries(window).await?;
        let (plugin, _) = handlers
            .iter()
            .find(|(_, a)| a.name == auth_name)
            .ok_or(AuthPluginNotFound(auth_name.to_string()))?;

        let event = self
            .send_to_plugin_and_wait(
                plugin_context,
                &plugin,
                &InternalEventPayload::CallHttpAuthenticationRequest(req),
            )
            .await?;
        match event.payload {
            InternalEventPayload::CallHttpAuthenticationResponse(resp) => Ok(resp),
            InternalEventPayload::EmptyResponse(_) => {
                Err(PluginErr("Auth plugin returned empty".to_string()))
            }
            InternalEventPayload::ErrorResponse(e) => Err(PluginErr(e.error)),
            e => Err(PluginErr(format!("Auth plugin returned invalid event {:?}", e))),
        }
    }

    pub async fn get_template_function_summaries<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
    ) -> Result<Vec<GetTemplateFunctionSummaryResponse>> {
        let plugin_context = PluginContext::new(window);
        let reply_events = self
            .send_and_wait(
                &plugin_context,
                &InternalEventPayload::GetTemplateFunctionSummaryRequest(EmptyPayload {}),
            )
            .await?;

        let mut results = Vec::new();
        for event in reply_events {
            if let InternalEventPayload::GetTemplateFunctionSummaryResponse(resp) = event.payload {
                results.push(resp.clone());
            }
        }

        // Add Rust-based functions
        results.push(GetTemplateFunctionSummaryResponse {
            plugin_ref_id: "__NATIVE__".to_string(), // Meh
            functions: vec![template_function_secure(), template_function_keyring()],
        });

        Ok(results)
    }

    pub async fn call_template_function(
        &self,
        plugin_context: &PluginContext,
        fn_name: &str,
        values: HashMap<String, JsonPrimitive>,
        purpose: RenderPurpose,
    ) -> TemplateResult<String> {
        let req = CallTemplateFunctionRequest {
            name: fn_name.to_string(),
            args: CallTemplateFunctionArgs { purpose, values },
        };

        let events = self
            .send_and_wait(plugin_context, &InternalEventPayload::CallTemplateFunctionRequest(req))
            .await
            .map_err(|e| RenderError(format!("Failed to call template function {e:}")))?;

        let value =
            events.into_iter().find_map(|e| match e.payload {
                // Error returned
                InternalEventPayload::CallTemplateFunctionResponse(
                    CallTemplateFunctionResponse { error: Some(error), .. },
                ) => Some(Err(error)),
                // Value or null returned
                InternalEventPayload::CallTemplateFunctionResponse(
                    CallTemplateFunctionResponse { value, .. },
                ) => Some(Ok(value.unwrap_or_default())),
                // Generic error returned
                InternalEventPayload::ErrorResponse(ErrorResponse { error }) => Some(Err(error)),
                _ => None,
            });

        match value {
            None => Err(RenderError(format!("Template function {fn_name}(…) not found "))),
            Some(Ok(v)) => Ok(v),
            Some(Err(e)) => Err(RenderError(e)),
        }
    }

    pub async fn import_data<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
        content: &str,
    ) -> Result<ImportResponse> {
        let reply_events = self
            .send_and_wait(
                &PluginContext::new(window),
                &InternalEventPayload::ImportRequest(ImportRequest {
                    content: content.to_string(),
                }),
            )
            .await?;

        // TODO: Don't just return the first valid response
        let result = reply_events.into_iter().find_map(|e| match e.payload {
            InternalEventPayload::ImportResponse(resp) => Some(resp),
            _ => None,
        });

        match result {
            None => Err(PluginErr("No importers found for file contents".to_string())),
            Some(resp) => Ok(resp),
        }
    }

    pub async fn filter_data<R: Runtime>(
        &self,
        window: &WebviewWindow<R>,
        filter: &str,
        content: &str,
        content_type: &str,
    ) -> Result<FilterResponse> {
        let ct = content_type.to_lowercase();
        let plugin_name = if ct.contains("xml") || ct.contains("html") {
            "@yaak/filter-xpath"
        } else {
            "@yaak/filter-jsonpath"
        };

        let plugin = self
            .get_plugin_by_name(plugin_name)
            .await
            .ok_or(PluginNotFoundErr(plugin_name.to_string()))?;

        let event = self
            .send_to_plugin_and_wait(
                &PluginContext::new(window),
                &plugin,
                &InternalEventPayload::FilterRequest(FilterRequest {
                    filter: filter.to_string(),
                    content: content.to_string(),
                }),
            )
            .await?;

        match event.payload {
            InternalEventPayload::FilterResponse(resp) => Ok(resp),
            InternalEventPayload::EmptyResponse(_) => {
                Err(PluginErr("Filter returned empty".to_string()))
            }
            e => Err(PluginErr(format!("Export returned invalid event {:?}", e))),
        }
    }
}

async fn read_plugins_dir(dir: &PathBuf) -> Result<Vec<String>> {
    let mut result = read_dir(dir).await?;
    let mut dirs: Vec<String> = vec![];
    while let Ok(Some(entry)) = result.next_entry().await {
        if entry.path().is_dir() {
            #[cfg(target_os = "windows")]
            dirs.push(fix_windows_paths(&entry.path()));
            #[cfg(not(target_os = "windows"))]
            dirs.push(entry.path().to_string_lossy().to_string());
        }
    }
    Ok(dirs)
}

#[cfg(target_os = "windows")]
fn fix_windows_paths(p: &PathBuf) -> String {
    use dunce;
    use path_slash::PathBufExt;
    use regex::Regex;

    // 1. Remove UNC prefix for Windows paths to pass to sidecar
    let safe_path = dunce::simplified(p.as_path()).to_string_lossy().to_string();

    // 2. Remove the drive letter
    let safe_path = Regex::new("^[a-zA-Z]:").unwrap().replace(safe_path.as_str(), "");

    // 3. Convert backslashes to forward
    let safe_path = PathBuf::from(safe_path.to_string()).to_slash_lossy().to_string();

    safe_path
}