fix: add auto-refresh for headless scaling and convergence operations

TODO.md

@@ -15,7 +15,6 @@
 - headless client start/stop/restart buttons on `/quma/headless` go past the card length
 - currency items (USD, EUR) displayed as roubles instead of as currency balances (Stash)
 - typo in update changelog: versions url should use `/#versions` hash anchor, not `/versions` path segment (current URL 404s on Forge)
-- no auto-refresh when scaling/converging clients
 - typo in update changelog: versions url should be like this `https://forge.sp-tarkov.com/mod/2310/wtt-commonlib/#versions` (`#version` is the important part)
 
 ## Convoy

src/headless/service.rs

@@ -6,7 +6,7 @@ use bollard::container::LogOutput;
 use futures_util::StreamExt;
 use parking_lot::Mutex;
 use serde::Serialize;
-use tokio::sync::RwLock;
+use tokio::sync::{broadcast, RwLock};
 
 use crate::client::{ClientHealth, ClientState};
 use crate::config::Config;
@@ -17,6 +17,7 @@ use crate::fika::client::FikaClient;
 use crate::forge::client::ForgeClient;
 use crate::headless::{HeadlessError, OperationTracker};
 use crate::spt::headless::EHeadlessStatus;
+use crate::web::sse::ServerEvent;
 
 #[derive(Debug, Clone, Copy)]
 pub enum LifecycleAction {
@@ -51,6 +52,7 @@ pub struct HeadlessService {
     pub(crate) fika_config_lock: Arc<Mutex<()>>,
     pub(crate) forge: ForgeClient,
     pub(crate) operations: OperationTracker,
+    pub(crate) events: broadcast::Sender<ServerEvent>,
 }
 
 impl HeadlessService {
@@ -67,6 +69,7 @@ impl HeadlessService {
         fika_client: Option<Arc<FikaClient>>,
         fika_config_lock: Arc<Mutex<()>>,
         forge: ForgeClient,
+        events: broadcast::Sender<ServerEvent>,
     ) -> Self {
         Self {
             container_mgr,
@@ -81,6 +84,7 @@ impl HeadlessService {
             fika_config_lock,
             forge,
             operations: OperationTracker::new(),
+            events,
         }
     }
 
@@ -125,6 +129,7 @@ impl HeadlessService {
         let converging = Arc::clone(&self.converging);
         let db = Arc::clone(&self.db);
         let ops = self.operations.clone();
+        let events = self.events.clone();
 
         tokio::spawn(async move {
             match crate::client::converge::converge(
@@ -140,8 +145,14 @@ impl HeadlessService {
             )
             .await
             {
-                Ok(()) => ops.complete(&op_id),
-                Err(e) => ops.fail(&op_id, e.to_string()),
+                Ok(()) => {
+                    ops.complete(&op_id);
+                    let _ = events.send(ServerEvent::HeadlessChanged);
+                }
+                Err(e) => {
+                    ops.fail(&op_id, e.to_string());
+                    let _ = events.send(ServerEvent::HeadlessChanged);
+                }
             }
         });
         op_id
@@ -547,13 +558,15 @@ impl HeadlessService {
         let converging = Arc::clone(&self.converging);
         let db = Arc::clone(&self.db);
         let ops = self.operations.clone();
+        let events = self.events.clone();
 
         let mut updated_config = headless_config;
         updated_config.clients.remove((index - 1) as usize);
 
         tokio::spawn(async move {
             if let Err(e) = crate::client::converge::remove_all_managed_containers(&mgr).await {
                 ops.fail(&op_id, format!("Failed to remove containers: {e}"));
+                let _ = events.send(ServerEvent::HeadlessChanged);
                 return;
             }
 
@@ -568,12 +581,14 @@ impl HeadlessService {
                         }
                         if let Err(e) = fresh_config.save(&config_path) {
                             ops.fail(&op_id, format!("Failed to save config: {e}"));
+                            let _ = events.send(ServerEvent::HeadlessChanged);
                             return;
                         }
                         *config_handle.write() = fresh_config;
                     }
                     Err(e) => {
                         ops.fail(&op_id, format!("Failed to reload config: {e}"));
+                        let _ = events.send(ServerEvent::HeadlessChanged);
                         return;
                     }
                 }
@@ -597,8 +612,14 @@ impl HeadlessService {
             )
             .await
             {
-                Ok(()) => ops.complete(&op_id),
-                Err(e) => ops.fail(&op_id, e.to_string()),
+                Ok(()) => {
+                    ops.complete(&op_id);
+                    let _ = events.send(ServerEvent::HeadlessChanged);
+                }
+                Err(e) => {
+                    ops.fail(&op_id, e.to_string());
+                    let _ = events.send(ServerEvent::HeadlessChanged);
+                }
             }
         });
 
@@ -647,6 +668,7 @@ impl HeadlessService {
         let converging = Arc::clone(&self.converging);
         let db = Arc::clone(&self.db);
         let ops = self.operations.clone();
+        let events = self.events.clone();
 
         tokio::spawn(async move {
             if converging
@@ -659,12 +681,14 @@ impl HeadlessService {
                 .is_err()
             {
                 ops.fail(&op_id, "Convergence already in progress".into());
+                let _ = events.send(ServerEvent::HeadlessChanged);
                 return;
             }
 
             if let Err(e) = crate::client::converge::remove_all_managed_containers(&mgr).await {
                 converging.store(false, std::sync::atomic::Ordering::Release);
                 ops.fail(&op_id, format!("Failed to remove containers: {e}"));
+                let _ = events.send(ServerEvent::HeadlessChanged);
                 return;
             }
 
@@ -688,8 +712,14 @@ impl HeadlessService {
             )
             .await
             {
-                Ok(()) => ops.complete(&op_id),
-                Err(e) => ops.fail(&op_id, e.to_string()),
+                Ok(()) => {
+                    ops.complete(&op_id);
+                    let _ = events.send(ServerEvent::HeadlessChanged);
+                }
+                Err(e) => {
+                    ops.fail(&op_id, e.to_string());
+                    let _ = events.send(ServerEvent::HeadlessChanged);
+                }
             }
         });
 

src/web/handlers/clients.rs

@@ -19,6 +19,7 @@ use crate::web::auth::{require_auth, require_permission, SessionUser};
 use crate::web::error::WebError;
 use crate::web::flash::{set_flash, take_flash, FlashMessage, FlashType};
 use crate::web::nav::NavContext;
+use crate::web::sse::ServerEvent;
 use crate::web::state::AppState;
 
 #[allow(unused_imports)]
@@ -368,11 +369,14 @@ pub async fn client_restart(
         .client_lifecycle(index, LifecycleAction::Restart)
         .await
     {
-        Ok(()) => set_flash(
-            &session,
-            &format!("Client {index} restarting"),
-            FlashType::Success,
-        ),
+        Ok(()) => {
+            let _ = state.events.send(ServerEvent::HeadlessChanged);
+            set_flash(
+                &session,
+                &format!("Client {index} restarting"),
+                FlashType::Success,
+            );
+        }
         Err(e) => set_flash(&session, &e.to_string(), FlashType::Error),
     }
 
@@ -414,6 +418,7 @@ pub async fn client_graceful_restart(
 
     match service.graceful_restart(index).await {
         Ok(result) => {
+            let _ = state.events.send(ServerEvent::HeadlessChanged);
             use crate::headless::service::GracefulResult;
             match result {
                 GracefulResult::Exited => {
@@ -474,11 +479,14 @@ pub async fn client_stop(
     };
 
     match service.client_lifecycle(index, LifecycleAction::Stop).await {
-        Ok(()) => set_flash(
-            &session,
-            &format!("Client {index} stopped"),
-            FlashType::Success,
-        ),
+        Ok(()) => {
+            let _ = state.events.send(ServerEvent::HeadlessChanged);
+            set_flash(
+                &session,
+                &format!("Client {index} stopped"),
+                FlashType::Success,
+            );
+        }
         Err(e) => set_flash(&session, &e.to_string(), FlashType::Error),
     }
 
@@ -522,11 +530,14 @@ pub async fn client_start(
         .client_lifecycle(index, LifecycleAction::Start)
         .await
     {
-        Ok(()) => set_flash(
-            &session,
-            &format!("Client {index} started"),
-            FlashType::Success,
-        ),
+        Ok(()) => {
+            let _ = state.events.send(ServerEvent::HeadlessChanged);
+            set_flash(
+                &session,
+                &format!("Client {index} started"),
+                FlashType::Success,
+            );
+        }
         Err(e) => set_flash(&session, &e.to_string(), FlashType::Error),
     }
 

src/web/mod.rs

@@ -1064,6 +1064,7 @@ pub async fn start_server(ctx: ServerContext, api_token: String) -> Result<()> {
                 fika_client.clone(),
                 Arc::clone(&fika_config_lock),
                 forge.clone(),
+                events_tx.clone(),
             ))
         } else {
             None

src/web/sse.rs

@@ -15,6 +15,7 @@ pub enum ServerEvent {
     RaidStarted,
     RaidEnded,
     IntegrityChanged,
+    HeadlessChanged,
 }
 
 pub async fn events_stream(
@@ -37,6 +38,7 @@ pub async fn events_stream(
                         ServerEvent::RaidStarted => "event: raidStarted\ndata: \n\n",
                         ServerEvent::RaidEnded => "event: raidEnded\ndata: \n\n",
                         ServerEvent::IntegrityChanged => "event: integrityChanged\ndata: \n\n",
+                        ServerEvent::HeadlessChanged => "event: headlessChanged\ndata: \n\n",
                     };
                     return Some((Ok::<_, actix_web::Error>(web::Bytes::from(msg)), rx));
                 }

templates/headless.html

@@ -12,30 +12,35 @@ <h1>Headless Clients</h1>
         <h2>Clients</h2>
         {% if user.can("headless.manage") %}
         <div class="flex gap-1">
-            <form method="post" action="/quma/headless/scale" style="display:inline">
+            <form method="post" action="/quma/headless/scale" style="display:inline"
+                  hx-post="/quma/headless/scale" hx-target="#operation-status" hx-swap="outerHTML">
                 <input type="hidden" name="csrf_token" value="{{ csrf_token }}">
                 <input type="number" name="count" min="0" max="16" value="{{ headless_target_count }}" style="width: 4rem; margin-right: 0.5rem">
                 <input type="hidden" name="force" value="false">
                 <button type="submit" class="btn btn-sm btn-primary">Scale</button>
             </form>
-            <form method="post" action="/quma/headless/create" style="display:inline">
+            <form method="post" action="/quma/headless/create" style="display:inline"
+                  hx-post="/quma/headless/create" hx-target="#operation-status" hx-swap="outerHTML">
                 <input type="hidden" name="csrf_token" value="{{ csrf_token }}">
                 <button type="submit" class="btn btn-sm btn-success">+ Add Client</button>
             </form>
-            <form method="post" action="/quma/headless/converge" style="display:inline">
+            <form method="post" action="/quma/headless/converge" style="display:inline"
+                  hx-post="/quma/headless/converge" hx-target="#operation-status" hx-swap="outerHTML">
                 <input type="hidden" name="csrf_token" value="{{ csrf_token }}">
                 <button type="submit" class="btn btn-sm">Force Converge</button>
             </form>
             <form method="post" action="/quma/headless/rebuild" style="display:inline"
-                  onsubmit="return confirm('Rebuild all headless clients? This will destroy and recreate all containers and overlays.')">
+                  hx-post="/quma/headless/rebuild" hx-target="#operation-status" hx-swap="outerHTML"
+                  hx-confirm="Rebuild all headless clients? This will destroy and recreate all containers and overlays.">
                 <input type="hidden" name="csrf_token" value="{{ csrf_token }}">
                 <button type="submit" class="btn btn-sm btn-danger">Rebuild All</button>
             </form>
         </div>
         {% endif %}
     </div>
 
-    <div hx-get="/quma/api/headless/status-partial" hx-trigger="load, every 5s" hx-swap="innerHTML" style="margin-top: 0.5rem">
+    <div id="operation-status"></div>
+    <div id="headless-clients-container" hx-get="/quma/api/headless/status-partial" hx-trigger="load, every 5s" hx-swap="innerHTML" style="margin-top: 0.5rem">
         {% if headless_converging %}
         <div style="background-color: var(--warning-bg); border-left: 4px solid var(--warning); padding: 0.75rem; border-radius: 4px; margin-bottom: 0.5rem">
             <div class="flex gap-1" style="align-items: center">
@@ -293,4 +298,17 @@ <h3 style="margin-top: 1.5rem">NUMA Scheduling</h3>
         <button type="submit" class="btn">Save Headless Settings</button>
     </div>
 </form>
+
+<script>
+(function() {
+    if (window.headlessEventSource) {
+        window.headlessEventSource.close();
+    }
+    const eventSource = new EventSource('/quma/events');
+    eventSource.addEventListener('headlessChanged', () => {
+        htmx.trigger('#headless-clients-container', 'load');
+    });
+    window.headlessEventSource = eventSource;
+})();
+</script>
 {% endblock %}