MTP: Enable MTP2MTP transfers

Streams!
This prepares features like S3Bucket→FTP.

Also in this commit: handle read-only drives and display a nice error message

Author: vdavid

apps/desktop/src-tauri/src/file_system/volume/local_posix_test.rs

@@ -101,6 +101,14 @@ fn test_supports_watching_returns_true() {
     assert!(volume.supports_watching());
 }
 
+#[test]
+fn test_supports_streaming_returns_false() {
+    // LocalPosixVolume uses the default implementation which returns false.
+    // Streaming is primarily for MTP-to-MTP transfers.
+    let volume = LocalPosixVolume::new("Test", "/tmp");
+    assert!(!volume.supports_streaming());
+}
+
 #[test]
 fn test_write_operations() {
     use std::fs;

apps/desktop/src-tauri/src/file_system/volume/mod.rs

@@ -91,6 +91,21 @@ impl std::fmt::Display for VolumeError {
 
 impl std::error::Error for VolumeError {}
 
+/// A stream of bytes read from a volume.
+///
+/// This is a synchronous, blocking iterator-style interface for reading
+/// file data in chunks. Used for streaming transfers between volumes.
+pub trait VolumeReadStream: Send {
+    /// Returns the next chunk of data, or None if complete.
+    fn next_chunk(&mut self) -> Option<Result<Vec<u8>, VolumeError>>;
+
+    /// Total size of the file in bytes.
+    fn total_size(&self) -> u64;
+
+    /// Bytes read so far (for progress tracking).
+    fn bytes_read(&self) -> u64;
+}
+
 impl From<std::io::Error> for VolumeError {
     fn from(err: std::io::Error) -> Self {
         match err.kind() {
@@ -218,6 +233,35 @@ pub trait Volume: Send + Sync {
     fn get_space_info(&self) -> Result<SpaceInfo, VolumeError> {
         Err(VolumeError::NotSupported)
     }
+
+    // ========================================
+    // Streaming: Optional, default not supported
+    // ========================================
+
+    /// Returns true if this volume supports streaming read/write operations.
+    fn supports_streaming(&self) -> bool {
+        false
+    }
+
+    /// Opens a streaming reader for the given path.
+    ///
+    /// Returns a VolumeReadStream that yields chunks of data.
+    /// The stream must be fully consumed or dropped before other operations.
+    fn open_read_stream(&self, path: &Path) -> Result<Box<dyn VolumeReadStream>, VolumeError> {
+        let _ = path;
+        Err(VolumeError::NotSupported)
+    }
+
+    /// Writes data from a stream to the given path.
+    ///
+    /// # Arguments
+    /// * `dest` - Destination path (file will be created/overwritten)
+    /// * `size` - Total size in bytes (required for protocols like MTP)
+    /// * `stream` - Source data stream
+    fn write_from_stream(&self, dest: &Path, size: u64, stream: Box<dyn VolumeReadStream>) -> Result<u64, VolumeError> {
+        let _ = (dest, size, stream);
+        Err(VolumeError::NotSupported)
+    }
 }
 
 // Implementations

apps/desktop/src-tauri/src/file_system/volume/mtp.rs

@@ -3,10 +3,11 @@
 //! Wraps MTP device storage as a Volume, enabling MTP browsing through
 //! the standard file listing pipeline (same icons, sorting, view modes as local files).
 
-use super::{ConflictInfo, CopyScanResult, SourceItemInfo, SpaceInfo, Volume, VolumeError};
+use super::{ConflictInfo, CopyScanResult, SourceItemInfo, SpaceInfo, Volume, VolumeError, VolumeReadStream};
 use crate::file_system::FileEntry;
 use crate::mtp::connection::{MtpConnectionError, connection_manager};
 use log::debug;
+use mtp_rs::FileDownload;
 use std::path::{Path, PathBuf};
 
 /// A volume backed by an MTP device storage.
@@ -380,6 +381,109 @@ impl Volume for MtpVolume {
             })
             .map_err(map_mtp_error)
     }
+
+    fn supports_streaming(&self) -> bool {
+        true
+    }
+
+    fn open_read_stream(&self, path: &Path) -> Result<Box<dyn VolumeReadStream>, VolumeError> {
+        let mtp_path = self.to_mtp_path(path);
+        let device_id = self.device_id.clone();
+        let storage_id = self.storage_id;
+
+        let handle = tokio::runtime::Handle::current();
+
+        // Get the file download stream from connection manager
+        let (download, total_size) = handle
+            .block_on(async {
+                connection_manager()
+                    .open_download_stream(&device_id, storage_id, &mtp_path)
+                    .await
+            })
+            .map_err(map_mtp_error)?;
+
+        Ok(Box::new(MtpReadStream {
+            handle,
+            download: Some(download),
+            total_size,
+            bytes_read: 0,
+        }))
+    }
+
+    fn write_from_stream(
+        &self,
+        dest: &Path,
+        size: u64,
+        mut stream: Box<dyn VolumeReadStream>,
+    ) -> Result<u64, VolumeError> {
+        let dest_folder = dest.parent().map(|p| self.to_mtp_path(p)).unwrap_or_default();
+        let filename = dest
+            .file_name()
+            .and_then(|n| n.to_str())
+            .ok_or_else(|| VolumeError::IoError("Invalid filename".into()))?
+            .to_string();
+
+        let device_id = self.device_id.clone();
+        let storage_id = self.storage_id;
+
+        // IMPORTANT: Collect all chunks BEFORE entering block_on to avoid nested runtime error.
+        // MtpReadStream::next_chunk() uses block_on internally, so we can't call it from
+        // within another block_on (which upload_from_stream would do).
+        let mut chunks: Vec<bytes::Bytes> = Vec::new();
+        while let Some(result) = stream.next_chunk() {
+            let data = result?;
+            chunks.push(bytes::Bytes::from(data));
+        }
+
+        let handle = tokio::runtime::Handle::current();
+
+        handle
+            .block_on(async {
+                connection_manager()
+                    .upload_from_chunks(&device_id, storage_id, &dest_folder, &filename, size, chunks)
+                    .await
+            })
+            .map_err(map_mtp_error)
+    }
+}
+
+/// Streaming reader for MTP files.
+///
+/// Wraps the mtp-rs FileDownload to provide sync iteration.
+pub struct MtpReadStream {
+    /// Tokio runtime handle for blocking on async operations.
+    handle: tokio::runtime::Handle,
+    /// The underlying async download (wrapped in Option for take semantics).
+    download: Option<FileDownload>,
+    /// Total file size.
+    total_size: u64,
+    /// Bytes read so far.
+    bytes_read: u64,
+}
+
+impl VolumeReadStream for MtpReadStream {
+    fn next_chunk(&mut self) -> Option<Result<Vec<u8>, VolumeError>> {
+        let download = self.download.as_mut()?;
+
+        self.handle.block_on(async {
+            match download.next_chunk().await {
+                Some(Ok(bytes)) => {
+                    self.bytes_read += bytes.len() as u64;
+                    Some(Ok(bytes.to_vec()))
+                }
+                Some(Err(e)) => Some(Err(VolumeError::IoError(e.to_string()))),
+                None => None,
+            }
+        })
+    }
+
+    fn total_size(&self) -> u64 {
+        self.total_size
+    }
+
+    fn bytes_read(&self) -> u64 {
+        self.bytes_read
+    }
 }
 
 /// Maps MTP connection errors to Volume errors.
@@ -461,4 +565,11 @@ mod tests {
         let vol = MtpVolume::new("mtp-20-5", 65537, "Test");
         assert!(!vol.supports_watching());
     }
+
+    #[test]
+    fn test_supports_streaming_returns_true() {
+        // MTP volumes support streaming for direct MTP-to-MTP transfers.
+        let vol = MtpVolume::new("mtp-20-5", 65537, "Test");
+        assert!(vol.supports_streaming());
+    }
 }

apps/desktop/src-tauri/src/file_system/write_operations/volume_copy.rs

@@ -517,9 +517,10 @@ fn copy_volumes_with_progress(
 /// Copies a single path from source volume to destination volume.
 ///
 /// Determines the appropriate strategy based on volume types:
+/// - If both support streaming: Use streaming for direct transfer
 /// - If source is local: dest.import_from_local()
 /// - If dest is local: source.export_to_local()
-/// - Otherwise: export to temp, then import from temp
+/// - Otherwise: Not supported
 fn copy_single_path(
     source_volume: &Arc<dyn Volume>,
     source_path: &Path,
@@ -536,6 +537,22 @@ fn copy_single_path(
     let source_is_local = source_volume.root().starts_with("/") && !source_volume.root().starts_with("/mtp-volume/");
     let dest_is_local = dest_volume.root().starts_with("/") && !dest_volume.root().starts_with("/mtp-volume/");
 
+    // Try streaming path for non-local volumes that support it
+    if !source_is_local && !dest_is_local {
+        if source_volume.supports_streaming() && dest_volume.supports_streaming() {
+            log::debug!(
+                "copy_single_path: using streaming for {} -> {}",
+                source_path.display(),
+                dest_path.display()
+            );
+            let stream = source_volume.open_read_stream(source_path)?;
+            let size = stream.total_size();
+            return dest_volume.write_from_stream(dest_path, size, stream);
+        }
+        // Neither supports streaming - not supported
+        return Err(VolumeError::NotSupported);
+    }
+
     if source_is_local && !dest_is_local {
         // Source is local, dest is not (e.g., Local → MTP)
         // Use import_from_local on destination
@@ -554,7 +571,7 @@ fn copy_single_path(
             dest_volume.root().join(dest_path)
         };
         source_volume.export_to_local(source_path, &local_dest)
-    } else if source_is_local && dest_is_local {
+    } else {
         // Both are local, use export which resolves paths internally
         // Note: export_to_local takes a path relative to the volume root for source,
         // and an absolute local path for destination
@@ -564,10 +581,6 @@ fn copy_single_path(
             dest_volume.root().join(dest_path)
         };
         source_volume.export_to_local(source_path, &local_dest)
-    } else {
-        // Both are non-local (e.g., MTP → MTP) - not supported directly
-        // Would need to go through temp directory
-        Err(VolumeError::NotSupported)
     }
 }