Add local in-memory pubsub

crates/geth-node/Cargo.toml

@@ -20,6 +20,7 @@ geth-document = { path = "../geth-document" }
 geth-iroh = { path = "../geth-iroh" }
 geth-keychain = { path = "../geth-keychain" }
 geth-kv = { path = "../geth-kv" }
+geth-pubsub = { path = "../geth-pubsub" }
 geth-resource = { path = "../geth-resource" }
 geth-secrets = { path = "../geth-secrets" }
 geth-ssh-identity = { path = "../geth-ssh-identity" }

crates/geth-node/src/lib.rs

@@ -13,6 +13,7 @@ use geth_document::DocumentResource;
 use geth_iroh::{EndpointStatus, GethIrohConfig, GethIrohEndpoint, GethRelayMode};
 use geth_keychain::{KeychainOp, KeychainOpKind};
 use geth_kv::{KvEntry, KvResource};
+use geth_pubsub::PubsubMessage;
 use geth_resource::ResourceDescriptor;
 use geth_secrets::{BearerAccess, ResourceMasterSecret};
 use geth_ssh_identity::{
@@ -29,7 +30,9 @@ use geth_types::{
     AuthOpId, Capability, KeyId, NodeId, PrincipalId, ResourceId, ResourceKind, ResourceName,
     SshCertId, SshCertRequestId, UnixMillis,
 };
+use std::collections::VecDeque;
 use std::path::Path;
+use std::sync::{Arc, Mutex};
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 use tokio::net::{UnixListener, UnixStream};
 
@@ -73,6 +76,10 @@ pub enum NodeError {
     ResourceNotFound(String),
     #[error("secrets error: {0}")]
     Secrets(#[from] geth_secrets::SecretsError),
+    #[error("pubsub error: {0}")]
+    Pubsub(#[from] geth_pubsub::PubsubError),
+    #[error("runtime state lock poisoned")]
+    RuntimeLockPoisoned,
     #[error("invalid ssh certificate kind: {0}")]
     InvalidSshCertKind(String),
     #[error("invalid ssh certificate request status: {0}")]
@@ -93,8 +100,21 @@ pub struct LocalNode {
     pub agent_id: String,
     pub node_id: String,
     pub iroh_status: EndpointStatus,
+    runtime: Arc<NodeRuntime>,
 }
 
+#[derive(Debug)]
+struct NodeRuntime {
+    pubsub: Mutex<PubsubRuntime>,
+}
+
+#[derive(Debug, Default)]
+struct PubsubRuntime {
+    messages: VecDeque<PubsubMessage>,
+}
+
+const PUBSUB_RING_LIMIT: usize = 256;
+
 pub fn init_node(paths: &GethPaths) -> Result<LocalNode, NodeError> {
     paths.ensure_base_dirs()?;
     if !paths.config_file().exists() {
@@ -117,6 +137,9 @@ pub fn init_node(paths: &GethPaths) -> Result<LocalNode, NodeError> {
         agent_id,
         node_id,
         iroh_status: EndpointStatus::scaffolded(),
+        runtime: Arc::new(NodeRuntime {
+            pubsub: Mutex::new(PubsubRuntime::default()),
+        }),
     })
 }
 
@@ -784,6 +807,44 @@ pub fn handle_request(
                 document: document_resource_from_stored(&stored),
             })
         }
+        ControlRequest::PubsubPub { topic, message } => {
+            geth_pubsub::validate_topic(&topic)?;
+            geth_pubsub::validate_message(&message)?;
+            let message = PubsubMessage {
+                topic: topic.clone().into(),
+                message,
+                published_at: UnixMillis(geth_store::now_ms()),
+            };
+            let mut runtime = node
+                .runtime
+                .pubsub
+                .lock()
+                .map_err(|_| NodeError::RuntimeLockPoisoned)?;
+            runtime.messages.push_back(message.clone());
+            while runtime.messages.len() > PUBSUB_RING_LIMIT {
+                runtime.messages.pop_front();
+            }
+            Ok(ControlResponse::PubsubPublished { message })
+        }
+        ControlRequest::PubsubSub { topic } => {
+            geth_pubsub::validate_topic(&topic)?;
+            let runtime = node
+                .runtime
+                .pubsub
+                .lock()
+                .map_err(|_| NodeError::RuntimeLockPoisoned)?;
+            let messages = runtime
+                .messages
+                .iter()
+                .filter(|message| message.topic.as_str() == topic)
+                .cloned()
+                .collect();
+            Ok(ControlResponse::PubsubMessages {
+                topic,
+                messages,
+                note: geth_pubsub::pubsub_storage_warning().to_owned(),
+            })
+        }
         ControlRequest::ModuleStub { module, command } => {
             Ok(ControlResponse::NotImplemented { module, command })
         }