goagent: bare bones api

for go
foo
2024-02-17 17:20:08 +02:00 · 2024-02-15 15:25:45 +02:00 · 2024-02-15 11:21:57 +02:00 · 2024-02-05 01:25:37 +02:00 · 2024-02-04 20:13:35 +02:00 · 2024-02-03 13:11:31 +02:00
50 changed files with 1316 additions and 1048 deletions
--- a/.editorconfig
+++ b/.editorconfig
@ -6,6 +6,10 @@ end_of_line = lf
 indent_style = space
 indent_size = 2

+[*.go]
+indent_style = tab
+indent_size = 2
+
 [*.nix]
 indent_style = space
 indent_size = 2
--- a/.envrc
+++ b/.envrc
@ -1 +1 @@
-use flake
+use flake . --impure
--- a/.gitignore
+++ b/.gitignore
@ -1,7 +1,7 @@
 /target/
-/.db/
 /.vagrant/
 /.direnv/
+/.devenv/
 /deps/
 /_build/
 /priv/static/assets/
--- a/Cargo.lock
+++ b/Cargo.lock
@ -32,6 +32,12 @@ version = "1.0.79"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "080e9890a082662b09c1ad45f567faeeb47f22b5fb23895fbe1e651e718e25ca"

+[[package]]
+name = "array-init"
+version = "2.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3d62b7694a562cdf5a74227903507c56ab2cc8bdd1f781ed5cb4cf9c9f810bfc"
+
 [[package]]
 name = "async-nats"
 version = "0.33.0"
@ -41,7 +47,7 @@ dependencies = [
 "base64",
 "bytes",
 "futures",
- "http",
+ "http 0.2.11",
 "memchr",
 "nkeys",
 "nuid",
@ -99,6 +105,30 @@ version = "1.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8c3c1a368f70d6cf7302d78f8f7093da241fb8e8807c05cc9e51a125895a6d5b"

+[[package]]
+name = "binrw"
+version = "0.13.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "173901312e9850391d4d7c1318c4e099fdc037d61870fca427429830efdb4e5f"
+dependencies = [
+ "array-init",
+ "binrw_derive",
+ "bytemuck",
+]
+
+[[package]]
+name = "binrw_derive"
+version = "0.13.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cb515fdd6f8d3a357c8e19b8ec59ef53880807864329b1cb1cba5c53bf76557e"
+dependencies = [
+ "either",
+ "owo-colors",
+ "proc-macro2",
+ "quote",
+ "syn 1.0.109",
+]
+
 [[package]]
 name = "bitflags"
 version = "1.3.2"
@ -120,6 +150,12 @@ dependencies = [
 "generic-array",
 ]

+[[package]]
+name = "bytemuck"
+version = "1.14.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a2ef034f05691a48569bd920a96c81b9d91bbad1ab5ac7c4616c1f6ef36cb79f"
+
 [[package]]
 name = "byteorder"
 version = "1.5.0"
@ -225,7 +261,7 @@ checksum = "f46882e17999c6cc590af592290432be3bce0428cb0d5f8b6715e4dc7b383eb3"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn",
+ "syn 2.0.48",
 ]

 [[package]]
@ -287,6 +323,12 @@ dependencies = [
 "subtle",
 ]

+[[package]]
+name = "either"
+version = "1.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a26ae43d7bcc3b814de94796a5e736d4029efb0ee900c12e2d54c993ad1a1e07"
+
 [[package]]
 name = "errno"
 version = "0.3.8"
@ -427,6 +469,23 @@ dependencies = [
 "itoa",
 ]

+[[package]]
+name = "http"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b32afd38673a8016f7c9ae69e5af41a58f81b1d31689040f2f1959594ce194ea"
+dependencies = [
+ "bytes",
+ "fnv",
+ "itoa",
+]
+
+[[package]]
+name = "httparse"
+version = "1.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d897f394bad6a705d5f4104762e116a75639e470d80901eed05a860a95cb1904"
+
 [[package]]
 name = "idna"
 version = "0.5.0"
@ -593,6 +652,12 @@ version = "0.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39"

+[[package]]
+name = "owo-colors"
+version = "3.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c1b04fb49957986fdce4d6ee7a65027d55d4b6d2265e5848bbb507b58ccfdb6f"
+
 [[package]]
 name = "parking_lot"
 version = "0.12.1"
@ -648,7 +713,7 @@ checksum = "4359fd9c9171ec6e8c62926d6faaf553a8dc3f64e1507e76da7911b4f6a04405"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn",
+ "syn 2.0.48",
 ]

 [[package]]
@ -706,6 +771,7 @@ version = "0.1.0"
 dependencies = [
 "anyhow",
 "async-nats",
+ "binrw",
 "bytes",
 "chrono",
 "futures",
@ -717,6 +783,7 @@ dependencies = [
 "thiserror",
 "tokio",
 "tokio-stream",
+ "tokio-tungstenite",
 "tokio-util",
 "tracing",
 "tracing-subscriber",
@ -962,7 +1029,7 @@ checksum = "46fe8f8603d81ba86327b23a2e9cdf49e1255fb94a4c5f297f6ee0547178ea2c"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn",
+ "syn 2.0.48",
 ]

 [[package]]
@ -993,7 +1060,18 @@ checksum = "0b2e6b945e9d3df726b65d6ee24060aff8e3533d431f677a9695db04eff9dfdb"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn",
+ "syn 2.0.48",
+]
+
+[[package]]
+name = "sha1"
+version = "0.10.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba"
+dependencies = [
+ "cfg-if",
+ "cpufeatures",
+ "digest",
 ]

 [[package]]
@ -1094,6 +1172,17 @@ version = "2.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "81cdd64d312baedb58e21336b31bc043b77e01cc99033ce76ef539f78e965ebc"

+[[package]]
+name = "syn"
+version = "1.0.109"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "72b64191b275b66ffe2469e8af2c1cfe3bafa67b529ead792a6d0160888b4237"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "unicode-ident",
+]
+
 [[package]]
 name = "syn"
 version = "2.0.48"
@ -1136,7 +1225,7 @@ checksum = "fa0faa943b50f3db30a20aa7e265dbc66076993efed8463e8de414e5d06d3471"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn",
+ "syn 2.0.48",
 ]

 [[package]]
@ -1220,7 +1309,7 @@ checksum = "5b8a1e28f2deaa14e508979454cb3a223b10b938b45af148bc0986de36f1923b"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn",
+ "syn 2.0.48",
 ]

 [[package]]
@ -1255,6 +1344,18 @@ dependencies = [
 "tokio",
 ]

+[[package]]
+name = "tokio-tungstenite"
+version = "0.21.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c83b561d025642014097b66e6c1bb422783339e0909e4429cde4749d1990bc38"
+dependencies = [
+ "futures-util",
+ "log",
+ "tokio",
+ "tungstenite",
+]
+
 [[package]]
 name = "tokio-util"
 version = "0.7.10"
@ -1288,7 +1389,7 @@ checksum = "34704c8d6ebcbc939824180af020566b01a7c01f80641264eba0999f6c2b6be7"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn",
+ "syn 2.0.48",
 ]

 [[package]]
@ -1326,6 +1427,25 @@ dependencies = [
 "tracing-log",
 ]

+[[package]]
+name = "tungstenite"
+version = "0.21.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9ef1a641ea34f399a848dea702823bbecfb4c486f911735368f1f137cb8257e1"
+dependencies = [
+ "byteorder",
+ "bytes",
+ "data-encoding",
+ "http 1.0.0",
+ "httparse",
+ "log",
+ "rand",
+ "sha1",
+ "thiserror",
+ "url",
+ "utf-8",
+]
+
 [[package]]
 name = "typenum"
 version = "1.17.0"
@ -1370,6 +1490,12 @@ dependencies = [
 "percent-encoding",
 ]

+[[package]]
+name = "utf-8"
+version = "0.7.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9"
+
 [[package]]
 name = "valuable"
 version = "0.1.0"
--- a/agent/Cargo.toml
+++ b/agent/Cargo.toml
@ -6,6 +6,7 @@ edition = "2021"
 [dependencies]
 anyhow = "1.0.79"
 async-nats = "0.33.0"
+binrw = "0.13.3"
 bytes = "1.5.0"
 chrono = { version = "0.4.33", default-features = false, features = ["now", "serde"] }
 futures = { version = "0.3.30", default-features = false, features = ["std"] }
@ -17,6 +18,7 @@ sysinfo = { version = "0.30.5", default-features = false }
 thiserror = "1.0.56"
 tokio = { version = "1.35.1", features = ["full"] }
 tokio-stream = { version = "0.1.14", default-features = false }
+tokio-tungstenite = "0.21.0"
 tokio-util = { version = "0.7.10", features = ["codec"] }
 tracing = "0.1.40"
 tracing-subscriber = { version = "0.3.18", features = ["fmt"] }
--- a/agent/server.conf
+++ b/agent/server.conf
@ -13,10 +13,10 @@ authorization: {
            password: demo_agent_password
            permissions: {
                publish: [
-                    "agents.v1.demo_agent.>"
+                    "agents.v1.1.>"
                ]
                subscribe: [
-                    "agents.v1.demo_agent.>"
+                    "agents.v1.1.>"
                    "_INBOX_demo_agent.>"
                ]
            }
--- a/agent/src/channels/messages.rs
+++ b/agent/src/channels/messages.rs
@ -0,0 +1,83 @@
+use binrw::helpers::until_eof;
+use binrw::prelude::*;
+
+#[derive(Debug, BinRead)]
+enum BinaryMessage {
+    #[br(magic = 0u8)]
+    Push(PushBinaryMessage),
+    #[br(magic = 1u8)]
+    Reply(ReplyBinaryMessage),
+    #[br(magic = 2u8)]
+    Broadcast(BroadcastBinaryMessage),
+}
+
+#[derive(Debug, BinRead)]
+struct PushBinaryMessage {
+    ref_size: u8,
+    join_ref_size: u8,
+    _topic_size: u8,
+    _event_size: u8,
+    #[br(count = ref_size)]
+    r#ref: Vec<u8>,
+    #[br(count = join_ref_size)]
+    join_ref: Vec<u8>,
+    #[br(count = _topic_size)]
+    topic: Vec<u8>,
+    #[br(count = _event_size)]
+    event: Vec<u8>,
+    #[br(parse_with = until_eof)]
+    payload: Vec<u8>,
+}
+
+#[derive(Debug, BinRead)]
+struct ReplyBinaryMessage {
+    join_ref_size: u8,
+    ref_size: u8,
+    _topic_size: u8,
+    _status_size: u8,
+    #[br(count = join_ref_size)]
+    join_ref: Vec<u8>,
+    #[br(count = ref_size)]
+    r#ref: Vec<u8>,
+    #[br(count = _topic_size)]
+    topic: Vec<u8>,
+    #[br(count = _status_size)]
+    status: Vec<u8>,
+    #[br(parse_with = until_eof)]
+    payload: Vec<u8>,
+}
+
+#[derive(Debug, BinRead)]
+struct BroadcastBinaryMessage {
+    _topic_size: u8,
+    _event_size: u8,
+    #[br(count = _topic_size)]
+    topic: Vec<u8>,
+    #[br(count = _event_size)]
+    event: Vec<u8>,
+    #[br(parse_with = until_eof)]
+    payload: Vec<u8>,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parses_binary() {
+        // broadcast!("agents:1", "subject", {:binary, "hello"})
+        let raw_msg = vec![
+            2, 8, 7, 97, 103, 101, 110, 116, 115, 58, 49, 115, 117, 98, 106, 101, 99, 116, 104,
+            101, 108, 108, 111,
+        ];
+
+        parse_binary_message(&raw_msg[..]).unwrap();
+    }
+
+    #[test]
+    fn parses_json() {
+        // broadcast!("agents:1", "subject", "hello")
+        let raw_msg = r#"[null, "0", "agents:1", "subject", "hello"]"#;
+        parse_json_message(raw_msg).unwrap();
+    }
+}
--- a/agent/src/channels/mod.rs
+++ b/agent/src/channels/mod.rs
@ -0,0 +1,137 @@
+mod messages;
+
+use std::marker::PhantomData;
+
+use futures::{SinkExt, StreamExt};
+// use futures::{SinkExt, StreamExt};
+use serde::{
+    de::{self, Visitor},
+    ser::SerializeTuple,
+    Deserialize, Serialize,
+};
+// use tokio_tungstenite::tungstenite::Message as WSMessage;
+
+#[derive(Debug)]
+struct Message<T> {
+    join_ref: Option<String>,
+    r#ref: String,
+    topic_name: String,
+    event_name: String,
+    payload: T,
+}
+
+impl<T> Serialize for Message<T>
+where
+    T: Serialize,
+{
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let mut msg = serializer.serialize_tuple(5)?;
+        msg.serialize_element(&self.join_ref)?;
+        msg.serialize_element(&self.r#ref)?;
+        msg.serialize_element(&self.topic_name)?;
+        msg.serialize_element(&self.event_name)?;
+        msg.serialize_element(&self.payload)?;
+        msg.end()
+    }
+}
+
+impl<'de, T> Deserialize<'de> for Message<T>
+where
+    T: Deserialize<'de>,
+{
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        struct TupleVisitor<T>(PhantomData<T>);
+
+        impl<'de, T> Visitor<'de> for TupleVisitor<T>
+        where
+            T: Deserialize<'de>,
+        {
+            type Value = Message<T>;
+
+            fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
+                formatter.write_str("a sequence of values")
+            }
+
+            fn visit_seq<A>(self, mut seq: A) -> Result<Self::Value, A::Error>
+            where
+                A: serde::de::SeqAccess<'de>,
+            {
+                Ok(Message {
+                    join_ref: seq.next_element()?.ok_or_else(|| {
+                        de::Error::custom("missing join_ref from sequence at position 0")
+                    })?,
+                    r#ref: seq.next_element()?.ok_or_else(|| {
+                        de::Error::custom("missing `ref` from sequence at position 1")
+                    })?,
+                    topic_name: seq.next_element()?.ok_or_else(|| {
+                        de::Error::custom("missing `topic_name` from sequence at position 2")
+                    })?,
+                    event_name: seq.next_element()?.ok_or_else(|| {
+                        de::Error::custom("missing `event_name` from sequence at position 3")
+                    })?,
+                    payload: seq.next_element()?.ok_or_else(|| {
+                        de::Error::custom("missing `payload` from sequence at position 4")
+                    })?,
+                })
+            }
+        }
+
+        deserializer.deserialize_tuple(5, TupleVisitor(PhantomData))
+    }
+}
+
+pub async fn testchannel() -> anyhow::Result<()> {
+    let (ws_stream, _) =
+        tokio_tungstenite::connect_async("ws://localhost:4000/agent/websocket?vsn=2.0.0").await?;
+
+    let (mut send, mut recv) = ws_stream.split();
+
+    let msg = Message {
+        join_ref: Some(String::from("0")),
+        r#ref: String::from("0"),
+        topic_name: String::from("agents:1"),
+        event_name: String::from("phx_join"),
+        payload: String::from("foo"),
+    };
+
+    tokio::spawn(async move {
+        while let Some(msg) = recv.next().await {
+            let msg = msg.unwrap();
+            match msg {
+                tokio_tungstenite::tungstenite::Message::Binary(bytes) => {
+                    let _ = self::messages::parse_binary_message(&bytes[..]).unwrap();
+                }
+                tokio_tungstenite::tungstenite::Message::Text(text) => {
+                    let _ = self::messages::parse_json_message(&text).unwrap();
+                }
+                _ => todo!(),
+            }
+        }
+    });
+
+    send.send(tokio_tungstenite::tungstenite::Message::Text(
+        serde_json::to_string(&msg).unwrap(),
+    ))
+    .await?;
+
+    // let msg = serde_json::to_string(&Message {
+    //     join_ref: Some(String::from("0")),
+    //     r#ref: "1",
+    //     topic_name: String::from("agents:1"),
+    //     event_name: String::from("shout"),
+    //     payload: Some(String::from("foo")),
+    // })?;
+
+    // send.send(WSMessage::Text(msg)).await?;
+
+    // let msg = Message::heartbeat().to_json()?;
+    // send.send(WSMessage::Text(msg)).await?;
+
+    Ok(())
+}
--- a/agent/src/main.rs
+++ b/agent/src/main.rs
@ -1,5 +1,8 @@
 use tokio::signal::ctrl_c;

+use crate::channels::testchannel;
+
+mod channels;
 mod health;
 mod messaging;
 mod pty;
@ -20,7 +23,8 @@ async fn main() -> anyhow::Result<()> {
 }

 async fn run() -> anyhow::Result<()> {
-    let client = messaging::Client::connect("demo_agent").await?;
+    // TODO: Configurable client
+    let client = messaging::Client::connect("1").await?;

    init_health_subsystem(client.clone());
    tracing::info!("initialized health subsystem");
@ -28,6 +32,8 @@ async fn run() -> anyhow::Result<()> {
    crate::messaging::init_services(client).await?;
    tracing::info!("initialized services");

+    testchannel().await;
+
    ctrl_c().await?;
    Ok(())
 }
--- a/agent/src/messaging/v1/mod.rs
+++ b/agent/src/messaging/v1/mod.rs
@ -16,12 +16,14 @@ pub struct Client {

 impl Client {
    pub async fn connect(id: &str) -> Result<Self, super::MessagingError> {
+        let prefix = format!("agents.v1.{}", id);
+
        let nats = async_nats::ConnectOptions::with_user_and_password(
            String::from("demo_agent"),
            String::from("demo_agent_password"),
        )
        .name(format!("Prymn Agent {id}"))
-        .custom_inbox_prefix(format!("_INBOX_{id}"))
+        .custom_inbox_prefix(format!("{prefix}._INBOX_"))
        .connect("localhost")
        .await
        .map_err(|err| {
@ -31,7 +33,7 @@ impl Client {

        Ok(Self {
            id: Arc::new(String::from(id)),
-            prefix: Arc::new(format!("agents.v1.{}", id)),
+            prefix: Arc::new(prefix),
            nats,
        })
    }
--- a/agent/src/messaging/v1/terminal.rs
+++ b/agent/src/messaging/v1/terminal.rs
@ -81,8 +81,8 @@ impl OpenTerminalMessage {
                        }
                    }
                    _close = close_stream.next() => {
-                        tracing::info!("closing terminal");
                        child.kill().await.expect("the child to exit normally");
+                        tracing::info!("closing terminal");
                        break;
                    }
                    io_result = child.wait() => {
--- a/app/lib/prymn/agents.ex
+++ b/app/lib/prymn/agents.ex
@ -1,150 +1,2 @@
 defmodule Prymn.Agents do
-  @moduledoc ~S"""
-  Prymn Agents are programs that manage a remote client machine. Prymn backend
-  communicates with them using GRPC calls. GRPC connections are started using
-  the Prymn.Agents.ConnectionSupervisor and are book-kept using the
-  Prymn.Agents.Registry.
-
-  Agents are only valid when a `Prymn.Servers.Server` is considered registered.
-
-  """
-
-  require Logger
-  alias Prymn.Agents.{Connection, Health, Agent}
-  alias PrymnProto.Prymn.Agent.Stub
-  alias PrymnProto.Prymn.{ExecRequest, SysUpdateRequest}
-
-  @doc """
-  Establish a connection with a Server if one does not already exist, and
-  return an Agent that interfaces with the rest of the system.
-
-  ## Examples
-      iex> Prymn.Servers.get_server_by_ip!("127.0.0.1") |> Prymn.Agents.from_server()
-      %Prymn.Agents.Agent{}
-  """
-  def from_server(%Prymn.Servers.Server{status: :registered} = server) do
-    case start_connection(server.public_ip) do
-      {:ok, _pid} -> Agent.new(server.public_ip)
-      {:error, error} -> {:error, error}
-    end
-  end
-
-  def from_server(%Prymn.Servers.Server{}) do
-    Logger.error("Tried to establish a connection with an unregistered server.")
-    {:error, :unauthorized_action}
-  end
-
-  @doc """
-  Establish a connection with a Server if one does not already exist for a
-  given App. Returns an [Agent] that interfaces with the rest of the system.
-  """
-  def from_app(%Prymn.Apps.App{} = app) do
-    app = Prymn.Repo.preload(app, :server)
-    from_server(app.server)
-  end
-
-  @doc """
-  Starts a new connection with `host_address` if one does not exist.
-
-  ## Examples
-      iex> Prymn.Agents.start_connection("127.0.0.1")
-      {:ok, <PID:1234>}
-      iex> Prymn.Agents.start_connection("127.0.0.1")
-      {:ok, <PID:1234>}
-  """
-  def start_connection(host_address) do
-    spec = {Connection, host_address}
-
-    case DynamicSupervisor.start_child(Prymn.Agents.ConnectionSupervisor, spec) do
-      {:ok, pid} -> {:ok, pid}
-      {:error, {:already_started, pid}} -> {:ok, pid}
-      {:error, error} -> {:error, error}
-    end
-  end
-
-  @doc """
-  Subscribe to the host's Health using Phoenix.PubSub.
-  Broadcasted messages are the Health struct:
-
-    %Prymn.Agents.Health{}
-  """
-  def subscribe_to_health(%Agent{} = agent) do
-    :ok = Health.subscribe(agent.host_address)
-    agent
-  end
-
-  def subscribe_to_health(host_address) do
-    :ok = Health.subscribe(host_address)
-  end
-
-  # TODO
-  # def alive?(host_address) do
-  # end
-
-  @doc """
-  Return the last known health status of the Agent, or `nil` if it doesn't
-  exist.
-  """
-  def get_health(host_address) do
-    Health.lookup(host_address)
-  end
-
-  @doc """
-  Get the system's information (CPU, Memory usage, etc.).
-  """
-  def get_sys_info(%Agent{} = agent) do
-    with {:ok, channel} <- get_channel(agent),
-         {:ok, result} <- Stub.get_sys_info(channel, %Google.Protobuf.Empty{}) do
-      result
-    else
-      {:error, error} -> {:error, error}
-    end
-  end
-
-  @doc """
-  Run a command.
-  """
-  def exec(%Agent{} = agent, %ExecRequest{} = request) do
-    with {:ok, channel} <- get_channel(agent),
-         {:ok, result} <- Stub.exec(channel, request) do
-      result
-    else
-      {:error, error} -> {:error, error}
-    end
-  end
-
-  def exec(%Agent{} = agent, request) when is_map(request),
-    do: exec(agent, struct(ExecRequest, request))
-
-  @doc """
-  Perform a system update.
-  """
-  def sys_update(%Agent{} = agent, %SysUpdateRequest{} = request) do
-    with {:ok, channel} <- get_channel(agent),
-         {:ok, result} <- Stub.sys_update(channel, request) do
-      result
-    else
-      {:error, error} -> {:error, error}
-    end
-  end
-
-  def sys_update(%Agent{} = agent, request) when is_map(request),
-    do: sys_update(agent, struct(SysUpdateRequest, request))
-
-  def terminal(%Agent{} = agent) do
-    # TODO: Find a better solve for bi-directional GRPC stream
-    with {:ok, channel} <- get_channel(agent),
-         stream <- Stub.terminal(channel) do
-      stream
-    else
-      {:error, error} -> {:error, error}
-    end
-  end
-
-  defp get_channel(%Agent{} = agent) do
-    case start_connection(agent.host_address) do
-      {:ok, pid} -> {:ok, Connection.get_channel(pid)}
-      {:error, error} -> {:error, error}
-    end
-  end
 end
--- a/app/lib/prymn/agents/agent.ex
+++ b/app/lib/prymn/agents/agent.ex
@ -1,13 +0,0 @@
-defmodule Prymn.Agents.Agent do
-  @moduledoc false
-
-  defstruct [:host_address]
-
-  @type t :: %__MODULE__{
-          host_address: String.t()
-        }
-
-  def new(host_address) when is_binary(host_address) do
-    %__MODULE__{host_address: host_address}
-  end
-end
--- a/app/lib/prymn/agents/connection.ex
+++ b/app/lib/prymn/agents/connection.ex
@ -1,140 +0,0 @@
-defmodule Prymn.Agents.Connection do
-  @moduledoc false
-
-  alias Prymn.Agents.Health
-  alias PrymnProto.Prymn.Agent.Stub
-  require Logger
-
-  use GenServer, restart: :transient
-
-  @timeout :timer.minutes(2)
-
-  def start_link(host_address) do
-    GenServer.start_link(__MODULE__, host_address, name: via(host_address))
-  end
-
-  def get_channel(server) when is_pid(server) do
-    GenServer.call(server, :get_channel)
-  end
-
-  ##
-  ## Server callbacks
-  ##
-
-  @impl true
-  def init(host) do
-    # Process.flag(:trap_exit, true)
-    pid = self()
-
-    # Start a connection without blocking the GenServer
-    Task.start_link(fn ->
-      case GRPC.Stub.connect(host, 50_012, []) do
-        {:ok, channel} -> send(pid, channel)
-        {:error, error} -> send(pid, {:connect_error, error})
-      end
-
-      # Keep receiving and sending back any messages to the GenServer forever
-      receive_loop(pid)
-    end)
-
-    {:ok, {host, nil}}
-  end
-
-  @impl true
-  def handle_continue(:health, {_, channel} = state) do
-    pid = self()
-
-    Task.start_link(fn ->
-      case Stub.health(channel, %Google.Protobuf.Empty{}) do
-        {:ok, stream} ->
-          # Read from the stream forever and send data back to parent
-          stream
-          |> Stream.each(fn {_, data} -> send(pid, data) end)
-          |> Enum.take_while(fn _ -> true end)
-
-        {:error, _rpcerror} ->
-          send(pid, {:connect_error, :rpc_error})
-      end
-    end)
-
-    {:noreply, state}
-  end
-
-  @impl true
-  def handle_call(:get_channel, _, {_, channel} = state) do
-    {:reply, channel, state, @timeout}
-  end
-
-  @impl true
-  def handle_info(%GRPC.Channel{} = channel, {host, _}) do
-    {:noreply, {host, channel}, {:continue, :health}}
-  end
-
-  def handle_info({:connect_error, reason}, {host, _} = state) do
-    health = Health.lookup(host, default: true)
-
-    case reason do
-      :timeout -> Health.make_timed_out(health)
-      :rpc_error -> Health.make_disconnected(health)
-    end
-    |> Health.update_and_broadcast()
-
-    # NOTE: Here we terminate normally, which means we won't be retrying. Maybe we want to?
-    {:stop, :normal, state}
-  end
-
-  def handle_info(%PrymnProto.Prymn.HealthResponse{} = response, {host, _} = state) do
-    response
-    |> Health.make_from_proto(host)
-    |> Health.update_and_broadcast()
-
-    {:noreply, state, @timeout}
-  end
-
-  def handle_info(%GRPC.RPCError{} = response, state) do
-    Logger.debug("received a GRPC error: #{inspect(response)}")
-    {:noreply, state}
-  end
-
-  def handle_info({:gun_up, _pid, _protocol}, state) do
-    # NOTE: If it's possible for the GRPC connection to be down when we receive
-    # this message, maybe we should restart the connection
-    {:noreply, state, {:continue, :health}}
-  end
-
-  def handle_info({:gun_down, _pid, _proto, _reason, _}, {host, _} = state) do
-    Health.lookup(host, default: true)
-    |> Health.make_disconnected()
-    |> Health.update_and_broadcast()
-
-    {:noreply, state, @timeout}
-  end
-
-  def handle_info(:timeout, state) do
-    {:stop, {:shutdown, :timeout}, state}
-  end
-
-  def handle_info(msg, state) do
-    Logger.debug("received unhandled message #{inspect(msg)}")
-    {:noreply, state}
-  end
-
-  @impl true
-  def terminate(reason, {host, channel}) do
-    Logger.debug("terminating Agent connection (host: #{host}, reason: #{inspect(reason)})")
-    Health.delete(host)
-    if channel, do: GRPC.Stub.disconnect(channel)
-  end
-
-  defp via(name) do
-    {:via, Registry, {Prymn.Agents.Registry, name}}
-  end
-
-  defp receive_loop(pid) do
-    receive do
-      msg -> send(pid, msg)
-    end
-
-    receive_loop(pid)
-  end
-end
--- a/app/lib/prymn/agents/health.ex
+++ b/app/lib/prymn/agents/health.ex
@ -1,89 +0,0 @@
-defmodule Prymn.Agents.Health do
-  @moduledoc """
-  The Health struct keeps simple health information of whether or not the
-  target host machine is up to date, has any tasks running, its resources are
-  getting depleted, or if it's unable be reached.
-  """
-
-  defstruct [:host, :version, :status, tasks: [], message: "Unknown"]
-
-  alias PrymnProto.Prymn.HealthResponse
-
-  @type t :: %{
-          host: String.t(),
-          version: String.t(),
-          status: atom(),
-          message: String.t()
-        }
-
-  def start() do
-    :ets.new(__MODULE__, [:set, :public, :named_table, read_concurrency: true])
-  end
-
-  def subscribe(host) do
-    Phoenix.PubSub.subscribe(Prymn.PubSub, "health:#{host}")
-  end
-
-  def broadcast!(%__MODULE__{host: host} = health) do
-    Phoenix.PubSub.broadcast!(Prymn.PubSub, "health:#{host}", health)
-  end
-
-  def update_and_broadcast(nil) do
-    nil
-  end
-
-  def update_and_broadcast(%__MODULE__{host: host} = health) do
-    :ets.insert(__MODULE__, {host, health})
-    broadcast!(health)
-  end
-
-  def delete(host_address) do
-    :ets.delete(__MODULE__, host_address)
-  end
-
-  def lookup(host_address, opts \\ []) do
-    default = Keyword.get(opts, :default, false)
-
-    case :ets.lookup(__MODULE__, host_address) do
-      [{^host_address, value}] -> value
-      [] when default -> %__MODULE__{host: host_address}
-      [] -> nil
-    end
-  end
-
-  def make_timed_out(%__MODULE__{} = health) do
-    %__MODULE__{health | status: :unreachable, message: "Connect timed out"}
-  end
-
-  def make_disconnected(%__MODULE__{} = health) do
-    %__MODULE__{health | status: :disconnected, message: "Disconnected"}
-  end
-
-  def make_from_proto(%HealthResponse{system: system, version: version, tasks: tasks}, host) do
-    %__MODULE__{host: host, status: :connected}
-    |> do_version(version)
-    |> do_system(system)
-    |> do_tasks(tasks)
-  end
-
-  defp do_version(health, version) do
-    %__MODULE__{health | version: version}
-  end
-
-  defp do_system(health, system) do
-    case system.status do
-      "normal" -> %__MODULE__{health | message: "Connected"}
-      status -> %__MODULE__{health | message: "Alert: #{status}"}
-    end
-  end
-
-  defp do_tasks(health, tasks) do
-    tasks =
-      Enum.map(tasks, fn {task_key, task_value} ->
-        progress = Float.round(task_value.progress, 2)
-        {task_key, %{task_value | progress: "#{progress}%"}}
-      end)
-
-    %__MODULE__{health | tasks: tasks}
-  end
-end
--- a/app/lib/prymn/agents/supervisor.ex
+++ b/app/lib/prymn/agents/supervisor.ex
@ -1,27 +0,0 @@
-defmodule Prymn.Agents.Supervisor do
-  @moduledoc false
-
-  use Supervisor
-
-  @dynamic_supervisor Prymn.Agents.ConnectionSupervisor
-
-  def start_link(init_arg) do
-    Supervisor.start_link(__MODULE__, init_arg, name: __MODULE__)
-  end
-
-  @impl true
-  def init(_init_arg) do
-    children = [
-      # The registry will be used to register `Connection` processes with their
-      # name as their address
-      {Registry, keys: :unique, name: Prymn.Agents.Registry},
-      # Dynamically start `Connection` processes
-      {DynamicSupervisor, name: @dynamic_supervisor, strategy: :one_for_one, max_seconds: 60}
-    ]
-
-    # Register a "health" table that stores in-memory any agent health data
-    Prymn.Agents.Health.start()
-
-    Supervisor.init(children, strategy: :one_for_one)
-  end
-end
--- a/app/lib/prymn/application.ex
+++ b/app/lib/prymn/application.ex
@ -14,8 +14,8 @@ defmodule Prymn.Application do
      {Phoenix.PubSub, name: Prymn.PubSub},
      {Finch, name: Prymn.Finch},
      {Oban, Application.fetch_env!(:prymn, Oban)},
-      Prymn.Agents.Supervisor,
-      {Task.Supervisor, name: Prymn.TaskSupervisor},
+      Prymn.Messaging.Supervisor,
+      # {Task.Supervisor, name: Prymn.TaskSupervisor},
      PrymnWeb.Endpoint
    ]

--- a/app/lib/prymn/apps/wordpress.ex
+++ b/app/lib/prymn/apps/wordpress.ex
@ -4,7 +4,6 @@ defmodule Prymn.Apps.Wordpress do
  import Ecto.Changeset
  alias Prymn.Apps.App
  alias Prymn.Agents
-  alias PrymnProto.Prymn.{ExecRequest, ExecResponse}

  @primary_key false
  embedded_schema do
@ -50,150 +49,150 @@ defmodule Prymn.Apps.Wordpress do
  #   })
  #   |> Health.update_and_broadcast()

-  def deploy(%App{type: "wordpress"} = app, agent, notify_fun) do
+  def deploy(%App{type: "wordpress"} = _app, _agent, _notify_fun) do
    # TODO: Run sanity checks
    #   e.g Database does not exist, domain does not exist, etc.
-    deploy(app, agent, notify_fun, 1)
+    # deploy(app, agent, notify_fun, 1)
  end

-  defp deploy(%App{wordpress: %__MODULE__{} = wp} = app, agent, notify_fun, 1 = step) do
-    # TODO: We need a mechanism to wait for the agent to connect before proceeding,
-    #       this is executed faster than the connection (which happens on deploy_app in Worker)
+  # defp deploy(%App{wordpress: %__MODULE__{} = wp} = app, agent, notify_fun, 1 = step) do
+  #   # TODO: We need a mechanism to wait for the agent to connect before proceeding,
+  #   #       this is executed faster than the connection (which happens on deploy_app in Worker)

-    Agents.exec(agent, %ExecRequest{
-      user: "root",
-      program: "mysql",
-      args: [
-        "-e",
-        # TODO: Sanitize the string to protect from injection
-        "create user '#{wp.db_user}'@'#{wp.db_host}' identified by '#{wp.db_pass}';"
-      ]
-    })
-    |> then(&get_results/1)
-    |> case do
-      {_, _, exit_code} = data when exit_code != 0 ->
-        notify_fun.(:error, data, 1 / 5 * 100)
+  #   Agents.exec(agent, %ExecRequest{
+  #     user: "root",
+  #     program: "mysql",
+  #     args: [
+  #       "-e",
+  #       # TODO: Sanitize the string to protect from injection
+  #       "create user '#{wp.db_user}'@'#{wp.db_host}' identified by '#{wp.db_pass}';"
+  #     ]
+  #   })
+  #   |> then(&get_results/1)
+  #   |> case do
+  #     {_, _, exit_code} = data when exit_code != 0 ->
+  #       notify_fun.(:error, data, 1 / 5 * 100)

-      data ->
-        notify_fun.(:progress, data, step / @max_steps * 100)
-        deploy(app, agent, notify_fun, step + 1)
-    end
-  end
+  #     data ->
+  #       notify_fun.(:progress, data, step / @max_steps * 100)
+  #       deploy(app, agent, notify_fun, step + 1)
+  #   end
+  # end

-  defp deploy(%App{wordpress: %__MODULE__{} = wp} = app, agent, notify_fun, 2 = step) do
-    Agents.exec(agent, %ExecRequest{
-      user: "root",
-      program: "mysql",
-      args: [
-        "-e",
-        # TODO: Sanitize the string to protect from injection
-        "grant all privileges on #{wp.db_name}.* to '#{wp.db_user}'@'#{wp.db_host}';"
-      ]
-    })
-    |> then(&get_results/1)
-    |> case do
-      {_, _, exit_code} = data when exit_code != 0 ->
-        notify_fun.(:error, data, step / @max_steps * 100)
+  # defp deploy(%App{wordpress: %__MODULE__{} = wp} = app, agent, notify_fun, 2 = step) do
+  #   Agents.exec(agent, %ExecRequest{
+  #     user: "root",
+  #     program: "mysql",
+  #     args: [
+  #       "-e",
+  #       # TODO: Sanitize the string to protect from injection
+  #       "grant all privileges on #{wp.db_name}.* to '#{wp.db_user}'@'#{wp.db_host}';"
+  #     ]
+  #   })
+  #   |> then(&get_results/1)
+  #   |> case do
+  #     {_, _, exit_code} = data when exit_code != 0 ->
+  #       notify_fun.(:error, data, step / @max_steps * 100)

-      data ->
-        notify_fun.(:progress, data, step / @max_steps * 100)
-        deploy(app, agent, notify_fun, step + 1)
-    end
-  end
+  #     data ->
+  #       notify_fun.(:progress, data, step / @max_steps * 100)
+  #       deploy(app, agent, notify_fun, step + 1)
+  #   end
+  # end

-  defp deploy(%App{wordpress: %__MODULE__{} = wp} = app, agent, notify_fun, 3 = step) do
-    Agents.exec(agent, %ExecRequest{
-      user: "vagrant",
-      program: "wp",
-      args: ["core", "download", "--path=#{wp.path}"]
-    })
-    |> then(&get_results/1)
-    |> case do
-      {_, _, exit_code} = data when exit_code != 0 ->
-        notify_fun.(:error, data, step / @max_steps * 100)
+  # defp deploy(%App{wordpress: %__MODULE__{} = wp} = app, agent, notify_fun, 3 = step) do
+  #   Agents.exec(agent, %ExecRequest{
+  #     user: "vagrant",
+  #     program: "wp",
+  #     args: ["core", "download", "--path=#{wp.path}"]
+  #   })
+  #   |> then(&get_results/1)
+  #   |> case do
+  #     {_, _, exit_code} = data when exit_code != 0 ->
+  #       notify_fun.(:error, data, step / @max_steps * 100)

-      data ->
-        notify_fun.(:progress, data, step / @max_steps * 100)
-        deploy(app, agent, notify_fun, step + 1)
-    end
-  end
+  #     data ->
+  #       notify_fun.(:progress, data, step / @max_steps * 100)
+  #       deploy(app, agent, notify_fun, step + 1)
+  #   end
+  # end

-  defp deploy(%App{wordpress: %__MODULE__{} = wp} = app, agent, notify_fun, 4 = step) do
-    Agents.exec(agent, %ExecRequest{
-      user: "vagrant",
-      program: "wp",
-      args: [
-        "config",
-        "create",
-        "--path=#{wp.path}",
-        "--dbhost=#{wp.db_host}",
-        "--dbname=#{wp.db_name}",
-        "--dbuser=#{wp.db_user}",
-        "--dbpass=#{wp.db_pass}"
-      ]
-    })
-    |> then(&get_results/1)
-    |> case do
-      {_, _, exit_code} = data when exit_code != 0 ->
-        notify_fun.(:error, data, step / @max_steps * 100)
+  # defp deploy(%App{wordpress: %__MODULE__{} = wp} = app, agent, notify_fun, 4 = step) do
+  #   Agents.exec(agent, %ExecRequest{
+  #     user: "vagrant",
+  #     program: "wp",
+  #     args: [
+  #       "config",
+  #       "create",
+  #       "--path=#{wp.path}",
+  #       "--dbhost=#{wp.db_host}",
+  #       "--dbname=#{wp.db_name}",
+  #       "--dbuser=#{wp.db_user}",
+  #       "--dbpass=#{wp.db_pass}"
+  #     ]
+  #   })
+  #   |> then(&get_results/1)
+  #   |> case do
+  #     {_, _, exit_code} = data when exit_code != 0 ->
+  #       notify_fun.(:error, data, step / @max_steps * 100)

-      data ->
-        notify_fun.(:progress, data, step / @max_steps * 100)
-        deploy(app, agent, notify_fun, step + 1)
-    end
-  end
+  #     data ->
+  #       notify_fun.(:progress, data, step / @max_steps * 100)
+  #       deploy(app, agent, notify_fun, step + 1)
+  #   end
+  # end

-  defp deploy(%App{wordpress: %__MODULE__{} = wp} = app, agent, notify_fun, 5 = step) do
-    Agents.exec(agent, %ExecRequest{
-      user: "vagrant",
-      program: "wp",
-      args: ["db", "create", "--path=#{wp.path}"]
-    })
-    |> then(&get_results/1)
-    |> case do
-      {_, _, exit_code} = data when exit_code != 0 ->
-        notify_fun.(:error, data, step / @max_steps * 100)
+  # defp deploy(%App{wordpress: %__MODULE__{} = wp} = app, agent, notify_fun, 5 = step) do
+  #   Agents.exec(agent, %ExecRequest{
+  #     user: "vagrant",
+  #     program: "wp",
+  #     args: ["db", "create", "--path=#{wp.path}"]
+  #   })
+  #   |> then(&get_results/1)
+  #   |> case do
+  #     {_, _, exit_code} = data when exit_code != 0 ->
+  #       notify_fun.(:error, data, step / @max_steps * 100)

-      data ->
-        notify_fun.(:progress, data, step / @max_steps * 100)
-        deploy(app, agent, notify_fun, step + 1)
-    end
-  end
+  #     data ->
+  #       notify_fun.(:progress, data, step / @max_steps * 100)
+  #       deploy(app, agent, notify_fun, step + 1)
+  #   end
+  # end

-  defp deploy(%App{name: name, wordpress: %__MODULE__{} = wp}, agent, notify_fun, 6 = step) do
-    Agents.exec(agent, %ExecRequest{
-      user: "vagrant",
-      program: "wp",
-      args: [
-        "core",
-        "install",
-        "--path=#{wp.path}",
-        "--url=http://site.test",
-        "--title=#{name}",
-        "--admin_user=#{wp.admin_username}",
-        "--admin_email=#{wp.admin_email}"
-      ]
-    })
-    |> then(&get_results/1)
-    |> case do
-      {_, _, exit_code} = data when exit_code != 0 ->
-        notify_fun.(:error, data, step / @max_steps * 100)
+  # defp deploy(%App{name: name, wordpress: %__MODULE__{} = wp}, agent, notify_fun, 6 = step) do
+  #   Agents.exec(agent, %ExecRequest{
+  #     user: "vagrant",
+  #     program: "wp",
+  #     args: [
+  #       "core",
+  #       "install",
+  #       "--path=#{wp.path}",
+  #       "--url=http://site.test",
+  #       "--title=#{name}",
+  #       "--admin_user=#{wp.admin_username}",
+  #       "--admin_email=#{wp.admin_email}"
+  #     ]
+  #   })
+  #   |> then(&get_results/1)
+  #   |> case do
+  #     {_, _, exit_code} = data when exit_code != 0 ->
+  #       notify_fun.(:error, data, step / @max_steps * 100)

-      data ->
-        notify_fun.(:complete, data, step / @max_steps * 100)
-    end
-  end
+  #     data ->
+  #       notify_fun.(:complete, data, step / @max_steps * 100)
+  #   end
+  # end

-  defp get_results(stream) do
-    Enum.reduce_while(stream, {"", "", nil}, fn
-      {:ok, %ExecResponse{out: {:exit_code, exit_code}}}, {stdout, stderr, _} ->
-        {:halt, {stdout, stderr, exit_code}}
+  # defp get_results(stream) do
+  #   Enum.reduce_while(stream, {"", "", nil}, fn
+  #     {:ok, %ExecResponse{out: {:exit_code, exit_code}}}, {stdout, stderr, _} ->
+  #       {:halt, {stdout, stderr, exit_code}}

-      {:ok, %ExecResponse{out: {:stdout, stdout}}}, {acc_stdout, stderr, exit_code} ->
-        {:cont, {acc_stdout <> stdout, stderr, exit_code}}
+  #     {:ok, %ExecResponse{out: {:stdout, stdout}}}, {acc_stdout, stderr, exit_code} ->
+  #       {:cont, {acc_stdout <> stdout, stderr, exit_code}}

-      {:ok, %ExecResponse{out: {:stderr, stderr}}}, {stdout, acc_stderr, exit_code} ->
-        {:cont, {stdout, acc_stderr <> stderr, exit_code}}
-    end)
-  end
+  #     {:ok, %ExecResponse{out: {:stderr, stderr}}}, {stdout, acc_stderr, exit_code} ->
+  #       {:cont, {stdout, acc_stderr <> stderr, exit_code}}
+  #   end)
+  # end
 end
--- a/app/lib/prymn/messaging.ex
+++ b/app/lib/prymn/messaging.ex
@ -0,0 +1,2 @@
+defmodule Prymn.Messaging do
+end
--- a/app/lib/prymn/messaging/connection.ex
+++ b/app/lib/prymn/messaging/connection.ex
@ -0,0 +1,62 @@
+defmodule Prymn.Messaging.Connection do
+  use GenServer
+
+  require Logger
+
+  @dialyzer {:nowarn_function, init: 1}
+
+  def start_link(%{name: name} = init_arg) do
+    GenServer.start_link(__MODULE__, init_arg, name: name)
+  end
+
+  def publish(server, subject, payload) do
+    GenServer.call(server, {:pub, subject, payload})
+  end
+
+  def subscribe(server, subject, reply) do
+    GenServer.call(server, {:sub, subject, reply})
+  end
+
+  @impl true
+  def init(_init_arg) do
+    Process.flag(:trap_exit, true)
+
+    connect_opts = %{
+      host: "localhost",
+      username: "prymn_admin",
+      password: "prymn_admin",
+      auth_required: true
+    }
+
+    case Gnat.start_link(connect_opts) do
+      {:ok, pid} ->
+        {:ok, pid}
+
+      {:error, reason} ->
+        # Let the supervisor restart the Connection after a short delay
+        Logger.info("Initial NATS connection failed. Restarting...")
+        Process.sleep(1000)
+        {:stop, reason}
+    end
+  end
+
+  @impl true
+  def handle_call({:pub, subject, payload}, _from, conn_pid) do
+    :ok = Gnat.pub(conn_pid, subject, payload)
+
+    {:reply, :ok, conn_pid}
+  end
+
+  def handle_call({:sub, subject, reply}, _from, conn_pid) do
+    {:ok, cons_pid} = GenServer.start_link(Prymn.Messaging.FooConsumer, reply)
+    {:ok, sub} = Gnat.sub(conn_pid, cons_pid, subject)
+
+    {:reply, {:ok, sub}, conn_pid}
+  end
+
+  @impl true
+  def handle_info({:EXIT, pid, reason}, conn_pid) when conn_pid == pid do
+    Logger.info("NATS connection lost (#{reason})")
+    {:stop, {:shutdown, :connection_closed}, conn_pid}
+  end
+end
--- a/app/lib/prymn/messaging/connection_manager.ex
+++ b/app/lib/prymn/messaging/connection_manager.ex
@ -0,0 +1,38 @@
+defmodule Prymn.Messaging.ConnectionManager do
+  use GenServer
+
+  defstruct subscriptions: %{}
+
+  def publish(subject, payload) do
+    GenServer.call(__MODULE__, {:pub, subject, payload})
+  end
+
+  def subscribe(subject) do
+    GenServer.call(__MODULE__, {:pub, subject})
+  end
+
+  @impl true
+  def init(_init_arg) do
+    children = [
+      {Prymn.Messaging.Connection, %{name: :nats}}
+    ]
+
+    Process.monitor(:nats)
+
+    {:ok, %__MODULE__{}}
+  end
+
+  @impl true
+  def handle_call({:pub, subject, payload}, _from, state) do
+    Prymn.Messaging.Connection.publish(:nats, subject, payload)
+    {:reply, :ok, state}
+  end
+
+  def handle_call({:sub, subject}, {pid, _}, %__MODULE__{subscriptions: subsciptions} = state) do
+    Prymn.Messaging.Connection.subscribe(:nats, subject, pid)
+
+    # Map.put(subsciptions)
+
+    {:reply, :ok, state}
+  end
+end
--- a/app/lib/prymn/messaging/foo_consumer.ex
+++ b/app/lib/prymn/messaging/foo_consumer.ex
@ -0,0 +1,15 @@
+defmodule Prymn.Messaging.FooConsumer do
+  use GenServer
+
+  @impl true
+  def init(reply) do
+    {:ok, reply}
+  end
+
+  @impl true
+  def handle_info({:msg, %{topic: subject, body: body}}, reply) do
+    dbg("received on fooconsumer: #{subject}")
+    send(reply, body)
+    {:noreply, reply}
+  end
+end
--- a/app/lib/prymn/messaging/supervisor.ex
+++ b/app/lib/prymn/messaging/supervisor.ex
@ -0,0 +1,17 @@
+defmodule Prymn.Messaging.Supervisor do
+  use Supervisor
+
+  def start_link(opts) do
+    Supervisor.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  @impl true
+  def init(_init_arg) do
+    children =
+      [
+        # Prymn.Messaging.HealthConsumer
+      ]
+
+    Supervisor.init(children, strategy: :one_for_one)
+  end
+end
--- a/app/lib/prymn/worker.ex
+++ b/app/lib/prymn/worker.ex
@ -13,7 +13,7 @@ defmodule Prymn.Worker do
  defp deploy_app(app_id) do
    pid = self()
    app = Apps.get_app!(app_id)
-    agent = Agents.from_app(app)
+    # agent = Agents.from_app(app)

    Task.start_link(fn ->
      notify_fun = fn
@ -22,7 +22,7 @@ defmodule Prymn.Worker do
        :error, data, _progress -> send(pid, {:error, data})
      end

-      Apps.Wordpress.deploy(app, agent, notify_fun)
+      # Apps.Wordpress.deploy(app, agent, notify_fun)
    end)
  end

--- a/app/lib/prymn_proto/.gitkeep
+++ b/app/lib/prymn_proto/.gitkeep
--- a/app/lib/prymn_web/channels/agent_channel.ex
+++ b/app/lib/prymn_web/channels/agent_channel.ex
@ -0,0 +1,26 @@
+defmodule PrymnWeb.AgentChannel do
+  use PrymnWeb, :channel
+
+  @impl true
+  def join("agents:" <> _agent_id, _payload, socket) do
+    {:ok, socket}
+  end
+
+  @impl true
+  def handle_in(event, _payload, socket) do
+    dbg(event)
+    {:noreply, socket}
+  end
+
+  # Channels can be used in a request/response fashion
+  # by sending replies to requests from the client
+  # @impl true
+  # def handle_in("ping", payload, socket) do
+  #   {:reply, {:ok, payload}, socket}
+  # end
+
+  # def handle_in("binary", {:binary, data}, socket) do
+  #   dbg(data)
+  #   {:noreply, socket}
+  # end
+end
--- a/app/lib/prymn_web/channels/agent_socket.ex
+++ b/app/lib/prymn_web/channels/agent_socket.ex
@ -0,0 +1,42 @@
+defmodule PrymnWeb.AgentSocket do
+  use Phoenix.Socket
+
+  # A Socket handler
+  #
+  # It's possible to control the websocket connection and
+  # assign values that can be accessed by your channel topics.
+
+  channel "agents:*", PrymnWeb.AgentChannel
+
+  # Socket params are passed from the client and can
+  # be used to verify and authenticate a user. After
+  # verification, you can put default assigns into
+  # the socket that will be set for all channels, ie
+  #
+  #     {:ok, assign(socket, :user_id, verified_user_id)}
+  #
+  # To deny connection, return `:error` or `{:error, term}`. To control the
+  # response the client receives in that case, [define an error handler in the
+  # websocket
+  # configuration](https://hexdocs.pm/phoenix/Phoenix.Endpoint.html#socket/3-websocket-configuration).
+  #
+  # See `Phoenix.Token` documentation for examples in
+  # performing token verification on connect.
+  @impl true
+  def connect(_params, socket, _connect_info) do
+    {:ok, socket}
+  end
+
+  # Socket id's are topics that allow you to identify all sockets for a given user:
+  #
+  #     def id(socket), do: "user_socket:#{socket.assigns.user_id}"
+  #
+  # Would allow you to broadcast a "disconnect" event and terminate
+  # all active sockets and channels for a given user:
+  #
+  #     PrymnWeb.Endpoint.broadcast("user_socket:#{user.id}", "disconnect", %{})
+  #
+  # Returning `nil` makes this socket anonymous.
+  @impl true
+  def id(_socket), do: nil
+end
--- a/app/lib/prymn_web/components/system_info.ex
+++ b/app/lib/prymn_web/components/system_info.ex
@ -3,17 +3,17 @@ defmodule PrymnWeb.SystemInfo do

  require Logger
  alias Phoenix.LiveView.AsyncResult
-  alias PrymnProto.Prymn.SysInfoResponse

  @impl true
  def update(assigns, socket) do
    {:ok,
     socket
     |> assign(:agent, assigns.agent)
-     |> assign(:sys_info, AsyncResult.loading())
-     |> start_async(:get_sys_info, fn ->
-       Prymn.Agents.get_sys_info(assigns.agent)
-     end)}
+     |> assign(:sys_info, AsyncResult.loading())}
+
+    # |> start_async(:get_sys_info, fn ->
+    #   Prymn.Agents.get_sys_info(assigns.agent)
+    # end)}
  end

  @impl true
@ -60,31 +60,32 @@ defmodule PrymnWeb.SystemInfo do
    """
  end

-  def handle_async(:get_sys_info, {:ok, %SysInfoResponse{} = sys_info}, socket) do
-    %{sys_info: sys_info_result, agent: agent} = socket.assigns
+  # @impl true
+  # def handle_async(:get_sys_info, {:ok, %SysInfoResponse{} = sys_info}, socket) do
+  #   %{sys_info: sys_info_result, agent: agent} = socket.assigns

-    {:noreply,
-     socket
-     |> assign(:sys_info, AsyncResult.ok(sys_info_result, sys_info))
-     |> start_async(:get_sys_info, fn ->
-       # 10 seconds is >5 which is gun's timeout duration (which might have a race
-       # condition if they are equal)
-       Process.sleep(:timer.seconds(10))
-       Prymn.Agents.get_sys_info(agent)
-     end)}
-  end
+  #   {:noreply,
+  #    socket
+  #    |> assign(:sys_info, AsyncResult.ok(sys_info_result, sys_info))
+  #    |> start_async(:get_sys_info, fn ->
+  #      # 10 seconds is >5 which is gun's timeout duration (which might have a race
+  #      # condition if they are equal)
+  #      Process.sleep(:timer.seconds(10))
+  #      Prymn.Agents.get_sys_info(agent)
+  #    end)}
+  # end

-  def handle_async(:get_sys_info, {:ok, {:error, grpc_error}}, socket) do
-    %{sys_info: sys_info_result} = socket.assigns
+  # def handle_async(:get_sys_info, {:ok, {:error, grpc_error}}, socket) do
+  #   %{sys_info: sys_info_result} = socket.assigns

-    {:noreply,
-     socket
-     |> assign(:sys_info, AsyncResult.failed(sys_info_result, grpc_error))}
-  end
+  #   {:noreply,
+  #    socket
+  #    |> assign(:sys_info, AsyncResult.failed(sys_info_result, grpc_error))}
+  # end

-  def handle_async(:get_sys_info, {:exit, _reason}, socket) do
-    {:noreply, socket}
-  end
+  # def handle_async(:get_sys_info, {:exit, _reason}, socket) do
+  #   {:noreply, socket}
+  # end

  defp calculate_cpu_usage(cpus) do
    (Enum.reduce(cpus, 0, fn x, acc -> x.usage + acc end) / Enum.count(cpus))
@ -96,13 +97,14 @@ defmodule PrymnWeb.SystemInfo do
  end

  defp calculate_disk_used_percent(disks) do
-    alias PrymnProto.Prymn.SysInfoResponse.Disk
+    0
+    # alias PrymnProto.Prymn.SysInfoResponse.Disk

-    {used, total} =
-      Enum.reduce(disks, {0, 0}, fn %Disk{} = disk, {used, total} ->
-        {used + disk.total_bytes - disk.avail_bytes, total + disk.total_bytes}
-      end)
+    # {used, total} =
+    #   Enum.reduce(disks, {0, 0}, fn %Disk{} = disk, {used, total} ->
+    #     {used + disk.total_bytes - disk.avail_bytes, total + disk.total_bytes}
+    #   end)

-    Float.round(100 * used / total, 2)
+    # Float.round(100 * used / total, 2)
  end
 end
--- a/app/lib/prymn_web/components/terminal.ex
+++ b/app/lib/prymn_web/components/terminal.ex
@ -1,7 +1,7 @@
 defmodule PrymnWeb.Terminal do
  use PrymnWeb, :live_component

-  alias PrymnProto.Prymn.TerminalRequest
+  alias Prymn.Agents

  @impl true
  def mount(socket) do
@ -46,75 +46,49 @@ defmodule PrymnWeb.Terminal do
  end

  @impl true
-  def handle_event("open_terminal", _params, socket) do
-    agent = Prymn.Agents.from_server(socket.assigns.server)
-    pid = self()
+  def handle_event("open_terminal", _params, %{assigns: assigns} = socket) do
+    # TODO: make up a terminal id
+    Agents.open_terminal(assigns.agent)

-    Task.Supervisor.start_child(Prymn.TaskSupervisor, fn ->
-      # FIXME: Have to wrap this in a Task because gun sends unsolicited messages
-      # to calling process
-      stream = Prymn.Agents.terminal(agent)
-
-      {:ok, mux_pid} =
-        Task.Supervisor.start_child(Prymn.TaskSupervisor, fn -> receive_loop(stream) end)
-
-      send_update(pid, PrymnWeb.Terminal, id: "terminal", mux_pid: mux_pid, open: true)
-
-      case GRPC.Stub.recv(stream, timeout: :infinity) do
-        {:ok, stream} ->
-          Enum.map(stream, fn
-            {:ok, %{output: data}} ->
-              send(mux_pid, :data)
-              send_update(pid, PrymnWeb.Terminal, id: "terminal", data: data)
-
-            {:error, _err} ->
-              send_update(pid, PrymnWeb.Terminal, id: "terminal", open: false)
-          end)
-
-        {:error, error} ->
-          dbg(error)
-      end
-    end)
-
-    {:noreply, socket}
+    {:noreply, assign(socket, :open, true)}
  end

  def handle_event("close_terminal", _params, socket) do
-    send(socket.assigns.mux_pid, :close)
-    {:noreply, assign(socket, :open, false)}
+    Agents.close_terminal(socket.assigns.agent)
+    {:noreply, assign(socket, open: false, data: "")}
  end

-  def handle_event("data_event", data, socket) when is_binary(data) do
-    send(socket.assigns.mux_pid, {:data_event, data})
+  def handle_event("data_event", data, %{assigns: assigns} = socket) when is_binary(data) do
+    Agents.send_terminal_input(assigns.agent, data)
    {:noreply, socket}
  end

  def handle_event("resize_event", %{"cols" => cols, "rows" => rows}, socket) do
-    send(socket.assigns.mux_pid, {:resize_event, rows, cols})
+    Agents.resize_terminal(socket.assigns.agent, rows, cols)
    {:noreply, socket}
  end

-  defp receive_loop(stream) do
-    receive do
-      {:data_event, data} ->
-        GRPC.Stub.send_request(stream, %TerminalRequest{input: data})
-        receive_loop(stream)
+  # defp receive_loop(stream) do
+  #   receive do
+  #     {:data_event, data} ->
+  #       GRPC.Stub.send_request(stream, %TerminalRequest{input: data})
+  #       receive_loop(stream)

-      {:resize_event, rows, cols} ->
-        GRPC.Stub.send_request(stream, %TerminalRequest{
-          resize: %TerminalRequest.Resize{rows: rows, cols: cols}
-        })
+  #     {:resize_event, rows, cols} ->
+  #       GRPC.Stub.send_request(stream, %TerminalRequest{
+  #         resize: %TerminalRequest.Resize{rows: rows, cols: cols}
+  #       })

-        receive_loop(stream)
+  #       receive_loop(stream)

-      :data ->
-        receive_loop(stream)
+  #     :data ->
+  #       receive_loop(stream)

-      :close ->
-        GRPC.Stub.send_request(stream, %TerminalRequest{input: ""}, end_stream: true)
-    after
-      120_000 ->
-        GRPC.Stub.send_request(stream, %TerminalRequest{input: ""}, end_stream: true)
-    end
-  end
+  #     :close ->
+  #       GRPC.Stub.send_request(stream, %TerminalRequest{input: ""}, end_stream: true)
+  #   after
+  #     120_000 ->
+  #       GRPC.Stub.send_request(stream, %TerminalRequest{input: ""}, end_stream: true)
+  #   end
+  # end
 end
--- a/app/lib/prymn_web/endpoint.ex
+++ b/app/lib/prymn_web/endpoint.ex
@ -13,6 +13,10 @@ defmodule PrymnWeb.Endpoint do

  socket "/live", Phoenix.LiveView.Socket, websocket: [connect_info: [session: @session_options]]

+  socket "/agent", PrymnWeb.AgentSocket,
+    websocket: true,
+    longpoll: false
+
  # Serve at "/" the static files from "priv/static" directory.
  #
  # You should set gzip to true if you are running phx.digest
--- a/app/lib/prymn_web/live/server_live/index.ex
+++ b/app/lib/prymn_web/live/server_live/index.ex
@ -8,23 +8,12 @@ defmodule PrymnWeb.ServerLive.Index do
  @impl true
  def mount(_params, _session, socket) do
    servers = Servers.list_servers()
-
-    healths =
-      if connected?(socket) do
-        for %Servers.Server{status: :registered, public_ip: ip} = server <- servers, into: %{} do
-          Agents.from_server(server)
-          |> Agents.subscribe_to_health()
-
-          {ip, Agents.get_health(ip)}
-        end
-      else
-        %{}
-      end
+    agents = Agents.from_servers(servers)

    {:ok,
     socket
     |> assign(:servers, servers)
-     |> assign(:healths, healths)}
+     |> assign(:agents, agents)}
  end

  @impl true
@ -40,27 +29,14 @@ defmodule PrymnWeb.ServerLive.Index do
          <Button.primary patch={~p"/servers/new"}>Connect a Server</Button.primary>
        </:actions>
      </.header>
-      <div class="space-y-5" phx-update="replace" id="servers">
-        <.link
-          :for={server <- @servers}
-          navigate={~p"/servers/#{server}"}
-          class="group block rounded-lg bg-gray-100 p-5 shadow-sm shadow-gray-300 hover:bg-black hover:text-white"
-        >
-          <div class="flex flex-row flex-wrap justify-between">
+      <div class="mt-10 space-y-5" phx-update="replace" id="servers">
+        <.link :for={server <- @servers} navigate={~p"/servers/#{server}"} class="group flex">
+          <.status_bar agent={@agents[server.id]} />
+          <div class="flex-1 rounded-r-lg bg-gray-100 p-5 transition-colors group-hover:bg-black group-hover:text-white">
            <h2 class="text-xl"><%= server.name %></h2>
-            <.server_status status={server.status} health={@healths[server.public_ip]} />
-          </div>
-          <div class="flex flex-row flex-wrap justify-between lg:text-sm">
-            <span>IP: <%= server.public_ip || "N/A" %></span>
-            <%= if @healths[server.public_ip] do %>
-              <span
-                :for={{name, task} <- @healths[server.public_ip].tasks || []}
-                class="text-right text-xs text-slate-700"
-              >
-                <div>In progress: <%= name %></div>
-                <div><%= task.progress %></div>
-              </span>
-            <% end %>
+            <div class="flex flex-row flex-wrap justify-between lg:text-sm">
+              <span>IP: <%= server.public_ip || "N/A" %></span>
+            </div>
          </div>
        </.link>
      </div>
@ -89,44 +65,26 @@ defmodule PrymnWeb.ServerLive.Index do
     |> update(:servers, fn servers -> [server | servers] end)}
  end

-  def handle_info(%Agents.Health{} = health, socket) do
-    healths = Map.put(socket.assigns.healths, health.host, health)
-    {:noreply, assign(socket, :healths, healths)}
-  end
+  # def handle_info(%Agents.Agent{} = agent, socket) do
+  #   id = String.to_integer(agent.id)
+  #   {:noreply, update(socket, :agents, &Map.put(&1, id, agent))}
+  # end

  def handle_info(msg, state) do
    Logger.debug("received unexpected message #{inspect(msg)}")
    {:noreply, state}
  end

-  defp server_status(assigns) do
-    case {assigns.status, assigns.health} do
-      {:unregistered, _} ->
-        ~H"""
-        <span class="self-center text-sm text-gray-500">Needs registration</span>
-        """
+  defp status_bar(assigns) do
+    assigns =
+      assign(assigns, :class, [
+        "w-3 rounded-l-lg",
+        assigns.agent.status == :connected && "bg-teal-500",
+        assigns.agent.status == :disconnected && "bg-red-500"
+      ])

-      {:registered, nil} ->
-        ~H"""
-        <.spinner size="md" />
-        """
-
-      {:registered, %Agents.Health{status: :connected}} ->
-        ~H"""
-        <span class="self-center text-sm text-green-600">Connected</span>
-        """
-
-      {:registered, %Agents.Health{status: :disconnected}} ->
-        ~H"""
-        <span class="self-center text-sm text-red-600">Disconnected</span>
-        """
-
-      {:registered, %Agents.Health{message: message}} ->
-        assigns = assign(assigns, :message, message)
-
-        ~H"""
-        <span class="self-center text-sm text-yellow-900"><%= @message %></span>
-        """
-    end
+    ~H"""
+    <div class={@class}></div>
+    """
  end
 end
--- a/app/lib/prymn_web/live/server_live/show.ex
+++ b/app/lib/prymn_web/live/server_live/show.ex
@ -2,7 +2,7 @@ defmodule PrymnWeb.ServerLive.Show do
  use PrymnWeb, :live_view

  require Logger
-  alias Prymn.{Agents, Servers}
+  alias Prymn.{Agents, Servers, Messaging}

  @impl true
  def mount(_params, _session, socket) do
@ -20,7 +20,7 @@ defmodule PrymnWeb.ServerLive.Show do
          <.dropdown title="Select a different server">
            <:button variant="tertiary">Server <%= @server.name %></:button>
            <:item
-              :for={server <- Enum.filter(@servers, fn s -> s.id != @server.id end)}
+              :for={server <- Enum.filter(@servers, &(&1.id != @server.id))}
              patch={~p"/servers/#{server}"}
            >
              <%= server.name %>
@ -29,7 +29,7 @@ defmodule PrymnWeb.ServerLive.Show do
          <Button.tertiary title="Edit server name" phx-click={show_edit_server_name()}>
            <.icon class="h-4 w-4" name="hero-pencil-solid" />
          </Button.tertiary>
-          <.indicator message={@health.message} />
+          <.indicator message="test" />
        </div>
        <form class="hidden items-center" id="server-name-edit" phx-submit={submit_edit_server_name()}>
          <input class="outline-none" type="text" name="name" value={@server.name} required />
@ -48,10 +48,6 @@ defmodule PrymnWeb.ServerLive.Show do
        </div>
      </div>
      <span class="text-sm opacity-75"><%= @server.public_ip %></span>
-      <div :for={{name, task} <- @health.tasks} class="my-3 text-sm text-slate-700">
-        <p>Background task in progress: <%= name %></p>
-        <p><%= task.progress %> complete</p>
-      </div>
      <div :if={@server.status == :unregistered} class="my-10">
        <p class="mb-9">
          Connect to your server using root credentials and execute the following command:
@ -96,7 +92,7 @@ defmodule PrymnWeb.ServerLive.Show do
          <h2 class="my-5 text-xl">
            Terminal
          </h2>
-          <.live_component id="terminal" module={PrymnWeb.Terminal} server={@server} />
+          <.live_component id="terminal" module={PrymnWeb.Terminal} agent={@agent} />
        </section>
      </div>
      <.back navigate={~p"/servers"}>Back to servers</.back>
@ -107,65 +103,65 @@ defmodule PrymnWeb.ServerLive.Show do
  @impl true
  def handle_params(%{"id" => id}, _, socket) do
    server = Servers.get_server!(id)
-
-    socket =
-      if connected?(socket) and server.status == :registered do
-        agent = Agents.from_server(server)
-        Agents.subscribe_to_health(agent)
-        assign(socket, :agent, agent)
-      else
-        socket
-      end
-
-    health = Agents.get_health(server.public_ip)
+    agent = Agents.from_server(server)

    {:noreply,
     socket
     |> assign(:page_title, server.name)
-     |> assign(:health, health || %{message: "Connecting...", tasks: []})
     |> assign(:server, server)
+     |> assign(:agent, agent)
     |> assign(:dry_run, false)
     |> assign(:update_output, [])
     # TODO: Do not assign this to the socket - instead generate it in the HTML
     |> assign(:registration_command, Servers.create_setup_command(server))}
  end

-  @impl true
-  def handle_info(%PrymnProto.Prymn.SysUpdateResponse{} = response, socket) do
-    output = String.split(response.output, "\n")
-    socket = assign(socket, :update_output, output)
-    {:noreply, socket}
-  end
+  # @impl true
+  # def handle_info(%Agents.Agent{} = agent, socket) do
+  #   {:noreply, assign(socket, :agent, agent)}
+  # end

-  def handle_info(%Agents.Health{host: host} = health, socket) do
-    socket =
-      if host == socket.assigns.server.public_ip,
-        do: assign(socket, :health, health),
-        else: socket
+  # def handle_info(%Messaging.Messages.TerminalOutput{output: output}, socket) do
+  #   send_update(PrymnWeb.Terminal, id: "terminal", data: output)
+  #   {:noreply, socket}
+  # end

-    {:noreply, socket}
-  end
+  # @impl true
+  #  def handle_info(%PrymnProto.Prymn.SysUpdateResponse{} = response, socket) do
+  #    output = String.split(response.output, "\n")
+  #    socket = assign(socket, :update_output, output)
+  #    {:noreply, socket}
+  #  end
+
+  # def handle_info(%Agents.Health{host: host} = health, socket) do
+  #  socket =
+  #    if host == socket.assigns.server.public_ip,
+  #      do: assign(socket, :health, health),
+  #      else: socket
+
+  #  {:noreply, socket}
+  # end

  @impl true
  def handle_event("system_update", _params, socket) do
-    server_name = get_in(socket.assigns, [:server, Access.key(:name)])
-    pid = self()
+    # server_name = get_in(socket.assigns, [:server, Access.key(:name)])
+    # pid = self()

-    if agent = socket.assigns[:agent] do
-      # TODO: This is ugly
-      Task.start_link(fn ->
-        Agents.sys_update(agent, %{dry_run: socket.assigns.dry_run})
-        |> Stream.each(fn
-          {:ok, msg} -> send(pid, msg)
-          {:error, error} -> Logger.error("error during system update call: #{inspect(error)}")
-        end)
-        |> Enum.to_list()
-      end)
+    # if agent = socket.assigns[:agent] do
+    #   # TODO: This is ugly
+    #   Task.start_link(fn ->
+    #     Agents.sys_update(agent, %{dry_run: socket.assigns.dry_run})
+    #     |> Stream.each(fn
+    #       {:ok, msg} -> send(pid, msg)
+    #       {:error, error} -> Logger.error("error during system update call: #{inspect(error)}")
+    #     end)
+    #     |> Enum.to_list()
+    #   end)

-      put_flash(socket, :info, "Started a system update on server #{server_name}.")
-    else
-      put_flash(socket, :error, "Could not perform the update.")
-    end
+    #   put_flash(socket, :info, "Started a system update on server #{server_name}.")
+    # else
+    #   put_flash(socket, :error, "Could not perform the update.")
+    # end

    {:noreply, socket}
  end
--- a/flake.lock
+++ b/flake.lock
@ -20,10 +20,65 @@
        "type": "github"
      }
    },
+    "devenv": {
+      "inputs": {
+        "flake-compat": "flake-compat",
+        "nix": "nix",
+        "nixpkgs": "nixpkgs",
+        "pre-commit-hooks": "pre-commit-hooks"
+      },
+      "locked": {
+        "lastModified": 1706918346,
+        "narHash": "sha256-UYbmL0db7+yQNpQ3nyW5077kmtB3fT/M0h/LhosODm4=",
+        "owner": "cachix",
+        "repo": "devenv",
+        "rev": "e4019e17e9818cc3f107cef515d23d255e43e29a",
+        "type": "github"
+      },
+      "original": {
+        "owner": "cachix",
+        "repo": "devenv",
+        "type": "github"
+      }
+    },
+    "flake-compat": {
+      "flake": false,
+      "locked": {
+        "lastModified": 1673956053,
+        "narHash": "sha256-4gtG9iQuiKITOjNQQeQIpoIB6b16fm+504Ch3sNKLd8=",
+        "owner": "edolstra",
+        "repo": "flake-compat",
+        "rev": "35bb57c0c8d8b62bbfd284272c928ceb64ddbde9",
+        "type": "github"
+      },
+      "original": {
+        "owner": "edolstra",
+        "repo": "flake-compat",
+        "type": "github"
+      }
+    },
    "flake-utils": {
      "inputs": {
        "systems": "systems"
      },
+      "locked": {
+        "lastModified": 1685518550,
+        "narHash": "sha256-o2d0KcvaXzTrPRIo0kOLV0/QXHhDQ5DTi+OxcjO8xqY=",
+        "owner": "numtide",
+        "repo": "flake-utils",
+        "rev": "a1720a10a6cfe8234c0e93907ffe81be440f4cef",
+        "type": "github"
+      },
+      "original": {
+        "owner": "numtide",
+        "repo": "flake-utils",
+        "type": "github"
+      }
+    },
+    "flake-utils_2": {
+      "inputs": {
+        "systems": "systems_2"
+      },
      "locked": {
        "lastModified": 1701680307,
        "narHash": "sha256-kAuep2h5ajznlPMD9rnQyffWG8EM/C73lejGofXvdM8=",
@ -38,7 +93,117 @@
        "type": "github"
      }
    },
+    "gitignore": {
+      "inputs": {
+        "nixpkgs": [
+          "devenv",
+          "pre-commit-hooks",
+          "nixpkgs"
+        ]
+      },
+      "locked": {
+        "lastModified": 1660459072,
+        "narHash": "sha256-8DFJjXG8zqoONA1vXtgeKXy68KdJL5UaXR8NtVMUbx8=",
+        "owner": "hercules-ci",
+        "repo": "gitignore.nix",
+        "rev": "a20de23b925fd8264fd7fad6454652e142fd7f73",
+        "type": "github"
+      },
+      "original": {
+        "owner": "hercules-ci",
+        "repo": "gitignore.nix",
+        "type": "github"
+      }
+    },
+    "lowdown-src": {
+      "flake": false,
+      "locked": {
+        "lastModified": 1633514407,
+        "narHash": "sha256-Dw32tiMjdK9t3ETl5fzGrutQTzh2rufgZV4A/BbxuD4=",
+        "owner": "kristapsdz",
+        "repo": "lowdown",
+        "rev": "d2c2b44ff6c27b936ec27358a2653caaef8f73b8",
+        "type": "github"
+      },
+      "original": {
+        "owner": "kristapsdz",
+        "repo": "lowdown",
+        "type": "github"
+      }
+    },
+    "nix": {
+      "inputs": {
+        "lowdown-src": "lowdown-src",
+        "nixpkgs": [
+          "devenv",
+          "nixpkgs"
+        ],
+        "nixpkgs-regression": "nixpkgs-regression"
+      },
+      "locked": {
+        "lastModified": 1676545802,
+        "narHash": "sha256-EK4rZ+Hd5hsvXnzSzk2ikhStJnD63odF7SzsQ8CuSPU=",
+        "owner": "domenkozar",
+        "repo": "nix",
+        "rev": "7c91803598ffbcfe4a55c44ac6d49b2cf07a527f",
+        "type": "github"
+      },
+      "original": {
+        "owner": "domenkozar",
+        "ref": "relaxed-flakes",
+        "repo": "nix",
+        "type": "github"
+      }
+    },
    "nixpkgs": {
+      "locked": {
+        "lastModified": 1678875422,
+        "narHash": "sha256-T3o6NcQPwXjxJMn2shz86Chch4ljXgZn746c2caGxd8=",
+        "owner": "NixOS",
+        "repo": "nixpkgs",
+        "rev": "126f49a01de5b7e35a43fd43f891ecf6d3a51459",
+        "type": "github"
+      },
+      "original": {
+        "owner": "NixOS",
+        "ref": "nixpkgs-unstable",
+        "repo": "nixpkgs",
+        "type": "github"
+      }
+    },
+    "nixpkgs-regression": {
+      "locked": {
+        "lastModified": 1643052045,
+        "narHash": "sha256-uGJ0VXIhWKGXxkeNnq4TvV3CIOkUJ3PAoLZ3HMzNVMw=",
+        "owner": "NixOS",
+        "repo": "nixpkgs",
+        "rev": "215d4d0fd80ca5163643b03a33fde804a29cc1e2",
+        "type": "github"
+      },
+      "original": {
+        "owner": "NixOS",
+        "repo": "nixpkgs",
+        "rev": "215d4d0fd80ca5163643b03a33fde804a29cc1e2",
+        "type": "github"
+      }
+    },
+    "nixpkgs-stable": {
+      "locked": {
+        "lastModified": 1685801374,
+        "narHash": "sha256-otaSUoFEMM+LjBI1XL/xGB5ao6IwnZOXc47qhIgJe8U=",
+        "owner": "NixOS",
+        "repo": "nixpkgs",
+        "rev": "c37ca420157f4abc31e26f436c1145f8951ff373",
+        "type": "github"
+      },
+      "original": {
+        "owner": "NixOS",
+        "ref": "nixos-23.05",
+        "repo": "nixpkgs",
+        "type": "github"
+      }
+    },
+    "nixpkgs_2": {
      "locked": {
        "lastModified": 1703134684,
        "narHash": "sha256-SQmng1EnBFLzS7WSRyPM9HgmZP2kLJcPAz+Ug/nug6o=",
@ -54,11 +219,40 @@
        "type": "github"
      }
    },
+    "pre-commit-hooks": {
+      "inputs": {
+        "flake-compat": [
+          "devenv",
+          "flake-compat"
+        ],
+        "flake-utils": "flake-utils",
+        "gitignore": "gitignore",
+        "nixpkgs": [
+          "devenv",
+          "nixpkgs"
+        ],
+        "nixpkgs-stable": "nixpkgs-stable"
+      },
+      "locked": {
+        "lastModified": 1704725188,
+        "narHash": "sha256-qq8NbkhRZF1vVYQFt1s8Mbgo8knj+83+QlL5LBnYGpI=",
+        "owner": "cachix",
+        "repo": "pre-commit-hooks.nix",
+        "rev": "ea96f0c05924341c551a797aaba8126334c505d2",
+        "type": "github"
+      },
+      "original": {
+        "owner": "cachix",
+        "repo": "pre-commit-hooks.nix",
+        "type": "github"
+      }
+    },
    "root": {
      "inputs": {
        "crane": "crane",
-        "flake-utils": "flake-utils",
-        "nixpkgs": "nixpkgs",
+        "devenv": "devenv",
+        "flake-utils": "flake-utils_2",
+        "nixpkgs": "nixpkgs_2",
        "rust-overlay": "rust-overlay"
      }
    },
@ -99,6 +293,21 @@
        "repo": "default",
        "type": "github"
      }
+    },
+    "systems_2": {
+      "locked": {
+        "lastModified": 1681028828,
+        "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
+        "owner": "nix-systems",
+        "repo": "default",
+        "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
+        "type": "github"
+      },
+      "original": {
+        "owner": "nix-systems",
+        "repo": "default",
+        "type": "github"
+      }
    }
  },
  "root": "root",
--- a/flake.nix
+++ b/flake.nix
@ -1,8 +1,14 @@
 {
  description = "Prymn";

+  nixConfig = {
+    extra-trusted-public-keys = "devenv.cachix.org-1:w1cLUi8dv3hnoSPGAuibQv+f9TZLr6cv/Hm9XgU50cw=";
+    extra-substituters = "https://devenv.cachix.org";
+  };
+
  inputs = {
    nixpkgs.url = "github:NixOS/nixpkgs/nixpkgs-unstable";
+    devenv.url = "github:cachix/devenv";
    flake-utils.url = "github:numtide/flake-utils";
    rust-overlay = {
      url = "github:oxalica/rust-overlay";
@ -12,33 +18,45 @@
    crane = {
      url = "github:ipetkov/crane";
      inputs.nixpkgs.follows = "nixpkgs";
-      inputs.flake-utils.follows = "flake-utils";
-      inputs.rust-overlay.follows = "rust-overlay";
    };
  };

-  outputs = { self, nixpkgs, flake-utils, rust-overlay, crane }:
+  outputs = { self, nixpkgs, devenv, flake-utils, rust-overlay, crane }@inputs:
    flake-utils.lib.eachDefaultSystem (system:
      let
        overlays = [ (import ./nix/overlay.nix) (import rust-overlay) ];
        pkgs = import nixpkgs { inherit system overlays; };
-        scripts = pkgs.callPackage ./nix/scripts.nix { };
        rustBuilder = import ./nix/rust.nix { inherit crane pkgs system; };
      in
      {
-        devShells.default = with pkgs; mkShell {
-          packages = [
-            elixir
-            elixir-ls
-            rustToolchain
-            protobuf
-            protoc-gen-elixir
-            scripts.prymn_db
-          ] ++ lib.optionals stdenv.isLinux [ inotify-tools ];
+        packages = {
+          devenv-up = self.devShells.${system}.default.config.procfileScript;
+        };

-          shellHook = ''
-            export PROJECT_ROOT_DIR="$PWD"
-          '';
+        devShells.default = devenv.lib.mkShell {
+          inherit pkgs inputs;
+
+          modules = [
+            ({ pkgs, config, ... }: {
+              packages = [
+                pkgs.elixir
+                pkgs.elixir-ls
+                pkgs.rustToolchain
+                pkgs.go
+              ] ++ pkgs.lib.optionals pkgs.stdenv.isLinux [ pkgs.inotify-tools ];
+
+              services.postgres = {
+                enable = true;
+                package = pkgs.postgresql_16;
+                initialDatabases = [{ name = "prymn_dev"; }];
+                initialScript = "CREATE USER postgres SUPERUSER;";
+              };
+
+              enterShell = ''
+                echo happy deving
+              '';
+            })
+          ];
        };

        checks.rustTest = rustBuilder.test;
--- a/goagent/cmd/agent/main.go
+++ b/goagent/cmd/agent/main.go
@ -0,0 +1,35 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+
+	"nikos.codes/prymn/prymn/goagent/pkg/gophx"
+)
+
+func main() {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	socket, err := gophx.Connect(ctx)
+	if err != nil {
+		fmt.Println(err)
+	}
+
+	channel := gophx.NewChannel("agents:1", nil, socket)
+
+	channel.On("foo_bar", func(payload any) {
+		fmt.Println("received msg: ", payload)
+	})
+
+	handleSignals(cancel)
+}
+
+func handleSignals(cancel context.CancelFunc) {
+	signalCh := make(chan os.Signal, 1)
+	signal.Notify(signalCh)
+	<-signalCh
+	cancel()
+}
--- a/goagent/go.mod
+++ b/goagent/go.mod
@ -0,0 +1,5 @@
+module nikos.codes/prymn/prymn/goagent
+
+go 1.21.5
+
+require nhooyr.io/websocket v1.8.10 // indirect
--- a/goagent/go.sum
+++ b/goagent/go.sum
@ -0,0 +1,2 @@
+nhooyr.io/websocket v1.8.10 h1:mv4p+MnGrLDcPlBoWsvPP7XCzTYMXP9F9eIGoKbgx7Q=
+nhooyr.io/websocket v1.8.10/go.mod h1:rN9OFWIUwuxg4fR5tELlYC04bXYowCP9GX47ivo2l+c=
--- a/goagent/pkg/gophx/channel.go
+++ b/goagent/pkg/gophx/channel.go
@ -0,0 +1,45 @@
+package gophx
+
+type Channel struct {
+	topic    string
+	bindings map[string]func(any)
+}
+
+func NewChannel(topic string, params any, socket *Socket) *Channel {
+	// TODO: rejoin when socket is reconnected
+	channel := &Channel{
+		bindings: make(map[string]func(any)),
+		topic:    topic,
+	}
+
+	msg := message{
+		joinRef: 0,
+		ref:     0,
+		topic:   topic,
+		event:   "phx_join",
+		payload: params,
+	}
+
+	data, err := msg.serializeJSON()
+	if err != nil {
+		panic(err)
+	}
+
+	socket.addChannel(channel)
+	socket.send(data)
+	return channel
+}
+
+func (channel *Channel) On(event string, callback func(any)) {
+	channel.bindings[event] = callback
+}
+
+func (channel *Channel) Push(event string, params any) {
+
+}
+
+func (channel *Channel) handleMessage(msg message) {
+	if binding, ok := channel.bindings[msg.event]; ok {
+		go binding(msg.payload)
+	}
+}
--- a/goagent/pkg/gophx/message.go
+++ b/goagent/pkg/gophx/message.go
@ -0,0 +1,36 @@
+package gophx
+
+import (
+	"encoding/json"
+)
+
+type message struct {
+	joinRef int64
+	ref     int64
+	topic   string
+	event   string
+	payload any
+}
+
+func (m message) serializeJSON() ([]byte, error) {
+	data, err := json.Marshal([]any{m.joinRef, m.ref, m.topic, m.event, m.payload})
+	if err != nil {
+		return nil, err
+	}
+
+	return data, nil
+}
+
+func deserializeJSON(data []byte, msg *message) error {
+	m := message{}
+	slice := []any{&m.joinRef, &m.ref, &m.topic, &m.event, &m.payload}
+
+	err := json.Unmarshal(data, &slice)
+	if err != nil {
+		return err
+	}
+
+	*msg = m
+
+	return nil
+}
--- a/goagent/pkg/gophx/socket.go
+++ b/goagent/pkg/gophx/socket.go
@ -0,0 +1,99 @@
+package gophx
+
+import (
+	"context"
+	"fmt"
+
+	"nhooyr.io/websocket"
+)
+
+type Socket struct {
+	sendChan  chan []byte
+	websocket *websocket.Conn
+	channels  map[string]*Channel
+}
+
+func Connect(ctx context.Context) (*Socket, error) {
+	serverUrl := "ws://localhost:4000/agent/websocket?vsn=2.0.0"
+
+	c, _, err := websocket.Dial(ctx, serverUrl, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	socket := Socket{
+		sendChan:  make(chan []byte),
+		channels:  make(map[string]*Channel),
+		websocket: c,
+	}
+
+	go socket.readMessages(ctx)
+
+	go writeMessages(ctx, socket.sendChan, c)
+
+	return &socket, nil
+}
+
+func (s *Socket) addChannel(ch *Channel) {
+	s.channels[ch.topic] = ch
+}
+
+func (s *Socket) send(data []byte) {
+	s.sendChan <- data
+}
+
+func (s *Socket) readMessages(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			typ, data, err := s.websocket.Read(ctx)
+			if err != nil {
+				// TODO: try to reconnect
+				fmt.Println("socket closed?", err)
+				return
+			}
+
+			switch typ {
+			case websocket.MessageBinary:
+				fmt.Println("received binary", data)
+			case websocket.MessageText:
+				msg := message{}
+
+				err := deserializeJSON(data, &msg)
+				if err != nil {
+					fmt.Println(err)
+				}
+
+				s.sendToChannel(msg)
+			}
+		}
+	}
+}
+
+func (s *Socket) sendToChannel(msg message) {
+	ch, ok := s.channels[msg.topic]
+	if !ok {
+		return
+	}
+
+	ch.handleMessage(msg)
+}
+
+func writeMessages(ctx context.Context, sendChan chan []byte, ws *websocket.Conn) {
+	for {
+		select {
+		case message, ok := <-sendChan:
+			if !ok {
+				fmt.Println("sending channel closed")
+				return
+			}
+
+			err := ws.Write(ctx, websocket.MessageText, message)
+			if err != nil {
+				fmt.Println(err)
+			}
+		}
+	}
+}
--- a/mix.exs
+++ b/mix.exs
@ -6,7 +6,6 @@ defmodule Prymn.MixProject do
      app: :prymn,
      version: "0.1.0",
      elixir: "~> 1.15",
-      compilers: [:proto | Mix.compilers()],
      elixirc_paths: elixirc_paths(Mix.env()),
      config_path: Path.expand("app/config/config.exs", __DIR__),
      test_paths: ["app/test"],
@ -51,10 +50,8 @@ defmodule Prymn.MixProject do
      {:jason, "~> 1.2"},
      {:dns_cluster, "~> 0.1.1"},
      {:plug_cowboy, "~> 2.5"},
-      {:grpc, "~> 0.7"},
-      {:protobuf, "~> 0.12.0"},
-      {:google_protos, "~> 0.3.0"},
      {:oban, "~> 2.17"},
+      {:gnat, "~> 1.7"},

      # Test
      {:floki, ">= 0.30.0", only: :test},
@ -66,7 +63,6 @@ defmodule Prymn.MixProject do
      {:tailwind_formatter, "~> 0.3.6", runtime: Mix.env() == :dev},
      {:phoenix_live_reload, "~> 1.2", only: :dev},
      {:dialyxir, "~> 1.4", only: [:dev], runtime: false},
-      {:prymn_proto_compiler, path: "proto_compiler", runtime: false}
    ]
  end

--- a/mix.lock
+++ b/mix.lock
@ -13,6 +13,7 @@
  "dns_cluster": {:hex, :dns_cluster, "0.1.1", "73b4b2c3ec692f8a64276c43f8c929733a9ab9ac48c34e4c0b3d9d1b5cd69155", [:mix], [], "hexpm", "03a3f6ff16dcbb53e219b99c7af6aab29eb6b88acf80164b4bd76ac18dc890b3"},
  "ecto": {:hex, :ecto, "3.11.1", "4b4972b717e7ca83d30121b12998f5fcdc62ba0ed4f20fd390f16f3270d85c3e", [:mix], [{:decimal, "~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "ebd3d3772cd0dfcd8d772659e41ed527c28b2a8bde4b00fe03e0463da0f1983b"},
  "ecto_sql": {:hex, :ecto_sql, "3.11.1", "e9abf28ae27ef3916b43545f9578b4750956ccea444853606472089e7d169470", [:mix], [{:db_connection, "~> 2.4.1 or ~> 2.5", [hex: :db_connection, repo: "hexpm", optional: false]}, {:ecto, "~> 3.11.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.6.0", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.16.0 or ~> 0.17.0 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "ce14063ab3514424276e7e360108ad6c2308f6d88164a076aac8a387e1fea634"},
+  "ed25519": {:hex, :ed25519, "1.4.1", "479fb83c3e31987c9cad780e6aeb8f2015fb5a482618cdf2a825c9aff809afc4", [:mix], [], "hexpm", "0dacb84f3faa3d8148e81019ca35f9d8dcee13232c32c9db5c2fb8ff48c80ec7"},
  "elixir_make": {:hex, :elixir_make, "0.7.7", "7128c60c2476019ed978210c245badf08b03dbec4f24d05790ef791da11aa17c", [:mix], [{:castore, "~> 0.1 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}], "hexpm", "5bc19fff950fad52bbe5f211b12db9ec82c6b34a9647da0c2224b8b8464c7e6c"},
  "erlex": {:hex, :erlex, "0.2.6", "c7987d15e899c7a2f34f5420d2a2ea0d659682c06ac607572df55a43753aa12e", [:mix], [], "hexpm", "2ed2e25711feb44d52b17d2780eabf998452f6efda104877a3881c2f8c0c0c75"},
  "esbuild": {:hex, :esbuild, "0.8.1", "0cbf919f0eccb136d2eeef0df49c4acf55336de864e63594adcea3814f3edf41", [:mix], [{:castore, ">= 0.0.0", [hex: :castore, repo: "hexpm", optional: false]}, {:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "25fc876a67c13cb0a776e7b5d7974851556baeda2085296c14ab48555ea7560f"},
@ -21,15 +22,15 @@
  "finch": {:hex, :finch, "0.16.0", "40733f02c89f94a112518071c0a91fe86069560f5dbdb39f9150042f44dcfb1a", [:mix], [{:castore, "~> 0.1 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: false]}, {:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.3", [hex: :mint, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 0.2.6 or ~> 1.0", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "f660174c4d519e5fec629016054d60edd822cdfe2b7270836739ac2f97735ec5"},
  "floki": {:hex, :floki, "0.35.2", "87f8c75ed8654b9635b311774308b2760b47e9a579dabf2e4d5f1e1d42c39e0b", [:mix], [], "hexpm", "6b05289a8e9eac475f644f09c2e4ba7e19201fd002b89c28c1293e7bd16773d9"},
  "gettext": {:hex, :gettext, "0.24.0", "6f4d90ac5f3111673cbefc4ebee96fe5f37a114861ab8c7b7d5b30a1108ce6d8", [:mix], [{:expo, "~> 0.5.1", [hex: :expo, repo: "hexpm", optional: false]}], "hexpm", "bdf75cdfcbe9e4622dd18e034b227d77dd17f0f133853a1c73b97b3d6c770e8b"},
-  "google_protos": {:hex, :google_protos, "0.3.0", "15faf44dce678ac028c289668ff56548806e313e4959a3aaf4f6e1ebe8db83f4", [:mix], [{:protobuf, "~> 0.10", [hex: :protobuf, repo: "hexpm", optional: false]}], "hexpm", "1f6b7fb20371f72f418b98e5e48dae3e022a9a6de1858d4b254ac5a5d0b4035f"},
-  "grpc": {:hex, :grpc, "0.7.0", "a86eab356b0b84406b526786a947ca50e9b9eae87108c873b51e321f8a71e8ed", [:mix], [{:cowboy, "~> 2.10", [hex: :cowboy, repo: "hexpm", optional: false]}, {:cowlib, "~> 2.12", [hex: :cowlib, repo: "hexpm", optional: false]}, {:gun, "~> 2.0", [hex: :gun, repo: "hexpm", optional: false]}, {:mint, "~> 1.5", [hex: :mint, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "632a9507da8d3c12b112b197db4d60da3c95bad02594d37711eeb622d032f254"},
-  "gun": {:hex, :gun, "2.0.1", "160a9a5394800fcba41bc7e6d421295cf9a7894c2252c0678244948e3336ad73", [:make, :rebar3], [{:cowlib, "2.12.1", [hex: :cowlib, repo: "hexpm", optional: false]}], "hexpm", "a10bc8d6096b9502205022334f719cc9a08d9adcfbfc0dbee9ef31b56274a20b"},
+  "gnat": {:hex, :gnat, "1.7.1", "491144f9c3aec00e9941d69538e2fd2836271e220315c8d2d87907c20ca7abc8", [:mix], [{:cowlib, "~> 2.0", [hex: :cowlib, repo: "hexpm", optional: false]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 0.5 or ~> 1.0", [hex: :nimble_parsec, repo: "hexpm", optional: false]}, {:nkeys, "~> 0.2", [hex: :nkeys, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "a5629088d9bdb16d982eb48fd431cf6c5a71e9b026281781983501237ab5b911"},
  "hpax": {:hex, :hpax, "0.1.2", "09a75600d9d8bbd064cdd741f21fc06fc1f4cf3d0fcc335e5aa19be1a7235c84", [:mix], [], "hexpm", "2c87843d5a23f5f16748ebe77969880e29809580efdaccd615cd3bed628a8c13"},
  "jason": {:hex, :jason, "1.4.1", "af1504e35f629ddcdd6addb3513c3853991f694921b1b9368b0bd32beb9f1b63", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "fbb01ecdfd565b56261302f7e1fcc27c4fb8f32d56eab74db621fc154604a7a1"},
  "mime": {:hex, :mime, "2.0.5", "dc34c8efd439abe6ae0343edbb8556f4d63f178594894720607772a041b04b02", [:mix], [], "hexpm", "da0d64a365c45bc9935cc5c8a7fc5e49a0e0f9932a761c55d6c52b142780a05c"},
  "mint": {:hex, :mint, "1.5.2", "4805e059f96028948870d23d7783613b7e6b0e2fb4e98d720383852a760067fd", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:hpax, "~> 0.1.1", [hex: :hpax, repo: "hexpm", optional: false]}], "hexpm", "d77d9e9ce4eb35941907f1d3df38d8f750c357865353e21d335bdcdf6d892a02"},
  "nimble_options": {:hex, :nimble_options, "1.1.0", "3b31a57ede9cb1502071fade751ab0c7b8dbe75a9a4c2b5bbb0943a690b63172", [:mix], [], "hexpm", "8bbbb3941af3ca9acc7835f5655ea062111c9c27bcac53e004460dfd19008a99"},
+  "nimble_parsec": {:hex, :nimble_parsec, "1.4.0", "51f9b613ea62cfa97b25ccc2c1b4216e81df970acd8e16e8d1bdc58fef21370d", [:mix], [], "hexpm", "9c565862810fb383e9838c1dd2d7d2c437b3d13b267414ba6af33e50d2d1cf28"},
  "nimble_pool": {:hex, :nimble_pool, "1.0.0", "5eb82705d138f4dd4423f69ceb19ac667b3b492ae570c9f5c900bb3d2f50a847", [:mix], [], "hexpm", "80be3b882d2d351882256087078e1b1952a28bf98d0a287be87e4a24a710b67a"},
+  "nkeys": {:hex, :nkeys, "0.2.2", "b1ab3324ed4f3a2c9658d7e80feeef86b4d15fbfd12ca5c8cf068289f582fcfa", [:mix], [{:ed25519, "~> 1.3", [hex: :ed25519, repo: "hexpm", optional: false]}], "hexpm", "3578802427b8d1d11ea6dd785c2ab774f527e2c3e449e67bd34612ab71ca471d"},
  "oban": {:hex, :oban, "2.17.1", "42d6221a1c17b63d81c19e3bad9ea82b59e39c47c1f9b7670ee33628569a449b", [:mix], [{:ecto_sql, "~> 3.6", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:ecto_sqlite3, "~> 0.9", [hex: :ecto_sqlite3, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.16", [hex: :postgrex, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "c02686ada7979b00e259c0efbafeae2749f8209747b3460001fe695c5bdbeee6"},
  "phoenix": {:hex, :phoenix, "1.7.10", "02189140a61b2ce85bb633a9b6fd02dff705a5f1596869547aeb2b2b95edd729", [:mix], [{:castore, ">= 0.0.0", [hex: :castore, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix_pubsub, "~> 2.1", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:phoenix_template, "~> 1.0", [hex: :phoenix_template, repo: "hexpm", optional: false]}, {:phoenix_view, "~> 2.0", [hex: :phoenix_view, repo: "hexpm", optional: true]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 2.6", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:plug_crypto, "~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:websock_adapter, "~> 0.5.3", [hex: :websock_adapter, repo: "hexpm", optional: false]}], "hexpm", "cf784932e010fd736d656d7fead6a584a4498efefe5b8227e9f383bf15bb79d0"},
  "phoenix_ecto": {:hex, :phoenix_ecto, "4.4.3", "86e9878f833829c3f66da03d75254c155d91d72a201eb56ae83482328dc7ca93", [:mix], [{:ecto, "~> 3.5", [hex: :ecto, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 2.14.2 or ~> 3.0 or ~> 4.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}, {:plug, "~> 1.9", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "d36c401206f3011fefd63d04e8ef626ec8791975d9d107f9a0817d426f61ac07"},
@ -43,7 +44,6 @@
  "plug_cowboy": {:hex, :plug_cowboy, "2.6.1", "9a3bbfceeb65eff5f39dab529e5cd79137ac36e913c02067dba3963a26efe9b2", [:mix], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:cowboy_telemetry, "~> 0.3", [hex: :cowboy_telemetry, repo: "hexpm", optional: false]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "de36e1a21f451a18b790f37765db198075c25875c64834bcc82d90b309eb6613"},
  "plug_crypto": {:hex, :plug_crypto, "2.0.0", "77515cc10af06645abbfb5e6ad7a3e9714f805ae118fa1a70205f80d2d70fe73", [:mix], [], "hexpm", "53695bae57cc4e54566d993eb01074e4d894b65a3766f1c43e2c61a1b0f45ea9"},
  "postgrex": {:hex, :postgrex, "0.17.4", "5777781f80f53b7c431a001c8dad83ee167bcebcf3a793e3906efff680ab62b3", [:mix], [{:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "6458f7d5b70652bc81c3ea759f91736c16a31be000f306d3c64bcdfe9a18b3cc"},
-  "protobuf": {:hex, :protobuf, "0.12.0", "58c0dfea5f929b96b5aa54ec02b7130688f09d2de5ddc521d696eec2a015b223", [:mix], [{:jason, "~> 1.2", [hex: :jason, repo: "hexpm", optional: true]}], "hexpm", "75fa6cbf262062073dd51be44dd0ab940500e18386a6c4e87d5819a58964dc45"},
  "ranch": {:hex, :ranch, "1.8.0", "8c7a100a139fd57f17327b6413e4167ac559fbc04ca7448e9be9057311597a1d", [:make, :rebar3], [], "hexpm", "49fbcfd3682fab1f5d109351b61257676da1a2fdbe295904176d5e521a2ddfe5"},
  "swoosh": {:hex, :swoosh, "1.14.3", "949e6bf6dd469449238a94ec6f19ec10b63fc8753de7f3ebe3d3aeaf772f4c6b", [:mix], [{:cowboy, "~> 1.1 or ~> 2.4", [hex: :cowboy, repo: "hexpm", optional: true]}, {:ex_aws, "~> 2.1", [hex: :ex_aws, repo: "hexpm", optional: true]}, {:finch, "~> 0.6", [hex: :finch, repo: "hexpm", optional: true]}, {:gen_smtp, "~> 0.13 or ~> 1.0", [hex: :gen_smtp, repo: "hexpm", optional: true]}, {:hackney, "~> 1.9", [hex: :hackney, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mail, "~> 0.2", [hex: :mail, repo: "hexpm", optional: true]}, {:mime, "~> 1.1 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug, "~> 1.9", [hex: :plug, repo: "hexpm", optional: true]}, {:plug_cowboy, ">= 1.0.0", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:req, "~> 0.4 or ~> 1.0", [hex: :req, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.2 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "6c565103fc8f086bdd96e5c948660af8e20922b7a90a75db261f06a34f805c8b"},
  "tailwind": {:hex, :tailwind, "0.2.2", "9e27288b568ede1d88517e8c61259bc214a12d7eed271e102db4c93fcca9b2cd", [:mix], [{:castore, ">= 0.0.0", [hex: :castore, repo: "hexpm", optional: false]}], "hexpm", "ccfb5025179ea307f7f899d1bb3905cd0ac9f687ed77feebc8f67bdca78565c4"},
--- a/nix/overlay.nix
+++ b/nix/overlay.nix
@ -6,6 +6,4 @@ final: prev: {
  elixir = prev.beam.packages.erlang_26.elixir_1_15;

  elixir-ls = prev.beam.packages.erlang_26.elixir-ls.override { elixir = final.elixir; };
-
-  protoc-gen-elixir = prev.callPackage ./protoc-gen-elixir.nix { elixir = final.elixir; };
 }
--- a/nix/protoc-gen-elixir.nix
+++ b/nix/protoc-gen-elixir.nix
@ -1,28 +0,0 @@
-{ elixir, beamPackages, fetchFromGitHub }:
-
-beamPackages.mixRelease rec {
-  pname = "protoc-gen-elixir";
-
-  version = "v0.12.0";
-
-  inherit elixir;
-
-  src = fetchFromGitHub {
-    owner = "elixir-protobuf";
-    repo = "protobuf";
-    rev = "ce9a031a5cae97336d4674670d313d54f1f80bf6";
-    sha256 = "wLU3iM9jI/Zc96/HfPUjNvjteGryWos6IobIb/4zqpw=";
-  };
-
-  mixFodDeps = beamPackages.fetchMixDeps {
-    pname = "mix-deps-${pname}";
-    inherit src version;
-    sha256 = "H7yiBHoxuiqWcNbWwPU5X0Nnv8f6nM8z/ZAfZAGPZjE=";
-  };
-
-  installPhase = ''
-    mix escript.build
-    mkdir -p $out/bin
-    mv ./protoc-gen-elixir $out/bin
-  '';
-}
--- a/nix/scripts.nix
+++ b/nix/scripts.nix
@ -1,47 +0,0 @@
-{ writeShellApplication, postgresql, ... }:
-
-{
-  prymn_db = writeShellApplication {
-    name = "prymn_db";
-
-    runtimeInputs = [ postgresql ];
-
-    text = ''
-      #!/usr/bin/env bash
-
-      set -e
-
-      export PGDATA=$PROJECT_ROOT_DIR/.db
-      export PGHOST=/tmp
-      export DB_LOG=$PROJECT_ROOT_DIR/.db/log
-
-      start_db() {
-        if [ ! -d "$PGDATA" ]; then
-            initdb "$PGDATA" --auth=trust
-        fi
-
-        if ! pg_ctl status >/dev/null; then
-            pg_ctl start -l "$DB_LOG" -o "-c unix_socket_directories=$PGHOST"
-            echo "starting your dev database..."
-        fi
-
-        user_exists=$(psql --csv -t -d postgres -c "SELECT count(*) FROM pg_user WHERE usename='postgres'")
-        if [ "$user_exists" != "1" ]; then
-            createuser -s -h "$PGHOST" postgres
-        fi
-      }
-
-      command="''${1-default}"
-      case $command in
-        shell)
-          psql -U postgres
-          exit 0
-          ;;
-        start | default | *)
-          start_db
-          exit 0
-          ;;
-      esac
-    '';
-  };
-}
--- a/proto/agent.proto
+++ b/proto/agent.proto
@ -1,92 +0,0 @@
-syntax = "proto3";
-
-import "google/protobuf/empty.proto";
-
-package prymn;
-
-message SystemHealth {
-    // Comma-separated statuses
-    string status = 1;
-}
-
-message TaskHealth {
-    string started_on = 1;
-    float progress = 2;
-}
-
-message HealthResponse {
-    string version = 1;
-    SystemHealth system = 2;
-    map<string, TaskHealth> tasks = 3;
-}
-
-message SysInfoResponse {
-    message Cpu {
-        uint64 freq_mhz = 1;
-        float  usage = 2;
-    }
-
-    message Disk {
-        string name = 1;
-        uint64 total_bytes = 2;
-        uint64 avail_bytes = 3;
-        string mount_point = 4;
-    }
-
-    uint64 uptime = 1;
-    string hostname = 2;
-    string os = 3;
-    uint64 mem_total_bytes = 4;
-    uint64 mem_avail_bytes = 5;
-    uint64 swap_total_bytes = 6;
-    uint64 swap_free_bytes = 7;
-    repeated Cpu  cpus = 8;
-    repeated Disk disks = 9;
-    uint32 updates_available = 10;
-}
-
-message ExecRequest {
-    string user = 1;
-    string program = 2;
-    repeated string args = 3;
-}
-
-message ExecResponse {
-    oneof out {
-        string stdout = 1;
-        string stderr = 2;
-        string error = 3;
-        int32 exit_code = 4;
-    }
-}
-
-message SysUpdateRequest {
-    bool dry_run = 1;
-}
-
-message SysUpdateResponse {
-    string output = 1;
-    int32 progress = 2;
-}
-
-message TerminalRequest {
-    message Resize {
-        uint32 rows = 1;
-        uint32 cols = 2;
-    }
-
-    bytes input = 1;
-    optional Resize resize = 2;
-}
-
-message TerminalResponse {
-    bytes output = 1;
-}
-
-service Agent {
-    rpc Health(google.protobuf.Empty) returns (stream HealthResponse);
-    rpc Exec(ExecRequest) returns (stream ExecResponse);
-    rpc Terminal(stream TerminalRequest) returns (stream TerminalResponse);
-    rpc GetSysInfo(google.protobuf.Empty) returns (SysInfoResponse);
-    rpc SysUpdate(SysUpdateRequest) returns (stream SysUpdateResponse);
-}
--- a/proto_compiler/.formatter.exs
+++ b/proto_compiler/.formatter.exs
@ -1,4 +0,0 @@
-# Used by "mix format"
-[
-  inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"]
-]
--- a/proto_compiler/.gitignore
+++ b/proto_compiler/.gitignore
@ -1,26 +0,0 @@
-# The directory Mix will write compiled artifacts to.
-/_build/
-
-# If you run "mix test --cover", coverage assets end up here.
-/cover/
-
-# The directory Mix downloads your dependencies sources to.
-/deps/
-
-# Where third-party dependencies like ExDoc output generated docs.
-/doc/
-
-# Ignore .fetch files in case you like to edit your project deps locally.
-/.fetch
-
-# If the VM crashes, it generates a dump, let's ignore it too.
-erl_crash.dump
-
-# Also ignore archive artifacts (built via "mix archive.build").
-*.ez
-
-# Ignore package tarball (built via "mix hex.build").
-proto-*.tar
-
-# Temporary files, for example, from tests.
-/tmp/
--- a/proto_compiler/lib/mix/tasks/compile.proto.ex
+++ b/proto_compiler/lib/mix/tasks/compile.proto.ex
@ -1,34 +0,0 @@
-defmodule Mix.Tasks.Compile.Proto do
-  use Mix.Task.Compiler
-
-  @manifest "compile.proto"
-
-  @impl true
-  def run(_args) do
-    output = "./app/lib/prymn_proto"
-    sources = Path.wildcard("proto/*.proto")
-    targets = Path.wildcard(output <> "/*.pb.ex")
-
-    if Mix.Utils.stale?(sources, targets) do
-      {_, 0} = do_protoc(sources, output)
-    end
-
-    :ok
-  end
-
-  @impl true
-  def manifests(), do: [manifest()]
-  defp manifest(), do: Path.join(Mix.Project.manifest_path(), @manifest)
-
-  defp do_protoc(sources, output) do
-    System.cmd(
-      "protoc",
-      [
-        "--elixir_out=plugins=grpc:" <> output,
-        "--elixir_opt=package_prefix=prymn_proto",
-        "-I",
-        "proto"
-      ] ++ sources
-    )
-  end
-end
--- a/proto_compiler/mix.exs
+++ b/proto_compiler/mix.exs
@ -1,22 +0,0 @@
-defmodule PrymnProtoCompiler.MixProject do
-  use Mix.Project
-
-  def project do
-    [
-      app: :prymn_proto_compiler,
-      version: "0.1.0",
-      elixir: "~> 1.15",
-      deps: deps()
-    ]
-  end
-
-  # Run "mix help compile.app" to learn about applications.
-  def application do
-    []
-  end
-
-  # Run "mix help deps" to learn about dependencies.
-  defp deps do
-    []
-  end
-end
Author	SHA1	Message	Date
Nikos Papadakis	dfd446f770	goagent: bare bones api	2024-02-17 17:20:08 +02:00
Nikos Papadakis	b4edfd7585	for go	2024-02-15 15:25:45 +02:00
Nikos Papadakis	1a5225ef6f	foo	2024-02-15 11:21:57 +02:00
Nikos Papadakis	6846e96f86	socket	2024-02-05 01:25:37 +02:00
Nikos Papadakis	6db2478786	todo: agents as genservers, messaging connection manager	2024-02-04 20:13:35 +02:00
Nikos Papadakis	3e066fd23a	subscription	2024-02-03 13:11:31 +02:00
Nikos Papadakis	32727b2ab8	add devenv + nats	2024-02-03 13:11:23 +02:00
Nikos Papadakis	59945bd2de	terminal	2024-02-02 20:53:52 +02:00
Nikos Papadakis	e0850c1d2b	make health somewhat work	2024-02-01 23:54:23 +02:00
Nikos Papadakis	b4cd5642ed	app refactoring	2024-02-01 17:34:26 +02:00