Implement live syncback feature and enhance API for service data transfer

- Added a new `performSyncback` function in the `App` module to handle the synchronization of service data from the game to the server.
- Introduced a `SyncbackConfirm` UI component to confirm syncback actions from the user.
- Updated the API to include a new `/api/syncback` endpoint for handling syncback requests, validating protocol and server versions.
- Enhanced the `SyncbackSignal` mechanism to facilitate communication between the server and the syncback process.
- Added tests to ensure the functionality of the live syncback feature and its integration with the CLI.

These changes improve the overall synchronization capabilities of the application, allowing for real-time updates and better project management.

src/cli/serve.rs

@@ -1,13 +1,26 @@
 use std::{
-    net::{IpAddr, Ipv4Addr},
-    path::PathBuf,
+    collections::HashMap,
+    io::Cursor,
+    mem::forget,
+    net::{IpAddr, Ipv4Addr, SocketAddr},
+    path::{Path, PathBuf},
+    process::{Command, Stdio},
     sync::Arc,
 };
 
+use anyhow::Context;
 use clap::Parser;
 use memofs::Vfs;
+use rbx_dom_weak::{types::Ref, types::Variant, InstanceBuilder, WeakDom};
 
-use crate::{serve_session::ServeSession, web::LiveServer};
+use crate::{
+    serve_session::ServeSession,
+    syncback::syncback_loop,
+    web::{
+        interface::{ServerExitReason, SyncbackPayload},
+        LiveServer,
+    },
+};
 
 use super::resolve_path;
 
@@ -35,30 +48,190 @@ impl ServeCommand {
     pub fn run(self) -> anyhow::Result<()> {
         let project_path = resolve_path(&self.project);
 
-        let (vfs, critical_errors) = Vfs::new_default_with_errors();
-
-        let session = Arc::new(ServeSession::new(vfs, project_path, Some(critical_errors))?);
-
-        let ip = self
-            .address
-            .or_else(|| session.serve_address())
-            .unwrap_or(DEFAULT_BIND_ADDRESS.into());
-
-        let port = self
-            .port
-            .or_else(|| session.project_port())
-            .unwrap_or(DEFAULT_PORT);
-
-        let server = LiveServer::new(session);
+        let (ip, port) = {
+            let vfs = Vfs::new_oneshot();
+            let session = ServeSession::new_oneshot(vfs, project_path.clone())?;
+            let ip = self
+                .address
+                .or_else(|| session.serve_address())
+                .unwrap_or(DEFAULT_BIND_ADDRESS.into());
+            let port = self
+                .port
+                .or_else(|| session.project_port())
+                .unwrap_or(DEFAULT_PORT);
+            forget(session);
+            (ip, port)
+        };
 
+        let addr: SocketAddr = (ip, port).into();
         let host = if ip.is_loopback() {
             "localhost".to_owned()
         } else {
             ip.to_string()
         };
-        log::info!("Listening: http://{}:{}", host, port);
-        server.start((ip, port).into());
 
-        Ok(())
+        loop {
+            let (vfs, critical_errors) = Vfs::new_default_with_errors();
+            let session =
+                Arc::new(ServeSession::new(vfs, project_path.clone(), Some(critical_errors))?);
+            let server = LiveServer::new(session);
+
+            log::info!("Listening: http://{}:{}", host, port);
+
+            match server.start(addr) {
+                ServerExitReason::SyncbackRequested(payload) => {
+                    log::info!("Live syncback requested, running...");
+                    run_live_syncback(&project_path, payload)?;
+                    log::info!("Syncback complete, restarting serve...");
+                    continue;
+                }
+            }
+        }
+    }
+}
+
+fn run_live_syncback(project_path: &Path, payload: SyncbackPayload) -> anyhow::Result<()> {
+    let new_dom = build_dom_from_chunks(payload)?;
+
+    let vfs = Vfs::new_oneshot();
+    let session_old = ServeSession::new_oneshot(vfs, project_path.to_path_buf())?;
+
+    let mut dom_old = session_old.tree();
+
+    let syncback_timer = std::time::Instant::now();
+    log::info!("Beginning live syncback (clean mode)...");
+
+    let result = syncback_loop(
+        session_old.vfs(),
+        &mut dom_old,
+        new_dom,
+        session_old.root_project(),
+        false, // incremental = false → clean mode
+    )?;
+
+    log::debug!(
+        "Syncback finished in {:.02}s",
+        syncback_timer.elapsed().as_secs_f32()
+    );
+
+    let base_path = session_old.root_project().folder_location();
+    drop(dom_old);
+
+    log::info!("Writing to the file system...");
+    result
+        .fs_snapshot
+        .write_to_vfs_parallel(base_path, session_old.vfs())?;
+
+    log::info!(
+        "Finished live syncback: wrote {} files/folders, removed {}.",
+        result.fs_snapshot.added_paths().len(),
+        result.fs_snapshot.removed_paths().len()
+    );
+
+    refresh_git_index(base_path);
+
+    forget(session_old);
+
+    Ok(())
+}
+
+fn build_dom_from_chunks(payload: SyncbackPayload) -> anyhow::Result<WeakDom> {
+    let mut dom = WeakDom::new(InstanceBuilder::new("DataModel"));
+    let root_ref = dom.root_ref();
+
+    let mut global_ref_map: HashMap<Ref, Ref> = HashMap::new();
+
+    for chunk in &payload.services {
+        let chunk_dom = rbx_binary::from_reader(Cursor::new(&chunk.data))
+            .with_context(|| format!("Failed to parse rbxm for service {}", chunk.class_name))?;
+
+        let service_ref = dom.insert(root_ref, InstanceBuilder::new(&chunk.class_name));
+
+        for &child_ref in chunk_dom.root().children() {
+            deep_clone_into(&chunk_dom, &mut dom, child_ref, service_ref, &mut global_ref_map);
+        }
+    }
+
+    fixup_ref_properties(&mut dom, &global_ref_map);
+
+    Ok(dom)
+}
+
+fn deep_clone_into(
+    source: &WeakDom,
+    target: &mut WeakDom,
+    source_ref: Ref,
+    target_parent: Ref,
+    ref_map: &mut HashMap<Ref, Ref>,
+) {
+    let inst = source.get_by_ref(source_ref).unwrap();
+    let mut builder = InstanceBuilder::new(inst.class.as_str()).with_name(inst.name.as_str());
+
+    for (key, value) in &inst.properties {
+        builder = builder.with_property(key.as_str(), value.clone());
+    }
+
+    let new_ref = target.insert(target_parent, builder);
+    ref_map.insert(source_ref, new_ref);
+
+    for &child_ref in inst.children() {
+        deep_clone_into(source, target, child_ref, new_ref, ref_map);
+    }
+}
+
+fn fixup_ref_properties(dom: &mut WeakDom, ref_map: &HashMap<Ref, Ref>) {
+    let all_refs: Vec<Ref> = ref_map.values().copied().collect();
+    for inst_ref in all_refs {
+        let props_to_fix: Vec<(String, Ref)> = {
+            let inst = dom.get_by_ref(inst_ref).unwrap();
+            inst.properties
+                .iter()
+                .filter_map(|(key, value)| {
+                    if let Variant::Ref(r) = value {
+                        if let Some(&mapped) = ref_map.get(r) {
+                            Some((key.to_string(), mapped))
+                        } else {
+                            None
+                        }
+                    } else {
+                        None
+                    }
+                })
+                .collect()
+        };
+
+        if !props_to_fix.is_empty() {
+            let inst = dom.get_by_ref_mut(inst_ref).unwrap();
+            for (key, new_ref) in props_to_fix {
+                inst.properties.insert(key.into(), Variant::Ref(new_ref));
+            }
+        }
+    }
+}
+
+fn refresh_git_index(project_dir: &Path) {
+    let mut check_dir = Some(project_dir);
+    let mut is_git_repo = false;
+    while let Some(dir) = check_dir {
+        if dir.join(".git").exists() {
+            is_git_repo = true;
+            break;
+        }
+        check_dir = dir.parent();
+    }
+
+    if is_git_repo {
+        log::info!("Refreshing git index...");
+        match Command::new("git")
+            .args(["update-index", "--refresh", "-q"])
+            .current_dir(project_dir)
+            .stdin(Stdio::null())
+            .stdout(Stdio::null())
+            .stderr(Stdio::null())
+            .status()
+        {
+            Ok(_) => log::info!("Git index refreshed."),
+            Err(e) => log::warn!("Failed to run git update-index --refresh: {}", e),
+        }
     }
 }

src/web/api.rs

@@ -29,8 +29,9 @@ use crate::{
     web::{
         interface::{
             ErrorResponse, Instance, InstanceMetadata, MessagesPacket, OpenResponse, ReadResponse,
-            ServerInfoResponse, SocketPacket, SocketPacketBody, SocketPacketType, SubscribeMessage,
-            WriteRequest, WriteResponse, PROTOCOL_VERSION, SERVER_VERSION,
+            ServerInfoResponse, SocketPacket, SocketPacketBody, SocketPacketType,
+            SubscribeMessage, SyncbackPayload, SyncbackRequest, WriteRequest, WriteResponse,
+            PROTOCOL_VERSION, SERVER_VERSION,
         },
         util::{deserialize_msgpack, msgpack, msgpack_ok, serialize_msgpack},
     },
@@ -122,6 +123,7 @@ fn variant_to_json(variant: &Variant) -> Option<serde_json::Value> {
 pub async fn call(
     serve_session: Arc<ServeSession>,
     mut request: Request<Incoming>,
+    syncback_signal: Arc<super::SyncbackSignal>,
 ) -> Response<Full<Bytes>> {
     let service = ApiService::new(serve_session);
 
@@ -153,6 +155,9 @@ pub async fn call(
             service.handle_api_open(request).await
         }
         (&Method::POST, "/api/write") => service.handle_api_write(request).await,
+        (&Method::POST, "/api/syncback") => {
+            handle_api_syncback(request, &service, syncback_signal).await
+        }
         (&Method::GET, "/api/validate-tree") => service.handle_api_validate_tree().await,
 
         (_method, path) => msgpack(
@@ -162,6 +167,96 @@ pub async fn call(
     }
 }
 
+async fn handle_api_syncback(
+    request: Request<Incoming>,
+    service: &ApiService,
+    syncback_signal: Arc<super::SyncbackSignal>,
+) -> Response<Full<Bytes>> {
+    let body = match request.into_body().collect().await {
+        Ok(collected) => collected.to_bytes(),
+        Err(err) => {
+            return msgpack(
+                ErrorResponse::bad_request(format!("Failed to read request body: {err}")),
+                StatusCode::BAD_REQUEST,
+            );
+        }
+    };
+
+    let syncback_request: SyncbackRequest = match deserialize_msgpack(&body) {
+        Ok(req) => req,
+        Err(err) => {
+            return msgpack(
+                ErrorResponse::bad_request(format!(
+                    "Failed to deserialize syncback request: {err}"
+                )),
+                StatusCode::BAD_REQUEST,
+            );
+        }
+    };
+
+    if syncback_request.protocol_version != PROTOCOL_VERSION {
+        return msgpack(
+            ErrorResponse::bad_request(format!(
+                "Protocol version mismatch: expected {}, got {}",
+                PROTOCOL_VERSION, syncback_request.protocol_version
+            )),
+            StatusCode::BAD_REQUEST,
+        );
+    }
+
+    let server_semver = SERVER_VERSION;
+    let server_parts: Vec<&str> = server_semver.split('.').collect();
+    let client_parts: Vec<&str> = syncback_request.server_version.split('.').collect();
+    let server_major_minor = server_parts.get(..2);
+    let client_major_minor = client_parts.get(..2);
+    if server_major_minor != client_major_minor {
+        return msgpack(
+            ErrorResponse::bad_request(format!(
+                "Server version mismatch: server is {}, plugin expects {}",
+                SERVER_VERSION, syncback_request.server_version
+            )),
+            StatusCode::BAD_REQUEST,
+        );
+    }
+
+    if let Some(place_id) = syncback_request.place_id {
+        if let Some(expected) = service.serve_session.serve_place_ids() {
+            if !expected.contains(&place_id) {
+                return msgpack(
+                    ErrorResponse::bad_request(format!(
+                        "Place ID {} is not in the servePlaceIds whitelist",
+                        place_id
+                    )),
+                    StatusCode::FORBIDDEN,
+                );
+            }
+        }
+        if let Some(blocked) = service.serve_session.blocked_place_ids() {
+            if blocked.contains(&place_id) {
+                return msgpack(
+                    ErrorResponse::bad_request(format!(
+                        "Place ID {} is in the blockedPlaceIds list",
+                        place_id
+                    )),
+                    StatusCode::FORBIDDEN,
+                );
+            }
+        }
+    }
+
+    log::info!(
+        "Live syncback requested with {} service chunks",
+        syncback_request.services.len()
+    );
+
+    let payload = SyncbackPayload {
+        services: syncback_request.services,
+    };
+    syncback_signal.fire(payload);
+
+    msgpack_ok(&serde_json::json!({"status": "syncback_initiated"}))
+}
+
 pub struct ApiService {
     serve_session: Arc<ServeSession>,
     suppressed_paths: Arc<Mutex<HashMap<PathBuf, (usize, usize)>>>,

src/web/interface.rs

@@ -27,6 +27,36 @@ pub(crate) const SERVER_VERSION: &str = env!("CARGO_PKG_VERSION");
 /// Current protocol version, which is required to match.
 pub const PROTOCOL_VERSION: u64 = 6;
 
+/// A single service's children serialized as rbxm binary data.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ServiceChunk {
+    pub class_name: String,
+    #[serde(with = "serde_bytes")]
+    pub data: Vec<u8>,
+}
+
+/// Incoming request from the plugin for live syncback.
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SyncbackRequest {
+    pub protocol_version: u64,
+    pub server_version: String,
+    pub place_id: Option<u64>,
+    pub services: Vec<ServiceChunk>,
+}
+
+/// Payload passed from the API handler to the serve loop after validation.
+#[derive(Debug)]
+pub struct SyncbackPayload {
+    pub services: Vec<ServiceChunk>,
+}
+
+/// Why the live server exited its accept loop.
+pub enum ServerExitReason {
+    SyncbackRequested(SyncbackPayload),
+}
+
 /// Message returned by Rojo API when a change has occurred.
 #[derive(Debug, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]

src/web/mod.rs

@@ -10,7 +10,7 @@ mod util;
 
 use std::convert::Infallible;
 use std::net::SocketAddr;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 
 use hyper::body::Incoming;
 use hyper::server::conn::http1;
@@ -19,52 +19,103 @@ use hyper::Request;
 use hyper_util::rt::TokioIo;
 use tokio::net::TcpListener;
 use tokio::runtime::Runtime;
+use tokio::sync::Notify;
 
 use crate::serve_session::ServeSession;
 
+use self::interface::{ServerExitReason, SyncbackPayload};
+
+/// Shared signal for the syncback endpoint to deposit its payload and notify
+/// the accept loop to shut down.
+pub struct SyncbackSignal {
+    payload: Mutex<Option<SyncbackPayload>>,
+    notify: Notify,
+}
+
+impl SyncbackSignal {
+    pub fn new() -> Self {
+        Self {
+            payload: Mutex::new(None),
+            notify: Notify::new(),
+        }
+    }
+
+    pub fn fire(&self, payload: SyncbackPayload) {
+        *self.payload.lock().unwrap() = Some(payload);
+        self.notify.notify_one();
+    }
+
+    pub fn take_payload(&self) -> Option<SyncbackPayload> {
+        self.payload.lock().unwrap().take()
+    }
+}
+
 pub struct LiveServer {
     serve_session: Arc<ServeSession>,
+    syncback_signal: Arc<SyncbackSignal>,
 }
 
 impl LiveServer {
     pub fn new(serve_session: Arc<ServeSession>) -> Self {
-        LiveServer { serve_session }
+        LiveServer {
+            serve_session,
+            syncback_signal: Arc::new(SyncbackSignal::new()),
+        }
     }
 
-    pub fn start(self, address: SocketAddr) {
+    pub fn start(self, address: SocketAddr) -> ServerExitReason {
         let serve_session = Arc::clone(&self.serve_session);
+        let syncback_signal = Arc::clone(&self.syncback_signal);
 
         let rt = Runtime::new().unwrap();
-        rt.block_on(async move {
+        let exit_reason = rt.block_on(async move {
             let listener = TcpListener::bind(address).await.unwrap();
 
             loop {
-                let (stream, _) = listener.accept().await.unwrap();
-                let io = TokioIo::new(stream);
-                let serve_session = Arc::clone(&serve_session);
-
-                tokio::spawn(async move {
-                    let service = service_fn(move |req: Request<Incoming>| {
+                tokio::select! {
+                    result = listener.accept() => {
+                        let (stream, _) = result.unwrap();
+                        let io = TokioIo::new(stream);
                         let serve_session = Arc::clone(&serve_session);
+                        let syncback_signal = Arc::clone(&syncback_signal);
 
-                        async move {
-                            if req.uri().path().starts_with("/api") {
-                                Ok::<_, Infallible>(api::call(serve_session, req).await)
-                            } else {
-                                Ok::<_, Infallible>(ui::call(serve_session, req).await)
+                        tokio::spawn(async move {
+                            let service = service_fn(move |req: Request<Incoming>| {
+                                let serve_session = Arc::clone(&serve_session);
+                                let syncback_signal = Arc::clone(&syncback_signal);
+
+                                async move {
+                                    if req.uri().path().starts_with("/api") {
+                                        Ok::<_, Infallible>(
+                                            api::call(serve_session, req, syncback_signal).await,
+                                        )
+                                    } else {
+                                        Ok::<_, Infallible>(ui::call(serve_session, req).await)
+                                    }
+                                }
+                            });
+
+                            if let Err(err) = http1::Builder::new()
+                                .serve_connection(io, service)
+                                .with_upgrades()
+                                .await
+                            {
+                                log::error!("Error serving connection: {err}");
                             }
-                        }
-                    });
-
-                    if let Err(err) = http1::Builder::new()
-                        .serve_connection(io, service)
-                        .with_upgrades()
-                        .await
-                    {
-                        log::error!("Error serving connection: {err}");
+                        });
                     }
-                });
+                    _ = syncback_signal.notify.notified() => {
+                        break;
+                    }
+                }
             }
+
+            let payload = syncback_signal
+                .take_payload()
+                .expect("Syncback signal fired but no payload was deposited");
+            ServerExitReason::SyncbackRequested(payload)
         });
+
+        exit_reason
     }
 }