//! High-performance data server for binary market data files.
//!
//! Provides thread-safe (`Arc`-based) loading and caching of tick data,
//! designed to serve hundreds of single-threaded VMs in parallel without
//! redundant I/O. Each file is read and parsed once, then shared across
//! all consumers via `Arc<[T]>` chunk references.
//!
//! # Architecture
//!
//! ```text
//!           Arc<DataServer>
//!           ┌──────────────────────────┐
//!           │ SymbolIndex   FileCache  │
//!           │ (HashMap)     (RwLock)   │
//!           └─────┬────────────┬───────┘
//!                 │            │
//!     VM Thread 1 │  VM Thread N
//!     Generator ──┘  Generator ──┘
//! ```

pub mod cache;
pub mod loader;
pub mod records;

use cache::{ChunkVec, FileCache, FileKey};
use records::{DataFormat, M1Parsed, TickParsed};

use regex::Regex;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, OnceLock};

/// Global data server instance, initialized once before any VM threads start.
static DATA_SERVER: OnceLock<Arc<DataServer>> = OnceLock::new();

/// Initializes the global data server for the given base path.
///
/// Scans the directory for data files and builds the symbol index.
/// Must be called once before any `get_data_server()` calls (typically
/// at program startup). Returns the `Arc` for optional local use.
///
/// # Panics
///
/// Panics if called more than once (the `OnceLock` is already set).
pub fn init_data_server(base_path: impl AsRef<Path>) -> Arc<DataServer> {
    let server = Arc::new(DataServer::new(base_path));
    DATA_SERVER
        .set(server.clone())
        .expect("DataServer already initialized");
    server
}

/// Returns the global data server, or `None` if not yet initialized.
pub fn get_data_server() -> Option<Arc<DataServer>> {
    DATA_SERVER.get().cloned()
}

/// Pre-scanned directory index mapping symbols to their sorted data files.
struct SymbolIndex {
    base_path: PathBuf,
    /// Symbol name (e.g. "EURUSD") → sorted list of file keys (chronological).
    symbols: HashMap<Arc<str>, Vec<FileKey>>,
}

impl SymbolIndex {
    /// Scans `base_path` for files matching `SYMBOL_YYYY_MM.(m1|tick)` and
    /// builds a grouped, chronologically sorted index.
    ///
    /// Mirrors Delphi's `TDataServer.UpdateSymbols`.
    fn scan(base_path: &Path) -> Self {
        let re = Regex::new(r"^(.+)_(\d{4})_(\d{2})\.(m1|tick)$").unwrap();
        let mut map: HashMap<Arc<str>, Vec<FileKey>> = HashMap::new();

        if let Ok(entries) = std::fs::read_dir(base_path) {
            for entry in entries.flatten() {
                let file_name = entry.file_name();
                let name = file_name.to_string_lossy();

                if let Some(caps) = re.captures(&name) {
                    let symbol: Arc<str> = Arc::from(&caps[1]);
                    let year: u16 = caps[2].parse().unwrap_or(0);
                    let month: u8 = caps[3].parse().unwrap_or(0);
                    let format = match &caps[4] {
                        "m1" => DataFormat::M1,
                        "tick" => DataFormat::Tick,
                        _ => continue,
                    };

                    if year > 0 && (1..=12).contains(&month) {
                        let key = FileKey {
                            symbol: symbol.clone(),
                            year,
                            month,
                            format,
                        };
                        map.entry(symbol).or_default().push(key);
                    }
                }
            }
        }

        // Sort each symbol's files chronologically (year, month, format)
        for files in map.values_mut() {
            files.sort_by(|a, b| {
                (a.year, a.month, a.format as u8).cmp(&(b.year, b.month, b.format as u8))
            });
        }

        Self {
            base_path: base_path.to_path_buf(),
            symbols: map,
        }
    }

    /// Returns the filesystem path for a given file key.
    fn file_path(&self, key: &FileKey) -> PathBuf {
        let ext = match key.format {
            DataFormat::M1 => "m1",
            DataFormat::Tick => "tick",
        };
        self.base_path
            .join(format!("{}_{:04}_{:02}.{}", key.symbol, key.year, key.month, ext))
    }
}

/// High-performance data server for binary market data files.
///
/// Thread-safe (`Send + Sync`). Designed to be wrapped in `Arc` and shared
/// across hundreds of VM threads. Each file is loaded and parsed once,
/// then served from the in-memory cache.
pub struct DataServer {
    index: SymbolIndex,
    cache: FileCache,
}

impl std::fmt::Debug for DataServer {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "DataServer({:?}, {} symbols)",
            self.index.base_path,
            self.index.symbols.len()
        )
    }
}

impl DataServer {
    /// Creates a new data server, scanning the base path for data files.
    pub fn new(base_path: impl AsRef<Path>) -> Self {
        Self {
            index: SymbolIndex::scan(base_path.as_ref()),
            cache: FileCache::new(),
        }
    }

    /// Returns a sorted list of all known symbol names.
    pub fn symbols(&self) -> Vec<Arc<str>> {
        let mut syms: Vec<_> = self.index.symbols.keys().cloned().collect();
        syms.sort();
        syms
    }

    /// Returns the number of data files for a given symbol and format,
    /// or `None` if the symbol is unknown.
    pub fn file_count(&self, symbol: &str, format: DataFormat) -> Option<usize> {
        self.index.symbols.get(symbol).map(|files| {
            files.iter().filter(|f| f.format == format).count()
        })
    }

    /// Returns a chunk iterator over all M1 data for a symbol.
    ///
    /// The iterator yields `Arc<[M1Parsed]>` chunks in chronological order,
    /// prefetching the next file while the current one is consumed.
    pub fn stream_m1(self: &Arc<Self>, symbol: &str) -> Option<SymbolChunkIter<M1Parsed>> {
        let files: Vec<FileKey> = self
            .index
            .symbols
            .get(symbol)?
            .iter()
            .filter(|f| f.format == DataFormat::M1)
            .cloned()
            .collect();

        if files.is_empty() {
            return None;
        }

        // Eagerly load the first file and prefetch the second
        self.ensure_m1_loaded(&files[0]);
        if files.len() > 1 {
            self.prefetch_m1(&files[1]);
        }

        Some(SymbolChunkIter {
            server: Arc::clone(self),
            files,
            file_idx: 0,
            chunk_idx: 0,
            current_file_chunks: None,
        })
    }

    /// Returns a chunk iterator over all tick data for a symbol.
    pub fn stream_tick(self: &Arc<Self>, symbol: &str) -> Option<SymbolChunkIter<TickParsed>> {
        let files: Vec<FileKey> = self
            .index
            .symbols
            .get(symbol)?
            .iter()
            .filter(|f| f.format == DataFormat::Tick)
            .cloned()
            .collect();

        if files.is_empty() {
            return None;
        }

        self.ensure_tick_loaded(&files[0]);
        if files.len() > 1 {
            self.prefetch_tick(&files[1]);
        }

        Some(SymbolChunkIter {
            server: Arc::clone(self),
            files,
            file_idx: 0,
            chunk_idx: 0,
            current_file_chunks: None,
        })
    }

    // -- M1 loading ----------------------------------------------------------

    /// Ensures the given M1 file is loaded into the cache (blocking).
    fn ensure_m1_loaded(&self, key: &FileKey) {
        if self.cache.is_cached(key) {
            return;
        }
        if self.cache.is_loading(key) {
            self.cache.wait_for_load(key);
            if self.cache.is_cached(key) {
                return;
            }
        }
        self.load_m1_sync(key);
    }

    /// Synchronously loads an M1 file into the cache.
    fn load_m1_sync(&self, key: &FileKey) {
        if !self.cache.try_claim_loading(key) {
            // Another thread is loading or already cached — wait for it
            self.cache.wait_for_load(key);
            return;
        }
        let path = self.index.file_path(key);
        match loader::load_m1_file(&path) {
            Ok(chunks) => self.cache.insert_m1(key.clone(), chunks),
            Err(e) => {
                eprintln!("Warning: failed to load {}: {}", path.display(), e);
                // Insert empty to unblock waiters and prevent retries
                self.cache.insert_m1(key.clone(), vec![]);
            }
        }
    }

    /// Prefetches an M1 file in a background thread.
    ///
    /// Requires `Arc<Self>` to safely share the server with the spawned thread.
    fn prefetch_m1(self: &Arc<Self>, key: &FileKey) {
        if self.cache.is_cached(key) || !self.cache.try_claim_loading(key) {
            return;
        }
        let path = self.index.file_path(key);
        let cache_key = key.clone();
        let server = Arc::clone(self);
        std::thread::spawn(move || {
            match loader::load_m1_file(&path) {
                Ok(chunks) => server.cache.insert_m1(cache_key, chunks),
                Err(e) => {
                    eprintln!("Warning: prefetch failed for {}: {}", path.display(), e);
                    server.cache.insert_m1(cache_key, vec![]);
                }
            }
        });
    }

    // -- Tick loading --------------------------------------------------------

    /// Ensures the given tick file is loaded into the cache (blocking).
    fn ensure_tick_loaded(&self, key: &FileKey) {
        if self.cache.is_cached(key) {
            return;
        }
        if self.cache.is_loading(key) {
            self.cache.wait_for_load(key);
            if self.cache.is_cached(key) {
                return;
            }
        }
        self.load_tick_sync(key);
    }

    /// Synchronously loads a tick file into the cache.
    fn load_tick_sync(&self, key: &FileKey) {
        if !self.cache.try_claim_loading(key) {
            self.cache.wait_for_load(key);
            return;
        }
        let path = self.index.file_path(key);
        match loader::load_tick_file(&path) {
            Ok(chunks) => self.cache.insert_tick(key.clone(), chunks),
            Err(e) => {
                eprintln!("Warning: failed to load {}: {}", path.display(), e);
                self.cache.insert_tick(key.clone(), vec![]);
            }
        }
    }

    /// Prefetches a tick file in a background thread.
    fn prefetch_tick(self: &Arc<Self>, key: &FileKey) {
        if self.cache.is_cached(key) || !self.cache.try_claim_loading(key) {
            return;
        }
        let path = self.index.file_path(key);
        let cache_key = key.clone();
        let server = Arc::clone(self);
        std::thread::spawn(move || {
            match loader::load_tick_file(&path) {
                Ok(chunks) => server.cache.insert_tick(cache_key, chunks),
                Err(e) => {
                    eprintln!("Warning: prefetch failed for {}: {}", path.display(), e);
                    server.cache.insert_tick(cache_key, vec![]);
                }
            }
        });
    }
}

/// Iterator yielding chunks of parsed records for a single symbol.
///
/// Traverses all files for the symbol in chronological order, yielding
/// each 1024-record chunk as an `Arc<[T]>`. When advancing to the next
/// file, the file after that is prefetched in the background.
///
/// This is the Rust equivalent of Delphi's `ProcessFile` chain with its
/// `LoadDataFile(nextFileInfo)` prefetch call.
pub struct SymbolChunkIter<T> {
    server: Arc<DataServer>,
    files: Vec<FileKey>,
    file_idx: usize,
    chunk_idx: usize,
    /// Cached reference to the current file's chunks to avoid repeated
    /// cache lookups within the same file.
    current_file_chunks: Option<ChunkVec<T>>,
}

impl SymbolChunkIter<M1Parsed> {
    /// Returns the next chunk of M1 records, or `None` when exhausted.
    pub fn next_chunk(&mut self) -> Option<Arc<[M1Parsed]>> {
        loop {
            // Try to get chunks for the current file
            if self.current_file_chunks.is_none() && self.file_idx < self.files.len() {
                let key = &self.files[self.file_idx];
                self.server.ensure_m1_loaded(key);
                self.current_file_chunks = self.server.cache.get_m1(key);
                self.chunk_idx = 0;
            }

            let chunks = self.current_file_chunks.as_ref()?;

            if self.chunk_idx < chunks.len() {
                let chunk = chunks[self.chunk_idx].clone();
                self.chunk_idx += 1;
                return Some(chunk);
            }

            // Current file exhausted — advance to next file
            self.file_idx += 1;
            self.current_file_chunks = None;

            if self.file_idx >= self.files.len() {
                return None; // All files exhausted
            }

            // Prefetch the file AFTER the next one (N+2), since N+1 was
            // already prefetched when we started N.
            if self.file_idx + 1 < self.files.len() {
                self.server.prefetch_m1(&self.files[self.file_idx + 1]);
            }
        }
    }
}

impl SymbolChunkIter<TickParsed> {
    /// Returns the next chunk of tick records, or `None` when exhausted.
    pub fn next_chunk(&mut self) -> Option<Arc<[TickParsed]>> {
        loop {
            if self.current_file_chunks.is_none() && self.file_idx < self.files.len() {
                let key = &self.files[self.file_idx];
                self.server.ensure_tick_loaded(key);
                self.current_file_chunks = self.server.cache.get_tick(key);
                self.chunk_idx = 0;
            }

            let chunks = self.current_file_chunks.as_ref()?;

            if self.chunk_idx < chunks.len() {
                let chunk = chunks[self.chunk_idx].clone();
                self.chunk_idx += 1;
                return Some(chunk);
            }

            self.file_idx += 1;
            self.current_file_chunks = None;

            if self.file_idx >= self.files.len() {
                return None;
            }

            if self.file_idx + 1 < self.files.len() {
                self.server.prefetch_tick(&self.files[self.file_idx + 1]);
            }
        }
    }
}

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

    #[test]
    fn test_symbol_index_scan() {
        // Create a temp directory with mock files
        let dir = tempfile::tempdir().unwrap();
        let names = [
            "EURUSD_2017_01.m1",
            "EURUSD_2017_02.m1",
            "EURUSD_2017_03.m1",
            "EURUSD_2017_01.tick",
            "GBPUSD_2020_06.m1",
            "invalid_file.txt",
            "AAPL.US_2006_08.m1",
        ];
        for name in &names {
            std::fs::write(dir.path().join(name), b"").unwrap();
        }

        let index = SymbolIndex::scan(dir.path());

        // EURUSD has both m1 and tick files
        let eur = index.symbols.get("EURUSD").unwrap();
        assert_eq!(eur.len(), 4); // 3 m1 + 1 tick
        // Should be sorted chronologically, M1 before Tick within same month
        assert_eq!(eur[0].month, 1);
        assert_eq!(eur[0].format, DataFormat::M1);
        // The tick file for month 1 comes right after the m1 file for month 1
        assert_eq!(eur[1].month, 1);
        assert_eq!(eur[1].format, DataFormat::Tick);

        // AAPL.US with dot in symbol name
        let aapl = index.symbols.get("AAPL.US").unwrap();
        assert_eq!(aapl.len(), 1);
        assert_eq!(aapl[0].year, 2006);

        // GBPUSD
        assert!(index.symbols.contains_key("GBPUSD"));

        // Invalid file should not appear
        assert!(!index.symbols.contains_key("invalid_file"));
    }

    #[test]
    fn test_file_path_construction() {
        let index = SymbolIndex {
            base_path: PathBuf::from("/mnt/tickdata/Pepperstone"),
            symbols: HashMap::new(),
        };

        let key = FileKey {
            symbol: Arc::from("EURUSD"),
            year: 2017,
            month: 3,
            format: DataFormat::M1,
        };

        assert_eq!(
            index.file_path(&key),
            PathBuf::from("/mnt/tickdata/Pepperstone/EURUSD_2017_03.m1")
        );

        let tick_key = FileKey {
            symbol: Arc::from("AAPL.US"),
            year: 2006,
            month: 8,
            format: DataFormat::Tick,
        };

        assert_eq!(
            index.file_path(&tick_key),
            PathBuf::from("/mnt/tickdata/Pepperstone/AAPL.US_2006_08.tick")
        );
    }

    #[test]
    fn test_data_server_symbols() {
        let dir = tempfile::tempdir().unwrap();
        for name in ["EURUSD_2017_01.m1", "GBPUSD_2020_06.m1"] {
            std::fs::write(dir.path().join(name), b"").unwrap();
        }

        let server = DataServer::new(dir.path());
        let syms = server.symbols();
        assert_eq!(syms.len(), 2);
        assert_eq!(syms[0].as_ref(), "EURUSD");
        assert_eq!(syms[1].as_ref(), "GBPUSD");
    }
}