msy 0.4.6

pub mod dual;
#[cfg(feature = "gcs")]
pub mod gcs;
pub mod local;
pub mod router;
#[cfg(feature = "s3")]
pub mod s3;
pub mod server;
#[cfg(feature = "ssh")]
pub mod ssh;

use crate::error::Result;
use crate::sync::scanner::FileEntry;
use async_trait::async_trait;
use futures::stream::{BoxStream, StreamExt};
use std::path::Path;
use std::time::SystemTime;

/// Transport-agnostic file information
///
/// Unlike std::fs::Metadata, this works for both local and remote files
#[derive(Debug, Clone, Copy)]
pub struct FileInfo {
	pub size: u64,
	pub modified: SystemTime,
}

/// Result of a file transfer operation
#[derive(Debug, Clone, Copy)]
pub struct TransferResult {
	/// Actual bytes written (may differ from file size for delta sync)
	pub bytes_written: u64,
	/// Number of delta operations (None if full file copy)
	pub delta_operations: Option<usize>,
	/// Bytes of literal data transferred via delta (None if full file copy)
	pub literal_bytes: Option<u64>,
	/// Bytes transferred over network (compressed size if compression used)
	pub transferred_bytes: Option<u64>,
	/// Whether compression was used
	pub compression_used: bool,
}

impl TransferResult {
	pub fn new(bytes_written: u64) -> Self {
		Self { bytes_written, delta_operations: None, literal_bytes: None, transferred_bytes: None, compression_used: false }
	}

	pub fn with_delta(bytes_written: u64, delta_operations: usize, literal_bytes: u64) -> Self {
		Self {
			bytes_written,
			delta_operations: Some(delta_operations),
			literal_bytes: Some(literal_bytes),
			transferred_bytes: None,
			compression_used: false,
		}
	}

	pub fn with_compression(bytes_written: u64, transferred_bytes: u64) -> Self {
		Self {
			bytes_written,
			delta_operations: None,
			literal_bytes: None,
			transferred_bytes: Some(transferred_bytes),
			compression_used: true,
		}
	}

	/// Returns true if this transfer used delta sync
	pub fn used_delta(&self) -> bool {
		self.delta_operations.is_some()
	}

	/// Calculate compression ratio (percentage of file that was literal data)
	/// Returns None if full file copy
	#[allow(dead_code)]
	pub fn compression_ratio(&self) -> Option<f64> {
		if let (Some(literal), true) = (self.literal_bytes, self.bytes_written > 0) {
			Some((literal as f64 / self.bytes_written as f64) * 100.0)
		} else {
			None
		}
	}
}

/// Transport abstraction for local and remote file operations
///
/// This trait provides a unified interface for file operations that works
/// across both local filesystems and remote systems (SSH, SFTP, etc.)
#[async_trait]
#[allow(dead_code)] // Methods will be used when we implement SSH transport
pub trait Transport: Send + Sync {
	/// Set scanning options (respect gitignore, include .git)
	///
	/// This configures how the transport scans directories. Options include:
	/// - `respect_gitignore`: Whether to filter files based on .gitignore rules
	/// - `include_git_dir`: Whether to include .git directories in scans
	///
	/// Default implementation does nothing (for transports that don't support it).
	fn set_scan_options(&mut self, _options: crate::sync::scanner::ScanOptions) {
		// Default: no-op for transports that don't support scan options
	}

	/// Prepare the transport for transferring a known number of files
	///
	/// Called after scanning to allow transports to optimize for the workload.
	/// For example, SSH transport can expand its connection pool based on file count.
	///
	/// Default implementation does nothing (for transports that don't need preparation).
	async fn prepare_for_transfer(&self, _file_count: usize) -> Result<()> {
		Ok(())
	}

	/// Scan a directory and return all entries
	///
	/// This recursively scans the directory. Behavior is controlled by scan options:
	/// - By default: respects .gitignore patterns and excludes .git directories
	/// - With archive mode: includes all files including .git
	async fn scan(&self, path: &Path) -> Result<Vec<FileEntry>>;

	/// Scan a directory and return a stream of entries
	///
	/// This recursively scans the directory, respecting .gitignore patterns
	/// and excluding .git directories.
	///
	/// The stream yields `Result<FileEntry>`.
	async fn scan_streaming(&self, path: &Path) -> Result<BoxStream<'static, Result<FileEntry>>> {
		// Default implementation: collect vector and stream it (inefficient but compatible)
		let entries = self.scan(path).await?;
		Ok(futures::stream::iter(entries.into_iter().map(Ok)).boxed())
	}

	/// Check if a path exists
	async fn exists(&self, path: &Path) -> Result<bool>;

	/// Get metadata for a path (for comparison during sync)
	async fn metadata(&self, path: &Path) -> Result<std::fs::Metadata>;

	/// Get file information (size and mtime) in a transport-agnostic way
	///
	/// This works for both local and remote files, unlike metadata() which returns
	/// std::fs::Metadata that can't be constructed for remote files.
	async fn file_info(&self, path: &Path) -> Result<FileInfo> {
		// Default implementation uses metadata()
		let meta = self.metadata(path).await?;
		let modified = meta
			.modified()
			.map_err(|e| crate::error::SyncError::Io(std::io::Error::new(e.kind(), format!("Failed to get mtime for {}: {}", path.display(), e))))?;
		Ok(FileInfo { size: meta.len(), modified })
	}

	/// Create all parent directories for a path
	async fn create_dir_all(&self, path: &Path) -> Result<()>;

	/// Create multiple directories in batch (optimization for remote transports)
	///
	/// Default implementation calls create_dir_all for each path.
	/// SSH transport overrides this with a single batched command.
	async fn create_dirs_batch(&self, paths: &[&Path]) -> Result<()> {
		for path in paths {
			self.create_dir_all(path).await?;
		}
		Ok(())
	}

	/// Copy a file from source to destination
	///
	/// This preserves modification time and handles parent directory creation.
	/// Returns the number of bytes actually written.
	async fn copy_file(&self, source: &Path, dest: &Path) -> Result<TransferResult>;

	/// Sync a file using delta sync if destination exists
	///
	/// This uses the rsync algorithm to transfer only changed blocks when
	/// the destination file already exists. Falls back to full copy if
	/// destination doesn't exist or delta sync isn't beneficial.
	/// Returns the number of bytes actually transferred.
	async fn sync_file_with_delta(&self, source: &Path, dest: &Path) -> Result<TransferResult> {
		// Default implementation: fall back to full copy
		self.copy_file(source, dest).await
	}

	/// Remove a file or directory
	async fn remove(&self, path: &Path, is_dir: bool) -> Result<()>;

	/// Create a hard link
	///
	/// Creates a hard link at `dest` pointing to `source`.
	/// Both paths must be on the same filesystem.
	async fn create_hardlink(&self, source: &Path, dest: &Path) -> Result<()>;

	/// Create a symbolic link
	///
	/// Creates a symbolic link at `dest` pointing to `target`.
	async fn create_symlink(&self, target: &Path, dest: &Path) -> Result<()>;

	/// Read file contents into a vector
	///
	/// This is used for cross-transport operations (e.g., remote→local).
	/// Default implementation reads from local filesystem.
	async fn read_file(&self, path: &Path) -> Result<Vec<u8>> {
		tokio::fs::read(path)
			.await
			.map_err(|e| crate::error::SyncError::Io(std::io::Error::new(e.kind(), format!("Failed to read file {}: {}", path.display(), e))))
	}

	/// Compute checksum of a file using streaming (avoids loading entire file into memory)
	///
	/// This method allows each transport to implement efficient checksum computation
	/// without loading the entire file into RAM. For local files, reads in chunks.
	/// For remote files, can use remote commands or streaming transfer.
	///
	/// Default implementation uses the IntegrityVerifier's compute_file_checksum,
	/// which already uses streaming for local files.
	async fn compute_checksum(&self, path: &Path, verifier: &crate::integrity::IntegrityVerifier) -> Result<crate::integrity::Checksum> {
		// Default implementation: use IntegrityVerifier (works for local files)
		let path = path.to_path_buf();
		let verifier = verifier.clone();
		tokio::task::spawn_blocking(move || verifier.compute_file_checksum(&path))
			.await
			.map_err(|e| crate::error::SyncError::Io(std::io::Error::other(e.to_string())))?
	}

	/// Write file contents from a vector
	///
	/// This is used for cross-transport operations (e.g., remote→local).
	/// Default implementation writes to local filesystem.
	async fn write_file(&self, path: &Path, data: &[u8], mtime: std::time::SystemTime) -> Result<()> {
		use tokio::io::AsyncWriteExt;

		// Create parent directories
		if let Some(parent) = path.parent() {
			tokio::fs::create_dir_all(parent).await?;
		}

		// Write file
		let mut file = tokio::fs::File::create(path).await?;
		file.write_all(data).await?;
		file.flush().await?;
		drop(file);

		// Set mtime
		filetime::set_file_mtime(path, filetime::FileTime::from_system_time(mtime))?;

		Ok(())
	}

	/// Get modification time for a file
	///
	/// This is used for cross-transport operations where metadata() doesn't work.
	/// Default implementation uses local filesystem.
	async fn get_mtime(&self, path: &Path) -> Result<std::time::SystemTime> {
		let metadata = tokio::fs::metadata(path).await?;
		metadata
			.modified()
			.map_err(|e| crate::error::SyncError::Io(std::io::Error::new(e.kind(), format!("Failed to get mtime for {}: {}", path.display(), e))))
	}

	/// Copy file using streaming (for large files)
	///
	/// Reads and writes in chunks to avoid loading entire file into memory.
	/// Calls progress_callback with (bytes_transferred, total_bytes) after each chunk.
	/// Returns total bytes transferred.
	async fn copy_file_streaming(
		&self, source: &Path, dest: &Path, progress_callback: Option<std::sync::Arc<dyn Fn(u64, u64) + Send + Sync>>,
	) -> Result<TransferResult> {
		// Default implementation: fall back to read_file/write_file for simplicity
		// Implementations can override for true streaming
		let data = self.read_file(source).await?;
		let total_size = data.len() as u64;
		let mtime = self.get_mtime(source).await?;

		if let Some(callback) = &progress_callback {
			callback(0, total_size);
		}
		self.write_file(dest, &data, mtime).await?;
		if let Some(callback) = &progress_callback {
			callback(total_size, total_size);
		}

		Ok(TransferResult::new(total_size))
	}

	/// Check available disk space at the destination
	///
	/// Verifies that at least `bytes_needed` (plus 10% buffer) is available.
	/// For remote transports, this may involve executing commands via SSH/SFTP.
	async fn check_disk_space(&self, path: &Path, bytes_needed: u64) -> Result<()> {
		// Default implementation: use local disk space check
		crate::resource::check_disk_space(path, bytes_needed)
	}

	/// Set extended attributes on a file
	///
	/// For remote transports, executes platform-specific commands via SSH.
	/// On Linux: uses setfattr
	/// On macOS: uses xattr -w
	async fn set_xattrs(&self, path: &Path, xattrs: &[(String, Vec<u8>)]) -> Result<()> {
		// Default implementation: use local xattr crate
		#[cfg(unix)]
		{
			let path = path.to_path_buf();
			let xattrs = xattrs.to_vec();

			tokio::task::spawn_blocking(move || {
				for (name, value) in xattrs {
					if let Err(e) = xattr::set(&path, &name, &value) {
						tracing::warn!("Failed to set xattr {} on {}: {}", name, path.display(), e);
					}
				}
			})
			.await
			.map_err(|e| crate::error::SyncError::Io(std::io::Error::other(e.to_string())))?;
		}
		#[cfg(not(unix))]
		{
			let _ = (path, xattrs);
		}
		Ok(())
	}

	/// Set POSIX ACLs on a file
	///
	/// For remote transports, executes setfacl via SSH.
	async fn set_acls(&self, path: &Path, acls_text: &str) -> Result<()> {
		// Default implementation: use local exacl crate
		#[cfg(all(unix, feature = "acl"))]
		{
			use exacl::{AclEntry, setfacl};
			use std::str::FromStr;

			let path = path.to_path_buf();
			let acls_text = acls_text.to_string();

			tokio::task::spawn_blocking(move || {
				let mut acl_entries = Vec::new();
				for line in acls_text.lines() {
					let line = line.trim();
					if line.is_empty() {
						continue;
					}
					match AclEntry::from_str(line) {
						Ok(entry) => acl_entries.push(entry),
						Err(e) => {
							tracing::warn!("Failed to parse ACL entry '{}' for {}: {}", line, path.display(), e);
							continue;
						}
					}
				}

				if !acl_entries.is_empty()
					&& let Err(e) = setfacl(&[&path], &acl_entries, None)
				{
					tracing::warn!("Failed to set ACLs on {}: {}", path.display(), e);
				}
			})
			.await
			.map_err(|e| crate::error::SyncError::Io(std::io::Error::other(e.to_string())))?;
		}
		#[cfg(not(all(unix, feature = "acl")))]
		{
			let _ = (path, acls_text);
		}
		Ok(())
	}

	/// Set BSD file flags (macOS only)
	///
	/// For remote transports, executes chflags via SSH.
	async fn set_bsd_flags(&self, path: &Path, flags: u32) -> Result<()> {
		// Default implementation: use local libc chflags
		#[cfg(target_os = "macos")]
		{
			use std::ffi::CString;
			let path = path.to_path_buf();

			tokio::task::spawn_blocking(move || {
				let c_path = match CString::new(path.to_str().unwrap_or("")) {
					Ok(p) => p,
					Err(e) => {
						tracing::warn!("Failed to create C string for {}: {}", path.display(), e);
						return;
					}
				};

				let result = unsafe { libc::chflags(c_path.as_ptr(), flags as _) };
				if result != 0 {
					tracing::warn!("Failed to set BSD flags on {}: {}", path.display(), std::io::Error::last_os_error());
				}
			})
			.await
			.map_err(|e| crate::error::SyncError::Io(std::io::Error::other(e.to_string())))?;
		}
		#[cfg(not(target_os = "macos"))]
		{
			let _ = (path, flags);
		}
		Ok(())
	}

	/// Bulk transfer multiple files efficiently using tar streaming
	///
	/// This method transfers many files in a single operation, which is much faster
	/// than individual file transfers for SSH/remote connections.
	///
	/// Arguments:
	/// - `source_base`: Base path on source (e.g., /path/to/source)
	/// - `dest_base`: Base path on destination (e.g., /path/to/dest)
	/// - `relative_paths`: List of relative paths to transfer (relative to source_base)
	///
	/// Returns total bytes transferred.
	///
	/// Default implementation falls back to individual copy_file calls.
	/// SSH transport overrides with efficient tar streaming.
	async fn bulk_copy_files(&self, source_base: &Path, dest_base: &Path, relative_paths: &[&Path]) -> Result<u64> {
		// Default implementation: fall back to individual copies
		let mut total_bytes = 0u64;
		for rel_path in relative_paths {
			let source = source_base.join(rel_path);
			let dest = dest_base.join(rel_path);
			match self.copy_file(&source, &dest).await {
				Ok(result) => total_bytes += result.bytes_written,
				Err(e) => {
					tracing::warn!("Failed to copy {}: {}", source.display(), e);
				}
			}
		}
		Ok(total_bytes)
	}
}

// Implement Transport for Arc<T> where T: Transport
// This allows sharing transports across tasks in parallel execution
#[async_trait]
impl<T: Transport + ?Sized> Transport for std::sync::Arc<T> {
	async fn scan(&self, path: &Path) -> Result<Vec<FileEntry>> {
		(**self).scan(path).await
	}

	async fn scan_streaming(&self, path: &Path) -> Result<BoxStream<'static, Result<FileEntry>>> {
		(**self).scan_streaming(path).await
	}

	async fn exists(&self, path: &Path) -> Result<bool> {
		(**self).exists(path).await
	}

	async fn metadata(&self, path: &Path) -> Result<std::fs::Metadata> {
		(**self).metadata(path).await
	}

	async fn file_info(&self, path: &Path) -> Result<FileInfo> {
		(**self).file_info(path).await
	}

	async fn create_dir_all(&self, path: &Path) -> Result<()> {
		(**self).create_dir_all(path).await
	}

	async fn create_dirs_batch(&self, paths: &[&Path]) -> Result<()> {
		(**self).create_dirs_batch(paths).await
	}

	async fn copy_file(&self, source: &Path, dest: &Path) -> Result<TransferResult> {
		(**self).copy_file(source, dest).await
	}

	async fn sync_file_with_delta(&self, source: &Path, dest: &Path) -> Result<TransferResult> {
		(**self).sync_file_with_delta(source, dest).await
	}

	async fn remove(&self, path: &Path, is_dir: bool) -> Result<()> {
		(**self).remove(path, is_dir).await
	}

	async fn create_hardlink(&self, source: &Path, dest: &Path) -> Result<()> {
		(**self).create_hardlink(source, dest).await
	}

	async fn create_symlink(&self, target: &Path, dest: &Path) -> Result<()> {
		(**self).create_symlink(target, dest).await
	}

	async fn read_file(&self, path: &Path) -> Result<Vec<u8>> {
		(**self).read_file(path).await
	}

	async fn write_file(&self, path: &Path, data: &[u8], mtime: std::time::SystemTime) -> Result<()> {
		(**self).write_file(path, data, mtime).await
	}

	async fn get_mtime(&self, path: &Path) -> Result<std::time::SystemTime> {
		(**self).get_mtime(path).await
	}

	async fn copy_file_streaming(
		&self, source: &Path, dest: &Path, progress_callback: Option<std::sync::Arc<dyn Fn(u64, u64) + Send + Sync>>,
	) -> Result<TransferResult> {
		(**self).copy_file_streaming(source, dest, progress_callback).await
	}

	async fn check_disk_space(&self, path: &Path, bytes_needed: u64) -> Result<()> {
		(**self).check_disk_space(path, bytes_needed).await
	}

	async fn set_xattrs(&self, path: &Path, xattrs: &[(String, Vec<u8>)]) -> Result<()> {
		(**self).set_xattrs(path, xattrs).await
	}

	async fn set_acls(&self, path: &Path, acls_text: &str) -> Result<()> {
		(**self).set_acls(path, acls_text).await
	}

	async fn set_bsd_flags(&self, path: &Path, flags: u32) -> Result<()> {
		(**self).set_bsd_flags(path, flags).await
	}

	async fn bulk_copy_files(&self, source_base: &Path, dest_base: &Path, relative_paths: &[&Path]) -> Result<u64> {
		(**self).bulk_copy_files(source_base, dest_base, relative_paths).await
	}
}