feat(bsk): support upload and download

crates/bsk-cli/src/daemon/ipc.rs

@@ -27,7 +27,8 @@ use bsk_protocol::system::{
     VersionSkewEntry,
 };
 use bsk_protocol::tools::{
-    HelpOutcome, RequestHelpResult, ReturnFailure, WaitMsParams, WaitMsResult,
+    DownloadParams, DownloadResult, HelpOutcome, RequestHelpResult, ReturnFailure,
+    TransferBeginParams, TransferIdParams, UploadParams, WaitMsParams, WaitMsResult,
 };
 use bsk_protocol::{
     CancelParams, CancelResult, ErrorCode, Method, PingResult, ResponseBody, RpcError, RpcId,
@@ -223,6 +224,11 @@ pub fn full_handler(status: DaemonStatus, state: Arc<DaemonState>) -> RpcHandler
                     Ok(v) => ResponseBody::Ok(v),
                     Err(e) => ResponseBody::Err(e),
                 },
+                Method::TransferBegin => handle_transfer_begin(&state, params),
+                Method::TransferChunk => handle_transfer_chunk(&state, params),
+                Method::TransferFinish => handle_transfer_finish(&state, params),
+                Method::TransferRead => handle_transfer_read(&state, params),
+                Method::TransferRelease => handle_transfer_release(&state, params),
                 Method::ToolTabList
                 | Method::ToolTabCreate
                 | Method::ToolTabClose
@@ -246,6 +252,8 @@ pub fn full_handler(status: DaemonStatus, state: Arc<DaemonState>) -> RpcHandler
                 | Method::ToolFill
                 | Method::ToolPress
                 | Method::ToolSelect
+                | Method::ToolUpload
+                | Method::ToolDownload
                 | Method::ToolEvaluate
                 | Method::ToolWaitForNavigation
                 | Method::ToolRequestHelp
@@ -338,24 +346,179 @@ async fn handle_tool_dispatch(
             });
         }
     };
+    let mut params = params;
+    let mut download_transfer_id: Option<String> = None;
+    if method == Method::ToolUpload {
+        let mut upload: UploadParams = match serde_json::from_value(params) {
+            Ok(v) => v,
+            Err(err) => return ResponseBody::Err(invalid_params(err.to_string())),
+        };
+        if upload.files.is_empty() || upload.files.len() > super::file_transfer::MAX_UPLOAD_FILES {
+            return ResponseBody::Err(invalid_params(format!(
+                "upload requires 1..={} files",
+                super::file_transfer::MAX_UPLOAD_FILES
+            )));
+        }
+        let ids: Vec<String> = upload.files.iter().map(|f| f.transfer_id.clone()).collect();
+        let paths = match state.transfers.resolve_uploads(&session_id.0, &ids) {
+            Ok(v) => v,
+            Err(err) => return ResponseBody::Err(err),
+        };
+        for (file, path) in upload.files.iter_mut().zip(paths) {
+            file.staged_path = Some(path.to_string_lossy().into_owned());
+        }
+        params = serde_json::to_value(upload).unwrap_or(Value::Null);
+    } else if method == Method::ToolDownload {
+        let mut download: DownloadParams = match serde_json::from_value(params) {
+            Ok(v) => v,
+            Err(err) => return ResponseBody::Err(invalid_params(err.to_string())),
+        };
+        let staging = match state.transfers.begin_download(&session_id.0) {
+            Ok(v) => v,
+            Err(err) => return ResponseBody::Err(err),
+        };
+        download.staging_path = Some(staging.directory.to_string_lossy().into_owned());
+        download_transfer_id = Some(staging.transfer_id);
+        params = serde_json::to_value(download).unwrap_or(Value::Null);
+    }
     let entry = inflight_guard.entry();
     // `record_stop` must reach the extension while `record_await` holds the
     // serial busy lock — finishing the recording unblocks await.
     let outcome = if method == Method::ToolRecordStop {
         state
             .tool_queues
-            .dispatch_unlocked(&session_id, method, params, timeout, Some(entry))
+            .dispatch_unlocked(&session_id, method.clone(), params, timeout, Some(entry))
             .await
     } else {
         state
             .tool_queues
-            .dispatch(&session_id, method, params, timeout, Some(entry))
+            .dispatch(&session_id, method.clone(), params, timeout, Some(entry))
             .await
     };
     drop(inflight_guard);
     match outcome {
+        Ok(v) if method == Method::ToolDownload => {
+            let id = download_transfer_id.expect("download transfer allocated");
+            let mut result: DownloadResult = match serde_json::from_value(v) {
+                Ok(v) => v,
+                Err(err) => {
+                    state
+                        .transfers
+                        .release(TransferIdParams { transfer_id: id });
+                    return ResponseBody::Err(RpcError {
+                        code: ErrorCode::ProtocolError,
+                        message: format!("invalid tool.download result: {err}"),
+                        data: None,
+                    });
+                }
+            };
+            let Some(path) = result.staged_path.take() else {
+                state
+                    .transfers
+                    .release(TransferIdParams { transfer_id: id });
+                return ResponseBody::Err(RpcError {
+                    code: ErrorCode::ProtocolError,
+                    message: "tool.download returned no staged_path".into(),
+                    data: None,
+                });
+            };
+            match state
+                .transfers
+                .finish_download(&id, std::path::Path::new(&path))
+            {
+                Ok(size) => {
+                    result.byte_size = size;
+                    result.transfer_id = Some(id);
+                    ResponseBody::Ok(serde_json::to_value(result).unwrap_or(Value::Null))
+                }
+                Err(err) => {
+                    state
+                        .transfers
+                        .release(TransferIdParams { transfer_id: id });
+                    ResponseBody::Err(err)
+                }
+            }
+        }
         Ok(v) => ResponseBody::Ok(v),
-        Err(err) => ResponseBody::Err(err.into_rpc()),
+        Err(err) => {
+            if let Some(id) = download_transfer_id {
+                state
+                    .transfers
+                    .release(TransferIdParams { transfer_id: id });
+            }
+            ResponseBody::Err(err.into_rpc())
+        }
+    }
+}
+
+fn handle_transfer_begin(state: &Arc<DaemonState>, params: Value) -> ResponseBody {
+    let p: TransferBeginParams = match serde_json::from_value(params) {
+        Ok(v) => v,
+        Err(err) => return ResponseBody::Err(invalid_params(err.to_string())),
+    };
+    if state
+        .sessions
+        .get(&SessionId(p.session_id.clone()))
+        .is_none()
+    {
+        return ResponseBody::Err(RpcError {
+            code: ErrorCode::NotFound,
+            message: format!("session {} unknown", p.session_id),
+            data: None,
+        });
+    }
+    match state.transfers.begin_upload(p) {
+        Ok(v) => ResponseBody::Ok(serde_json::to_value(v).unwrap_or(Value::Null)),
+        Err(err) => ResponseBody::Err(err),
+    }
+}
+
+fn handle_transfer_chunk(state: &Arc<DaemonState>, params: Value) -> ResponseBody {
+    let p = match serde_json::from_value(params) {
+        Ok(v) => v,
+        Err(err) => return ResponseBody::Err(invalid_params(err.to_string())),
+    };
+    match state.transfers.write_chunk(p) {
+        Ok(v) => ResponseBody::Ok(serde_json::to_value(v).unwrap_or(Value::Null)),
+        Err(err) => ResponseBody::Err(err),
+    }
+}
+
+fn handle_transfer_finish(state: &Arc<DaemonState>, params: Value) -> ResponseBody {
+    let p = match serde_json::from_value(params) {
+        Ok(v) => v,
+        Err(err) => return ResponseBody::Err(invalid_params(err.to_string())),
+    };
+    match state.transfers.finish_upload(p) {
+        Ok(v) => ResponseBody::Ok(serde_json::to_value(v).unwrap_or(Value::Null)),
+        Err(err) => ResponseBody::Err(err),
+    }
+}
+
+fn handle_transfer_read(state: &Arc<DaemonState>, params: Value) -> ResponseBody {
+    let p = match serde_json::from_value(params) {
+        Ok(v) => v,
+        Err(err) => return ResponseBody::Err(invalid_params(err.to_string())),
+    };
+    match state.transfers.read_chunk(p) {
+        Ok(v) => ResponseBody::Ok(serde_json::to_value(v).unwrap_or(Value::Null)),
+        Err(err) => ResponseBody::Err(err),
+    }
+}
+
+fn handle_transfer_release(state: &Arc<DaemonState>, params: Value) -> ResponseBody {
+    let p: TransferIdParams = match serde_json::from_value(params) {
+        Ok(v) => v,
+        Err(err) => return ResponseBody::Err(invalid_params(err.to_string())),
+    };
+    ResponseBody::Ok(serde_json::to_value(state.transfers.release(p)).unwrap_or(Value::Null))
+}
+
+fn invalid_params(message: impl Into<String>) -> RpcError {
+    RpcError {
+        code: ErrorCode::InvalidParams,
+        message: message.into(),
+        data: None,
     }
 }
 
@@ -798,6 +961,7 @@ async fn handle_session_stop(
     .await
     {
         Ok(stop) => {
+            state.transfers.release_session(&session_id.0);
             let result = CliSessionStopResult {
                 stopped: vec![session_id.0],
                 failed: Vec::new(),
@@ -897,6 +1061,7 @@ async fn handle_session_stop_all(
         .await
         {
             Ok(stop) => {
+                state.transfers.release_session(&id.0);
                 stopped.push(id.0);
                 returned_tab_ids.extend(stop.returned_tab_ids);
                 return_failures.extend(stop.return_failures);