Added file watching

Author: Caleb-T-Owens

Cargo.lock

@@ -103,12 +103,51 @@ export async function invoke<T>(command: string, params: Record<string, unknown>
 	}
 }
 
+let webListener: WebListener | undefined;
+
 export function listen<T>(event: EventName, handle: EventCallback<T>) {
 	if (import.meta.env.VITE_BUILD_TARGET === 'electron') {
+		if (!webListener) {
+			webListener = new WebListener();
+		}
+
 		// TODO: Listening in electron
-		return async () => {};
+		return webListener.listen({ name: event, handle });
 	} else {
 		const unlisten = listenTauri(event, handle);
 		return async () => await unlisten.then((unlistenFn) => unlistenFn());
 	}
 }
+
+class WebListener {
+	private socket: WebSocket | undefined;
+	private count = 0;
+	private handlers: { name: EventName; handle: EventCallback<any> }[] = [];
+
+	listen(handler: { name: EventName; handle: EventCallback<any> }): () => void {
+		this.handlers.push(handler);
+		this.count++;
+		if (!this.socket) {
+			this.socket = new WebSocket('ws://localhost:6978/ws');
+			this.socket.addEventListener('message', (event) => {
+				const data: { name: string; payload: any } = JSON.parse(event.data);
+				for (const handler of this.handlers) {
+					if (handler.name === data.name) {
+						// The id is an artifact from tauri, we don't use it so
+						// I've used a random value
+						handler.handle({ event: data.name, payload: data.payload, id: 69 });
+					}
+				}
+			});
+		}
+
+		return () => {
+			this.handlers = this.handlers.filter((h) => h !== handler);
+			this.count--;
+			if (this.count === 0) {
+				this.socket?.close();
+				this.socket = undefined;
+			}
+		};
+	}
+}

apps/desktop/src/lib/backend/ipc.ts

@@ -17,7 +17,11 @@ doctest = false
 
 [dependencies]
 serde.workspace = true
-axum = "0.8.4"
+axum = { version = "0.8.4", features = ["ws"] }
+futures-util = { version = "0.3", default-features = false, features = [
+    "sink",
+    "std",
+] }
 tower = "0.5.2"
 tower-http = { version = "0.6.5", features = ["cors"] }
 tokio = { workspace = true, features = ["full"] }
@@ -55,15 +59,20 @@ but-graph.workspace = true
 gitbutler-oplog.workspace = true
 but-hunk-dependency.workspace = true
 serde-error = "0.1.3"
-gix = { workspace = true, features = ["worktree-mutation", "blocking-http-transport-curl"] }
+gix = { workspace = true, features = [
+    "worktree-mutation",
+    "blocking-http-transport-curl",
+] }
 git2.workspace = true
 gitbutler-feedback.workspace = true
 gitbutler-error.workspace = true
 gitbutler-diff.workspace = true
+gitbutler-watcher.workspace = true
 but-rules.workspace = true
 but-action.workspace = true
 reqwest = { version = "0.12", features = ["json"] }
 gitbutler-forge.workspace = true
 open = "5.3"
 url = "2.5"
 gitbutler-id.workspace = true
+uuid.workspace = true

crates/but-server/Cargo.toml

@@ -1,7 +1,16 @@
-use std::sync::Arc;
+use std::{collections::HashMap, sync::Arc};
 
-use axum::{Json, Router, routing::get};
+use axum::{
+    Json, Router,
+    extract::{
+        WebSocketUpgrade,
+        ws::{Message, WebSocket},
+    },
+    response::IntoResponse,
+    routing::{any, get},
+};
 use but_settings::AppSettingsWithDiskSync;
+use futures_util::{SinkExt, StreamExt as _};
 use serde::{Deserialize, Serialize};
 use serde_json::json;
 use tokio::sync::Mutex;
@@ -34,11 +43,13 @@ mod virtual_branches;
 mod workspace;
 mod zip;
 
+#[derive(Clone)]
 pub(crate) struct RequestContext {
     app_settings: Arc<AppSettingsWithDiskSync>,
     user_controller: Arc<gitbutler_user::Controller>,
     project_controller: Arc<gitbutler_project::Controller>,
     active_projects: Arc<Mutex<ActiveProjects>>,
+    broadcaster: Arc<Mutex<Broadcaster>>,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -55,6 +66,13 @@ pub(crate) struct Request {
     params: serde_json::Value,
 }
 
+#[derive(Debug, Serialize, Clone)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct FrontendEvent {
+    name: String,
+    payload: serde_json::Value,
+}
+
 pub async fn run() {
     let cors = CorsLayer::new()
         .allow_methods(Any)
@@ -70,25 +88,35 @@ pub async fn run() {
         .expect("missing config dir")
         .join("gitbutler-server");
 
-    let app_settings = Arc::new(
-        AppSettingsWithDiskSync::new(config_dir.clone()).expect("failed to create app settings"),
-    );
-    let user_controller = Arc::new(gitbutler_user::Controller::from_path(&app_data_dir));
-    let project_controller = Arc::new(gitbutler_project::Controller::from_path(&app_data_dir));
-    let active_projects = Arc::new(Mutex::new(ActiveProjects::new()));
+    let broadcaster = Arc::new(Mutex::new(Broadcaster {
+        senders: HashMap::new(),
+    }));
+
+    let ctx = RequestContext {
+        app_settings: Arc::new(
+            AppSettingsWithDiskSync::new(config_dir.clone())
+                .expect("failed to create app settings"),
+        ),
+        user_controller: Arc::new(gitbutler_user::Controller::from_path(&app_data_dir)),
+        project_controller: Arc::new(gitbutler_project::Controller::from_path(&app_data_dir)),
+        active_projects: Arc::new(Mutex::new(ActiveProjects::new())),
+        broadcaster: broadcaster.clone(),
+    };
 
     // build our application with a single route
     let app = Router::new()
         .route(
             "/",
-            get(|| async { "Hello, World!" }).post(move |req| {
-                let ctx = RequestContext {
-                    app_settings: Arc::clone(&app_settings),
-                    user_controller: Arc::clone(&user_controller),
-                    project_controller: Arc::clone(&project_controller),
-                    active_projects: Arc::clone(&active_projects),
-                };
-                handle_command(req, ctx)
+            get(|| async { "Hello, World!" }).post({
+                let ctx = ctx.clone();
+                move |req| handle_command(req, ctx)
+            }),
+        )
+        .route(
+            "/ws",
+            any({
+                let broadcaster = broadcaster.clone();
+                async move |req| handle_ws_request(req, broadcaster).await
             }),
         )
         .layer(ServiceBuilder::new().layer(cors));
@@ -99,6 +127,54 @@ pub async fn run() {
     axum::serve(listener, app).await.unwrap();
 }
 
+struct Broadcaster {
+    senders: HashMap<uuid::Uuid, tokio::sync::mpsc::UnboundedSender<FrontendEvent>>,
+}
+
+impl Broadcaster {
+    fn send(&self, event: FrontendEvent) {
+        for sender in self.senders.values() {
+            let _ = sender.send(event.clone());
+        }
+    }
+}
+
+async fn handle_ws_request(
+    ws: WebSocketUpgrade,
+    broadcaster: Arc<Mutex<Broadcaster>>,
+) -> impl IntoResponse {
+    ws.on_upgrade(move |socket| handle_websocket(socket, broadcaster))
+}
+
+async fn handle_websocket(socket: WebSocket, broadcaster: Arc<Mutex<Broadcaster>>) {
+    let (send, mut recv) = tokio::sync::mpsc::unbounded_channel();
+    let id = uuid::Uuid::new_v4();
+    broadcaster.lock().await.senders.insert(id, send);
+
+    let (mut socket_send, mut socket_recv) = socket.split();
+    let thread = tokio::spawn(async move {
+        while let Some(event) = recv.recv().await {
+            socket_send
+                .send(Message::Text(serde_json::to_string(&event).unwrap().into()))
+                .await
+                .unwrap();
+        }
+    });
+
+    while let Some(Ok(msg)) = socket_recv.next().await {
+        #[allow(clippy::single_match)]
+        match msg {
+            Message::Close(_) => {
+                thread.abort();
+                break;
+            }
+            _ => {}
+        }
+    }
+
+    broadcaster.lock().await.senders.remove(&id);
+}
+
 async fn handle_command(
     Json(request): Json<Request>,
     ctx: RequestContext,