feat(plugins): introduce 'pipes', allowing users to pipe data to and control plugins from the command line (#3066)

* prototype - working with message from the cli

* prototype - pipe from the CLI to plugins

* prototype - pipe from the CLI to plugins and back again

* prototype - working with better cli interface

* prototype - working after removing unused stuff

* prototype - working with launching plugin if it is not launched, also fixed event ordering

* refactor: change message to cli-message

* prototype - allow plugins to send messages to each other

* fix: allow cli messages to send plugin parameters (and implement backpressure)

* fix: use input_pipe_id to identify cli pipes instead of their message name

* fix: come cleanups and add skip_cache parameter

* fix: pipe/client-server communication robustness

* fix: leaking messages between plugins while loading

* feat: allow plugins to specify how a new plugin instance is launched when sending messages

* fix: add permissions

* refactor: adjust cli api

* fix: improve cli plugin loading error messages

* docs: cli pipe

* fix: take plugin configuration into account when messaging between plugins

* refactor: pipe message protobuf interface

* refactor: update(event) -> pipe

* refactor - rename CliMessage to CliPipe

* fix: add is_private to pipes and change some naming

* refactor - cli client

* refactor: various cleanups

* style(fmt): rustfmt

* fix(pipes): backpressure across multiple plugins

* style: some cleanups

* style(fmt): rustfmt

* style: fix merge conflict mistake

* style(wording): clarify pipe permission

9 files changed, 295 insertions, 4 deletions

diff --git a/zellij-utils/src/plugin_api/action.proto b/zellij-utils/src/plugin_api/action.proto
index da10d82a8..0ed5b3b7a 100644
--- a/zellij-utils/src/plugin_api/action.proto
+++ b/zellij-utils/src/plugin_api/action.proto
@@ -53,9 +53,17 @@ message Action {
     IdAndName rename_tab_payload = 44;
     string rename_session_payload = 45;
     LaunchOrFocusPluginPayload launch_plugin_payload = 46;
+    CliPipePayload message_payload = 47;
   }
 }
 
+message CliPipePayload {
+  optional string name = 1;
+  string payload = 2;
+  repeated NameAndValue args = 3;
+  optional string plugin = 4;
+}
+
 message IdAndName {
   bytes name = 1;
   uint32 id = 2;
@@ -227,6 +235,7 @@ enum ActionName {
     BreakPaneLeft = 79;
     RenameSession = 80;
     LaunchPlugin = 81;
+    CliPipe = 82;
 }
 
 message Position {
diff --git a/zellij-utils/src/plugin_api/action.rs b/zellij-utils/src/plugin_api/action.rs
index 71eaaa130..82de4c7ee 100644
--- a/zellij-utils/src/plugin_api/action.rs
+++ b/zellij-utils/src/plugin_api/action.rs
@@ -1246,6 +1246,7 @@ impl TryFrom<Action> for ProtobufAction {
             | Action::Deny
             | Action::Copy
             | Action::DumpLayout
+            | Action::CliPipe { .. }
             | Action::SkipConfirm(..) => Err("Unsupported action"),
         }
     }
diff --git a/zellij-utils/src/plugin_api/mod.rs b/zellij-utils/src/plugin_api/mod.rs
index 40057fde0..0e26485e9 100644
--- a/zellij-utils/src/plugin_api/mod.rs
+++ b/zellij-utils/src/plugin_api/mod.rs
@@ -5,6 +5,7 @@ pub mod file;
 pub mod input_mode;
 pub mod key;
 pub mod message;
+pub mod pipe_message;
 pub mod plugin_command;
 pub mod plugin_ids;
 pub mod plugin_permission;
diff --git a/zellij-utils/src/plugin_api/pipe_message.proto b/zellij-utils/src/plugin_api/pipe_message.proto
new file mode 100644
index 000000000..5f488a758
--- /dev/null
+++ b/zellij-utils/src/plugin_api/pipe_message.proto
@@ -0,0 +1,23 @@
+syntax = "proto3";
+
+package api.pipe_message;
+
+message PipeMessage {
+    PipeSource source = 1;
+    optional string cli_source_id = 2;
+    optional uint32 plugin_source_id = 3;
+    string name = 4;
+    optional string payload = 5;
+    repeated Arg args = 6;
+    bool is_private = 7;
+}
+
+enum PipeSource {
+  Cli = 0;
+  Plugin = 1;
+}
+
+message Arg {
+  string key = 1;
+  string value = 2;
+}
diff --git a/zellij-utils/src/plugin_api/pipe_message.rs b/zellij-utils/src/plugin_api/pipe_message.rs
new file mode 100644
index 000000000..dbfe49305
--- /dev/null
+++ b/zellij-utils/src/plugin_api/pipe_message.rs
@@ -0,0 +1,71 @@
+pub use super::generated_api::api::pipe_message::{
+    Arg as ProtobufArg, PipeMessage as ProtobufPipeMessage, PipeSource as ProtobufPipeSource,
+};
+use crate::data::{PipeMessage, PipeSource};
+
+use std::convert::TryFrom;
+
+impl TryFrom<ProtobufPipeMessage> for PipeMessage {
+    type Error = &'static str;
+    fn try_from(protobuf_pipe_message: ProtobufPipeMessage) -> Result<Self, &'static str> {
+        let source = match (
+            ProtobufPipeSource::from_i32(protobuf_pipe_message.source),
+            protobuf_pipe_message.cli_source_id,
+            protobuf_pipe_message.plugin_source_id,
+        ) {
+            (Some(ProtobufPipeSource::Cli), Some(cli_source_id), _) => {
+                PipeSource::Cli(cli_source_id)
+            },
+            (Some(ProtobufPipeSource::Plugin), _, Some(plugin_source_id)) => {
+                PipeSource::Plugin(plugin_source_id)
+            },
+            _ => return Err("Invalid PipeSource or payload"),
+        };
+        let name = protobuf_pipe_message.name;
+        let payload = protobuf_pipe_message.payload;
+        let args = protobuf_pipe_message
+            .args
+            .into_iter()
+            .map(|arg| (arg.key, arg.value))
+            .collect();
+        let is_private = protobuf_pipe_message.is_private;
+        Ok(PipeMessage {
+            source,
+            name,
+            payload,
+            args,
+            is_private,
+        })
+    }
+}
+
+impl TryFrom<PipeMessage> for ProtobufPipeMessage {
+    type Error = &'static str;
+    fn try_from(pipe_message: PipeMessage) -> Result<Self, &'static str> {
+        let (source, cli_source_id, plugin_source_id) = match pipe_message.source {
+            PipeSource::Cli(input_pipe_id) => {
+                (ProtobufPipeSource::Cli as i32, Some(input_pipe_id), None)
+            },
+            PipeSource::Plugin(plugin_id) => {
+                (ProtobufPipeSource::Plugin as i32, None, Some(plugin_id))
+            },
+        };
+        let name = pipe_message.name;
+        let payload = pipe_message.payload;
+        let args: Vec<_> = pipe_message
+            .args
+            .into_iter()
+            .map(|(key, value)| ProtobufArg { key, value })
+            .collect();
+        let is_private = pipe_message.is_private;
+        Ok(ProtobufPipeMessage {
+            source,
+            cli_source_id,
+            plugin_source_id,
+            name,
+            payload,
+            args,
+            is_private,
+        })
+    }
+}
diff --git a/zellij-utils/src/plugin_api/plugin_command.proto b/zellij-utils/src/plugin_api/plugin_command.proto
index 53994a88c..6ffb0345b 100644
--- a/zellij-utils/src/plugin_api/plugin_command.proto
+++ b/zellij-utils/src/plugin_api/plugin_command.proto
@@ -87,6 +87,10 @@ enum CommandName {
   DeleteDeadSession = 73;
   DeleteAllDeadSessions = 74;
   RenameSession = 75;
+  UnblockCliPipeInput = 76;
+  BlockCliPipeInput = 77;
+  CliPipeOutput = 78;
+  MessageToPlugin = 79;
 }
 
 message PluginCommand {
@@ -137,9 +141,45 @@ message PluginCommand {
     WebRequestPayload web_request_payload = 44;
     string delete_dead_session_payload = 45;
     string rename_session_payload = 46;
+    string unblock_cli_pipe_input_payload = 47;
+    string block_cli_pipe_input_payload = 48;
+    CliPipeOutputPayload cli_pipe_output_payload = 49;
+    MessageToPluginPayload message_to_plugin_payload = 50;
   }
 }
 
+message CliPipeOutputPayload {
+  string pipe_name = 1;
+  string output = 2;
+}
+
+message MessageToPluginPayload {
+  optional string plugin_url = 1;
+  repeated ContextItem plugin_config = 2;
+  string message_name = 3;
+  optional string message_payload = 4;
+  repeated ContextItem message_args = 5;
+  optional NewPluginArgs new_plugin_args = 6;
+}
+
+message NewPluginArgs {
+    optional bool should_float = 1;
+    optional PaneId pane_id_to_replace = 2;
+    optional string pane_title = 3;
+    optional string cwd = 4;
+    bool skip_cache = 5;
+}
+
+message PaneId {
+  PaneType pane_type = 1;
+  uint32 id = 2;
+}
+
+enum PaneType {
+  Terminal = 0;
+  Plugin = 1;
+}
+
 message SwitchSessionPayload {
   optional string name = 1;
   optional uint32 tab_position = 2;
diff --git a/zellij-utils/src/plugin_api/plugin_command.rs b/zellij-utils/src/plugin_api/plugin_command.rs
index ed476687c..fa570a267 100644
--- a/zellij-utils/src/plugin_api/plugin_command.rs
+++ b/zellij-utils/src/plugin_api/plugin_command.rs
@@ -3,9 +3,11 @@ pub use super::generated_api::api::{
     event::{EventNameList as ProtobufEventNameList, Header},
     input_mode::InputMode as ProtobufInputMode,
     plugin_command::{
-        plugin_command::Payload, CommandName, ContextItem, EnvVariable, ExecCmdPayload,
-        HttpVerb as ProtobufHttpVerb, IdAndNewName, MovePayload, OpenCommandPanePayload,
-        OpenFilePayload, PluginCommand as ProtobufPluginCommand, PluginMessagePayload,
+        plugin_command::Payload, CliPipeOutputPayload, CommandName, ContextItem, EnvVariable,
+        ExecCmdPayload, HttpVerb as ProtobufHttpVerb, IdAndNewName, MessageToPluginPayload,
+        MovePayload, NewPluginArgs as ProtobufNewPluginArgs, OpenCommandPanePayload,
+        OpenFilePayload, PaneId as ProtobufPaneId, PaneType as ProtobufPaneType,
+        PluginCommand as ProtobufPluginCommand, PluginMessagePayload,
         RequestPluginPermissionPayload, ResizePayload, RunCommandPayload, SetTimeoutPayload,
         SubscribePayload, SwitchSessionPayload, SwitchTabToPayload, UnsubscribePayload,
         WebRequestPayload,
@@ -14,7 +16,10 @@ pub use super::generated_api::api::{
     resize::ResizeAction as ProtobufResizeAction,
 };
 
-use crate::data::{ConnectToSession, HttpVerb, PermissionType, PluginCommand};
+use crate::data::{
+    ConnectToSession, HttpVerb, MessageToPlugin, NewPluginArgs, PaneId, PermissionType,
+    PluginCommand,
+};
 
 use std::collections::BTreeMap;
 use std::convert::TryFrom;
@@ -42,6 +47,33 @@ impl Into<ProtobufHttpVerb> for HttpVerb {
     }
 }
 
+impl TryFrom<ProtobufPaneId> for PaneId {
+    type Error = &'static str;
+    fn try_from(protobuf_pane_id: ProtobufPaneId) -> Result<Self, &'static str> {
+        match ProtobufPaneType::from_i32(protobuf_pane_id.pane_type) {
+            Some(ProtobufPaneType::Terminal) => Ok(PaneId::Terminal(protobuf_pane_id.id)),
+            Some(ProtobufPaneType::Plugin) => Ok(PaneId::Plugin(protobuf_pane_id.id)),
+            None => Err("Failed to convert PaneId"),
+        }
+    }
+}
+
+impl TryFrom<PaneId> for ProtobufPaneId {
+    type Error = &'static str;
+    fn try_from(pane_id: PaneId) -> Result<Self, &'static str> {
+        match pane_id {
+            PaneId::Terminal(id) => Ok(ProtobufPaneId {
+                pane_type: ProtobufPaneType::Terminal as i32,
+                id,
+            }),
+            PaneId::Plugin(id) => Ok(ProtobufPaneId {
+                pane_type: ProtobufPaneType::Plugin as i32,
+                id,
+            }),
+        }
+    }
+}
+
 impl TryFrom<ProtobufPluginCommand> for PluginCommand {
     type Error = &'static str;
     fn try_from(protobuf_plugin_command: ProtobufPluginCommand) -> Result<Self, &'static str> {
@@ -641,6 +673,62 @@ impl TryFrom<ProtobufPluginCommand> for PluginCommand {
                 },
                 _ => Err("Mismatched payload for RenameSession"),
             },
+            Some(CommandName::UnblockCliPipeInput) => match protobuf_plugin_command.payload {
+                Some(Payload::UnblockCliPipeInputPayload(pipe_name)) => {
+                    Ok(PluginCommand::UnblockCliPipeInput(pipe_name))
+                },
+                _ => Err("Mismatched payload for UnblockPipeInput"),
+            },
+            Some(CommandName::BlockCliPipeInput) => match protobuf_plugin_command.payload {
+                Some(Payload::BlockCliPipeInputPayload(pipe_name)) => {
+                    Ok(PluginCommand::BlockCliPipeInput(pipe_name))
+                },
+                _ => Err("Mismatched payload for BlockPipeInput"),
+            },
+            Some(CommandName::CliPipeOutput) => match protobuf_plugin_command.payload {
+                Some(Payload::CliPipeOutputPayload(CliPipeOutputPayload { pipe_name, output })) => {
+                    Ok(PluginCommand::CliPipeOutput(pipe_name, output))
+                },
+                _ => Err("Mismatched payload for PipeOutput"),
+            },
+            Some(CommandName::MessageToPlugin) => match protobuf_plugin_command.payload {
+                Some(Payload::MessageToPluginPayload(MessageToPluginPayload {
+                    plugin_url,
+                    plugin_config,
+                    message_name,
+                    message_payload,
+                    message_args,
+                    new_plugin_args,
+                })) => {
+                    let plugin_config: BTreeMap<String, String> = plugin_config
+                        .into_iter()
+                        .map(|e| (e.name, e.value))
+                        .collect();
+                    let message_args: BTreeMap<String, String> = message_args
+                        .into_iter()
+                        .map(|e| (e.name, e.value))
+                        .collect();
+                    Ok(PluginCommand::MessageToPlugin(MessageToPlugin {
+                        plugin_url,
+                        plugin_config,
+                        message_name,
+                        message_payload,
+                        message_args,
+                        new_plugin_args: new_plugin_args.and_then(|protobuf_new_plugin_args| {
+                            Some(NewPluginArgs {
+                                should_float: protobuf_new_plugin_args.should_float,
+                                pane_id_to_replace: protobuf_new_plugin_args
+                                    .pane_id_to_replace
+                                    .and_then(|p_id| PaneId::try_from(p_id).ok()),
+                                pane_title: protobuf_new_plugin_args.pane_title,
+                                cwd: protobuf_new_plugin_args.cwd.map(|cwd| PathBuf::from(cwd)),
+                                skip_cache: protobuf_new_plugin_args.skip_cache,
+                            })
+                        }),
+                    }))
+                },
+                _ => Err("Mismatched payload for PipeOutput"),
+            },
             None => Err("Unrecognized plugin command"),
         }
     }
@@ -1069,6 +1157,54 @@ impl TryFrom<PluginCommand> for ProtobufPluginCommand {
                 name: CommandName::RenameSession as i32,
                 payload: Some(Payload::RenameSessionPayload(new_session_name)),
             }),
+            PluginCommand::UnblockCliPipeInput(pipe_name) => Ok(ProtobufPluginCommand {
+                name: CommandName::UnblockCliPipeInput as i32,
+                payload: Some(Payload::UnblockCliPipeInputPayload(pipe_name)),
+            }),
+            PluginCommand::BlockCliPipeInput(pipe_name) => Ok(ProtobufPluginCommand {
+                name: CommandName::BlockCliPipeInput as i32,
+                payload: Some(Payload::BlockCliPipeInputPayload(pipe_name)),
+            }),
+            PluginCommand::CliPipeOutput(pipe_name, output) => Ok(ProtobufPluginCommand {
+                name: CommandName::CliPipeOutput as i32,
+                payload: Some(Payload::CliPipeOutputPayload(CliPipeOutputPayload {
+                    pipe_name,
+                    output,
+                })),
+            }),
+            PluginCommand::MessageToPlugin(message_to_plugin) => {
+                let plugin_config: Vec<_> = message_to_plugin
+                    .plugin_config
+                    .into_iter()
+                    .map(|(name, value)| ContextItem { name, value })
+                    .collect();
+                let message_args: Vec<_> = message_to_plugin
+                    .message_args
+                    .into_iter()
+                    .map(|(name, value)| ContextItem { name, value })
+                    .collect();
+                Ok(ProtobufPluginCommand {
+                    name: CommandName::MessageToPlugin as i32,
+                    payload: Some(Payload::MessageToPluginPayload(MessageToPluginPayload {
+                        plugin_url: message_to_plugin.plugin_url,
+                        plugin_config,
+                        message_name: message_to_plugin.message_name,
+                        message_payload: message_to_plugin.message_payload,
+                        message_args,
+                        new_plugin_args: message_to_plugin.new_plugin_args.map(|m_t_p| {
+                            ProtobufNewPluginArgs {
+                                should_float: m_t_p.should_float,
+                                pane_id_to_replace: m_t_p
+                                    .pane_id_to_replace
+                                    .and_then(|p_id| ProtobufPaneId::try_from(p_id).ok()),
+                                pane_title: m_t_p.pane_title,
+                                cwd: m_t_p.cwd.map(|cwd| cwd.display().to_string()),
+                                skip_cache: m_t_p.skip_cache,
+                            }
+                        }),
+                    })),
+                })
+            },
         }
     }
 }
diff --git a/zellij-utils/src/plugin_api/plugin_permission.proto b/zellij-utils/src/plugin_api/plugin_permission.proto
index a796c7481..761384c1d 100644
--- a/zellij-utils/src/plugin_api/plugin_permission.proto
+++ b/zellij-utils/src/plugin_api/plugin_permission.proto
@@ -10,4 +10,6 @@ enum PermissionType {
   OpenTerminalsOrPlugins = 4;
   WriteToStdin = 5;
   WebAccess = 6;
+  ReadCliPipes = 7;
+  MessageAndLaunchOtherPlugins = 8;
 }
diff --git a/zellij-utils/src/plugin_api/plugin_permission.rs b/zellij-utils/src/plugin_api/plugin_permission.rs
index c9f0d49f5..4f258ac28 100644
--- a/zellij-utils/src/plugin_api/plugin_permission.rs
+++ b/zellij-utils/src/plugin_api/plugin_permission.rs
@@ -20,6 +20,10 @@ impl TryFrom<ProtobufPermissionType> for PermissionType {
             },
             ProtobufPermissionType::WriteToStdin => Ok(PermissionType::WriteToStdin),
             ProtobufPermissionType::WebAccess => Ok(PermissionType::WebAccess),
+            ProtobufPermissionType::ReadCliPipes => Ok(PermissionType::ReadCliPipes),
+            ProtobufPermissionType::MessageAndLaunchOtherPlugins => {
+                Ok(PermissionType::MessageAndLaunchOtherPlugins)
+            },
         }
     }
 }
@@ -41,6 +45,10 @@ impl TryFrom<PermissionType> for ProtobufPermissionType {
             },
             PermissionType::WriteToStdin => Ok(ProtobufPermissionType::WriteToStdin),
             PermissionType::WebAccess => Ok(ProtobufPermissionType::WebAccess),
+            PermissionType::ReadCliPipes => Ok(ProtobufPermissionType::ReadCliPipes),
+            PermissionType::MessageAndLaunchOtherPlugins => {
+                Ok(ProtobufPermissionType::MessageAndLaunchOtherPlugins)
+            },
         }
     }
 }