feat: separate out backend block execution (#135)

* feat: new backend-driven block execution runtime

* fix: rebase broken code

* format

* fix tests

* revert to ssh buffer reader from main

Author: ellie

backend/Cargo.lock

@@ -38,6 +38,8 @@ nix = { version = "0.29.0", features = ["signal"] }
 lazy_static = "1.5.0"
 shellexpand = "3.1.0"
 open = "5"
+reqwest = { version = "0.12", features = ["json"] }
+url = "2.4"
 klickhouse = { version = "0.14.0", features = ["tls"] }
 sha2 = "0.10"
 bigdecimal = "0.4"
@@ -66,6 +68,7 @@ minijinja = { version = "2.12.0", features = [
     "urlencode",
     "loop_controls",
     "key_interning",
+    "builtins"
 ] }
 walkdir = "2.5.0"
 thiserror = "2.0.11"
@@ -81,10 +84,10 @@ russh-config = "0.54.0" # For SSH config parsing with glob support
 dirs = "5.0" # For finding home directory
 glob = "0.3" # For general glob pattern matching
 base64 = "0.22"
+chrono = { version = "0.4", features = ["serde", "clock"] }
 async-trait = "0.1.89"
 notify-debouncer-full = { version = "0.6.0", features = ["serde"] }
 toml = "0.9.2"
-
 ts-rs = { version = "11.0.1", features = ["serde-json-impl", "uuid-impl"] }
 json-digest = "0.0.16"
 trash = "5.2.2"
@@ -97,10 +100,17 @@ features = [
     "runtime-tokio",
     "tls-native-tls",
     "time",
+    "chrono",
     "postgres",
     "mysql",
+    "sqlite",
     "uuid",
+    "chrono",
     "postgres",
+    "mysql",
+    "sqlite",
+    "uuid",
+    "bigdecimal",
     "bigdecimal",
 ]
 

backend/Cargo.toml

@@ -0,0 +1,98 @@
+use tauri::{ipc::Channel, AppHandle, Manager, State};
+use uuid::Uuid;
+
+use crate::commands::events::ChannelEventBus;
+use crate::runtime::blocks::handler::BlockOutput;
+use crate::runtime::blocks::registry::BlockRegistry;
+use crate::runtime::blocks::Block;
+use crate::runtime::workflow::context_builder::ContextBuilder;
+use crate::state::AtuinState;
+
+/// Convert editor document block to runtime Block enum
+fn document_to_block(block_data: &serde_json::Value) -> Result<Block, String> {
+    Block::from_document(block_data)
+}
+
+#[tauri::command]
+pub async fn execute_block(
+    state: State<'_, AtuinState>,
+    app_handle: AppHandle,
+    block_id: String,
+    runbook_id: String,
+    editor_document: Vec<serde_json::Value>,
+    output_channel: Channel<BlockOutput>,
+) -> Result<String, String> {
+    // Build execution context
+    let mut context = ContextBuilder::build_context(&block_id, &editor_document, &runbook_id)
+        .await
+        .map_err(|e| e.to_string())?;
+
+    // Add SSH pool to context
+    context.ssh_pool = Some(state.ssh_pool());
+
+    // Add output storage to context
+    context.output_storage = Some(state.runbook_output_variables.clone());
+
+    // Add PTY store to context
+    context.pty_store = Some(state.pty_store());
+
+    // Add event bus to context
+    let gc_sender = state.gc_event_sender();
+    let event_bus = std::sync::Arc::new(ChannelEventBus::new(gc_sender));
+    context.event_bus = Some(event_bus);
+
+    // Find the block in the document
+    let block_data = editor_document
+        .iter()
+        .find(|b| b.get("id").and_then(|v| v.as_str()) == Some(&block_id))
+        .ok_or("Block not found")?;
+
+    // Convert document block to runtime block
+    let block = document_to_block(block_data)?;
+
+    // Get event sender from state
+    let event_sender = state.event_sender();
+
+    // Create registry and execute
+    let registry = BlockRegistry::new();
+
+    match registry
+        .execute_block(&block, context, event_sender, Some(output_channel))
+        .await
+    {
+        Ok(handle) => {
+            let execution_id = handle.id;
+            // Store the execution handle for cancellation
+            if let Some(state) = app_handle.try_state::<AtuinState>() {
+                state
+                    .block_executions
+                    .write()
+                    .await
+                    .insert(execution_id, handle.clone());
+            }
+            Ok(execution_id.to_string())
+        }
+        Err(e) => Err(format!("Execution failed: {}", e)),
+    }
+}
+
+#[tauri::command]
+pub async fn cancel_block_execution(
+    app_handle: AppHandle,
+    execution_id: String,
+) -> Result<(), String> {
+    let execution_uuid = Uuid::parse_str(&execution_id).map_err(|e| e.to_string())?;
+
+    if let Some(state) = app_handle.try_state::<AtuinState>() {
+        let mut executions = state.block_executions.write().await;
+        if let Some(handle) = executions.remove(&execution_uuid) {
+            // Cancel the execution
+            handle.cancellation_token.cancel();
+            Ok(())
+        } else {
+            Err("Execution not found".to_string())
+        }
+    } else {
+        Err("State not available".to_string())
+    }
+}

backend/src/commands/blocks.rs

@@ -0,0 +1,58 @@
+use async_trait::async_trait;
+use tauri::{ipc::Channel, AppHandle, Manager};
+use tokio::sync::mpsc;
+
+use crate::runtime::events::{EventBus, GCEvent};
+use crate::state::AtuinState;
+
+/// Channel-based event bus implementation that forwards events to Tauri channels
+pub struct ChannelEventBus {
+    sender: mpsc::UnboundedSender<GCEvent>,
+}
+
+impl ChannelEventBus {
+    pub fn new(sender: mpsc::UnboundedSender<GCEvent>) -> Self {
+        Self { sender }
+    }
+}
+
+#[async_trait]
+impl EventBus for ChannelEventBus {
+    async fn emit(&self, event: GCEvent) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        self.sender
+            .send(event)
+            .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send + Sync>)?;
+        Ok(())
+    }
+}
+
+/// Subscribe to the Grand Central event stream
+#[tauri::command]
+pub async fn subscribe_to_events(
+    app_handle: AppHandle,
+    event_channel: Channel<GCEvent>,
+) -> Result<(), String> {
+    let state = app_handle
+        .try_state::<AtuinState>()
+        .ok_or("State not available")?;
+
+    // Get the event receiver from state
+    let mut receiver = state
+        .event_receiver
+        .lock()
+        .await
+        .take()
+        .ok_or("Event receiver already taken or not available")?;
+
+    // Spawn task to forward events to the channel
+    tokio::spawn(async move {
+        while let Some(event) = receiver.recv().await {
+            if let Err(e) = event_channel.send(event) {
+                eprintln!("Failed to send event to frontend: {}", e);
+                break;
+            }
+        }
+    });
+
+    Ok(())
+}

backend/src/commands/events.rs

@@ -1,5 +1,7 @@
 pub(crate) mod block_state;
+pub(crate) mod blocks;
 pub(crate) mod dependency;
+pub(crate) mod events;
 pub(crate) mod exec_log;
 pub(crate) mod feedback;
 pub(crate) mod kubernetes;

backend/src/commands/mod.rs

@@ -503,6 +503,9 @@ fn main() {
             commands::mysql::mysql_query,
             commands::mysql::mysql_execute,
             commands::kubernetes::kubernetes_get_execute,
+            commands::blocks::execute_block,
+            commands::blocks::cancel_block_execution,
+            commands::events::subscribe_to_events,
             commands::updates::check_for_updates,
             commands::workspaces::copy_welcome_workspace,
             commands::workspaces::reset_workspaces,

backend/src/main.rs

@@ -9,11 +9,13 @@ use bytes::Bytes;
 use eyre::{eyre, Result};
 use portable_pty::{CommandBuilder, MasterPty, PtySize};
 use serde::{Deserialize, Serialize};
+use ts_rs::TS;
 use uuid::Uuid;
 
 use crate::runtime::pty_store::PtyLike;
 
-#[derive(Clone, Deserialize, Serialize, Debug)]
+#[derive(Clone, Deserialize, Serialize, Debug, TS)]
+#[ts(export)]
 pub struct PtyMetadata {
     pub pid: Uuid,
     pub runbook: Uuid,

backend/src/pty.rs

@@ -2,6 +2,8 @@ use serde::{Deserialize, Serialize};
 use typed_builder::TypedBuilder;
 use uuid::Uuid;
 
+use super::FromDocument;
+
 #[derive(Debug, Serialize, Deserialize, Clone, TypedBuilder)]
 #[serde(rename_all = "camelCase")]
 pub struct Clickhouse {
@@ -20,3 +22,52 @@ pub struct Clickhouse {
     #[builder(default = 0)]
     pub auto_refresh: i32,
 }
+
+impl FromDocument for Clickhouse {
+    fn from_document(block_data: &serde_json::Value) -> Result<Self, String> {
+        let block_id = block_data
+            .get("id")
+            .and_then(|v| v.as_str())
+            .ok_or("Block has no id")?;
+
+        let props = block_data
+            .get("props")
+            .and_then(|p| p.as_object())
+            .ok_or("Block has no props")?;
+
+        let id = Uuid::parse_str(block_id).map_err(|e| e.to_string())?;
+
+        let clickhouse = Clickhouse::builder()
+            .id(id)
+            .name(
+                props
+                    .get("name")
+                    .and_then(|v| v.as_str())
+                    .unwrap_or("ClickHouse Query")
+                    .to_string(),
+            )
+            .query(
+                props
+                    .get("query")
+                    .and_then(|v| v.as_str())
+                    .unwrap_or("")
+                    .to_string(),
+            )
+            .uri(
+                props
+                    .get("uri")
+                    .and_then(|v| v.as_str())
+                    .unwrap_or("")
+                    .to_string(),
+            )
+            .auto_refresh(
+                props
+                    .get("autoRefresh")
+                    .and_then(|v| v.as_i64())
+                    .unwrap_or(0) as i32,
+            )
+            .build();
+
+        Ok(clickhouse)
+    }
+}