src-tauri/src/commands/integrations.rs

@@ -340,6 +340,7 @@ pub async fn initiate_oauth(
                 refresh_registry: Arc::new(tokio::sync::Mutex::new(
                     crate::kube::RefreshRegistry::new(),
                 )),
+                watchers: Arc::new(Mutex::new(std::collections::HashMap::new())),
             };
             while let Some(callback) = callback_rx.recv().await {
                 tracing::info!("Received OAuth callback for state: {}", callback.state);

src-tauri/src/commands/kube.rs

@@ -4055,3 +4055,78 @@ pub async fn edit_resource(
 
     Ok(())
 }
+
+#[tauri::command]
+pub async fn subscribe_to_k8s_events(
+    cluster_id: String,
+    namespace: String,
+    resource_type: String,
+    state: State<'_, AppState>,
+) -> Result<String, String> {
+    let _app_state = state.inner();
+    let _app_state = state.inner();
+
+    let rx = crate::kube::start_resource_watcher(_app_state, cluster_id, namespace, resource_type)
+        .await
+        .map_err(|e| format!("Failed to start watcher: {e}"))?;
+
+    let duration = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map_err(|e| format!("Failed to get duration: {e}"))?;
+    let unsubscribe_id = format!("watcher-{}", duration.as_millis());
+
+    state
+        .inner()
+        .watchers
+        .lock()
+        .unwrap()
+        .insert(unsubscribe_id.clone(), rx);
+
+    Ok(unsubscribe_id)
+}
+
+#[tauri::command]
+pub async fn subscribe_to_all_k8s_events(
+    cluster_id: String,
+    state: State<'_, AppState>,
+) -> Result<String, String> {
+    let _app_state = state.inner();
+    let _app_state = state.inner();
+
+    let rx = crate::kube::start_all_resources_watcher(_app_state, cluster_id)
+        .await
+        .map_err(|e| format!("Failed to start all watcher: {e}"))?;
+
+    let duration = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map_err(|e| format!("Failed to get duration: {e}"))?;
+    let unsubscribe_id = format!("watcher-all-{}", duration.as_millis());
+
+    state
+        .inner()
+        .watchers
+        .lock()
+        .unwrap()
+        .insert(unsubscribe_id.clone(), rx);
+
+    Ok(unsubscribe_id)
+}
+
+#[tauri::command]
+pub async fn unsubscribe_from_k8s_events(
+    unsubscribe_id: String,
+    state: State<'_, AppState>,
+) -> Result<(), String> {
+    let removed = state
+        .inner()
+        .watchers
+        .lock()
+        .unwrap()
+        .remove(&unsubscribe_id);
+
+    if removed.is_none() {
+        return Err(format!("Watcher {} not found", unsubscribe_id));
+    }
+
+    Ok(())
+}

src-tauri/src/kube/mod.rs

@@ -1,10 +1,12 @@
 pub mod client;
 pub mod portforward;
 pub mod refresh;
+pub mod watcher;
 
 pub use client::ClusterClient;
 pub use portforward::{PortForwardSession, PortForwardStatus};
 pub use refresh::RefreshRegistry;
+pub use watcher::{start_all_resources_watcher, start_resource_watcher, Watcher};
 
 #[cfg(test)]
 mod tests {

src-tauri/src/kube/watcher.rs

@@ -0,0 +1,88 @@
+use crate::state::AppState;
+use anyhow::Result;
+use tokio::sync::mpsc;
+use tracing::info;
+
+pub struct Watcher {
+    cluster_id: String,
+    namespace: String,
+    resource_type: String,
+    #[allow(dead_code)]
+    tx: mpsc::Sender<serde_json::Value>,
+}
+
+impl Watcher {
+    pub fn new(
+        cluster_id: String,
+        namespace: String,
+        resource_type: String,
+        tx: mpsc::Sender<serde_json::Value>,
+    ) -> Self {
+        Self {
+            cluster_id,
+            namespace,
+            resource_type,
+            tx,
+        }
+    }
+
+    pub async fn start(self) -> Result<()> {
+        info!(
+            "Starting watcher for {}/{} in namespace {}",
+            self.resource_type, self.cluster_id, self.namespace
+        );
+
+        // Placeholder for watcher implementation
+        // Requires k8s-openapi with watch feature and tokio-stream
+        loop {
+            tokio::time::sleep(tokio::time::Duration::from_secs(60)).await;
+        }
+    }
+}
+
+pub async fn start_resource_watcher(
+    _app_state: &AppState,
+    cluster_id: String,
+    namespace: String,
+    resource_type: String,
+) -> Result<mpsc::Receiver<serde_json::Value>> {
+    let (tx, rx) = mpsc::channel(100);
+
+    let watcher_tx = tx.clone();
+    let cluster_id = cluster_id.clone();
+    let namespace = namespace.clone();
+    let resource_type = resource_type.clone();
+
+    tokio::spawn(async move {
+        let watcher = Watcher::new(cluster_id, namespace, resource_type, watcher_tx);
+        if let Err(e) = watcher.start().await {
+            tracing::error!("Watcher failed: {}", e);
+        }
+    });
+
+    Ok(rx)
+}
+
+pub async fn start_all_resources_watcher(
+    _app_state: &AppState,
+    cluster_id: String,
+) -> Result<mpsc::Receiver<serde_json::Value>> {
+    let (tx, rx) = mpsc::channel(100);
+
+    let resources = vec!["pods", "services", "deployments", "replicasets", "daemonsets"];
+
+    for resource_type in resources {
+        let watcher_tx = tx.clone();
+        let cluster_id = cluster_id.clone();
+        let namespace = "default".to_string();
+
+        tokio::spawn(async move {
+            let watcher = Watcher::new(cluster_id, namespace, resource_type.to_string(), watcher_tx);
+            if let Err(e) = watcher.start().await {
+                tracing::error!("Watcher for {} failed: {}", resource_type, e);
+            }
+        });
+    }
+
+    Ok(rx)
+}

src-tauri/src/lib.rs

@@ -44,6 +44,7 @@ pub fn run() {
         clusters: Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())),
         port_forwards: Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())),
         refresh_registry: Arc::new(tokio::sync::Mutex::new(crate::kube::RefreshRegistry::new())),
+        watchers: Arc::new(Mutex::new(std::collections::HashMap::new())),
     };
     let stronghold_salt = format!(
         "tftsr-stronghold-salt-v1-{:x}",

src-tauri/src/state.rs

@@ -97,6 +97,8 @@ pub struct AppState {
     pub port_forwards: Arc<TokioMutex<HashMap<String, crate::kube::PortForwardSession>>>,
     /// Refresh registry for domain-based data fetching
     pub refresh_registry: Arc<TokioMutex<crate::kube::RefreshRegistry>>,
+    /// Resource watchers: unsubscribe_id -> receiver
+    pub watchers: Arc<Mutex<HashMap<String, tokio::sync::mpsc::Receiver<serde_json::Value>>>>,
 }
 
 /// Determine the application data directory.

src/lib/eventBus.ts

@@ -0,0 +1,116 @@
+import { invoke } from "@tauri-apps/api/core";
+
+export type EventCallback<T = any> = (data: T) => void;
+
+export interface EventUnsubscribe {
+  (): void;
+}
+
+export interface EventBus {
+  on<T = any>(event: string, callback: EventCallback<T>): EventUnsubscribe;
+  off(event: string, callback: EventCallback): void;
+  emit<T = any>(event: string, data?: T): void;
+  once<T = any>(event: string, callback: EventCallback<T>): EventUnsubscribe;
+}
+
+class SimpleEventBus implements EventBus {
+  private events: Record<string, Set<EventCallback>> = {};
+  private onceEvents: Record<string, Set<EventCallback>> = {};
+
+  on<T = any>(event: string, callback: EventCallback<T>): EventUnsubscribe {
+    if (!this.events[event]) {
+      this.events[event] = new Set();
+    }
+    this.events[event].add(callback);
+
+    return () => this.off(event, callback);
+  }
+
+  off(event: string, callback: EventCallback): void {
+    if (this.events[event]) {
+      this.events[event].delete(callback);
+    }
+  }
+
+  emit<T = any>(event: string, data?: T): void {
+    const callbacks = this.events[event];
+    if (callbacks) {
+      callbacks.forEach((callback) => callback(data as T));
+    }
+
+    const onceCallbacks = this.onceEvents[event];
+    if (onceCallbacks) {
+      onceCallbacks.forEach((callback) => callback(data as T));
+      delete this.onceEvents[event];
+    }
+  }
+
+  once<T = any>(event: string, callback: EventCallback<T>): EventUnsubscribe {
+    if (!this.onceEvents[event]) {
+      this.onceEvents[event] = new Set();
+    }
+    this.onceEvents[event].add(callback);
+
+    return () => {
+      if (this.onceEvents[event]) {
+        this.onceEvents[event].delete(callback);
+      }
+    };
+  }
+}
+
+export const eventBus: EventBus = new SimpleEventBus();
+
+export async function subscribeToK8sEvents(
+  clusterId: string,
+  namespace: string,
+  resourceType: string,
+  callback: EventCallback<any>
+): Promise<EventUnsubscribe> {
+  try {
+    const unsubscribeId = await invoke<string>("subscribe_to_k8s_events", {
+      clusterId,
+      namespace,
+      resourceType,
+    });
+
+    const handler = (data: any) => {
+      callback(data);
+    };
+
+    eventBus.on(`k8s:${clusterId}:${namespace}:${resourceType}`, handler);
+
+    return () => {
+      eventBus.off(`k8s:${clusterId}:${namespace}:${resourceType}`, handler);
+      invoke<void>("unsubscribe_from_k8s_events", { unsubscribeId });
+    };
+  } catch (error) {
+    console.error("Failed to subscribe to K8s events:", error);
+    return () => {};
+  }
+}
+
+export async function subscribeToAllEvents(
+  clusterId: string,
+  callback: EventCallback<any>
+): Promise<EventUnsubscribe> {
+  try {
+    const unsubscribeId = await invoke<string>("subscribe_to_all_k8s_events", {
+      clusterId,
+    });
+
+    const handler = (data: any) => {
+      callback(data);
+    };
+
+    eventBus.on(`k8s:${clusterId}:all`, handler);
+
+    return () => {
+      eventBus.off(`k8s:${clusterId}:all`, handler);
+      invoke<void>("unsubscribe_from_k8s_events", { unsubscribeId });
+    };
+  } catch (error) {
+    console.error("Failed to subscribe to all K8s events:", error);
+    return () => {};
+  }
+}