msy 0.4.2

//! Channel types for streaming sync pipeline.
//!
//! Three-task pipeline: Generator -> Sender -> Receiver
//! Using bounded channels for backpressure.

use crate::streaming::protocol::BlockChecksum;
use bytes::Bytes;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::sync::mpsc;

/// Channel size for Generator -> Sender (file entries)
pub const GENERATOR_CHANNEL_SIZE: usize = 1024;

/// Channel size for Sender -> Receiver (data chunks)
pub const SENDER_CHANNEL_SIZE: usize = 64;

/// Data chunk size for transfer
pub const DATA_CHUNK_SIZE: usize = 256 * 1024; // 256KB

/// Maximum delta chunk size (16MB - well under 64MB frame limit)
pub const DELTA_CHUNK_SIZE: usize = 16 * 1024 * 1024;

/// Minimum file size for delta sync
pub const DELTA_MIN_SIZE: u64 = 64 * 1024; // 64KB

// =============================================================================
// FileJob: Generator -> Sender
// =============================================================================

/// A file job sent from Generator to Sender.
/// Contains all information needed to read and transfer the file.
#[derive(Debug, Clone)]
pub struct FileJob {
    /// Path relative to sync root
    pub path: Arc<PathBuf>,

    /// File size in bytes
    pub size: u64,

    /// Modification time (Unix timestamp)
    pub mtime: i64,

    /// File mode/permissions
    pub mode: u32,

    /// Inode number (for hard link detection)
    pub inode: u64,

    /// Whether this file needs delta transfer
    pub need_delta: bool,

    /// Block checksums from destination (for delta computation)
    /// Only present if need_delta is true and file exists on dest
    pub checksums: Option<DeltaInfo>,
}

/// Delta information from destination file
#[derive(Debug, Clone)]
pub struct DeltaInfo {
    /// Block size used for checksums
    pub block_size: u32,

    /// Destination file size (needed to calculate last block size)
    pub file_size: u64,

    /// Block checksums
    pub checksums: Vec<BlockChecksum>,
}

// =============================================================================
// DataChunk: Sender -> wire
// =============================================================================

/// A chunk of file data ready for transmission.
#[derive(Debug)]
pub struct DataChunk {
    /// Path relative to sync root
    pub path: Arc<PathBuf>,

    /// Offset within file
    pub offset: u64,

    /// Data content
    pub data: Bytes,

    /// Whether this is the final chunk for this file
    pub is_final: bool,

    /// Whether this is delta data
    pub is_delta: bool,

    /// Whether the data is compressed
    pub is_compressed: bool,
}

// =============================================================================
// Pipeline messages
// =============================================================================

/// Message from Generator to the rest of the pipeline
#[derive(Debug)]
pub enum GeneratorMessage {
    /// A file that needs to be transferred
    File(FileJob),

    /// A directory that needs to be created
    Mkdir { path: Arc<PathBuf>, mode: u32 },

    /// A symlink that needs to be created
    Symlink { path: Arc<PathBuf>, target: String },

    /// A file or directory that needs to be deleted
    Delete { path: Arc<PathBuf>, is_dir: bool },

    /// End of file list - no more files coming
    FileEnd { total_files: u64, total_bytes: u64 },

    /// End of deletes
    DeleteEnd { count: u64 },
}

/// Sync direction
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SyncDirection {
    /// Local -> Remote (push)
    Push,
    /// Remote -> Local (pull)
    Pull,
}

// =============================================================================
// Channel types
// =============================================================================

/// Sender for file jobs from Generator
pub type FileJobSender = mpsc::Sender<GeneratorMessage>;

/// Receiver for file jobs in Sender task
pub type FileJobReceiver = mpsc::Receiver<GeneratorMessage>;

/// Create a bounded channel for Generator -> Sender communication
pub fn file_job_channel() -> (FileJobSender, FileJobReceiver) {
    mpsc::channel(GENERATOR_CHANNEL_SIZE)
}

// =============================================================================
// Destination state (from Initial Exchange)
// =============================================================================

/// State of a destination file, received during Initial Exchange
#[derive(Debug, Clone)]
pub struct DestFileState {
    /// File size
    pub size: u64,

    /// Modification time
    pub mtime: i64,

    /// File mode
    pub mode: u32,

    /// Whether this is a directory
    pub is_dir: bool,

    /// Block checksums for delta (if file is a delta candidate)
    pub delta_info: Option<DeltaInfo>,
}

/// Destination file index, built during Initial Exchange
#[derive(Debug, Default)]
pub struct DestIndex {
    /// Map of path -> dest state
    files: std::collections::HashMap<String, DestFileState>,
}

impl DestIndex {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn insert(&mut self, path: String, state: DestFileState) {
        self.files.insert(path, state);
    }

    pub fn get(&self, path: &str) -> Option<&DestFileState> {
        self.files.get(path)
    }

    pub fn remove(&mut self, path: &str) -> Option<DestFileState> {
        self.files.remove(path)
    }

    pub fn contains(&self, path: &str) -> bool {
        self.files.contains_key(path)
    }

    /// Get all remaining paths (for delete detection)
    pub fn remaining_paths(&self) -> impl Iterator<Item = (&String, &DestFileState)> {
        self.files.iter()
    }

    pub fn len(&self) -> usize {
        self.files.len()
    }

    pub fn is_empty(&self) -> bool {
        self.files.is_empty()
    }
}

// =============================================================================
// Sync statistics
// =============================================================================

/// Statistics for a sync operation
#[derive(Debug, Default, Clone)]
pub struct SyncStats {
    /// Files successfully transferred
    pub files_ok: u64,

    /// Files that failed
    pub files_err: u64,

    /// Total bytes transferred
    pub bytes_transferred: u64,

    /// Files transferred via delta
    pub delta_files: u64,

    /// Bytes saved by delta transfer
    pub delta_bytes_saved: u64,

    /// Directories created
    pub dirs_created: u64,

    /// Symlinks created
    pub symlinks_created: u64,

    /// Files/directories deleted
    pub deleted: u64,

    /// Hard links created
    pub hardlinks_created: u64,
}

impl SyncStats {
    pub fn new() -> Self {
        Self::default()
    }
}

// =============================================================================
// Tests
// =============================================================================

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_dest_index() {
        let mut index = DestIndex::new();

        index.insert(
            "file.txt".to_string(),
            DestFileState {
                size: 1024,
                mtime: 1234567890,
                mode: 0o644,
                is_dir: false,
                delta_info: None,
            },
        );

        assert!(index.contains("file.txt"));
        assert!(!index.contains("other.txt"));

        let state = index.get("file.txt").unwrap();
        assert_eq!(state.size, 1024);

        index.remove("file.txt");
        assert!(!index.contains("file.txt"));
    }

    #[tokio::test]
    async fn test_file_job_channel() {
        let (tx, mut rx) = file_job_channel();

        let job = GeneratorMessage::File(FileJob {
            path: Arc::new(PathBuf::from("test.txt")),
            size: 100,
            mtime: 0,
            mode: 0o644,
            inode: 0,
            need_delta: false,
            checksums: None,
        });

        tx.send(job).await.unwrap();
        drop(tx);

        let received = rx.recv().await.unwrap();
        match received {
            GeneratorMessage::File(job) => {
                assert_eq!(job.path.as_ref(), &PathBuf::from("test.txt"));
                assert_eq!(job.size, 100);
            }
            _ => panic!("Expected File message"),
        }
    }

    #[test]
    fn test_sync_stats() {
        let mut stats = SyncStats::new();
        stats.files_ok = 100;
        stats.bytes_transferred = 1024 * 1024;
        stats.delta_files = 10;
        stats.delta_bytes_saved = 512 * 1024;

        assert_eq!(stats.files_ok, 100);
        assert_eq!(stats.delta_files, 10);
    }
}