deploy 2.0.0-dev-53

Cargo.lock

@@ -926,7 +926,7 @@ dependencies = [
 
 [[package]]
 name = "cache"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "anyhow",
  "tokio",
@@ -1118,7 +1118,7 @@ dependencies = [
 
 [[package]]
 name = "command"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "komodo_client",
  "run_command",
@@ -1144,7 +1144,7 @@ checksum = "e47641d3deaf41fb1538ac1f54735925e275eaf3bf4d55c81b137fba797e5cbb"
 
 [[package]]
 name = "config"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "colored",
  "indexmap 2.11.4",
@@ -1429,7 +1429,7 @@ checksum = "2a2330da5de22e8a3cb63252ce2abb30116bf5265e89c0e01bc17015ce30a476"
 
 [[package]]
 name = "database"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "anyhow",
  "async-compression",
@@ -1707,7 +1707,7 @@ dependencies = [
 
 [[package]]
 name = "encoding"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "anyhow",
  "bytes",
@@ -1749,7 +1749,7 @@ dependencies = [
 
 [[package]]
 name = "environment"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "anyhow",
  "formatting",
@@ -1759,7 +1759,7 @@ dependencies = [
 
 [[package]]
 name = "environment_file"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "thiserror 2.0.17",
 ]
@@ -1855,7 +1855,7 @@ dependencies = [
 
 [[package]]
 name = "formatting"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "serror",
 ]
@@ -2027,7 +2027,7 @@ checksum = "e629b9b98ef3dd8afe6ca2bd0f89306cec16d43d907889945bc5d6687f2f13c7"
 
 [[package]]
 name = "git"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "anyhow",
  "cache",
@@ -2615,7 +2615,7 @@ dependencies = [
 
 [[package]]
 name = "interpolate"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "anyhow",
  "komodo_client",
@@ -2748,7 +2748,7 @@ dependencies = [
 
 [[package]]
 name = "komodo_cli"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "anyhow",
  "chrono",
@@ -2774,7 +2774,7 @@ dependencies = [
 
 [[package]]
 name = "komodo_client"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "anyhow",
  "async_timing_util",
@@ -2810,7 +2810,7 @@ dependencies = [
 
 [[package]]
 name = "komodo_core"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "anyhow",
  "arc-swap",
@@ -2883,7 +2883,7 @@ dependencies = [
 
 [[package]]
 name = "komodo_periphery"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "anyhow",
  "arc-swap",
@@ -3003,7 +3003,7 @@ checksum = "34080505efa8e45a4b816c349525ebe327ceaa8559756f0356cba97ef3bf7432"
 
 [[package]]
 name = "logger"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "anyhow",
  "komodo_client",
@@ -3275,7 +3275,7 @@ dependencies = [
 
 [[package]]
 name = "noise"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "anyhow",
  "arc-swap",
@@ -3695,7 +3695,7 @@ checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
 
 [[package]]
 name = "periphery_client"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "anyhow",
  "derive_variants",
@@ -4063,9 +4063,9 @@ dependencies = [
 
 [[package]]
 name = "regex"
-version = "1.11.3"
+version = "1.12.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8b5288124840bee7b386bc413c487869b360b2b4ec421ea56425128692f2a82c"
+checksum = "4a52d8d02cacdb176ef4678de6c052efb4b3da14b78e4db683a4252762be5433"
 dependencies = [
  "aho-corasick",
  "memchr",
@@ -4075,9 +4075,9 @@ dependencies = [
 
 [[package]]
 name = "regex-automata"
-version = "0.4.11"
+version = "0.4.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "833eb9ce86d40ef33cb1306d8accf7bc8ec2bfea4355cbdebb3df68b40925cad"
+checksum = "722166aa0d7438abbaa4d5cc2c649dac844e8c56d82fb3d33e9c34b5cd268fc6"
 dependencies = [
  "aho-corasick",
  "memchr",
@@ -4170,7 +4170,7 @@ dependencies = [
 
 [[package]]
 name = "response"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "anyhow",
  "axum",
@@ -4455,7 +4455,7 @@ dependencies = [
 
 [[package]]
 name = "secret_file"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "tokio",
 ]
@@ -5457,7 +5457,7 @@ dependencies = [
 
 [[package]]
 name = "transport"
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 dependencies = [
  "anyhow",
  "axum",

Cargo.toml

@@ -8,7 +8,7 @@ members = [
 ]
 
 [workspace.package]
-version = "2.0.0-dev-50"
+version = "2.0.0-dev-53"
 edition = "2024"
 authors = ["mbecker20 <becker.maxh@gmail.com>"]
 license = "GPL-3.0-or-later"
@@ -144,5 +144,5 @@ typeshare = "1.0.4"
 dashmap = "6.1.0"
 wildcard = "0.3.0"
 colored = "3.0.0"
-regex = "1.11.3"
+regex = "1.12.1"
 bytes = "1.10.1"

bin/cli/src/config.rs

@@ -28,7 +28,7 @@ pub fn cli_env() -> &'static Env {
     {
       Ok(env) => env,
       Err(e) => {
-        panic!("{e:?}");
+        panic!("{e:?}")
       }
     }
   })

bin/core/src/alert/discord.rs

@@ -50,7 +50,7 @@ pub async fn send_alert(
       match alert.level {
         SeverityLevel::Ok => {
           format!(
-            "{level} | **{name}**{region} is now **reachable**\n{link}"
+            "{level} | **{name}**{region} is now **connected**\n{link}"
           )
         }
         SeverityLevel::Critical => {
@@ -241,31 +241,33 @@ pub async fn send_alert(
     }
     AlertData::None {} => Default::default(),
   };
-  if !content.is_empty() {
-    let VariablesAndSecrets { variables, secrets } =
-      get_variables_and_secrets().await?;
-    let mut url_interpolated = url.to_string();
-
-    let mut interpolator =
-      Interpolator::new(Some(&variables), &secrets);
-
-    interpolator.interpolate_string(&mut url_interpolated)?;
-
-    send_message(&url_interpolated, &content)
-      .await
-      .map_err(|e| {
-        let replacers = interpolator
-          .secret_replacers
-          .into_iter()
-          .collect::<Vec<_>>();
-        let sanitized_error =
-          svi::replace_in_string(&format!("{e:?}"), &replacers);
-        anyhow::Error::msg(format!(
-          "Error with slack request: {sanitized_error}"
-        ))
-      })?;
+  
+  if content.is_empty() {
+    return Ok(());
   }
-  Ok(())
+
+  let VariablesAndSecrets { variables, secrets } =
+    get_variables_and_secrets().await?;
+  let mut url_interpolated = url.to_string();
+
+  let mut interpolator =
+    Interpolator::new(Some(&variables), &secrets);
+
+  interpolator.interpolate_string(&mut url_interpolated)?;
+
+  send_message(&url_interpolated, &content)
+    .await
+    .map_err(|e| {
+      let replacers = interpolator
+        .secret_replacers
+        .into_iter()
+        .collect::<Vec<_>>();
+      let sanitized_error =
+        svi::replace_in_string(&format!("{e:?}"), &replacers);
+      anyhow::Error::msg(format!(
+        "Error with slack request: {sanitized_error}"
+      ))
+    })
 }
 
 async fn send_message(

bin/core/src/alert/mod.rs

@@ -294,7 +294,7 @@ fn standard_alert_content(alert: &Alert) -> String {
       let link = resource_link(ResourceTargetVariant::Server, id);
       match alert.level {
         SeverityLevel::Ok => {
-          format!("{level} | {name}{region} is now reachable\n{link}")
+          format!("{level} | {name}{region} is now connected\n{link}")
         }
         SeverityLevel::Critical => {
           let err = err

bin/core/src/alert/ntfy.rs

@@ -9,10 +9,32 @@ pub async fn send_alert(
   alert: &Alert,
 ) -> anyhow::Result<()> {
   let content = standard_alert_content(alert);
-  if !content.is_empty() {
-    send_message(url, email, content).await?;
+  if content.is_empty() {
+    return Ok(());
   }
-  Ok(())
+
+  let VariablesAndSecrets { variables, secrets } =
+    get_variables_and_secrets().await?;
+  let mut url_interpolated = url.to_string();
+
+  let mut interpolator =
+    Interpolator::new(Some(&variables), &secrets);
+
+  interpolator.interpolate_string(&mut url_interpolated)?;
+
+  send_message(&url_interpolated, email, content)
+    .await
+    .map_err(|e| {
+      let replacers = interpolator
+        .secret_replacers
+        .into_iter()
+        .collect::<Vec<_>>();
+      let sanitized_error =
+        svi::replace_in_string(&format!("{e:?}"), &replacers);
+      anyhow::Error::msg(format!(
+        "Error with slack request: {sanitized_error}"
+      ))
+    })
 }
 
 async fn send_message(
@@ -22,7 +44,7 @@ async fn send_message(
 ) -> anyhow::Result<()> {
   let mut request = http_client()
     .post(url)
-    .header("Title", "ntfy Alert")
+    .header("Title", "Komodo Alert")
     .body(content);
 
   if let Some(email) = email {
@@ -43,9 +65,7 @@ async fn send_message(
       )
     })?;
     Err(anyhow!(
-      "Failed to send message to ntfy | {} | {}",
-      status,
-      text
+      "Failed to send message to ntfy | {status} | {text}",
     ))
   }
 }

bin/core/src/alert/pushover.rs

@@ -8,10 +8,30 @@ pub async fn send_alert(
   alert: &Alert,
 ) -> anyhow::Result<()> {
   let content = standard_alert_content(alert);
-  if !content.is_empty() {
-    send_message(url, content).await?;
+  if content.is_empty() {
+    return Ok(());
   }
-  Ok(())
+
+  let VariablesAndSecrets { variables, secrets } =
+    get_variables_and_secrets().await?;
+  let mut url_interpolated = url.to_string();
+
+  let mut interpolator =
+    Interpolator::new(Some(&variables), &secrets);
+
+  interpolator.interpolate_string(&mut url_interpolated)?;
+
+  send_message(&url_interpolated, content).await.map_err(|e| {
+    let replacers = interpolator
+      .secret_replacers
+      .into_iter()
+      .collect::<Vec<_>>();
+    let sanitized_error =
+      svi::replace_in_string(&format!("{e:?}"), &replacers);
+    anyhow::Error::msg(format!(
+      "Error with slack request: {sanitized_error}"
+    ))
+  })
 }
 
 async fn send_message(

bin/core/src/alert/slack.rs

@@ -64,11 +64,11 @@ pub async fn send_alert(
       match alert.level {
         SeverityLevel::Ok => {
           let text =
-            format!("{level} | *{name}*{region} is now *reachable*");
+            format!("{level} | *{name}*{region} is now *connected*");
           let blocks = vec![
             Block::header(level),
             Block::section(format!(
-              "*{name}*{region} is now *reachable*"
+              "*{name}*{region} is now *connnected*"
             )),
           ];
           (text, blocks.into())
@@ -468,31 +468,32 @@ pub async fn send_alert(
     }
     AlertData::None {} => Default::default(),
   };
-  if !text.is_empty() {
-    let VariablesAndSecrets { variables, secrets } =
-      get_variables_and_secrets().await?;
-    let mut url_interpolated = url.to_string();
-
-    let mut interpolator =
-      Interpolator::new(Some(&variables), &secrets);
-
-    interpolator.interpolate_string(&mut url_interpolated)?;
-
-    let slack = ::slack::Client::new(url_interpolated);
-    slack
-      .send_owned_message_single(&text, blocks.as_deref())
-      .await
-      .map_err(|e| {
-        let replacers = interpolator
-          .secret_replacers
-          .into_iter()
-          .collect::<Vec<_>>();
-        let sanitized_error =
-          svi::replace_in_string(&format!("{e:?}"), &replacers);
-        anyhow::Error::msg(format!(
-          "Error with slack request: {sanitized_error}"
-        ))
-      })?;
+  if text.is_empty() {
+    return Ok(());
   }
+  let VariablesAndSecrets { variables, secrets } =
+    get_variables_and_secrets().await?;
+  let mut url_interpolated = url.to_string();
+
+  let mut interpolator =
+    Interpolator::new(Some(&variables), &secrets);
+
+  interpolator.interpolate_string(&mut url_interpolated)?;
+
+  let slack = ::slack::Client::new(url_interpolated);
+  slack
+    .send_owned_message_single(&text, blocks.as_deref())
+    .await
+    .map_err(|e| {
+      let replacers = interpolator
+        .secret_replacers
+        .into_iter()
+        .collect::<Vec<_>>();
+      let sanitized_error =
+        svi::replace_in_string(&format!("{e:?}"), &replacers);
+      anyhow::Error::msg(format!(
+        "Error with slack request: {sanitized_error}"
+      ))
+    })?;
   Ok(())
 }

bin/core/src/connection/client.rs

@@ -142,7 +142,7 @@ async fn handle_passkey_login(
     };
 
     socket
-      .send(LoginMessage::V1Passkey(passkey))
+      .send_message(LoginMessage::V1Passkey(passkey))
       .await
       .context("Failed to send Login V1Passkey message")?;
 

bin/core/src/connection/mod.rs

@@ -10,8 +10,8 @@ use anyhow::anyhow;
 use cache::CloneCache;
 use database::mungos::{by_id::update_one_by_id, mongodb::bson::doc};
 use encoding::{
-  CastBytes as _, Decode as _, EncodedJsonMessage, EncodedOption,
-  EncodedResult, WithChannel,
+  CastBytes as _, Decode as _, EncodedJsonMessage, EncodedResponse,
+  WithChannel,
 };
 use komodo_client::entities::{
   builder::{AwsBuilderConfig, UrlBuilderConfig},
@@ -19,7 +19,7 @@ use komodo_client::entities::{
   server::Server,
 };
 use periphery_client::transport::{
-  EncodedTransportMessage, TransportMessage,
+  EncodedTransportMessage, ResponseMessage, TransportMessage,
 };
 use serror::serror_into_anyhow_error;
 use tokio::sync::RwLock;
@@ -253,10 +253,8 @@ impl<'a> From<&'a OwnedPeripheryConnectionArgs>
 }
 
 /// Sends None as InProgress ping.
-pub type ResponseChannels = CloneCache<
-  Uuid,
-  Sender<EncodedOption<EncodedResult<EncodedJsonMessage>>>,
->;
+pub type ResponseChannels =
+  CloneCache<Uuid, Sender<EncodedResponse<EncodedJsonMessage>>>;
 
 pub type TerminalChannels = CloneCache<Uuid, Sender<Vec<u8>>>;
 
@@ -365,7 +363,7 @@ impl PeripheryConnection {
         let Ok(message) = receiver.recv().await else {
           break;
         };
-        match ws_write.send_inner(message.into_bytes()).await {
+        match ws_write.send(message.into_bytes()).await {
           Ok(_) => receiver.clear_buffer(),
           Err(e) => {
             self.set_error(e).await;
@@ -380,7 +378,7 @@ impl PeripheryConnection {
 
     let handle_reads = async {
       loop {
-        match ws_read.recv_inner().await {
+        match ws_read.recv().await {
           Ok(WebsocketMessage::Message(message)) => {
             self.handle_incoming_message(message).await
           }
@@ -415,28 +413,28 @@ impl PeripheryConnection {
       }
     };
     match message {
-      TransportMessage::Response(data) => match data.decode() {
-        Ok(WithChannel {
-          channel: channel_id,
-          data,
-        }) => {
-          let Some(channel) = self.responses.get(&channel_id).await
-          else {
-            warn!(
-              "Failed to forward Response message | No response channel found at {channel_id}"
-            );
-            return;
-          };
-          if let Err(e) = channel.send(data).await {
-            warn!(
-              "Failed to send response | Channel failure at {channel_id} | {e:#}"
-            );
+      TransportMessage::Response(data) => {
+        match data.decode().map(ResponseMessage::into_inner) {
+          Ok(WithChannel { channel, data }) => {
+            let Some(response_channel) =
+              self.responses.get(&channel).await
+            else {
+              warn!(
+                "Failed to forward Response message | No response channel found at {channel}"
+              );
+              return;
+            };
+            if let Err(e) = response_channel.send(data).await {
+              warn!(
+                "Failed to forward Response | Response channel failure at {channel} | {e:#}"
+              );
+            }
+          }
+          Err(e) => {
+            warn!("Failed to read Response message | {e:#}");
           }
         }
-        Err(e) => {
-          warn!("Failed to read Response message | {e:#}");
-        }
-      },
+      }
       TransportMessage::Terminal(data) => match data.decode() {
         Ok(WithChannel {
           channel: channel_id,

bin/core/src/connection/server.rs

@@ -17,11 +17,12 @@ use komodo_client::{
     user::system_user,
   },
 };
-use periphery_client::transport::LoginMessage;
+use periphery_client::{
+  api::PeripheryConnectionQuery, transport::LoginMessage,
+};
 use resolver_api::Resolve;
 use serror::{AddStatusCode, AddStatusCodeError};
 use transport::{
-  PeripheryConnectionQuery,
   auth::{
     HeaderConnectionIdentifiers, LoginFlow, LoginFlowArgs,
     PublicKeyValidator, ServerLoginFlow,
@@ -124,7 +125,7 @@ async fn existing_server_handler(
     let mut socket = AxumWebsocket(socket);
 
     if let Err(e) = socket
-      .send(LoginMessage::OnboardingFlow(false))
+      .send_message(LoginMessage::OnboardingFlow(false))
       .await
       .context("Failed to send Login OnboardingFlow false message")
     {
@@ -157,7 +158,7 @@ async fn onboard_server_handler(
       format!("server={}", urlencoding::encode(&server_query));
     let mut socket = AxumWebsocket(socket);
 
-    if let Err(e) = socket.send(LoginMessage::OnboardingFlow(true)).await.context(
+    if let Err(e) = socket.send_message(LoginMessage::OnboardingFlow(true)).await.context(
       "Failed to send Login OnboardingFlow true message",
     ).context("Server onboarding error") {
       warn!("{e:#}");
@@ -214,7 +215,7 @@ async fn onboard_server_handler(
     };
 
     if let Err(e) = socket
-      .send(LoginMessage::Success)
+      .send_message(LoginMessage::Success)
       .await
       .context("Failed to send Login Onboarding Successful message")
     {

bin/core/src/main.rs

@@ -6,7 +6,7 @@ extern crate tracing;
 use std::{net::SocketAddr, str::FromStr};
 
 use anyhow::Context;
-use axum::Router;
+use axum::{Router, routing::get};
 use axum_server::{Handle, tls_rustls::RustlsConfig};
 use tower_http::{
   cors::{Any, CorsLayer},
@@ -90,6 +90,7 @@ async fn app() -> anyhow::Result<()> {
     .not_found_service(frontend_index.clone());
 
   let app = Router::new()
+    .route("/version", get(|| async { env!("CARGO_PKG_VERSION") }))
     .nest("/auth", api::auth::router())
     .nest("/user", api::user::router())
     .nest("/read", api::read::router())

bin/core/src/periphery/mod.rs

@@ -139,7 +139,7 @@ impl PeripheryClient {
         continue;
       };
 
-      return message.decode_into();
+      return message.decode();
     }
   }
 }

bin/periphery/src/api/container.rs

@@ -12,10 +12,9 @@ use periphery_client::api::container::*;
 use resolver_api::Resolve;
 
 use crate::{
-  docker::{
-    docker_client, stats::get_container_stats, stop_container_command,
-  },
+  docker::{stats::get_container_stats, stop_container_command},
   helpers::format_log_grep,
+  state::docker_client,
 };
 
 // ======
@@ -30,7 +29,12 @@ impl Resolve<super::Args> for InspectContainer {
     self,
     _: &super::Args,
   ) -> anyhow::Result<Container> {
-    docker_client().inspect_container(&self.name).await
+    let client = docker_client().load();
+    let client = client
+      .iter()
+      .next()
+      .context("Could not connect to docker client")?;
+    client.inspect_container(&self.name).await
   }
 }
 
@@ -106,7 +110,12 @@ impl Resolve<super::Args> for GetFullContainerStats {
     self,
     _: &super::Args,
   ) -> anyhow::Result<FullContainerStats> {
-    docker_client().full_container_stats(&self.name).await
+    let client = docker_client().load();
+    let client = client
+      .iter()
+      .next()
+      .context("Could not connect to docker client")?;
+    client.full_container_stats(&self.name).await
   }
 }
 
@@ -279,7 +288,12 @@ impl Resolve<super::Args> for StartAllContainers {
     self,
     _: &super::Args,
   ) -> anyhow::Result<Vec<Log>> {
-    let containers = docker_client()
+    let client = docker_client().load();
+    let client = client
+      .iter()
+      .next()
+      .context("Could not connect to docker client")?;
+    let containers = client
       .list_containers()
       .await
       .context("failed to list all containers on host")?;
@@ -308,7 +322,12 @@ impl Resolve<super::Args> for RestartAllContainers {
     self,
     _: &super::Args,
   ) -> anyhow::Result<Vec<Log>> {
-    let containers = docker_client()
+    let client = docker_client().load();
+    let client = client
+      .iter()
+      .next()
+      .context("Could not connect to docker client")?;
+    let containers = client
       .list_containers()
       .await
       .context("failed to list all containers on host")?;
@@ -337,7 +356,12 @@ impl Resolve<super::Args> for PauseAllContainers {
     self,
     _: &super::Args,
   ) -> anyhow::Result<Vec<Log>> {
-    let containers = docker_client()
+    let client = docker_client().load();
+    let client = client
+      .iter()
+      .next()
+      .context("Could not connect to docker client")?;
+    let containers = client
       .list_containers()
       .await
       .context("failed to list all containers on host")?;
@@ -366,7 +390,12 @@ impl Resolve<super::Args> for UnpauseAllContainers {
     self,
     _: &super::Args,
   ) -> anyhow::Result<Vec<Log>> {
-    let containers = docker_client()
+    let client = docker_client().load();
+    let client = client
+      .iter()
+      .next()
+      .context("Could not connect to docker client")?;
+    let containers = client
       .list_containers()
       .await
       .context("failed to list all containers on host")?;
@@ -395,7 +424,12 @@ impl Resolve<super::Args> for StopAllContainers {
     self,
     _: &super::Args,
   ) -> anyhow::Result<Vec<Log>> {
-    let containers = docker_client()
+    let client = docker_client().load();
+    let client = client
+      .iter()
+      .next()
+      .context("Could not connect to docker client")?;
+    let containers = client
       .list_containers()
       .await
       .context("failed to list all containers on host")?;

bin/periphery/src/api/docker.rs

@@ -1,5 +1,6 @@
 use std::sync::OnceLock;
 
+use anyhow::Context;
 use cache::TimeoutCache;
 use command::run_komodo_command;
 use komodo_client::entities::{
@@ -15,7 +16,7 @@ use komodo_client::entities::{
 use periphery_client::api::docker::*;
 use resolver_api::Resolve;
 
-use crate::docker::{docker_client, docker_login};
+use crate::{docker::docker_login, state::docker_client};
 
 // =====
 // IMAGE
@@ -24,7 +25,12 @@ use crate::docker::{docker_client, docker_login};
 impl Resolve<super::Args> for InspectImage {
   #[instrument(name = "InspectImage", level = "debug")]
   async fn resolve(self, _: &super::Args) -> anyhow::Result<Image> {
-    docker_client().inspect_image(&self.name).await
+    let client = docker_client().load();
+    let client = client
+      .iter()
+      .next()
+      .context("Could not connect to docker client")?;
+    client.inspect_image(&self.name).await
   }
 }
 
@@ -36,7 +42,12 @@ impl Resolve<super::Args> for ImageHistory {
     self,
     _: &super::Args,
   ) -> anyhow::Result<Vec<ImageHistoryResponseItem>> {
-    docker_client().image_history(&self.name).await
+    let client = docker_client().load();
+    let client = client
+      .iter()
+      .next()
+      .context("Could not connect to docker client")?;
+    client.image_history(&self.name).await
   }
 }
 
@@ -125,7 +136,12 @@ impl Resolve<super::Args> for PruneImages {
 impl Resolve<super::Args> for InspectNetwork {
   #[instrument(name = "InspectNetwork", level = "debug")]
   async fn resolve(self, _: &super::Args) -> anyhow::Result<Network> {
-    docker_client().inspect_network(&self.name).await
+    let client = docker_client().load();
+    let client = client
+      .iter()
+      .next()
+      .context("Could not connect to docker client")?;
+    client.inspect_network(&self.name).await
   }
 }
 
@@ -171,7 +187,12 @@ impl Resolve<super::Args> for PruneNetworks {
 impl Resolve<super::Args> for InspectVolume {
   #[instrument(name = "InspectVolume", level = "debug")]
   async fn resolve(self, _: &super::Args) -> anyhow::Result<Volume> {
-    docker_client().inspect_volume(&self.name).await
+    let client = docker_client().load();
+    let client = client
+      .iter()
+      .next()
+      .context("Could not connect to docker client")?;
+    client.inspect_volume(&self.name).await
   }
 }
 

bin/periphery/src/api/keys.rs

@@ -6,8 +6,8 @@ use periphery_client::api::keys::{
 use resolver_api::Resolve;
 
 use crate::{
-  config::{periphery_config, periphery_keys},
-  connection::core_public_keys,
+  config::periphery_config,
+  state::{core_public_keys, periphery_keys},
 };
 
 //

bin/periphery/src/api/mod.rs

@@ -1,6 +1,6 @@
 use command::run_komodo_command;
 use derive_variants::EnumVariants;
-use encoding::{EncodedJsonMessage, EncodedResult};
+use encoding::{EncodedJsonMessage, EncodedResponse};
 use futures::FutureExt;
 use komodo_client::entities::{
   config::{DockerRegistry, GitProvider},
@@ -17,9 +17,8 @@ use serde::{Deserialize, Serialize};
 
 use crate::{
   api::compose::list_compose_projects,
-  config::{periphery_config, periphery_keys},
-  docker::docker_client,
-  stats::stats_client,
+  config::periphery_config,
+  state::{docker_client, periphery_keys, stats_client},
 };
 
 pub mod terminal;
@@ -41,7 +40,7 @@ pub struct Args {
   Serialize, Deserialize, Debug, Clone, Resolve, EnumVariants,
 )]
 #[args(Args)]
-#[response(EncodedResult<EncodedJsonMessage>)]
+#[response(EncodedResponse<EncodedJsonMessage>)]
 #[error(anyhow::Error)]
 #[variant_derive(Debug)]
 #[serde(tag = "type", content = "params")]
@@ -189,19 +188,22 @@ impl Resolve<Args> for PollStatus {
   ) -> anyhow::Result<PollStatusResponse> {
     // Docker lists
     let docker_lists = async {
-      let docker = docker_client();
+      let client = docker_client().load();
+      let Some(client) = client.iter().next() else {
+        return Default::default();
+      };
       let containers =
-        docker.list_containers().await.unwrap_or_default();
+        client.list_containers().await.unwrap_or_default();
       // Todo: handle errors better
       (
         tokio::join!(
-          docker
+          client
             .list_networks(&containers)
             .map(Result::unwrap_or_default),
-          docker
+          client
             .list_images(&containers)
             .map(Result::unwrap_or_default),
-          docker
+          client
             .list_volumes(&containers)
             .map(Result::unwrap_or_default)
         ),
@@ -235,6 +237,7 @@ impl Resolve<Args> for PollStatus {
         container_terminals_disabled: config
           .disable_container_terminals,
         stats_polling_rate: config.stats_polling_rate,
+        docker_connected: docker_client().load().is_some(),
       },
       system_info: stats_client.info.clone(),
       system_stats,

bin/periphery/src/api/terminal.rs

@@ -15,7 +15,12 @@ use transport::channel::{BufferedChannel, Sender};
 use uuid::Uuid;
 
 use crate::{
-  config::periphery_config, connection::core_connections, terminal::*,
+  config::periphery_config,
+  state::{
+    TerminalChannel, core_connections, terminal_channels,
+    terminal_triggers,
+  },
+  terminal::*,
 };
 
 //

bin/periphery/src/config.rs

@@ -10,27 +10,6 @@ use komodo_client::entities::{
   config::periphery::{CliArgs, Env, PeripheryConfig},
   logger::{LogConfig, LogLevel},
 };
-use noise::key::RotatableKeyPair;
-
-/// Should call in startup to ensure Periphery errors without valid private key.
-pub fn periphery_keys() -> &'static RotatableKeyPair {
-  static PERIPHERY_KEYS: OnceLock<RotatableKeyPair> = OnceLock::new();
-  PERIPHERY_KEYS.get_or_init(|| {
-    let config = periphery_config();
-    if let Some(private_key_spec) = config.private_key.as_deref() {
-      RotatableKeyPair::from_private_key_spec(private_key_spec)
-    } else {
-      RotatableKeyPair::from_private_key_spec(
-        config
-          .root_directory
-          .join("keys/periphery.key")
-          .to_str()
-          .expect("Invalid root directory"),
-      )
-    }
-    .unwrap()
-  })
-}
 
 pub fn periphery_args() -> &'static CliArgs {
   static PERIPHERY_ARGS: OnceLock<CliArgs> = OnceLock::new();

bin/periphery/src/connection/client.rs

@@ -19,8 +19,9 @@ use transport::{
 
 use crate::{
   api::Args,
-  config::{periphery_config, periphery_keys},
-  connection::{core_connections, core_public_keys},
+  config::periphery_config,
+  connection::core_public_keys,
+  state::{core_connections, periphery_keys},
 };
 
 pub async fn handler(address: &str) -> anyhow::Result<()> {
@@ -161,7 +162,7 @@ async fn handle_onboarding(
 
   // Post onboarding login 1: Send public key
   socket
-    .send(LoginMessage::PublicKey(
+    .send_message(LoginMessage::PublicKey(
       periphery_keys().load().public.clone(),
     ))
     .await

bin/periphery/src/connection/mod.rs

@@ -1,17 +1,12 @@
-use std::{
-  sync::{Arc, OnceLock},
-  time::Duration,
-};
+use std::{sync::Arc, time::Duration};
 
 use anyhow::anyhow;
-use arc_swap::ArcSwap;
-use cache::CloneCache;
 use encoding::{
   CastBytes as _, Decode as _, Encode as _, WithChannel,
 };
-use noise::key::SpkiPublicKey;
 use periphery_client::transport::{
-  EncodedRequestMessage, EncodedTransportMessage, TransportMessage,
+  EncodedRequestMessage, EncodedTransportMessage, RequestMessage,
+  TransportMessage,
 };
 use resolver_api::Resolve;
 use transport::{
@@ -19,7 +14,7 @@ use transport::{
     ConnectionIdentifiers, LoginFlow, LoginFlowArgs,
     PublicKeyValidator,
   },
-  channel::{BufferedChannel, BufferedReceiver, Sender},
+  channel::{BufferedReceiver, Sender},
   websocket::{
     Websocket, WebsocketReceiverExt as _, WebsocketSender as _,
   },
@@ -27,84 +22,17 @@ use transport::{
 
 use crate::{
   api::{Args, PeripheryRequest},
-  config::{periphery_config, periphery_keys},
+  config::periphery_config,
+  state::{CorePublicKeys, core_public_keys, periphery_keys},
 };
 
 pub mod client;
 pub mod server;
 
-// Core Address / Host -> Channel
-pub type CoreConnection = BufferedChannel<EncodedTransportMessage>;
-pub type CoreConnections = CloneCache<String, Arc<CoreConnection>>;
-
-pub fn core_connections() -> &'static CoreConnections {
-  static CORE_CONNECTIONS: OnceLock<CoreConnections> =
-    OnceLock::new();
-  CORE_CONNECTIONS.get_or_init(Default::default)
-}
-
-pub fn core_public_keys() -> &'static CorePublicKeys {
-  static CORE_PUBLIC_KEYS: OnceLock<CorePublicKeys> = OnceLock::new();
-  CORE_PUBLIC_KEYS.get_or_init(CorePublicKeys::default)
-}
-
-pub struct CorePublicKeys(ArcSwap<Vec<SpkiPublicKey>>);
-
-impl Default for CorePublicKeys {
-  fn default() -> Self {
-    let keys = CorePublicKeys(Default::default());
-    keys.refresh();
-    keys
-  }
-}
-
-impl CorePublicKeys {
-  pub fn is_valid(&self, public_key: &str) -> bool {
-    let keys = self.0.load();
-    keys.is_empty() || keys.iter().any(|pk| pk.as_str() == public_key)
-  }
-
-  pub fn refresh(&self) {
-    let config = periphery_config();
-    let Some(core_public_keys) = config.core_public_keys.as_ref()
-    else {
-      return;
-    };
-    let core_public_keys = core_public_keys
-      .iter()
-      .flat_map(|public_key| {
-        let res = if let Some(path) = public_key.strip_prefix("file:")
-        {
-          SpkiPublicKey::from_file(path)
-        } else {
-          SpkiPublicKey::from_maybe_pem(public_key)
-        };
-        match (res, config.server_enabled) {
-          (Ok(public_key), _) => Some(public_key),
-          (Err(e), false) => {
-            // If only outbound connections, only warn.
-            // It will be written the next time `RotateCoreKeys` is executed.
-            warn!("{e:#}");
-            None
-          }
-          (Err(e), true) => {
-            // This is too dangerous to allow if server_enabled.
-            panic!("{e:#}");
-          }
-        }
-      })
-      .collect::<Vec<_>>();
-    self.0.store(Arc::new(core_public_keys));
-  }
-}
-
 impl PublicKeyValidator for &CorePublicKeys {
   type ValidationResult = ();
   async fn validate(&self, public_key: String) -> anyhow::Result<()> {
-    let keys = self.0.load();
-    if keys.is_empty()
-      || keys.iter().any(|pk| pk.as_str() == public_key)
-    {
+    if self.is_valid(&public_key).await {
       Ok(())
     } else {
       Err(
@@ -159,7 +87,7 @@ async fn handle_socket<W: Websocket>(
           break;
         }
       };
-      match ws_write.send_inner(message.into_bytes()).await {
+      match ws_write.send(message.into_bytes()).await {
         // Clears the stored message from receiver buffer.
         Ok(_) => receiver.clear_buffer(),
         Err(e) => {
@@ -173,7 +101,7 @@ async fn handle_socket<W: Websocket>(
 
   let handle_reads = async {
     loop {
-      let message = match ws_read.recv().await {
+      let message = match ws_read.recv_message().await {
         Ok(res) => res,
         Err(e) => {
           warn!("{e:#}");
@@ -208,14 +136,15 @@ fn handle_request(
     let WithChannel {
       channel,
       data: request,
-    }: WithChannel<PeripheryRequest> = match message.decode() {
-      Ok(res) => res,
-      Err(e) => {
-        // TODO: handle:
-        warn!("Failed to parse Request bytes | {e:#}");
-        return;
-      }
-    };
+    }: WithChannel<PeripheryRequest> =
+      match message.decode().and_then(RequestMessage::map_decode) {
+        Ok(res) => res,
+        Err(e) => {
+          // TODO: handle:
+          warn!("Failed to parse Request bytes | {e:#}");
+          return;
+        }
+      };
 
     let resolve_response = async {
       let response = match request.resolve(&args).await {

bin/periphery/src/connection/server.rs

@@ -18,10 +18,11 @@ use axum::{
   routing::get,
 };
 use axum_server::tls_rustls::RustlsConfig;
-use periphery_client::transport::LoginMessage;
+use periphery_client::{
+  api::CoreConnectionQuery, transport::LoginMessage,
+};
 use serror::{AddStatusCode, AddStatusCodeError};
 use transport::{
-  CoreConnectionQuery,
   auth::{
     ConnectionIdentifiers, HeaderConnectionIdentifiers,
     ServerLoginFlow,
@@ -33,7 +34,7 @@ use transport::{
 };
 
 use crate::{
-  api::Args, config::periphery_config, connection::core_connections,
+  api::Args, config::periphery_config, state::core_connections,
 };
 
 pub async fn run() -> anyhow::Result<()> {
@@ -45,6 +46,7 @@ pub async fn run() -> anyhow::Result<()> {
     .context("failed to parse listen address")?;
 
   let app = Router::new()
+    .route("/version", get(|| async { env!("CARGO_PKG_VERSION") }))
     .route("/", get(crate::connection::server::handler))
     .layer(middleware::from_fn(guard_request_by_ip))
     .into_make_service_with_connect_info::<SocketAddr>();
@@ -165,7 +167,7 @@ async fn handle_login(
   match (&config.core_public_keys, &config.passkeys) {
     (Some(_), _) | (_, None) => {
       socket
-        .send(LoginMessage::V1PasskeyFlow(false))
+        .send_message(LoginMessage::V1PasskeyFlow(false))
         .await
         .context("Failed to send Login V1PasskeyFlow message")?;
       super::handle_login::<_, ServerLoginFlow>(socket, identifiers)
@@ -188,7 +190,7 @@ async fn handle_passkey_login(
   };
   let res = async {
     socket
-      .send(LoginMessage::V1PasskeyFlow(true))
+      .send_message(LoginMessage::V1PasskeyFlow(true))
       .await
       .context("Failed to send login type indicator")?;
 
@@ -203,7 +205,7 @@ async fn handle_passkey_login(
       .any(|expected_passkey| expected_passkey.as_bytes() == passkey)
     {
       socket
-        .send(LoginMessage::Success)
+        .send_message(LoginMessage::Success)
         .await
         .context("Failed to send login type indicator")?;
       Ok(())

bin/periphery/src/docker/containers.rs

@@ -9,7 +9,9 @@ use komodo_client::entities::docker::{
   container::*,
 };
 
-use super::{DockerClient, stats::container_stats};
+use crate::state::container_stats;
+
+use super::DockerClient;
 
 impl DockerClient {
   pub async fn list_containers(

bin/periphery/src/docker/mod.rs

@@ -1,6 +1,4 @@
-use std::sync::OnceLock;
-
-use anyhow::anyhow;
+use anyhow::{Context, anyhow};
 use bollard::Docker;
 use command::run_komodo_command;
 use komodo_client::entities::{TerminationSignal, update::Log};
@@ -13,21 +11,15 @@ mod images;
 mod networks;
 mod volumes;
 
-pub fn docker_client() -> &'static DockerClient {
-  static DOCKER_CLIENT: OnceLock<DockerClient> = OnceLock::new();
-  DOCKER_CLIENT.get_or_init(Default::default)
-}
-
 pub struct DockerClient {
   docker: Docker,
 }
 
-impl Default for DockerClient {
-  fn default() -> DockerClient {
-    DockerClient {
-      docker: Docker::connect_with_defaults()
-        .expect("failed to connect to docker daemon"),
-    }
+impl DockerClient {
+  pub fn connect() -> anyhow::Result<DockerClient> {
+    let docker = Docker::connect_with_defaults()
+      .context("Failed to connect to docker api. Docker monitoring won't work and will return empty results.")?;
+    Ok(DockerClient { docker })
   }
 }
 

bin/periphery/src/docker/stats.rs

@@ -1,10 +1,6 @@
-use std::{
-  collections::HashMap,
-  sync::{Arc, OnceLock},
-};
+use std::{collections::HashMap, sync::Arc};
 
 use anyhow::{Context, anyhow};
-use arc_swap::ArcSwap;
 use async_timing_util::wait_until_timelength;
 use bollard::{models, query_parameters::StatsOptionsBuilder};
 use futures::StreamExt;
@@ -19,15 +15,10 @@ use komodo_client::entities::docker::{
 };
 use run_command::async_run_command;
 
-use crate::{config::periphery_config, docker::DockerClient};
-
-pub type ContainerStatsMap = HashMap<String, ContainerStats>;
-
-pub fn container_stats() -> &'static ArcSwap<ContainerStatsMap> {
-  static CONTAINER_STATS: OnceLock<ArcSwap<ContainerStatsMap>> =
-    OnceLock::new();
-  CONTAINER_STATS.get_or_init(Default::default)
-}
+use crate::{
+  config::periphery_config, docker::DockerClient,
+  state::container_stats,
+};
 
 pub fn spawn_polling_thread() {
   tokio::spawn(async move {

bin/periphery/src/main.rs

@@ -1,7 +1,10 @@
 use futures::{StreamExt, stream::FuturesUnordered};
 use komodo_client::entities::config::periphery::Command;
 
-use crate::config::{periphery_args, periphery_keys};
+use crate::{
+  config::periphery_args,
+  state::{core_public_keys, periphery_keys},
+};
 
 #[macro_use]
 extern crate tracing;
@@ -11,6 +14,7 @@ mod config;
 mod connection;
 mod docker;
 mod helpers;
+mod state;
 mod stats;
 mod terminal;
 
@@ -29,6 +33,9 @@ async fn app() -> anyhow::Result<()> {
   // Init + log public key. Will crash if invalid private key here.
   info!("Public Key: {}", periphery_keys().load().public);
 
+  // Init core public keys. Will crash if invalid core public keys here.
+  core_public_keys();
+
   rustls::crypto::aws_lc_rs::default_provider()
     .install_default()
     .expect("Failed to install default crypto provider");

bin/periphery/src/state.rs

@@ -0,0 +1,292 @@
+use std::{
+  collections::HashMap,
+  path::PathBuf,
+  str::FromStr,
+  sync::{Arc, OnceLock},
+};
+
+use anyhow::{Context, anyhow};
+use arc_swap::ArcSwap;
+use cache::CloneCache;
+use komodo_client::entities::docker::container::ContainerStats;
+use noise::key::{RotatableKeyPair, SpkiPublicKey};
+use periphery_client::transport::EncodedTransportMessage;
+use tokio::sync::{Mutex, RwLock, mpsc, oneshot};
+use tokio_util::sync::CancellationToken;
+use transport::channel::BufferedChannel;
+use uuid::Uuid;
+
+use crate::{
+  config::periphery_config,
+  docker::DockerClient,
+  stats::StatsClient,
+  terminal::{StdinMsg, Terminal},
+};
+
+/// Should call in startup to ensure Periphery errors without valid private key.
+pub fn periphery_keys() -> &'static RotatableKeyPair {
+  static PERIPHERY_KEYS: OnceLock<RotatableKeyPair> = OnceLock::new();
+  PERIPHERY_KEYS.get_or_init(|| {
+    let config = periphery_config();
+    if let Some(private_key_spec) = config.private_key.as_deref() {
+      RotatableKeyPair::from_private_key_spec(private_key_spec)
+    } else {
+      RotatableKeyPair::from_private_key_spec(&format!(
+        "file:{}",
+        config
+          .root_directory
+          .join("keys/periphery.key")
+          .to_str()
+          .expect("Invalid root directory")
+      ))
+    }
+    .unwrap()
+  })
+}
+
+pub fn core_public_keys() -> &'static CorePublicKeys {
+  static CORE_PUBLIC_KEYS: OnceLock<CorePublicKeys> = OnceLock::new();
+  CORE_PUBLIC_KEYS.get_or_init(CorePublicKeys::default)
+}
+
+pub struct CorePublicKeys {
+  keys: ArcSwap<Vec<SpkiPublicKey>>,
+  /// If any keys fail to write, store them here.
+  /// For Periphery -> Core connection, Periphery will
+  /// write the Core pub keys to these files as they connect.
+  to_write: ArcSwap<Vec<PathBuf>>,
+}
+
+impl Default for CorePublicKeys {
+  fn default() -> Self {
+    let keys = CorePublicKeys {
+      keys: Default::default(),
+      to_write: Default::default(),
+    };
+    keys.refresh();
+    keys
+  }
+}
+
+impl CorePublicKeys {
+  pub fn load(&self) -> arc_swap::Guard<Arc<Vec<SpkiPublicKey>>> {
+    self.keys.load()
+  }
+
+  pub async fn is_valid(&self, public_key: &str) -> bool {
+    // For Periphery -> Core connection, maybe init
+    // Core public key file if it doesn't exist.
+    self.maybe_write(public_key).await;
+    let keys = self.keys.load();
+    keys.is_empty() || keys.iter().any(|pk| pk.as_str() == public_key)
+  }
+
+  async fn maybe_write(&self, public_key: &str) {
+    let to_write = self.to_write.load();
+    let path = match to_write.as_slice() {
+      // Do nothing if empty
+      [] => return,
+      [path, _rest @ ..] => path,
+    };
+    let public_key = match SpkiPublicKey::from_maybe_pem(public_key) {
+      Ok(public_key) => public_key,
+      Err(e) => {
+        error!("Invalid incoming public key | {e:#}");
+        return;
+      }
+    };
+    if let Err(e) = public_key.write_pem_async(path).await {
+      warn!("Failed to pin incoming public key | {e:#}");
+    }
+    self.refresh();
+  }
+
+  pub fn refresh(&self) {
+    let config = periphery_config();
+    let Some(core_public_keys) = config.core_public_keys.as_ref()
+    else {
+      return;
+    };
+    let mut to_write = Vec::new();
+    let core_public_keys = core_public_keys
+      .iter()
+      .flat_map(|public_key| {
+        if let Some(path) = public_key.strip_prefix("file:")
+        {
+          match (SpkiPublicKey::from_file(path), config.server_enabled) {
+            (Ok(public_key), _) => Some(public_key),
+            (Err(e), false) => {
+              // If only outbound connections, only warn.
+              // It will be written when Core public key received.
+              warn!("{e:#}");
+              let Ok(path) = PathBuf::from_str(path);
+              to_write.push(path);
+              None
+            }
+            (Err(e), true) => {
+              // This is too dangerous to allow if server_enabled.
+              error!("{e:#}");
+              std::process::exit(1)
+            }
+          }
+        } else {
+          SpkiPublicKey::from_maybe_pem(public_key)
+            .context("Invalid hardcoded public key. If this is supposed to point to file, add 'file:' prefix.")
+            .inspect_err(|e| {
+              error!("{e:#}");
+              std::process::exit(1)
+            })
+            .ok()
+        }
+      })
+      .collect::<Vec<_>>();
+    self.keys.store(Arc::new(core_public_keys));
+    self.to_write.store(Arc::new(to_write));
+  }
+}
+
+// Core Address / Host -> Channel
+pub type CoreConnection = BufferedChannel<EncodedTransportMessage>;
+pub type CoreConnections = CloneCache<String, Arc<CoreConnection>>;
+
+pub fn core_connections() -> &'static CoreConnections {
+  static CORE_CONNECTIONS: OnceLock<CoreConnections> =
+    OnceLock::new();
+  CORE_CONNECTIONS.get_or_init(Default::default)
+}
+
+pub fn stats_client() -> &'static RwLock<StatsClient> {
+  static STATS_CLIENT: OnceLock<RwLock<StatsClient>> =
+    OnceLock::new();
+  STATS_CLIENT.get_or_init(|| RwLock::new(StatsClient::default()))
+}
+
+pub type PtyName = String;
+pub type PtyMap =
+  tokio::sync::RwLock<HashMap<PtyName, Arc<Terminal>>>;
+
+pub fn terminals() -> &'static PtyMap {
+  static TERMINALS: OnceLock<PtyMap> = OnceLock::new();
+  TERMINALS.get_or_init(Default::default)
+}
+
+pub type TerminalChannels = CloneCache<Uuid, Arc<TerminalChannel>>;
+
+pub fn terminal_channels() -> &'static TerminalChannels {
+  static TERMINAL_CHANNELS: OnceLock<TerminalChannels> =
+    OnceLock::new();
+  TERMINAL_CHANNELS.get_or_init(Default::default)
+}
+
+#[derive(Debug)]
+pub struct TerminalChannel {
+  pub sender: mpsc::Sender<StdinMsg>,
+  pub cancel: CancellationToken,
+}
+
+pub fn terminal_triggers() -> &'static TerminalTriggers {
+  static TERMINAL_TRIGGERS: OnceLock<TerminalTriggers> =
+    OnceLock::new();
+  TERMINAL_TRIGGERS.get_or_init(Default::default)
+}
+
+/// Periphery must wait for Core to finish setting
+/// up channel forwarding before sending message,
+/// or the first sent messages may be missed.
+#[derive(Default)]
+pub struct TerminalTriggers(CloneCache<Uuid, Arc<TerminalTrigger>>);
+
+impl TerminalTriggers {
+  pub async fn insert(&self, channel: Uuid) {
+    let (sender, receiver) = oneshot::channel();
+    let trigger = Arc::new(TerminalTrigger {
+      sender: Some(sender).into(),
+      receiver: Some(receiver).into(),
+    });
+    self.0.insert(channel, trigger).await;
+  }
+
+  pub async fn send(&self, channel: &Uuid) -> anyhow::Result<()> {
+    let trigger = self.0.get(channel).await.with_context(|| {
+      format!("No trigger found for channel {channel}")
+    })?;
+    trigger.send().await
+  }
+
+  pub async fn wait(&self, channel: &Uuid) -> anyhow::Result<()> {
+    let trigger = self.0.get(channel).await.with_context(|| {
+      format!("No trigger found for channel {channel}")
+    })?;
+    trigger.wait().await?;
+    self.0.remove(channel).await;
+    Ok(())
+  }
+}
+
+#[derive(Debug)]
+pub struct TerminalTrigger {
+  sender: Mutex<Option<oneshot::Sender<()>>>,
+  receiver: Mutex<Option<oneshot::Receiver<()>>>,
+}
+
+impl TerminalTrigger {
+  /// This consumes the Trigger Sender.
+  pub async fn send(&self) -> anyhow::Result<()> {
+    let mut sender = self.sender.lock().await;
+    let sender = sender
+      .take()
+      .context("Called TerminalTrigger 'send' more than once.")?;
+    sender
+      .send(())
+      .map_err(|_| anyhow!("TerminalTrigger sender already used"))
+  }
+
+  /// This consumes the Trigger Receiver.
+  pub async fn wait(&self) -> anyhow::Result<()> {
+    let mut receiver = self.receiver.lock().await;
+    let receiver = receiver
+      .take()
+      .context("Called TerminalTrigger 'wait' more than once.")?;
+    receiver.await.context("Failed to receive TerminalTrigger")
+  }
+}
+
+pub fn docker_client() -> &'static SwappableDockerClient {
+  static DOCKER_CLIENT: OnceLock<SwappableDockerClient> =
+    OnceLock::new();
+  DOCKER_CLIENT.get_or_init(SwappableDockerClient::init)
+}
+
+#[derive(Default)]
+pub struct SwappableDockerClient(ArcSwap<Option<DockerClient>>);
+
+impl SwappableDockerClient {
+  pub fn init() -> Self {
+    let docker = DockerClient::connect()
+      // Only logs on first init, although keeps trying to connect
+      .inspect_err(|e| warn!("{e:#}"))
+      .ok();
+    Self(ArcSwap::new(Arc::new(docker)))
+  }
+
+  pub fn load(&self) -> arc_swap::Guard<Arc<Option<DockerClient>>> {
+    let res = self.0.load();
+    if res.is_some() {
+      return res;
+    }
+    self.reload();
+    self.0.load()
+  }
+
+  pub fn reload(&self) {
+    self.0.store(Arc::new(DockerClient::connect().ok()));
+  }
+}
+
+pub type ContainerStatsMap = HashMap<String, ContainerStats>;
+
+pub fn container_stats() -> &'static ArcSwap<ContainerStatsMap> {
+  static CONTAINER_STATS: OnceLock<ArcSwap<ContainerStatsMap>> =
+    OnceLock::new();
+  CONTAINER_STATS.get_or_init(Default::default)
+}

bin/periphery/src/stats.rs

@@ -1,4 +1,4 @@
-use std::{cmp::Ordering, sync::OnceLock};
+use std::cmp::Ordering;
 
 use async_timing_util::wait_until_timelength;
 use komodo_client::entities::stats::{
@@ -6,15 +6,8 @@ use komodo_client::entities::stats::{
   SystemProcess, SystemStats,
 };
 use sysinfo::{ProcessRefreshKind, ProcessesToUpdate, System};
-use tokio::sync::RwLock;
 
-use crate::config::periphery_config;
-
-pub fn stats_client() -> &'static RwLock<StatsClient> {
-  static STATS_CLIENT: OnceLock<RwLock<StatsClient>> =
-    OnceLock::new();
-  STATS_CLIENT.get_or_init(|| RwLock::new(StatsClient::default()))
-}
+use crate::{config::periphery_config, state::stats_client};
 
 /// This should be called before starting the server in main.rs.
 /// Keeps the cached stats up to date

bin/periphery/src/terminal.rs

@@ -1,12 +1,7 @@
-use std::{
-  collections::{HashMap, VecDeque},
-  sync::{Arc, OnceLock},
-  time::Duration,
-};
+use std::{collections::VecDeque, sync::Arc, time::Duration};
 
 use anyhow::{Context, anyhow};
 use bytes::Bytes;
-use cache::CloneCache;
 use encoding::{Decode as _, WithChannel};
 use komodo_client::{
   api::write::TerminalRecreateMode,
@@ -14,88 +9,10 @@ use komodo_client::{
 };
 use periphery_client::transport::EncodedTerminalMessage;
 use portable_pty::{CommandBuilder, PtySize, native_pty_system};
-use tokio::sync::{Mutex, broadcast, mpsc, oneshot};
+use tokio::sync::{broadcast, mpsc};
 use tokio_util::sync::CancellationToken;
-use uuid::Uuid;
 
-#[derive(Debug)]
-pub struct TerminalChannel {
-  pub sender: mpsc::Sender<StdinMsg>,
-  pub cancel: CancellationToken,
-}
-
-pub type TerminalChannels = CloneCache<Uuid, Arc<TerminalChannel>>;
-
-pub fn terminal_channels() -> &'static TerminalChannels {
-  static TERMINAL_CHANNELS: OnceLock<TerminalChannels> =
-    OnceLock::new();
-  TERMINAL_CHANNELS.get_or_init(Default::default)
-}
-
-#[derive(Debug)]
-pub struct TerminalTrigger {
-  sender: Mutex<Option<oneshot::Sender<()>>>,
-  receiver: Mutex<Option<oneshot::Receiver<()>>>,
-}
-
-impl TerminalTrigger {
-  /// This consumes the Trigger Sender.
-  pub async fn send(&self) -> anyhow::Result<()> {
-    let mut sender = self.sender.lock().await;
-    let sender = sender
-      .take()
-      .context("Called TerminalTrigger 'send' more than once.")?;
-    sender.send(()).map_err(|_| anyhow!("Sender already used"))
-  }
-
-  /// This consumes the Trigger Receiver.
-  pub async fn wait(&self) -> anyhow::Result<()> {
-    let mut receiver = self.receiver.lock().await;
-    let receiver = receiver
-      .take()
-      .context("Called TerminalTrigger 'wait' more than once.")?;
-    receiver.await.context("Failed to receive TerminalTrigger")
-  }
-}
-
-/// Periphery must wait for Core to finish setting
-/// up channel forwarding before sending message,
-/// or the first sent messages may be missed.
-#[derive(Default)]
-pub struct TerminalTriggers(CloneCache<Uuid, Arc<TerminalTrigger>>);
-
-pub fn terminal_triggers() -> &'static TerminalTriggers {
-  static TERMINAL_TRIGGERS: OnceLock<TerminalTriggers> =
-    OnceLock::new();
-  TERMINAL_TRIGGERS.get_or_init(Default::default)
-}
-
-impl TerminalTriggers {
-  pub async fn insert(&self, channel: Uuid) {
-    let (sender, receiver) = oneshot::channel();
-    let trigger = Arc::new(TerminalTrigger {
-      sender: Some(sender).into(),
-      receiver: Some(receiver).into(),
-    });
-    self.0.insert(channel, trigger).await;
-  }
-
-  pub async fn send(&self, channel: &Uuid) -> anyhow::Result<()> {
-    let trigger = self.0.get(channel).await.with_context(|| {
-      format!("No trigger found for channel {channel}")
-    })?;
-    trigger.send().await
-  }
-
-  pub async fn wait(&self, channel: &Uuid) -> anyhow::Result<()> {
-    let trigger = self.0.get(channel).await.with_context(|| {
-      format!("No trigger found for channel {channel}")
-    })?;
-    trigger.wait().await?;
-    self.0.remove(channel).await;
-    Ok(())
-  }
-}
+use crate::state::{terminal_channels, terminal_triggers, terminals};
 
 pub async fn handle_message(message: EncodedTerminalMessage) {
   let WithChannel {
@@ -138,11 +55,6 @@ pub async fn handle_message(message: EncodedTerminalMessage) {
   };
 }
 
-type PtyName = String;
-type PtyMap = tokio::sync::RwLock<HashMap<PtyName, Arc<Terminal>>>;
-type StdinSender = mpsc::Sender<StdinMsg>;
-type StdoutReceiver = broadcast::Receiver<Bytes>;
-
 pub async fn create_terminal(
   name: String,
   command: String,
@@ -241,11 +153,6 @@ pub async fn delete_all_terminals() {
   tokio::time::sleep(Duration::from_millis(100)).await;
 }
 
-fn terminals() -> &'static PtyMap {
-  static TERMINALS: OnceLock<PtyMap> = OnceLock::new();
-  TERMINALS.get_or_init(Default::default)
-}
-
 #[derive(Clone, serde::Deserialize)]
 pub struct ResizeDimensions {
   rows: u16,
@@ -258,6 +165,9 @@ pub enum StdinMsg {
   Resize(ResizeDimensions),
 }
 
+pub type StdinSender = mpsc::Sender<StdinMsg>;
+pub type StdoutReceiver = broadcast::Receiver<Bytes>;
+
 pub struct Terminal {
   /// The command that was used as the root command, eg `shell`
   command: String,

client/core/rs/src/api/read/build.rs

@@ -236,4 +236,3 @@ pub struct ListCommonBuildExtraArgs {
 
 #[typeshare]
 pub type ListCommonBuildExtraArgsResponse = Vec<String>;
-

client/core/rs/src/entities/server.rs

@@ -356,6 +356,8 @@ pub struct PeripheryInformation {
   pub container_terminals_disabled: bool,
   /// The rate the system stats are being polled from the system
   pub stats_polling_rate: Timelength,
+  /// Whether Periphery is successfully connected to docker daemon.
+  pub docker_connected: bool,
 }
 
 /// Info about an active terminal on a server.

client/periphery/rs/src/api/mod.rs

@@ -23,6 +23,20 @@ pub mod terminal;
 
 //
 
+#[derive(Deserialize, Debug, Clone)]
+pub struct CoreConnectionQuery {
+  /// Core host (eg demo.komo.do)
+  pub core: String,
+}
+
+#[derive(Deserialize, Debug, Clone)]
+pub struct PeripheryConnectionQuery {
+  /// Server Id or name
+  pub server: String,
+}
+
+//
+
 #[derive(Serialize, Deserialize, Debug, Clone, Resolve)]
 #[response(GetHealthResponse)]
 #[error(anyhow::Error)]

client/periphery/rs/src/transport/login.rs

@@ -1,51 +1,23 @@
 use anyhow::{Context, anyhow};
 use derive_variants::{EnumVariants, ExtractVariant};
-use encoding::{CastBytes, Decode, Encode, EncodedResult};
+use encoding::{
+  CastBytes, Decode, Encode, EncodedResponse, impl_cast_bytes_vec,
+  impl_from_for_wrapper,
+};
 use noise::key::SpkiPublicKey;
 
-use crate::transport::{
-  DecodedTransportMessage, EncodedTransportMessage,
-};
-
-impl Encode<EncodedTransportMessage> for LoginMessage {
-  fn encode(self) -> EncodedTransportMessage {
-    DecodedTransportMessage::Login(Ok(self)).encode()
-  }
-}
+use crate::transport::{EncodedTransportMessage, TransportMessage};
 
 #[derive(Debug)]
 pub struct EncodedLoginMessage(
-  EncodedResult<InnerEncodedLoginMessage>,
+  EncodedResponse<InnerEncodedLoginMessage>,
 );
 
-impl CastBytes for EncodedLoginMessage {
-  fn from_vec(bytes: Vec<u8>) -> Self {
-    Self(EncodedResult::from_vec(bytes))
-  }
-  fn into_vec(self) -> Vec<u8> {
-    self.0.into_vec()
-  }
-}
-
-impl Encode<EncodedLoginMessage> for anyhow::Result<LoginMessage> {
-  fn encode(self) -> EncodedLoginMessage {
-    EncodedLoginMessage(self.map(LoginMessage::encode).encode())
-  }
-}
-
-impl Encode<EncodedLoginMessage>
-  for EncodedResult<InnerEncodedLoginMessage>
-{
-  fn encode(self) -> EncodedLoginMessage {
-    EncodedLoginMessage(self)
-  }
-}
-
-impl Decode<LoginMessage> for EncodedLoginMessage {
-  fn decode(self) -> anyhow::Result<LoginMessage> {
-    self.0.decode_into()
-  }
-}
+impl_from_for_wrapper!(
+  EncodedLoginMessage,
+  EncodedResponse<InnerEncodedLoginMessage>
+);
+impl_cast_bytes_vec!(EncodedLoginMessage, EncodedResponse);
 
 /// ```markdown
 /// | -- u8[] -- | --------- u8 ------------ |
@@ -54,14 +26,7 @@ impl Decode<LoginMessage> for EncodedLoginMessage {
 #[derive(Clone, Debug)]
 pub struct InnerEncodedLoginMessage(Vec<u8>);
 
-impl CastBytes for InnerEncodedLoginMessage {
-  fn from_vec(vec: Vec<u8>) -> Self {
-    Self(vec)
-  }
-  fn into_vec(self) -> Vec<u8> {
-    self.0
-  }
-}
+impl_cast_bytes_vec!(InnerEncodedLoginMessage, Vec);
 
 #[derive(EnumVariants, Clone)]
 #[variant_derive(Debug, Clone, Copy)]
@@ -93,8 +58,8 @@ pub enum LoginMessage {
   V1Passkey(Vec<u8>),
 }
 
-impl Encode<InnerEncodedLoginMessage> for LoginMessage {
-  fn encode(self) -> InnerEncodedLoginMessage {
+impl Encode<EncodedTransportMessage> for LoginMessage {
+  fn encode(self) -> EncodedTransportMessage {
     let variant_byte = self.extract_variant().as_byte();
     let mut bytes = match self {
       LoginMessage::Success => Vec::new(),
@@ -114,29 +79,37 @@ impl Encode<InnerEncodedLoginMessage> for LoginMessage {
       LoginMessage::V1Passkey(bytes) => bytes,
     };
     bytes.push(variant_byte);
-    InnerEncodedLoginMessage(bytes)
+    let inner = InnerEncodedLoginMessage(bytes);
+    let res = Ok(inner).encode();
+    TransportMessage::Login(EncodedLoginMessage(res)).encode()
   }
 }
 
-impl Decode<LoginMessage> for InnerEncodedLoginMessage {
-  /// Parses login messages, performing various validations.
+impl Decode<LoginMessage> for EncodedLoginMessage {
   fn decode(self) -> anyhow::Result<LoginMessage> {
-    let mut bytes = self.0;
+    let mut bytes = self
+      .0
+      .decode()?
+      .context("Should not receive Pending (2) Response message")?
+      .into_vec();
+
     let variant_byte = bytes
       .pop()
-      .context("Failed to parse login message | bytes are empty")?;
-    let variant = LoginMessageVariant::from_byte(variant_byte)?;
-    use LoginMessageVariant::*;
-    match variant {
-      Success => Ok(LoginMessage::Success),
+      .context("Failed to parse login message | Bytes are empty")?;
 
-      Nonce => Ok(LoginMessage::Nonce(
+    let variant = LoginMessageVariant::from_byte(variant_byte)?;
+
+    use LoginMessageVariant::*;
+    let message = match variant {
+      Success => LoginMessage::Success,
+
+      Nonce => LoginMessage::Nonce(
         bytes
           .try_into()
           .map_err(|_| anyhow!("Invalid connection nonce"))?,
-      )),
+      ),
 
-      Handshake => Ok(LoginMessage::Handshake(bytes)),
+      Handshake => LoginMessage::Handshake(bytes),
 
       OnboardingFlow => {
         let onboarding_flow = match bytes.as_slice() {
@@ -148,7 +121,7 @@ impl Decode<LoginMessage> for InnerEncodedLoginMessage {
             ));
           }
         };
-        Ok(LoginMessage::OnboardingFlow(onboarding_flow))
+        LoginMessage::OnboardingFlow(onboarding_flow)
       }
 
       PublicKey => {
@@ -159,7 +132,7 @@ impl Decode<LoginMessage> for InnerEncodedLoginMessage {
         }
         let public_key = String::from_utf8(bytes)
           .context("Public key is not valid utf-8")?;
-        Ok(LoginMessage::PublicKey(SpkiPublicKey::from(public_key)))
+        LoginMessage::PublicKey(SpkiPublicKey::from(public_key))
       }
 
       // V1
@@ -173,7 +146,7 @@ impl Decode<LoginMessage> for InnerEncodedLoginMessage {
             ));
           }
         };
-        Ok(LoginMessage::V1PasskeyFlow(passkey_login))
+        LoginMessage::V1PasskeyFlow(passkey_login)
       }
 
       V1Passkey => {
@@ -182,9 +155,11 @@ impl Decode<LoginMessage> for InnerEncodedLoginMessage {
             "Got empty LoginMessage V1Passkey bytes"
           ));
         }
-        Ok(LoginMessage::V1Passkey(bytes))
+        LoginMessage::V1Passkey(bytes)
       }
-    }
+    };
+
+    Ok(message)
   }
 }
 

client/periphery/rs/src/transport/mod.rs

@@ -2,160 +2,30 @@ use anyhow::{Context as _, anyhow};
 use derive_variants::{EnumVariants, ExtractVariant as _};
 use encoding::{
   CastBytes, Decode, Encode, EncodedChannel, EncodedJsonMessage,
-  EncodedOption, EncodedResult, WithChannel,
+  EncodedResponse, WithChannel, impl_cast_bytes_vec,
+  impl_from_for_wrapper,
 };
 
 mod login;
 pub use login::*;
 use serde::de::DeserializeOwned;
-
-#[derive(Debug, Clone)]
-pub struct EncodedTransportMessage(Vec<u8>);
-
-impl CastBytes for EncodedTransportMessage {
-  fn from_vec(vec: Vec<u8>) -> Self {
-    Self(vec)
-  }
-  fn into_vec(self) -> Vec<u8> {
-    self.0
-  }
-}
-
-// =================
-//  REQUEST MESSAGE
-// =================
-
-#[derive(Debug)]
-pub struct EncodedRequestMessage(EncodedChannel<EncodedJsonMessage>);
-
-impl CastBytes for EncodedRequestMessage {
-  fn from_vec(bytes: Vec<u8>) -> Self {
-    Self(EncodedChannel::from_vec(bytes))
-  }
-  fn into_vec(self) -> Vec<u8> {
-    self.0.into_vec()
-  }
-}
-
-impl Encode<EncodedRequestMessage>
-  for WithChannel<EncodedJsonMessage>
-{
-  fn encode(self) -> EncodedRequestMessage {
-    EncodedRequestMessage(self.encode())
-  }
-}
-
-impl<T: DeserializeOwned> Decode<WithChannel<T>>
-  for EncodedRequestMessage
-{
-  fn decode(self) -> anyhow::Result<WithChannel<T>> {
-    let WithChannel { channel, data } = self.0.decode()?;
-    let data = data.decode()?;
-    Ok(WithChannel { channel, data })
-  }
-}
-
-// ==================
-//  RESPONSE MESSAGE
-// ==================
-
-#[derive(Debug)]
-pub struct EncodedResponseMessage(
-  EncodedChannel<EncodedOption<EncodedResult<EncodedJsonMessage>>>,
-);
-
-impl CastBytes for EncodedResponseMessage {
-  fn from_vec(bytes: Vec<u8>) -> Self {
-    Self(EncodedChannel::from_vec(bytes))
-  }
-  fn into_vec(self) -> Vec<u8> {
-    self.0.into_vec()
-  }
-}
-
-impl Encode<EncodedResponseMessage>
-  for WithChannel<Option<anyhow::Result<EncodedJsonMessage>>>
-{
-  fn encode(self) -> EncodedResponseMessage {
-    let data = self
-      .map(|data| data.map(|data| data.encode()).encode())
-      .encode();
-    EncodedResponseMessage(data)
-  }
-}
-
-impl Encode<EncodedResponseMessage>
-  for WithChannel<EncodedResult<EncodedJsonMessage>>
-{
-  fn encode(self) -> EncodedResponseMessage {
-    let data = self.map(|data| Some(data).encode()).encode();
-    EncodedResponseMessage(data)
-  }
-}
-
-impl<T: DeserializeOwned>
-  Decode<WithChannel<Option<anyhow::Result<T>>>>
-  for EncodedResponseMessage
-{
-  fn decode(
-    self,
-  ) -> anyhow::Result<WithChannel<Option<anyhow::Result<T>>>> {
-    let WithChannel { channel, data } = self.0.decode()?;
-    let data = data
-      .decode()?
-      .map(|data| data.decode().and_then(|data| data.decode()));
-    Ok(WithChannel { channel, data })
-  }
-}
-
-impl
-  Decode<
-    WithChannel<EncodedOption<EncodedResult<EncodedJsonMessage>>>,
-  > for EncodedResponseMessage
-{
-  fn decode(
-    self,
-  ) -> anyhow::Result<
-    WithChannel<EncodedOption<EncodedResult<EncodedJsonMessage>>>,
-  > {
-    self.0.decode()
-  }
-}
-
-// ==================
-//  TERMINAL MESSAGE
-// ==================
-
-#[derive(Debug)]
-pub struct EncodedTerminalMessage(EncodedChannel<Vec<u8>>);
-
-impl CastBytes for EncodedTerminalMessage {
-  fn from_vec(bytes: Vec<u8>) -> Self {
-    Self(EncodedChannel::from_vec(bytes))
-  }
-  fn into_vec(self) -> Vec<u8> {
-    self.0.into_vec()
-  }
-}
-
-impl Encode<EncodedTerminalMessage> for WithChannel<Vec<u8>> {
-  fn encode(self) -> EncodedTerminalMessage {
-    EncodedTerminalMessage(self.encode())
-  }
-}
-
-impl Decode<WithChannel<Vec<u8>>> for EncodedTerminalMessage {
-  fn decode(self) -> anyhow::Result<WithChannel<Vec<u8>>> {
-    let WithChannel { channel, data } = self.0.decode()?;
-    let data = data.decode()?;
-    Ok(WithChannel { channel, data })
-  }
-}
+use uuid::Uuid;
 
 // ===================
 //  TRANSPORT MESSAGE
 // ===================
 
+#[derive(Debug, Clone)]
+pub struct EncodedTransportMessage(Vec<u8>);
+
+impl_cast_bytes_vec!(EncodedTransportMessage, Vec);
+
+/// When an EncodedTransportMessage is received,
+/// it is decoded into this type.
+///
+/// Note that inner bytes for top level message variants are left as is,
+/// as their decoding is left to specific handler.
+/// The main receiving hot loop should do minimal parsing.
 #[derive(Debug, EnumVariants)]
 #[variant_derive(Debug, Clone, Copy)]
 pub enum TransportMessage {
@@ -179,10 +49,8 @@ impl Encode<EncodedTransportMessage> for TransportMessage {
   }
 }
 
-impl<T: From<TransportMessage>> Decode<T>
-  for EncodedTransportMessage
-{
-  fn decode(self) -> anyhow::Result<T> {
+impl Decode<TransportMessage> for EncodedTransportMessage {
+  fn decode(self) -> anyhow::Result<TransportMessage> {
     let mut bytes = self.0;
     let variant_byte = bytes
       .pop()
@@ -203,77 +71,7 @@ impl<T: From<TransportMessage>> Decode<T>
           EncodedTerminalMessage(EncodedChannel::from_vec(bytes)),
         ),
       };
-    Ok(message.into())
-  }
-}
-
-// ===================
-//  DECODED TRANSPORT
-// ===================
-
-pub enum DecodedTransportMessage {
-  Login(anyhow::Result<LoginMessage>),
-  Request(WithChannel<EncodedJsonMessage>),
-  Response(
-    WithChannel<Option<anyhow::Result<EncodedJsonMessage>>>, // EncodedChannel<EncodedOption<EncodedResult<EncodedJsonMessage>>>,
-  ),
-  Terminal(WithChannel<Vec<u8>>),
-}
-
-impl Encode<TransportMessage> for DecodedTransportMessage {
-  fn encode(self) -> TransportMessage {
-    use DecodedTransportMessage::*;
-    match self {
-      Login(res) => TransportMessage::Login(res.encode()),
-      Request(data) => TransportMessage::Request(
-        EncodedRequestMessage(data.encode()),
-      ),
-      Response(data) => {
-        TransportMessage::Response(EncodedResponseMessage(
-          data
-            .map(|data| data.map(|data| data.encode()).encode())
-            .encode(),
-        ))
-      }
-      Terminal(data) => TransportMessage::Terminal(
-        EncodedTerminalMessage(data.encode()),
-      ),
-    }
-  }
-}
-
-impl Encode<EncodedTransportMessage> for DecodedTransportMessage {
-  fn encode(self) -> EncodedTransportMessage {
-    let res: TransportMessage = self.encode();
-    res.encode()
-  }
-}
-
-impl<T: From<DecodedTransportMessage>> Decode<T>
-  for TransportMessage
-{
-  fn decode(self) -> anyhow::Result<T> {
-    let message = match self {
-      TransportMessage::Login(encoded_result) => {
-        DecodedTransportMessage::Login(encoded_result.decode())
-      }
-      TransportMessage::Request(encoded_channel) => {
-        DecodedTransportMessage::Request(encoded_channel.0.decode()?)
-      }
-      TransportMessage::Response(encoded_channel) => {
-        let WithChannel { channel, data } =
-          encoded_channel.0.decode()?;
-        let data = data.decode()?.map(|data| data.decode());
-        DecodedTransportMessage::Response(WithChannel {
-          channel,
-          data,
-        })
-      }
-      TransportMessage::Terminal(encoded_channel) => {
-        DecodedTransportMessage::Terminal(encoded_channel.0.decode()?)
-      }
-    };
-    Ok(message.into())
+    Ok(message)
   }
 }
 
@@ -304,3 +102,132 @@ impl TransportMessageVariant {
     }
   }
 }
+
+// =================
+//  REQUEST MESSAGE
+// =================
+
+#[derive(Debug)]
+pub struct EncodedRequestMessage(EncodedChannel<EncodedJsonMessage>);
+
+impl_cast_bytes_vec!(EncodedRequestMessage, EncodedChannel);
+
+pub struct RequestMessage(WithChannel<EncodedJsonMessage>);
+
+impl RequestMessage {
+  pub fn map_decode<T: DeserializeOwned>(
+    self,
+  ) -> anyhow::Result<WithChannel<T>> {
+    self.0.map_decode()
+  }
+}
+
+impl RequestMessage {
+  pub fn new(channel: Uuid, json: EncodedJsonMessage) -> Self {
+    Self(WithChannel {
+      channel,
+      data: json,
+    })
+  }
+}
+
+impl Encode<EncodedTransportMessage> for RequestMessage {
+  fn encode(self) -> EncodedTransportMessage {
+    TransportMessage::Request(EncodedRequestMessage(self.0.encode()))
+      .encode()
+  }
+}
+
+impl Decode<RequestMessage> for EncodedRequestMessage {
+  fn decode(self) -> anyhow::Result<RequestMessage> {
+    self.0.decode().map(RequestMessage)
+  }
+}
+
+// ==================
+//  RESPONSE MESSAGE
+// ==================
+
+#[derive(Debug)]
+pub struct EncodedResponseMessage(
+  EncodedChannel<EncodedResponse<EncodedJsonMessage>>,
+);
+
+impl_from_for_wrapper!(
+  EncodedResponseMessage,
+  EncodedChannel<EncodedResponse<EncodedJsonMessage>>
+);
+
+impl_cast_bytes_vec!(EncodedResponseMessage, EncodedChannel);
+
+pub struct ResponseMessage(
+  WithChannel<EncodedResponse<EncodedJsonMessage>>,
+);
+
+impl ResponseMessage {
+  pub fn new(
+    channel: Uuid,
+    response: EncodedResponse<EncodedJsonMessage>,
+  ) -> Self {
+    Self(WithChannel {
+      channel,
+      data: response,
+    })
+  }
+
+  pub fn into_inner(
+    self,
+  ) -> WithChannel<EncodedResponse<EncodedJsonMessage>> {
+    self.0
+  }
+}
+
+impl Encode<EncodedTransportMessage> for ResponseMessage {
+  fn encode(self) -> EncodedTransportMessage {
+    TransportMessage::Response(EncodedResponseMessage(
+      self.0.encode(),
+    ))
+    .encode()
+  }
+}
+
+impl Decode<ResponseMessage> for EncodedResponseMessage {
+  fn decode(self) -> anyhow::Result<ResponseMessage> {
+    self.0.decode().map(ResponseMessage)
+  }
+}
+
+// ==================
+//  TERMINAL MESSAGE
+// ==================
+
+#[derive(Debug)]
+pub struct EncodedTerminalMessage(EncodedChannel<Vec<u8>>);
+
+impl TerminalMessage {
+  pub fn new(channel: Uuid, bytes: Vec<u8>) -> Self {
+    Self(WithChannel {
+      channel,
+      data: bytes,
+    })
+  }
+}
+
+impl_cast_bytes_vec!(EncodedTerminalMessage, EncodedChannel);
+
+pub struct TerminalMessage(WithChannel<Vec<u8>>);
+
+impl Encode<EncodedTransportMessage> for TerminalMessage {
+  fn encode(self) -> EncodedTransportMessage {
+    TransportMessage::Terminal(EncodedTerminalMessage(
+      self.0.encode(),
+    ))
+    .encode()
+  }
+}
+
+impl Decode<WithChannel<Vec<u8>>> for EncodedTerminalMessage {
+  fn decode(self) -> anyhow::Result<WithChannel<Vec<u8>>> {
+    self.0.decode_map()
+  }
+}

lib/encoding/src/channel.rs

@@ -13,24 +13,16 @@ use crate::{CastBytes, Decode, Encode};
 #[derive(Clone, Debug)]
 pub struct EncodedChannel<T>(T);
 
-impl<T> From<T> for EncodedChannel<T> {
-  fn from(value: T) -> Self {
-    Self(value)
-  }
-}
+impl_wrapper!(EncodedChannel);
 
-impl<T: CastBytes> CastBytes for EncodedChannel<T> {
-  fn from_bytes(bytes: Bytes) -> Self {
-    Self(T::from_bytes(bytes))
-  }
-  fn into_bytes(self) -> Bytes {
-    self.0.into_bytes()
-  }
-  fn from_vec(vec: Vec<u8>) -> Self {
-    Self(T::from_vec(vec))
-  }
-  fn into_vec(self) -> Vec<u8> {
-    self.0.into_vec()
+impl<B: CastBytes> EncodedChannel<B> {
+  pub fn decode_map<T>(self) -> anyhow::Result<WithChannel<T>>
+  where
+    B: Decode<T>,
+  {
+    let WithChannel { channel, data } = self.decode()?;
+    let data = data.decode()?;
+    Ok(WithChannel { channel, data })
   }
 }
 
@@ -46,14 +38,21 @@ impl<T> WithChannel<T> {
       data: map(self.data),
     }
   }
-}
 
-impl<T, E: Encode<T>> Encode<WithChannel<T>> for WithChannel<E> {
-  fn encode(self) -> WithChannel<T> {
-    WithChannel {
-      channel: self.channel,
-      data: self.data.encode(),
-    }
+  pub fn map_encode<B: CastBytes + Send>(self) -> EncodedChannel<B>
+  where
+    T: Encode<B>,
+  {
+    self.map(Encode::encode).encode()
+  }
+
+  pub fn map_decode<D>(self) -> anyhow::Result<WithChannel<D>>
+  where
+    T: CastBytes + Send + Decode<D>,
+  {
+    let WithChannel { channel, data } = self;
+    let data = data.decode()?;
+    Ok(WithChannel { channel, data })
   }
 }
 

lib/encoding/src/json.rs

@@ -1,7 +1,7 @@
 use anyhow::Context;
 use serde::{Serialize, de::DeserializeOwned};
 
-use crate::{CastBytes, Decode, Encode, EncodedResult};
+use crate::{CastBytes, Decode, Encode, EncodedResponse};
 
 /// ```markdown
 /// | --- u8[] --- |
@@ -10,6 +10,8 @@ use crate::{CastBytes, Decode, Encode, EncodedResult};
 #[derive(Clone, Debug)]
 pub struct EncodedJsonMessage(Vec<u8>);
 
+impl_identity!(EncodedJsonMessage);
+
 impl CastBytes for EncodedJsonMessage {
   fn from_vec(vec: Vec<u8>) -> Self {
     Self(vec)
@@ -27,9 +29,9 @@ where
   &'a T: Send,
 {
   fn encode(self) -> anyhow::Result<EncodedJsonMessage> {
-    let bytes = serde_json::to_vec(self.0)
-      .context("Failed to serialize data to bytes")?;
-    Ok(EncodedJsonMessage(bytes))
+    serde_json::to_vec(self.0)
+      .context("Failed to serialize data to bytes")
+      .map(EncodedJsonMessage)
   }
 }
 
@@ -41,7 +43,7 @@ impl<T: DeserializeOwned> Decode<T> for EncodedJsonMessage {
 }
 
 impl<T: Serialize + Send> From<T>
-  for EncodedResult<EncodedJsonMessage>
+  for EncodedResponse<EncodedJsonMessage>
 {
   fn from(value: T) -> Self {
     serde_json::to_vec(&value)

lib/encoding/src/lib.rs

@@ -2,15 +2,16 @@
 
 use bytes::Bytes;
 
+#[macro_use]
+mod macros;
+
 mod channel;
 mod json;
-mod option;
-mod result;
+mod response;
 
 pub use channel::*;
 pub use json::*;
-pub use option::*;
-pub use result::*;
+pub use response::*;
 
 pub trait Encode<Target>: Sized + Send {
   fn encode(self) -> Target;
@@ -22,30 +23,6 @@ pub trait Encode<Target>: Sized + Send {
   }
 }
 
-impl Encode<Bytes> for Bytes {
-  fn encode(self) -> Bytes {
-    self
-  }
-}
-
-impl Encode<Vec<u8>> for Vec<u8> {
-  fn encode(self) -> Vec<u8> {
-    self
-  }
-}
-
-impl Encode<Vec<u8>> for Bytes {
-  fn encode(self) -> Vec<u8> {
-    self.into()
-  }
-}
-
-impl Encode<Bytes> for Vec<u8> {
-  fn encode(self) -> Bytes {
-    self.into()
-  }
-}
-
 pub trait Decode<Target>: Sized {
   fn decode(self) -> anyhow::Result<Target>;
   fn decode_into<T>(self) -> anyhow::Result<T>
@@ -56,29 +33,8 @@ pub trait Decode<Target>: Sized {
   }
 }
 
-impl Decode<Bytes> for Bytes {
-  fn decode(self) -> anyhow::Result<Bytes> {
-    Ok(self)
-  }
-}
-
-impl Decode<Vec<u8>> for Vec<u8> {
-  fn decode(self) -> anyhow::Result<Vec<u8>> {
-    Ok(self)
-  }
-}
-
-impl Decode<Vec<u8>> for Bytes {
-  fn decode(self) -> anyhow::Result<Vec<u8>> {
-    Ok(self.into())
-  }
-}
-
-impl Decode<Bytes> for Vec<u8> {
-  fn decode(self) -> anyhow::Result<Bytes> {
-    Ok(self.into())
-  }
-}
+impl_identity!(Bytes);
+impl_identity!(Vec<u8>);
 
 /// Helps cast between the top level message types.
 /// Implement whichever ones are most convenient for the source type.
@@ -114,3 +70,4 @@ impl CastBytes for Vec<u8> {
     self
   }
 }
+

lib/encoding/src/macros.rs

@@ -0,0 +1,63 @@
+#[macro_export]
+macro_rules! impl_cast_bytes_vec {
+  ($typ:ty, $through:ident) => {
+    impl CastBytes for $typ {
+      fn from_vec(bytes: Vec<u8>) -> Self {
+        Self($through::from_vec(bytes))
+      }
+      fn into_vec(self) -> Vec<u8> {
+        self.0.into_vec()
+      }
+    }
+  };
+}
+
+#[macro_export]
+macro_rules! impl_from_for_wrapper {
+  ($typ:ty, $through:ty) => {
+    impl From<$through> for $typ {
+      fn from(value: $through) -> Self {
+        Self(value)
+      }
+    }
+  };
+}
+
+macro_rules! impl_identity {
+  ($typ:ty) => {
+    impl Encode<$typ> for $typ {
+      fn encode(self) -> $typ {
+        self
+      }
+    }
+    impl Decode<$typ> for $typ {
+      fn decode(self) -> anyhow::Result<$typ> {
+        Ok(self)
+      }
+    }
+  };
+}
+
+macro_rules! impl_wrapper {
+  ($struct:ident) => {
+    impl<T> From<T> for $struct<T> {
+      fn from(value: T) -> Self {
+        Self(value)
+      }
+    }
+    impl<T: CastBytes> CastBytes for $struct<T> {
+      fn from_bytes(bytes: Bytes) -> Self {
+        Self(T::from_bytes(bytes))
+      }
+      fn into_bytes(self) -> Bytes {
+        self.0.into_bytes()
+      }
+      fn from_vec(vec: Vec<u8>) -> Self {
+        Self(T::from_vec(vec))
+      }
+      fn into_vec(self) -> Vec<u8> {
+        self.0.into_vec()
+      }
+    }
+  };
+}

lib/encoding/src/option.rs

@@ -1,60 +0,0 @@
-use anyhow::Context;
-use bytes::Bytes;
-
-use crate::{CastBytes, Decode, Encode};
-
-/// Message wrapper to handle Error unwrapping
-/// anywhere in the en/decoding chain.
-/// ```markdown
-/// | -- u8[] -- | ----- u8 ------ |
-/// | <CONTENTS> | 0: Ok or _: Err |
-/// ```
-#[derive(Clone, Debug)]
-pub struct EncodedOption<T>(T);
-
-impl<T> From<T> for EncodedOption<T> {
-  fn from(value: T) -> Self {
-    Self(value)
-  }
-}
-
-impl<T: CastBytes> CastBytes for EncodedOption<T> {
-  fn from_bytes(bytes: Bytes) -> Self {
-    Self(T::from_bytes(bytes))
-  }
-  fn into_bytes(self) -> Bytes {
-    self.0.into_bytes()
-  }
-  fn from_vec(vec: Vec<u8>) -> Self {
-    Self(T::from_vec(vec))
-  }
-  fn into_vec(self) -> Vec<u8> {
-    self.0.into_vec()
-  }
-}
-
-impl<T: CastBytes + Send> Encode<EncodedOption<T>> for Option<T> {
-  fn encode(self) -> EncodedOption<T> {
-    match self {
-      Some(data) => {
-        let mut bytes = data.into_vec();
-        bytes.push(0);
-        EncodedOption(T::from_vec(bytes))
-      }
-      None => EncodedOption(T::from_vec(vec![1])),
-    }
-  }
-}
-
-impl<T: CastBytes> Decode<Option<T>> for EncodedOption<T> {
-  fn decode(self) -> anyhow::Result<Option<T>> {
-    let mut bytes = self.0.into_vec();
-    let option_byte =
-      bytes.pop().context("OptionWrapper bytes cannot be empty")?;
-    if option_byte == 0 {
-      Ok(Some(T::from_vec(bytes)))
-    } else {
-      Ok(None)
-    }
-  }
-}

lib/encoding/src/response.rs

@@ -0,0 +1,125 @@
+use anyhow::{Context, Result as AnyhowResult};
+use bytes::Bytes;
+use serror::{deserialize_error_bytes, serialize_error_bytes};
+
+use crate::{CastBytes, Decode, Encode};
+
+/// Message wrapper to handle Error unwrapping
+/// anywhere in the en/decoding chain.
+/// ```markdown
+/// | -- u8[] -- | ---------- u8 ----------- |
+/// | <CONTENTS> | 0: Ok, 1: Err, 2: Pending |
+/// ```
+#[derive(Clone, Debug)]
+pub struct EncodedResponse<T>(T);
+
+impl_wrapper!(EncodedResponse);
+
+pub enum Response<T> {
+  Ok(T),
+  Err(anyhow::Error),
+  Pending,
+}
+
+impl<T> Response<T> {
+  pub fn into_anyhow(self) -> AnyhowResult<Option<T>> {
+    self.into()
+  }
+
+  pub fn map<R>(self, map: impl FnOnce(T) -> R) -> Response<R> {
+    use Response::*;
+    match self {
+      Ok(t) => Ok(map(t)),
+      Err(e) => Err(e),
+      Pending => Pending,
+    }
+  }
+
+  pub fn map_encode<B: CastBytes + Send>(self) -> EncodedResponse<B>
+  where
+    T: Encode<B>,
+  {
+    self.map(Encode::encode).encode()
+  }
+
+  pub fn map_decode<D>(self) -> anyhow::Result<Option<D>>
+  where
+    T: CastBytes + Send + Decode<D>,
+  {
+    match self.map(Decode::decode) {
+      Response::Ok(res) => res.map(Some),
+      Response::Err(e) => Err(e),
+      Response::Pending => Ok(None),
+    }
+  }
+}
+
+impl<T: CastBytes + Send> Encode<EncodedResponse<T>> for Response<T> {
+  fn encode(self) -> EncodedResponse<T> {
+    use Response::*;
+    let bytes = match self {
+      Ok(data) => {
+        let mut bytes = data.into_vec();
+        bytes.push(0);
+        bytes
+      }
+      Err(e) => {
+        let mut bytes = serialize_error_bytes(&e);
+        bytes.push(1);
+        bytes
+      }
+      Pending => {
+        vec![2]
+      }
+    };
+    EncodedResponse(T::from_vec(bytes))
+  }
+}
+
+impl<T: CastBytes + Send> Encode<EncodedResponse<T>>
+  for AnyhowResult<T>
+{
+  fn encode(self) -> EncodedResponse<T> {
+    Response::from(self).encode()
+  }
+}
+
+impl<T: CastBytes> Encode<EncodedResponse<T>> for &anyhow::Error {
+  fn encode(self) -> EncodedResponse<T> {
+    let mut bytes = serialize_error_bytes(self);
+    bytes.push(1);
+    EncodedResponse::from_vec(bytes)
+  }
+}
+
+impl<T: CastBytes> Decode<Option<T>> for EncodedResponse<T> {
+  fn decode(self) -> AnyhowResult<Option<T>> {
+    let mut bytes = self.0.into_vec();
+    let result_byte =
+      bytes.pop().context("ResultWrapper bytes cannot be empty")?;
+    match result_byte {
+      0 => Ok(Some(T::from_vec(bytes))),
+      1 => Err(deserialize_error_bytes(&bytes)),
+      _ => Ok(None),
+    }
+  }
+}
+
+impl<T> From<AnyhowResult<T>> for Response<T> {
+  fn from(value: AnyhowResult<T>) -> Self {
+    match value {
+      Ok(t) => Self::Ok(t),
+      Err(e) => Self::Err(e),
+    }
+  }
+}
+
+impl<T> From<Response<T>> for AnyhowResult<Option<T>> {
+  fn from(value: Response<T>) -> Self {
+    match value {
+      Response::Ok(t) => Ok(Some(t)),
+      Response::Err(e) => Err(e),
+      Response::Pending => Ok(None),
+    }
+  }
+}

lib/encoding/src/result.rs

@@ -1,78 +0,0 @@
-use anyhow::Context;
-use bytes::Bytes;
-use serror::{deserialize_error_bytes, serialize_error_bytes};
-
-use crate::{CastBytes, Decode, Encode};
-
-/// Message wrapper to handle Error unwrapping
-/// anywhere in the en/decoding chain.
-/// ```markdown
-/// | -- u8[] -- | ----- u8 ------ |
-/// | <CONTENTS> | 0: Ok or _: Err |
-/// ```
-#[derive(Clone, Debug)]
-pub struct EncodedResult<T>(T);
-
-impl<T> From<T> for EncodedResult<T> {
-  fn from(value: T) -> Self {
-    Self(value)
-  }
-}
-
-impl<T: CastBytes> CastBytes for EncodedResult<T> {
-  fn from_bytes(bytes: Bytes) -> Self {
-    Self(T::from_bytes(bytes))
-  }
-  fn into_bytes(self) -> Bytes {
-    self.0.into_bytes()
-  }
-  fn from_vec(vec: Vec<u8>) -> Self {
-    Self(T::from_vec(vec))
-  }
-  fn into_vec(self) -> Vec<u8> {
-    self.0.into_vec()
-  }
-}
-
-impl<T: CastBytes + Send> Encode<EncodedResult<T>>
-  for anyhow::Result<T>
-{
-  fn encode(self) -> EncodedResult<T> {
-    let bytes = match self {
-      Ok(data) => {
-        let mut bytes = data.into_vec();
-        bytes.push(0);
-        bytes
-      }
-      Err(e) => {
-        let mut bytes = serialize_error_bytes(&e);
-        bytes.push(1);
-        bytes
-      }
-    };
-    EncodedResult(T::from_vec(bytes))
-  }
-}
-
-impl<T: CastBytes + Send> Encode<EncodedResult<T>>
-  for &anyhow::Error
-{
-  fn encode(self) -> EncodedResult<T> {
-    let mut bytes = serialize_error_bytes(self);
-    bytes.push(1);
-    EncodedResult::from_vec(bytes)
-  }
-}
-
-impl<T: CastBytes> Decode<T> for EncodedResult<T> {
-  fn decode(self) -> anyhow::Result<T> {
-    let mut bytes = self.0.into_vec();
-    let result_byte =
-      bytes.pop().context("ResultWrapper bytes cannot be empty")?;
-    if result_byte == 0 {
-      Ok(T::from_vec(bytes))
-    } else {
-      Err(deserialize_error_bytes(&bytes))
-    }
-  }
-}

lib/response/src/lib.rs

@@ -74,20 +74,3 @@ impl JsonString {
     }
   }
 }
-
-pub enum JsonBytes {
-  Ok(Vec<u8>),
-  Err(serde_json::Error),
-}
-
-impl<T> From<T> for JsonBytes
-where
-  T: Serialize,
-{
-  fn from(value: T) -> JsonBytes {
-    match serde_json::to_vec(&value) {
-      Ok(body) => JsonBytes::Ok(body),
-      Err(e) => JsonBytes::Err(e),
-    }
-  }
-}

lib/transport/src/auth.rs

@@ -55,7 +55,7 @@ impl LoginFlow for ServerLoginFlow {
 
     let res = async {
       socket
-        .send(LoginMessage::Nonce(nonce))
+        .send_message(LoginMessage::Nonce(nonce))
         .await
         .context("Failed to send connection nonce")?;
 
@@ -81,7 +81,7 @@ impl LoginFlow for ServerLoginFlow {
         .next_message()
         .context("Failed to write handshake_m2")?;
       socket
-        .send(LoginMessage::Handshake(handshake_m2))
+        .send_message(LoginMessage::Handshake(handshake_m2))
         .await
         .context("Failed to send handshake_m2")?;
 
@@ -107,7 +107,7 @@ impl LoginFlow for ServerLoginFlow {
     match res {
       Ok(res) => {
         socket
-          .send(LoginMessage::Success)
+          .send_message(LoginMessage::Success)
           .await
           .context("Failed to send login successful to client")?;
         Ok(res)
@@ -161,7 +161,7 @@ impl LoginFlow for ClientLoginFlow {
         .next_message()
         .context("Failed to write handshake m1")?;
       socket
-        .send(LoginMessage::Handshake(handshake_m1))
+        .send_message(LoginMessage::Handshake(handshake_m1))
         .await
         .context("Failed to send handshake_m1")?;
 
@@ -188,7 +188,7 @@ impl LoginFlow for ClientLoginFlow {
         .next_message()
         .context("Failed to write handshake_m3")?;
       socket
-        .send(LoginMessage::Handshake(handshake_m3))
+        .send_message(LoginMessage::Handshake(handshake_m3))
         .await
         .context("Failed to send handshake_m3")?;
 

lib/transport/src/channel.rs

@@ -1,10 +1,11 @@
 use anyhow::{Context, anyhow};
 use encoding::{
-  Encode, EncodedJsonMessage, EncodedResult, JsonMessage, WithChannel,
+  Encode, EncodedJsonMessage, EncodedResponse, JsonMessage,
 };
 use futures_util::FutureExt;
 use periphery_client::transport::{
-  DecodedTransportMessage, EncodedTransportMessage, TransportMessage,
+  EncodedTransportMessage, RequestMessage, ResponseMessage,
+  TerminalMessage,
 };
 use serde::Serialize;
 use tokio::sync::{Mutex, MutexGuard, mpsc};
@@ -86,36 +87,30 @@ impl Sender<EncodedTransportMessage> {
   where
     &'a T: Send,
   {
-    let data = JsonMessage(request).encode()?;
-    let message =
-      DecodedTransportMessage::Request(WithChannel { channel, data });
-    self.send_message(message).await
+    let json = JsonMessage(request).encode()?;
+    self.send_message(RequestMessage::new(channel, json)).await
   }
 
   pub async fn send_in_progress(
     &self,
     channel: Uuid,
   ) -> anyhow::Result<()> {
-    let message = DecodedTransportMessage::Response(WithChannel {
-      channel,
-      data: None,
-    });
-    self.send_message(message).await
+    self
+      .send_message(ResponseMessage::new(
+        channel,
+        encoding::Response::Pending.encode(),
+      ))
+      .await
   }
 
   pub async fn send_response(
     &self,
     channel: Uuid,
-    response: EncodedResult<EncodedJsonMessage>,
+    response: EncodedResponse<EncodedJsonMessage>,
   ) -> anyhow::Result<()> {
-    let message = TransportMessage::Response(
-      WithChannel {
-        channel,
-        data: response,
-      }
-      .encode(),
-    );
-    self.send_message(message).await
+    self
+      .send_message(ResponseMessage::new(channel, response))
+      .await
   }
 
   pub async fn send_terminal(
@@ -123,11 +118,9 @@ impl Sender<EncodedTransportMessage> {
     channel: Uuid,
     data: impl Into<Vec<u8>>,
   ) -> anyhow::Result<()> {
-    let message = DecodedTransportMessage::Terminal(WithChannel {
-      channel,
-      data: data.into(),
-    });
-    self.send_message(message).await
+    self
+      .send_message(TerminalMessage::new(channel, data.into()))
+      .await
   }
 }
 

lib/transport/src/lib.rs

@@ -1,30 +1,8 @@
-use ::bytes::Bytes;
-use serde::Deserialize;
-
 pub mod auth;
 pub mod channel;
 pub mod timeout;
 pub mod websocket;
 
-pub trait TransportHandler {
-  fn handle_incoming_bytes(
-    &self,
-    bytes: Bytes,
-  ) -> impl Future<Output = ()> + Send;
-}
-
-#[derive(Deserialize, Debug, Clone)]
-pub struct CoreConnectionQuery {
-  /// Core host (eg demo.komo.do)
-  pub core: String,
-}
-
-#[derive(Deserialize, Debug, Clone)]
-pub struct PeripheryConnectionQuery {
-  /// Server Id or name
-  pub server: String,
-}
-
 /// - Fixes ws addresses:
 ///   - `server.domain` => `wss://server.domain`
 ///   - `http://server.domain` => `ws://server.domain`

lib/transport/src/websocket/axum.rs

@@ -25,7 +25,7 @@ impl Websocket for AxumWebsocket {
     (AxumWebsocketSender(tx), AxumWebsocketReceiver::new(rx))
   }
 
-  async fn send_inner(&mut self, bytes: Bytes) -> anyhow::Result<()> {
+  async fn send(&mut self, bytes: Bytes) -> anyhow::Result<()> {
     self
       .0
       .send(axum::extract::ws::Message::Binary(bytes))
@@ -58,7 +58,7 @@ pub type InnerWebsocketSender =
 pub struct AxumWebsocketSender(pub InnerWebsocketSender);
 
 impl WebsocketSender for AxumWebsocketSender {
-  async fn send_inner(&mut self, bytes: Bytes) -> anyhow::Result<()> {
+  async fn send(&mut self, bytes: Bytes) -> anyhow::Result<()> {
     self
       .0
       .send(axum::extract::ws::Message::Binary(bytes))
@@ -130,7 +130,7 @@ impl WebsocketReceiver for AxumWebsocketReceiver {
     self.cancel = Some(cancel);
   }
 
-  async fn recv_inner(
+  async fn recv(
     &mut self,
   ) -> anyhow::Result<WebsocketMessage<Self::CloseFrame>> {
     let fut = try_next(&mut self.receiver);

lib/transport/src/websocket/login.rs

@@ -1,7 +1,9 @@
 use anyhow::anyhow;
 use encoding::{Decode as _, Encode as _};
 use noise::key::SpkiPublicKey;
-use periphery_client::transport::{LoginMessage, TransportMessage};
+use periphery_client::transport::{
+  EncodedLoginMessage, LoginMessage, TransportMessage,
+};
 
 use crate::{
   auth::AUTH_TIMEOUT,
@@ -13,8 +15,9 @@ pub trait LoginWebsocketExt: WebsocketExt {
     &mut self,
     e: &anyhow::Error,
   ) -> impl Future<Output = anyhow::Result<()>> + Send {
-    let message = TransportMessage::Login(e.encode_into());
-    self.send(message)
+    let message =
+      TransportMessage::Login(EncodedLoginMessage::from(e.encode()));
+    self.send_message(message)
   }
 
   fn recv_login_message(
@@ -22,7 +25,7 @@ pub trait LoginWebsocketExt: WebsocketExt {
   ) -> impl Future<Output = anyhow::Result<LoginMessage>> + Send {
     async {
       let TransportMessage::Login(message) =
-        self.recv().with_timeout(AUTH_TIMEOUT).await?
+        self.recv_message().with_timeout(AUTH_TIMEOUT).await?
       else {
         return Err(anyhow!(
           "Expected Login message, got other message type"

lib/transport/src/websocket/mod.rs

@@ -4,10 +4,11 @@ use anyhow::{Context, anyhow};
 use bytes::Bytes;
 use encoding::{
   CastBytes as _, Decode as _, Encode, EncodedJsonMessage,
-  EncodedResult, JsonMessage, WithChannel,
+  EncodedResponse, JsonMessage,
 };
 use periphery_client::transport::{
-  DecodedTransportMessage, EncodedTransportMessage, TransportMessage,
+  EncodedTransportMessage, RequestMessage, ResponseMessage,
+  TerminalMessage, TransportMessage,
 };
 use serde::Serialize;
 use tokio_util::sync::CancellationToken;
@@ -37,7 +38,7 @@ pub trait Websocket: Send {
   /// Abstraction over websocket splitting
   fn split(self) -> (impl WebsocketSender, impl WebsocketReceiver);
 
-  fn send_inner(
+  fn send(
     &mut self,
     bytes: Bytes,
   ) -> impl Future<Output = anyhow::Result<()>> + Send;
@@ -59,15 +60,15 @@ pub trait Websocket: Send {
 }
 
 pub trait WebsocketExt: Websocket {
-  fn send(
+  fn send_message(
     &mut self,
     message: impl Encode<EncodedTransportMessage>,
   ) -> impl Future<Output = anyhow::Result<()>> + Send {
-    self.send_inner(message.encode().into_bytes())
+    self.send(message.encode().into_bytes())
   }
 
-  /// Looping receiver for websocket messages which only returns on messages.
-  fn recv(
+  /// Looping receiver for websocket messages which only returns on TransportMessage.
+  fn recv_message(
     &mut self,
   ) -> MaybeWithTimeout<
     impl Future<Output = anyhow::Result<TransportMessage>> + Send,
@@ -91,7 +92,7 @@ impl<W: Websocket> WebsocketExt for W {}
 /// Traits for split websocket receiver
 pub trait WebsocketSender {
   /// Streamlined sending on bytes
-  fn send_inner(
+  fn send(
     &mut self,
     bytes: Bytes,
   ) -> impl Future<Output = anyhow::Result<()>> + Send;
@@ -103,11 +104,11 @@ pub trait WebsocketSender {
 }
 
 pub trait WebsocketSenderExt: WebsocketSender + Send {
-  fn send(
+  fn send_message(
     &mut self,
     message: impl Encode<EncodedTransportMessage>,
   ) -> impl Future<Output = anyhow::Result<()>> + Send {
-    self.send_inner(message.encode().into_vec().into())
+    self.send(message.encode().into_vec().into())
   }
 
   fn send_request<'a, T: Serialize + Send>(
@@ -119,12 +120,8 @@ pub trait WebsocketSenderExt: WebsocketSender + Send {
     &'a T: Send,
   {
     async move {
-      let data = JsonMessage(request).encode()?;
-      let message = DecodedTransportMessage::Request(WithChannel {
-        channel,
-        data,
-      });
-      self.send(message).await
+      let json = JsonMessage(request).encode()?;
+      self.send_message(RequestMessage::new(channel, json)).await
     }
   }
 
@@ -132,26 +129,18 @@ pub trait WebsocketSenderExt: WebsocketSender + Send {
     &mut self,
     channel: Uuid,
   ) -> impl Future<Output = anyhow::Result<()>> + Send {
-    let message = DecodedTransportMessage::Response(WithChannel {
+    self.send_message(ResponseMessage::new(
       channel,
-      data: None,
-    });
-    self.send(message)
+      encoding::Response::Pending.encode(),
+    ))
   }
 
   fn send_response(
     &mut self,
     channel: Uuid,
-    response: EncodedResult<EncodedJsonMessage>,
+    response: EncodedResponse<EncodedJsonMessage>,
   ) -> impl Future<Output = anyhow::Result<()>> + Send {
-    let message = TransportMessage::Response(
-      WithChannel {
-        channel,
-        data: response,
-      }
-      .encode(),
-    );
-    self.send(message)
+    self.send_message(ResponseMessage::new(channel, response))
   }
 
   fn send_terminal(
@@ -159,11 +148,7 @@ pub trait WebsocketSenderExt: WebsocketSender + Send {
     channel: Uuid,
     data: impl Into<Vec<u8>>,
   ) -> impl Future<Output = anyhow::Result<()>> + Send {
-    let message = DecodedTransportMessage::Terminal(WithChannel {
-      channel,
-      data: data.into(),
-    });
-    self.send(message)
+    self.send_message(TerminalMessage::new(channel, data.into()))
   }
 }
 
@@ -178,7 +163,7 @@ pub trait WebsocketReceiver: Send {
 
   /// Looping receiver for websocket messages which only returns
   /// on significant messages. Must implement cancel support.
-  fn recv_inner(
+  fn recv(
     &mut self,
   ) -> impl Future<
     Output = anyhow::Result<WebsocketMessage<Self::CloseFrame>>,
@@ -186,15 +171,15 @@ pub trait WebsocketReceiver: Send {
 }
 
 pub trait WebsocketReceiverExt: WebsocketReceiver {
-  /// Looping receiver for websocket messages which only returns on messages.
-  fn recv(
+  /// Looping receiver for websocket messages which only returns on TransportMessage.
+  fn recv_message(
     &mut self,
   ) -> MaybeWithTimeout<
     impl Future<Output = anyhow::Result<TransportMessage>> + Send,
   > {
     MaybeWithTimeout::new(async {
       match self
-        .recv_inner()
+        .recv()
         .await
         .context("Failed to read websocket message")?
       {

lib/transport/src/websocket/tungstenite.rs

@@ -51,7 +51,7 @@ impl Websocket for TungsteniteWebsocket {
     MaybeWithTimeout::new(try_next(&mut self.0))
   }
 
-  async fn send_inner(&mut self, bytes: Bytes) -> anyhow::Result<()> {
+  async fn send(&mut self, bytes: Bytes) -> anyhow::Result<()> {
     self
       .0
       .send(tungstenite::Message::Binary(bytes))
@@ -76,7 +76,7 @@ pub type InnerWebsocketSender = SplitSink<
 pub struct TungsteniteWebsocketSender(pub InnerWebsocketSender);
 
 impl WebsocketSender for TungsteniteWebsocketSender {
-  async fn send_inner(&mut self, bytes: Bytes) -> anyhow::Result<()> {
+  async fn send(&mut self, bytes: Bytes) -> anyhow::Result<()> {
     self
       .0
       .send(tungstenite::Message::Binary(bytes))
@@ -149,7 +149,7 @@ impl WebsocketReceiver for TungsteniteWebsocketReceiver {
     self.cancel = Some(cancel);
   }
 
-  async fn recv_inner(
+  async fn recv(
     &mut self,
   ) -> anyhow::Result<WebsocketMessage<Self::CloseFrame>> {
     let fut = try_next(&mut self.receiver);