msy 0.4.6

use super::{TransferResult, Transport};
use crate::error::{Result, SyncError};
use crate::fs_util::{has_hard_links, same_filesystem, supports_cow_reflinks};
use crate::integrity::{ChecksumType, IntegrityVerifier};
use crate::resource::format_bytes;
use crate::sync::scanner::{FileEntry, ScanOptions, Scanner};
use crate::temp_file::TempFileGuard;
use async_trait::async_trait;
use futures::stream::{BoxStream, Stream, StreamExt};
use std::fs::{self, File};
use std::path::Path;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::sync::mpsc;

#[cfg(unix)]
use std::os::unix::fs::MetadataExt;

/// Stream wrapper for mpsc::Receiver
struct ReceiverStream<T> {
	rx: mpsc::Receiver<T>,
}

impl<T> Stream for ReceiverStream<T> {
	type Item = T;

	fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
		self.rx.poll_recv(cx)
	}
}

/// Check if a file is sparse by comparing allocated blocks to file size
#[cfg(unix)]
fn is_file_sparse(metadata: &std::fs::Metadata) -> bool {
	let blocks = metadata.blocks();
	let file_size = metadata.len();
	let allocated_size = blocks * 512;

	// File is sparse if allocated size is significantly less than file size
	let threshold = 4096;
	file_size > threshold && allocated_size < file_size.saturating_sub(threshold)
}

#[cfg(not(unix))]
fn is_file_sparse(_metadata: &std::fs::Metadata) -> bool {
	false // Non-Unix platforms don't support sparse detection
}

/// Copy a sparse file while preserving holes
///
/// Tries to use SEEK_HOLE/SEEK_DATA for efficiency, falls back to block-based
/// zero detection if not supported.
#[cfg(unix)]
fn copy_sparse_file(source: &Path, dest: &Path) -> std::io::Result<u64> {
	// Try SEEK_HOLE/SEEK_DATA first (most efficient)
	match copy_sparse_file_seek(source, dest) {
		Ok(size) => Ok(size),
		Err(e) if e.raw_os_error() == Some(libc::EINVAL) => {
			// SEEK_DATA not supported, fall back to block-based approach
			copy_sparse_file_blocks(source, dest)
		}
		Err(e) => Err(e),
	}
}

/// Copy sparse file using SEEK_HOLE/SEEK_DATA (fast path)
#[cfg(unix)]
fn copy_sparse_file_seek(source: &Path, dest: &Path) -> std::io::Result<u64> {
	use std::io::{Read, Seek, SeekFrom, Write};
	use std::os::unix::io::AsRawFd;

	const SEEK_DATA: i32 = 3; // Find next data region
	const SEEK_HOLE: i32 = 4; // Find next hole

	let mut src_file = File::open(source)?;
	let src_meta = src_file.metadata()?;
	let file_size = src_meta.len();

	if dest.exists() {
		fs::remove_file(dest)?;
	}
	let mut dst_file = File::create(dest)?;

	let mut pos: i64 = 0;
	let file_size_i64 = file_size as i64;
	let src_fd = src_file.as_raw_fd();

	// Try SEEK_DATA first to check if supported
	let first_data = unsafe { libc::lseek(src_fd, 0, SEEK_DATA) };
	if first_data < 0 {
		let err = std::io::Error::last_os_error();
		if err.raw_os_error() == Some(libc::EINVAL) {
			return Err(err); // Not supported, caller will fall back
		}
		// ENXIO means file is all holes - just set size and return
		dst_file.set_len(file_size)?;
		return Ok(file_size);
	}

	// Seek back to start
	unsafe { libc::lseek(src_fd, 0, libc::SEEK_SET) };
	src_file.seek(SeekFrom::Start(0))?;

	while pos < file_size_i64 {
		let data_start = unsafe { libc::lseek(src_fd, pos, SEEK_DATA) };
		if data_start < 0 {
			break; // No more data (ENXIO)
		}
		if data_start >= file_size_i64 {
			break;
		}

		let hole_start = unsafe { libc::lseek(src_fd, data_start, SEEK_HOLE) };
		let data_end = if hole_start < 0 || hole_start > file_size_i64 { file_size_i64 } else { hole_start };

		let data_len = (data_end - data_start) as usize;
		src_file.seek(SeekFrom::Start(data_start as u64))?;
		dst_file.seek(SeekFrom::Start(data_start as u64))?;

		let mut remaining = data_len;
		let mut buffer = vec![0u8; 1024 * 1024];

		while remaining > 0 {
			let chunk_size = remaining.min(buffer.len());
			let read = src_file.read(&mut buffer[..chunk_size])?;
			if read == 0 {
				break;
			}
			dst_file.write_all(&buffer[..read])?;
			remaining = remaining.saturating_sub(read);
		}

		pos = data_end;
	}

	dst_file.set_len(file_size)?;
	dst_file.sync_all()?;
	Ok(file_size)
}

/// Copy sparse file by reading blocks and detecting zeros (slow path, portable)
#[cfg(unix)]
fn copy_sparse_file_blocks(source: &Path, dest: &Path) -> std::io::Result<u64> {
	use std::io::{Read, Seek, SeekFrom, Write};

	let mut src_file = File::open(source)?;
	let src_meta = src_file.metadata()?;
	let file_size = src_meta.len();

	if dest.exists() {
		fs::remove_file(dest)?;
	}
	let mut dst_file = File::create(dest)?;

	// Set file size FIRST using ftruncate
	// This creates a sparse file with the correct size
	dst_file.set_len(file_size)?;

	const BLOCK_SIZE: usize = 4096; // Typical filesystem block size
	let mut buffer = vec![0u8; BLOCK_SIZE];
	let mut pos = 0u64;

	while pos < file_size {
		let to_read = ((file_size - pos) as usize).min(BLOCK_SIZE);
		let read = src_file.read(&mut buffer[..to_read])?;
		if read == 0 {
			break;
		}

		// Check if block is all zeros (sparse hole)
		if buffer[..read].iter().all(|&b| b == 0) {
			// Skip this block (hole is already there from ftruncate)
			pos += read as u64;
		} else {
			// Write non-zero data
			dst_file.seek(SeekFrom::Start(pos))?;
			dst_file.write_all(&buffer[..read])?;
			pos += read as u64;
		}
	}

	dst_file.sync_all()?;
	Ok(file_size)
}

#[cfg(not(unix))]
fn copy_sparse_file(source: &Path, dest: &Path) -> std::io::Result<u64> {
	// On non-Unix platforms, fall back to regular copy
	fs::copy(source, dest)
}

/// Local filesystem transport
///
/// Implements the Transport trait for local filesystem operations.
/// This wraps the existing Phase 1 implementation in the async Transport interface.
pub struct LocalTransport {
	verifier: IntegrityVerifier,
	scan_options: ScanOptions,
}

impl LocalTransport {
	pub fn new() -> Self {
		// Default: no verification
		Self { verifier: IntegrityVerifier::new(ChecksumType::None, false), scan_options: ScanOptions::default() }
	}

	pub fn with_verifier(verifier: IntegrityVerifier) -> Self {
		Self { verifier, scan_options: ScanOptions::default() }
	}

	#[allow(dead_code)] // Public API
	pub fn with_scan_options(mut self, options: ScanOptions) -> Self {
		self.scan_options = options;
		self
	}
}

impl Default for LocalTransport {
	fn default() -> Self {
		Self::new()
	}
}

#[async_trait]
impl Transport for LocalTransport {
	fn set_scan_options(&mut self, options: ScanOptions) {
		self.scan_options = options;
	}

	async fn scan(&self, path: &Path) -> Result<Vec<FileEntry>> {
		// Use existing scanner (runs synchronously, wrapped in async)
		let path = path.to_path_buf();
		let options = self.scan_options;
		tokio::task::spawn_blocking(move || {
			let scanner = Scanner::new(&path).with_options(options);
			scanner.scan()
		})
		.await
		.map_err(|e| SyncError::Io(std::io::Error::other(e.to_string())))?
	}

	async fn scan_streaming(&self, path: &Path) -> Result<BoxStream<'static, Result<FileEntry>>> {
		let path = path.to_path_buf();
		let options = self.scan_options;
		let (tx, rx) = mpsc::channel(1000);

		// Spawn blocking task to run the scanner
		tokio::task::spawn_blocking(move || {
			let scanner = Scanner::new(&path).with_options(options);
			if let Ok(iter) = scanner.scan_streaming() {
				for entry in iter {
					if tx.blocking_send(entry).is_err() {
						break; // Receiver dropped
					}
				}
			} else {
				// Handle scanner creation error
				let _ = tx.blocking_send(Err(SyncError::Io(std::io::Error::other("Failed to start scanner"))));
			}
		});

		Ok(ReceiverStream { rx }.boxed())
	}

	async fn exists(&self, path: &Path) -> Result<bool> {
		Ok(tokio::fs::try_exists(path).await.unwrap_or(false))
	}

	async fn metadata(&self, path: &Path) -> Result<std::fs::Metadata> {
		tokio::fs::metadata(path).await.map_err(|e| SyncError::ReadDirError { path: path.to_path_buf(), source: e })
	}

	async fn create_dir_all(&self, path: &Path) -> Result<()> {
		tokio::fs::create_dir_all(path).await.map_err(SyncError::Io)
	}

	async fn copy_file(&self, source: &Path, dest: &Path) -> Result<TransferResult> {
		// Ensure parent directory exists
		if let Some(parent) = dest.parent() {
			self.create_dir_all(parent).await?;
		}

		// Copy file with checksum verification using spawn_blocking
		let source = source.to_path_buf();
		let dest = dest.to_path_buf();

		tokio::task::spawn_blocking(move || {
			// Check if source is sparse
			let source_meta = fs::metadata(&source).map_err(|e| SyncError::CopyError { path: source.clone(), source: e })?;

			let is_sparse = is_file_sparse(&source_meta);

			if is_sparse {
				// For sparse files, use std::fs::copy() which preserves sparseness on Unix
				tracing::debug!("Sparse file detected ({}), using sparse-aware copy", source.display());
				let bytes_written = fs::copy(&source, &dest).map_err(|e| SyncError::CopyError { path: source.clone(), source: e })?;

				// Strip xattrs (fs::copy may preserve them on some platforms)
				#[cfg(unix)]
				{
					if let Ok(xattr_list) = xattr::list(&dest) {
						for attr_name in xattr_list {
							let _ = xattr::remove(&dest, &attr_name);
						}
					}
				}

				// Preserve modification time
				if let Ok(mtime) = source_meta.modified() {
					let _ = filetime::set_file_mtime(&dest, filetime::FileTime::from_system_time(mtime));
				}

				tracing::debug!("Sparse copy complete: {} ({} bytes logical size)", source.display(), bytes_written);

				return Ok(bytes_written);
			}

			// Use fs::copy() which is optimized per-platform:
			// - macOS: clonefile() for COW reflinks on APFS (100x+ faster)
			// - Linux: copy_file_range() for zero-copy (kernel-side)
			// - Fallback: sendfile() or read/write
			// This is MUCH faster than manual read/write loop
			let bytes_written = fs::copy(&source, &dest).map_err(|e| SyncError::CopyError { path: source.clone(), source: e })?;

			// fs::copy() may preserve xattrs on some platforms (e.g., macOS).
			// Strip all xattrs so that Transferrer can selectively re-add them
			// based on preserve_xattrs setting.
			#[cfg(unix)]
			{
				if let Ok(xattr_list) = xattr::list(&dest) {
					for attr_name in xattr_list {
						let _ = xattr::remove(&dest, &attr_name);
					}
				}
			}

			tracing::debug!("Copied {} ({} bytes, fast copy)", source.display(), bytes_written);

			// Preserve modification time
			if let Ok(mtime) = source_meta.modified() {
				let _ = filetime::set_file_mtime(&dest, filetime::FileTime::from_system_time(mtime));
			}

			Ok(bytes_written)
		})
		.await
		.map_err(|e| SyncError::Io(std::io::Error::other(e.to_string())))
		.and_then(|r| r)
		.map(TransferResult::new)
	}

	async fn sync_file_with_delta(&self, source: &Path, dest: &Path) -> Result<TransferResult> {
		// Check if destination exists
		if !self.exists(dest).await? {
			tracing::debug!("Destination doesn't exist, using full copy");
			return self.copy_file(source, dest).await;
		}

		// Get file sizes
		let source_meta = self.metadata(source).await?;
		let dest_meta = self.metadata(dest).await?;
		let source_size = source_meta.len();
		let dest_size = dest_meta.len();

		// Size-based heuristic: use delta sync for files >10MB
		// Below this threshold, sequential copy is often faster than the overhead
		// of checksumming + delta generation + random I/O, even with O(1) rolling hash.
		// This threshold is tuned based on benchmarks showing delta sync is beneficial
		// for files as small as 10MB when changes are localized (e.g., 1MB change in 100MB).
		const DELTA_THRESHOLD: u64 = 10 * 1024 * 1024; // 10MB

		if dest_size < DELTA_THRESHOLD {
			tracing::debug!(
				"File size ({:.1} MB) below delta threshold ({} MB), using full copy",
				dest_size as f64 / 1024.0 / 1024.0,
				DELTA_THRESHOLD / 1024 / 1024
			);
			return self.copy_file(source, dest).await;
		}

		// Skip delta if destination is very small (full copy is faster)
		if dest_size < 4096 {
			tracing::debug!("Destination too small for delta sync, using full copy");
			return self.copy_file(source, dest).await;
		}

		tracing::info!("File size {:.1} MB, attempting delta sync", dest_size as f64 / 1024.0 / 1024.0);

		// Run delta sync in blocking task
		let source = source.to_path_buf();
		let dest = dest.to_path_buf();
		let verifier = self.verifier.clone();

		tokio::task::spawn_blocking(move || {
			use crate::delta::estimate_change_ratio;
			use std::io::{BufReader, Read, Seek, SeekFrom, Write};
			use std::time::Instant;

			let block_size = 64 * 1024; // 64KB blocks for good I/O performance
			let total_start = Instant::now();

			// Check if source file is sparse FIRST (before change ratio)
			// Sparse files need special handling to preserve holes
			let source_meta = fs::metadata(&source).map_err(|e| SyncError::CopyError { path: source.clone(), source: e })?;

			if is_file_sparse(&source_meta) {
				tracing::info!("Source file is sparse (allocated size < logical size), using sparse-aware copy");

				// Use SEEK_HOLE/SEEK_DATA to preserve sparseness
				let bytes_written = copy_sparse_file(&source, &dest).map_err(|e| SyncError::CopyError { path: source.clone(), source: e })?;

				tracing::debug!("Sparse file copy complete: {} bytes logical size", bytes_written);

				return Ok(TransferResult::new(bytes_written));
			}

			// Sample blocks to estimate change ratio
			// If >75% of file has changed, full copy is faster than delta sync
			let change_ratio_result = estimate_change_ratio(
				&source,
				&dest,
				block_size,
				Some(20),   // Sample 20 blocks
				Some(0.75), // 75% threshold
			);

			match change_ratio_result {
				Ok(ratio) => {
					tracing::info!("Change ratio: {} ({}/{} blocks changed)", ratio.change_ratio_percent(), ratio.blocks_changed, ratio.blocks_sampled);

					if !ratio.use_delta {
						tracing::info!(
							"Change ratio {} exceeds threshold {:.1}%, using full copy instead of delta sync",
							ratio.change_ratio_percent(),
							ratio.threshold * 100.0
						);

						// Fallback to full copy (not sparse, so fs::copy is fine)
						let bytes_written = fs::copy(&source, &dest).map_err(|e| SyncError::CopyError { path: source.clone(), source: e })?;

						return Ok(TransferResult::new(bytes_written));
					}

					tracing::info!("Change ratio {} below threshold, proceeding with delta sync", ratio.change_ratio_percent());
				}
				Err(e) => {
					tracing::warn!("Failed to estimate change ratio: {}. Proceeding with delta sync anyway.", e);
				}
			}

			// Choose delta sync strategy based on filesystem capabilities and file properties
			let supports_cow = supports_cow_reflinks(&dest);
			let same_fs = same_filesystem(&source, &dest);
			let has_hardlinks = has_hard_links(&dest);

			let use_cow_strategy = supports_cow && same_fs && !has_hardlinks;

			// Log strategy selection for debugging
			if use_cow_strategy {
				tracing::info!("Delta sync strategy: COW (clone + selective writes) - filesystem supports COW reflinks");
			} else {
				let reason = if !supports_cow {
					"filesystem does not support COW reflinks"
				} else if !same_fs {
					"source and dest on different filesystems"
				} else {
					"destination has hard links (preserving link integrity)"
				};

				tracing::info!("Delta sync strategy: in-place (full file rebuild) - {}", reason);
			}

			// Strategy 1: COW clone + selective writes (fast on APFS/BTRFS/XFS)
			// Strategy 2: In-place delta (for ext4, hard links, cross-filesystem)
			let temp_dest = dest.with_extension("sy.tmp");
			let temp_guard = TempFileGuard::new(&temp_dest);

			let (bytes_written, literal_bytes, changed_blocks) = if use_cow_strategy {
				// COW Strategy: Clone file (instant), then selectively overwrite changed blocks
				fs::copy(&dest, &temp_dest).map_err(|e| SyncError::DeltaSyncError {
					path: temp_dest.clone(),
					strategy: "COW (clone + selective writes)".to_string(),
					source: e,
					hint: "COW file cloning failed. This may happen if:\n  \
                           - Filesystem doesn't support reflinks (needs APFS, BTRFS, or XFS)\n  \
                           - Cross-filesystem operation detected\n  \
                           - Insufficient disk space\n  \
                           Falling back to in-place strategy may help."
						.to_string(),
				})?;

				// Strip xattrs from clone (fs::copy may preserve them)
				#[cfg(unix)]
				{
					if let Ok(xattr_list) = xattr::list(&temp_dest) {
						for attr_name in xattr_list {
							let _ = xattr::remove(&temp_dest, &attr_name);
						}
					}
				}

				// Open source and original dest for reading (sequential)
				let mut source_file =
					BufReader::with_capacity(256 * 1024, File::open(&source).map_err(|e| SyncError::CopyError { path: source.clone(), source: e })?);
				let mut dest_file =
					BufReader::with_capacity(256 * 1024, File::open(&dest).map_err(|e| SyncError::CopyError { path: dest.clone(), source: e })?);

				// Open temp file for writing (selective, seek-based)
				let mut temp_file = File::options().write(true).open(&temp_dest).map_err(|e| SyncError::CopyError { path: temp_dest.clone(), source: e })?;

				let mut source_buf = vec![0u8; block_size];
				let mut dest_buf = vec![0u8; block_size];
				let mut offset = 0u64;
				let mut bytes_written = 0u64;
				let mut literal_bytes = 0u64;
				let mut changed_blocks = 0usize;

				// Compare blocks and only write changed ones
				loop {
					let src_read = source_file.read(&mut source_buf).map_err(|e| SyncError::CopyError { path: source.clone(), source: e })?;
					if src_read == 0 {
						break; // EOF
					}

					let dst_read = dest_file.read(&mut dest_buf).map_err(|e| SyncError::CopyError { path: dest.clone(), source: e })?;

					// Compare blocks
					let blocks_match = src_read == dst_read && source_buf[..src_read] == dest_buf[..dst_read];

					if !blocks_match {
						// Block changed - seek and write to temp file
						temp_file.seek(SeekFrom::Start(offset)).map_err(|e| SyncError::CopyError { path: temp_dest.clone(), source: e })?;
						temp_file.write_all(&source_buf[..src_read]).map_err(|e| SyncError::CopyError { path: temp_dest.clone(), source: e })?;

						// Verify block if paranoid mode enabled
						if verifier.verify_on_write() {
							let mut verify_buf = vec![0u8; src_read];
							temp_file.seek(SeekFrom::Start(offset)).map_err(|e| SyncError::CopyError { path: temp_dest.clone(), source: e })?;
							temp_file.read_exact(&mut verify_buf).map_err(|e| SyncError::CopyError { path: temp_dest.clone(), source: e })?;

							if !verifier.verify_block(&source_buf[..src_read], &verify_buf)? {
								let expected = verifier.compute_data_checksum(&source_buf[..src_read])?;
								let actual = verifier.compute_data_checksum(&verify_buf)?;
								return Err(SyncError::BlockCorruption {
									path: temp_dest.clone(),
									block_number: (offset / block_size as u64) as usize,
									expected_checksum: expected.to_hex(),
									actual_checksum: actual.to_hex(),
								});
							}
						}

						literal_bytes += src_read as u64;
						changed_blocks += 1;
					}
					// If blocks match, we don't write anything! Clone already has the data.

					bytes_written += src_read as u64;
					offset += src_read as u64;
				}

				// Truncate temp file to source size (if source < dest)
				temp_file.set_len(bytes_written).map_err(|e| SyncError::CopyError { path: temp_dest.clone(), source: e })?;

				// Flush and sync temp file
				temp_file.flush().map_err(|e| SyncError::CopyError { path: temp_dest.clone(), source: e })?;
				drop(temp_file);

				(bytes_written, literal_bytes, changed_blocks)
			} else {
				// In-place Strategy: Create temp file, copy only changed blocks
				// This avoids slow fs::copy() on non-COW filesystems like ext4

				// Create empty temp file and allocate space
				let temp_file = File::create(&temp_dest).map_err(|e| SyncError::DeltaSyncError {
					path: temp_dest.clone(),
					strategy: "in-place (full file rebuild)".to_string(),
					source: e,
					hint: "Failed to create temporary file for delta sync.\n  \
                           Check write permissions and disk space on destination."
						.to_string(),
				})?;
				temp_file.set_len(source_size).map_err(|e| SyncError::DeltaSyncError {
					path: temp_dest.clone(),
					strategy: "in-place (full file rebuild)".to_string(),
					source: e,
					hint: format!(
						"Failed to allocate {} for temporary file.\n  \
                                  Check available disk space on destination.",
						format_bytes(source_size)
					),
				})?;
				drop(temp_file);

				// Open source and dest for reading
				let mut source_file =
					BufReader::with_capacity(256 * 1024, File::open(&source).map_err(|e| SyncError::CopyError { path: source.clone(), source: e })?);
				let mut dest_file =
					BufReader::with_capacity(256 * 1024, File::open(&dest).map_err(|e| SyncError::CopyError { path: dest.clone(), source: e })?);

				// Open temp for random writes
				let mut temp_file = File::options().write(true).open(&temp_dest).map_err(|e| SyncError::CopyError { path: temp_dest.clone(), source: e })?;

				let mut source_buf = vec![0u8; block_size];
				let mut dest_buf = vec![0u8; block_size];
				let mut offset = 0u64;
				let mut bytes_written = 0u64;
				let mut literal_bytes = 0u64;
				let mut changed_blocks = 0usize;

				// Compare blocks and write ALL blocks (changed + unchanged)
				// to build the complete new file
				loop {
					let src_read = source_file.read(&mut source_buf).map_err(|e| SyncError::CopyError { path: source.clone(), source: e })?;
					if src_read == 0 {
						break; // EOF
					}

					let dst_read = dest_file.read(&mut dest_buf).map_err(|e| SyncError::CopyError { path: dest.clone(), source: e })?;

					// Compare blocks
					let blocks_match = src_read == dst_read && source_buf[..src_read] == dest_buf[..dst_read];

					// Always seek and write (building complete file)
					temp_file.seek(SeekFrom::Start(offset)).map_err(|e| SyncError::CopyError { path: temp_dest.clone(), source: e })?;
					temp_file.write_all(&source_buf[..src_read]).map_err(|e| SyncError::CopyError { path: temp_dest.clone(), source: e })?;

					// Verify block if paranoid mode enabled
					if verifier.verify_on_write() {
						let mut verify_buf = vec![0u8; src_read];
						temp_file.seek(SeekFrom::Start(offset)).map_err(|e| SyncError::CopyError { path: temp_dest.clone(), source: e })?;
						temp_file.read_exact(&mut verify_buf).map_err(|e| SyncError::CopyError { path: temp_dest.clone(), source: e })?;

						if !verifier.verify_block(&source_buf[..src_read], &verify_buf)? {
							let expected = verifier.compute_data_checksum(&source_buf[..src_read])?;
							let actual = verifier.compute_data_checksum(&verify_buf)?;
							return Err(SyncError::BlockCorruption {
								path: temp_dest.clone(),
								block_number: (offset / block_size as u64) as usize,
								expected_checksum: expected.to_hex(),
								actual_checksum: actual.to_hex(),
							});
						}
					}

					if !blocks_match {
						literal_bytes += src_read as u64;
						changed_blocks += 1;
					}

					bytes_written += src_read as u64;
					offset += src_read as u64;
				}

				// Flush and sync temp file
				temp_file.flush().map_err(|e| SyncError::CopyError { path: temp_dest.clone(), source: e })?;
				drop(temp_file);

				(bytes_written, literal_bytes, changed_blocks)
			};

			let total_elapsed = total_start.elapsed();
			tracing::debug!("Local delta sync completed in {:?} ({} changed blocks)", total_elapsed, changed_blocks);

			let compression_ratio = if source_size > 0 { (literal_bytes as f64 / source_size as f64) * 100.0 } else { 0.0 };

			// Atomic rename
			fs::rename(&temp_dest, &dest).map_err(|e| SyncError::CopyError { path: dest.clone(), source: e })?;

			// Defuse temp file guard - file successfully renamed
			temp_guard.defuse();

			let total_blocks = bytes_written.div_ceil(block_size as u64) as usize;
			tracing::info!("Local delta sync: {} blocks compared, {} changed ({:.1}%)", total_blocks, changed_blocks, compression_ratio);

			Ok::<TransferResult, SyncError>(TransferResult::with_delta(bytes_written, changed_blocks, literal_bytes))
		})
		.await
		.map_err(|e| SyncError::Io(std::io::Error::other(e.to_string())))?
	}

	async fn remove(&self, path: &Path, is_dir: bool) -> Result<()> {
		if is_dir {
			tokio::fs::remove_dir_all(path).await.map_err(SyncError::Io)?;
		} else {
			tokio::fs::remove_file(path).await.map_err(SyncError::Io)?;
		}
		tracing::info!("Removed: {}", path.display());
		Ok(())
	}

	async fn create_hardlink(&self, source: &Path, dest: &Path) -> Result<()> {
		// Ensure parent directory exists
		if let Some(parent) = dest.parent() {
			tokio::fs::create_dir_all(parent).await.map_err(SyncError::Io)?;
		}

		// Create the hard link
		tokio::fs::hard_link(source, dest).await.map_err(SyncError::Io)?;

		tracing::debug!("Created hardlink: {} -> {}", dest.display(), source.display());
		Ok(())
	}

	async fn create_symlink(&self, target: &Path, dest: &Path) -> Result<()> {
		// Ensure parent directory exists
		if let Some(parent) = dest.parent() {
			tokio::fs::create_dir_all(parent).await.map_err(SyncError::Io)?;
		}

		// Remove existing file/symlink if present (force behavior like ln -sf)
		if dest.exists() || dest.is_symlink() {
			tokio::fs::remove_file(dest).await.ok(); // Ignore errors
		}

		// Create the symbolic link
		#[cfg(unix)]
		{
			tokio::fs::symlink(target, dest).await.map_err(SyncError::Io)?;
		}

		#[cfg(windows)]
		{
			// Windows requires different symlink APIs for files vs directories
			if tokio::fs::metadata(target).await.ok().map(|m| m.is_dir()).unwrap_or(false) {
				tokio::fs::symlink_dir(target, dest).await.map_err(SyncError::Io)?;
			} else {
				tokio::fs::symlink_file(target, dest).await.map_err(SyncError::Io)?;
			}
		}

		tracing::debug!("Created symlink: {} -> {}", dest.display(), target.display());
		Ok(())
	}

	async fn copy_file_streaming(
		&self, source: &Path, dest: &Path, progress_callback: Option<std::sync::Arc<dyn Fn(u64, u64) + Send + Sync>>,
	) -> Result<TransferResult> {
		// Ensure parent directory exists
		if let Some(parent) = dest.parent() {
			self.create_dir_all(parent).await?;
		}

		let source = source.to_path_buf();
		let dest = dest.to_path_buf();

		tokio::task::spawn_blocking(move || {
			use std::io::{Read, Write};

			// Get source metadata
			let source_meta = fs::metadata(&source).map_err(|e| SyncError::CopyError { path: source.clone(), source: e })?;

			let total_size = source_meta.len();

			// Open source for reading
			let mut src_file = File::open(&source).map_err(|e| SyncError::CopyError { path: source.clone(), source: e })?;

			// Create destination for writing
			let mut dst_file = File::create(&dest).map_err(|e| SyncError::CopyError { path: dest.clone(), source: e })?;

			// Streaming copy with progress updates
			const CHUNK_SIZE: usize = 1024 * 1024; // 1MB chunks
			let mut buffer = vec![0u8; CHUNK_SIZE];
			let mut bytes_transferred = 0u64;

			// Initial progress callback
			if let Some(callback) = &progress_callback {
				callback(0, total_size);
			}

			loop {
				let bytes_read = src_file.read(&mut buffer).map_err(|e| SyncError::CopyError { path: source.clone(), source: e })?;

				if bytes_read == 0 {
					break; // EOF
				}

				dst_file.write_all(&buffer[..bytes_read]).map_err(|e| SyncError::CopyError { path: dest.clone(), source: e })?;

				bytes_transferred += bytes_read as u64;

				// Update progress after each chunk
				if let Some(callback) = &progress_callback {
					callback(bytes_transferred, total_size);
				}
			}

			// Flush and sync
			dst_file.flush().map_err(|e| SyncError::CopyError { path: dest.clone(), source: e })?;
			dst_file.sync_all().map_err(|e| SyncError::CopyError { path: dest.clone(), source: e })?;
			drop(dst_file);

			// Strip xattrs (to match copy_file behavior)
			#[cfg(unix)]
			{
				if let Ok(xattr_list) = xattr::list(&dest) {
					for attr_name in xattr_list {
						let _ = xattr::remove(&dest, &attr_name);
					}
				}
			}

			// Preserve modification time
			if let Ok(mtime) = source_meta.modified() {
				let _ = filetime::set_file_mtime(&dest, filetime::FileTime::from_system_time(mtime));
			}

			tracing::debug!("Streaming copy complete: {} ({} bytes)", source.display(), bytes_transferred);

			Ok(TransferResult::new(bytes_transferred))
		})
		.await
		.map_err(|e| SyncError::Io(std::io::Error::other(e.to_string())))
		.and_then(|r| r)
	}
}

#[cfg(test)]
mod tests {
	use super::*;
	use std::path::PathBuf;
	use tempfile::TempDir;

	#[tokio::test]
	async fn test_local_transport_scan() {
		let temp = TempDir::new().unwrap();
		let root = temp.path();

		// Create test structure
		fs::create_dir(root.join("dir1")).unwrap();
		fs::write(root.join("file1.txt"), "content").unwrap();
		fs::write(root.join("dir1/file2.txt"), "content").unwrap();

		let transport = LocalTransport::new();
		let entries = transport.scan(root).await.unwrap();

		assert!(entries.len() >= 3);
		assert!(entries.iter().any(|e| e.relative_path.as_path() == Path::new("file1.txt")));
	}

	#[tokio::test]
	async fn test_local_transport_exists() {
		let temp = TempDir::new().unwrap();
		let root = temp.path();

		fs::write(root.join("exists.txt"), "content").unwrap();

		let transport = LocalTransport::new();
		assert!(transport.exists(&root.join("exists.txt")).await.unwrap());
		assert!(!transport.exists(&root.join("not_exists.txt")).await.unwrap());
	}

	#[tokio::test]
	async fn test_local_transport_copy_file() {
		let source_dir = TempDir::new().unwrap();
		let dest_dir = TempDir::new().unwrap();

		let source_file = source_dir.path().join("test.txt");
		fs::write(&source_file, "test content").unwrap();

		let transport = LocalTransport::new();
		let dest_file = dest_dir.path().join("test.txt");
		transport.copy_file(&source_file, &dest_file).await.unwrap();

		assert!(dest_file.exists());
		assert_eq!(fs::read_to_string(&dest_file).unwrap(), "test content");
	}

	#[tokio::test]
	async fn test_local_transport_create_dir_all() {
		let temp = TempDir::new().unwrap();
		let nested_path = temp.path().join("a/b/c");

		let transport = LocalTransport::new();
		transport.create_dir_all(&nested_path).await.unwrap();

		assert!(nested_path.exists());
		assert!(nested_path.is_dir());
	}

	#[tokio::test]
	async fn test_local_transport_remove_file() {
		let temp = TempDir::new().unwrap();
		let file = temp.path().join("remove.txt");
		fs::write(&file, "content").unwrap();

		let transport = LocalTransport::new();
		transport.remove(&file, false).await.unwrap();

		assert!(!file.exists());
	}

	#[tokio::test]
	async fn test_local_transport_remove_dir() {
		let temp = TempDir::new().unwrap();
		let dir = temp.path().join("remove_dir");
		fs::create_dir(&dir).unwrap();
		fs::write(dir.join("file.txt"), "content").unwrap();

		let transport = LocalTransport::new();
		transport.remove(&dir, true).await.unwrap();

		assert!(!dir.exists());
	}

	#[tokio::test]
	#[cfg(unix)] // Sparse files work differently on Windows
	async fn test_local_transport_sparse_file_copy() {
		use std::io::Write;
		use std::os::unix::fs::MetadataExt;

		let source_dir = TempDir::new().unwrap();
		let dest_dir = TempDir::new().unwrap();

		// Create a sparse file using dd
		let source_file = source_dir.path().join("sparse.dat");
		let output = std::process::Command::new("dd")
			.args([
				"if=/dev/zero",
				&format!("of={}", source_file.display()),
				"bs=1024",
				"count=0",
				"seek=10240", // 10MB sparse file
			])
			.output()
			.expect("Failed to create sparse file");

		if !output.status.success() {
			panic!("dd command failed");
		}

		// Write some actual data
		let mut file = std::fs::OpenOptions::new().write(true).open(&source_file).unwrap();
		file.write_all(&[0x42; 4096]).unwrap();
		drop(file);

		// Copy the file
		let transport = LocalTransport::new();
		let dest_file = dest_dir.path().join("sparse.dat");
		let result = transport.copy_file(&source_file, &dest_file).await.unwrap();

		// Verify copy succeeded
		assert!(dest_file.exists());
		assert_eq!(result.bytes_written, 10 * 1024 * 1024);

		// Verify destination is also sparse (or at least has same size)
		let dest_meta = fs::metadata(&dest_file).unwrap();
		assert_eq!(dest_meta.len(), 10 * 1024 * 1024);

		// Check if sparseness was preserved (depends on filesystem)
		let dest_blocks = dest_meta.blocks();
		let dest_allocated = dest_blocks * 512;
		if dest_allocated < dest_meta.len() {
			// Sparseness was preserved!
			eprintln!("✓ Sparse file copy preserved sparseness: {} allocated vs {} logical", dest_allocated, dest_meta.len());
		} else {
			eprintln!("⚠ Sparseness not preserved: {} allocated vs {} logical (filesystem dependent)", dest_allocated, dest_meta.len());
		}
	}

	// === Error Handling Tests ===

	#[tokio::test]
	async fn test_copy_file_nonexistent_source() {
		let dest_dir = TempDir::new().unwrap();
		let transport = LocalTransport::new();

		let nonexistent = PathBuf::from("/nonexistent/file.txt");
		let dest = dest_dir.path().join("test.txt");

		let result = transport.copy_file(&nonexistent, &dest).await;
		assert!(result.is_err(), "Should fail when source doesn't exist");
	}

	#[tokio::test]
	#[cfg(unix)] // Permission tests work differently on Windows
	async fn test_copy_file_permission_denied_destination() {
		use std::os::unix::fs::PermissionsExt;

		let source_dir = TempDir::new().unwrap();
		let dest_dir = TempDir::new().unwrap();

		// Create source file
		let source_file = source_dir.path().join("test.txt");
		fs::write(&source_file, "test content").unwrap();

		// Make destination directory read-only
		let mut perms = fs::metadata(dest_dir.path()).unwrap().permissions();
		perms.set_mode(0o444); // Read-only
		fs::set_permissions(dest_dir.path(), perms).unwrap();

		let transport = LocalTransport::new();
		let dest_file = dest_dir.path().join("test.txt");

		let result = transport.copy_file(&source_file, &dest_file).await;

		// Restore permissions for cleanup
		let mut perms = fs::metadata(dest_dir.path()).unwrap().permissions();
		perms.set_mode(0o755);
		let _ = fs::set_permissions(dest_dir.path(), perms);

		assert!(result.is_err(), "Should fail when destination is read-only");
	}

	#[tokio::test]
	async fn test_create_dir_all_nested() {
		let temp = TempDir::new().unwrap();
		let transport = LocalTransport::new();

		let nested_path = temp.path().join("a/b/c/d/e/f");
		transport.create_dir_all(&nested_path).await.unwrap();

		assert!(nested_path.exists());
		assert!(nested_path.is_dir());
	}

	#[tokio::test]
	#[cfg(unix)]
	async fn test_create_dir_permission_denied() {
		use std::os::unix::fs::PermissionsExt;

		let temp = TempDir::new().unwrap();
		let parent = temp.path().join("parent");
		fs::create_dir(&parent).unwrap();

		// Make parent read-only
		let mut perms = fs::metadata(&parent).unwrap().permissions();
		perms.set_mode(0o444);
		fs::set_permissions(&parent, perms).unwrap();

		let transport = LocalTransport::new();
		let child = parent.join("child");

		let result = transport.create_dir_all(&child).await;

		// Restore permissions for cleanup
		let mut perms = fs::metadata(&parent).unwrap().permissions();
		perms.set_mode(0o755);
		let _ = fs::set_permissions(&parent, perms);

		assert!(result.is_err(), "Should fail when parent is read-only");
	}

	#[tokio::test]
	async fn test_remove_nonexistent_file() {
		let temp = TempDir::new().unwrap();
		let transport = LocalTransport::new();

		let nonexistent = temp.path().join("nonexistent.txt");
		let result = transport.remove(&nonexistent, false).await;

		// Should error on nonexistent file
		assert!(result.is_err());
	}

	#[tokio::test]
	async fn test_metadata_nonexistent_file() {
		let temp = TempDir::new().unwrap();
		let transport = LocalTransport::new();

		let nonexistent = temp.path().join("nonexistent.txt");
		let result = transport.metadata(&nonexistent).await;

		assert!(result.is_err(), "Should fail for nonexistent file");
	}

	#[tokio::test]
	async fn test_scan_nonexistent_directory() {
		let transport = LocalTransport::new();
		let nonexistent = PathBuf::from("/nonexistent/directory");

		let result = transport.scan(&nonexistent).await;
		assert!(result.is_err(), "Should fail when directory doesn't exist");
	}

	#[tokio::test]
	#[cfg(unix)]
	async fn test_scan_permission_denied() {
		use std::os::unix::fs::PermissionsExt;

		let temp = TempDir::new().unwrap();
		let protected_dir = temp.path().join("protected");
		fs::create_dir(&protected_dir).unwrap();
		fs::write(protected_dir.join("file.txt"), "content").unwrap();

		// Make directory unreadable
		let mut perms = fs::metadata(&protected_dir).unwrap().permissions();
		perms.set_mode(0o000); // No permissions
		fs::set_permissions(&protected_dir, perms).unwrap();

		let transport = LocalTransport::new();
		let result = transport.scan(&protected_dir).await;

		// Restore permissions for cleanup
		let mut perms = fs::metadata(&protected_dir).unwrap().permissions();
		perms.set_mode(0o755);
		let _ = fs::set_permissions(&protected_dir, perms);

		assert!(result.is_err(), "Should fail when directory is not readable");
	}

	#[tokio::test]
	#[cfg(unix)]
	async fn test_hardlink_across_filesystems() {
		// This test attempts to create a hardlink across filesystems
		// It should fail gracefully
		let source_dir = TempDir::new().unwrap();
		let source_file = source_dir.path().join("source.txt");
		fs::write(&source_file, "content").unwrap();

		// Try to link to /tmp (likely different filesystem on many systems)
		let dest = PathBuf::from("/tmp/sy_test_hardlink_cross_fs.txt");

		let transport = LocalTransport::new();
		let result = transport.create_hardlink(&source_file, &dest).await;

		// Clean up if it somehow succeeded
		let _ = fs::remove_file(&dest);

		// On most systems this should fail (cross-device link)
		// But if both are on same filesystem, it might succeed
		// Either way, we're testing that it doesn't crash
		// Both outcomes are acceptable - we just verify no panic
		let _ = result;
	}

	#[tokio::test]
	async fn test_copy_file_streaming_with_progress() {
		use std::sync::Arc;
		use std::sync::atomic::{AtomicU64, Ordering};

		let source_dir = TempDir::new().unwrap();
		let dest_dir = TempDir::new().unwrap();

		// Create a 5MB test file
		let source_file = source_dir.path().join("large.dat");
		let data = vec![0x42u8; 5 * 1024 * 1024]; // 5MB
		fs::write(&source_file, &data).unwrap();

		// Track progress updates
		let progress_updates = Arc::new(AtomicU64::new(0));
		let progress_updates_clone = progress_updates.clone();
		let last_bytes = Arc::new(AtomicU64::new(0));
		let last_bytes_clone = last_bytes.clone();

		let progress_callback = Arc::new(move |bytes_transferred: u64, total: u64| {
			progress_updates_clone.fetch_add(1, Ordering::SeqCst);
			last_bytes_clone.store(bytes_transferred, Ordering::SeqCst);
			assert!(bytes_transferred <= total, "Transferred bytes should not exceed total");
		});

		// Copy file with streaming and progress
		let transport = LocalTransport::new();
		let dest_file = dest_dir.path().join("large.dat");
		let result = transport.copy_file_streaming(&source_file, &dest_file, Some(progress_callback)).await.unwrap();

		// Verify copy succeeded
		assert_eq!(result.bytes_written, 5 * 1024 * 1024);
		assert!(dest_file.exists());

		// Verify content matches
		let dest_data = fs::read(&dest_file).unwrap();
		assert_eq!(dest_data, data);

		// Verify progress was updated (should be at least 5 updates for 5MB file with 1MB chunks)
		let updates = progress_updates.load(Ordering::SeqCst);
		assert!(updates >= 5, "Expected at least 5 progress updates, got {}", updates);

		// Verify final progress shows complete transfer
		let final_bytes = last_bytes.load(Ordering::SeqCst);
		assert_eq!(final_bytes, 5 * 1024 * 1024, "Final progress should show complete transfer");
	}

	#[tokio::test]
	async fn test_copy_file_streaming_without_progress() {
		let source_dir = TempDir::new().unwrap();
		let dest_dir = TempDir::new().unwrap();

		// Create a test file
		let source_file = source_dir.path().join("test.txt");
		fs::write(&source_file, "test content").unwrap();

		// Copy without progress callback
		let transport = LocalTransport::new();
		let dest_file = dest_dir.path().join("test.txt");
		let result = transport.copy_file_streaming(&source_file, &dest_file, None).await.unwrap();

		// Verify copy succeeded
		assert_eq!(result.bytes_written, 12);
		assert!(dest_file.exists());
		assert_eq!(fs::read_to_string(&dest_file).unwrap(), "test content");
	}
}