use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};

#[cfg(test)]
use common::fs::read_json;
use common::save_on_disk::SaveOnDisk;
use fs_err as fs;
use serde::{Deserialize, Serialize};
use shard::files::APPLIED_SEQ_FILE;

use crate::operations::types::CollectionResult;

/// How often the `applied_seq` is persisted
pub const APPLIED_SEQ_SAVE_INTERVAL: u64 = 64;

/// Data structure, used for (de)serialization of the `applied_seq` file
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
struct AppliedSeq {
    pub op_num: u64,
}

impl AppliedSeq {
    fn new(op_num: u64) -> Self {
        Self { op_num }
    }
}

#[derive(Debug)]
pub struct AppliedSeqHandler {
    /// the AppliedSeq atomic file
    file: Option<SaveOnDisk<AppliedSeq>>,
    /// path of the underlying persisted file
    path: PathBuf,
    /// precise in-memory op_num (can be larger than value persisted in `file`)
    op_num: AtomicU64,
    /// tracking update for interval based persistence
    update_count: AtomicU64,
}

impl AppliedSeqHandler {
    /// Get the current in-memory op_num for the last_applied_seq.
    /// The value is likely larger than what is persisted in `file`.
    ///
    ///
    /// Returns None if the handler is not active.
    pub fn op_num(&self) -> Option<u64> {
        if self.file.is_some() {
            Some(self.op_num.load(Ordering::Relaxed))
        } else {
            None
        }
    }

    /// Get the op_num upper bound for the last_applied_seq adjusted to the persistence interval
    ///
    /// Returns None if the handler is not active.
    pub fn op_num_upper_bound(&self) -> Option<u64> {
        if self.file.is_some() {
            let adjusted = self.op_num.load(Ordering::Relaxed) + APPLIED_SEQ_SAVE_INTERVAL + 1;
            Some(adjusted)
        } else {
            None
        }
    }

    /// Path for the applied_seq json file
    pub fn path(&self) -> &Path {
        self.path.as_path()
    }

    #[cfg(test)]
    fn persisted_op_num(&self) -> u64 {
        let persisted = read_json::<AppliedSeq>(&self.path).unwrap().op_num;
        debug_assert!(persisted <= self.op_num.load(Ordering::Relaxed));
        persisted
    }

    /// Test helper: force set the op_num and immediately persist to disk.
    /// This bypasses the interval-based persistence.
    #[cfg(test)]
    pub fn force_set_and_persist(&self, op_num: u64) -> CollectionResult<()> {
        self.op_num.store(op_num, Ordering::Relaxed);
        self.save(op_num)
    }

    /// Load or create the underlying applied seq file.
    pub fn load_or_init(shard_path: &Path, wal_last_index: u64) -> Self {
        let update_count = AtomicU64::new(0);
        let path = shard_path.join(APPLIED_SEQ_FILE);
        let file_was_already_present = path.exists();

        let loaded_file: Result<SaveOnDisk<AppliedSeq>, _> =
            SaveOnDisk::load_or_init(&path, || AppliedSeq::new(wal_last_index));
        match loaded_file {
            Ok(file) => {
                let persisted_applied_seq = file.read().op_num;
                debug_assert!(
                    persisted_applied_seq <= wal_last_index,
                    "last_applied_seq:{persisted_applied_seq} cannot be larger than last_wal_index:{wal_last_index}"
                );
                Self {
                    file: Some(file),
                    path,
                    op_num: AtomicU64::new(persisted_applied_seq),
                    update_count,
                }
            }
            Err(err) => {
                if file_was_already_present {
                    log::error!("Error while loading existing applied_seq at {path:?} {err}");
                    // delete file as it is malformed
                    if let Err(err) = fs::remove_file(&path) {
                        log::error!("Could not delete malformed applied_seq file {path:?} {err}");
                        Self {
                            file: None,
                            path,
                            op_num: AtomicU64::new(wal_last_index),
                            update_count,
                        }
                    } else {
                        // try again to create the file from scratch
                        Self::load_or_init(shard_path, wal_last_index)
                    }
                } else {
                    log::error!("Error while creating new applied_seq at {path:?} {err}");
                    Self {
                        file: None,
                        path,
                        op_num: AtomicU64::new(wal_last_index),
                        update_count,
                    }
                }
            }
        }
    }

    fn save(&self, op_num: u64) -> CollectionResult<()> {
        if let Some(file) = self.file.as_ref() {
            file.write(|current| current.op_num = op_num)?;
        }
        Ok(())
    }

    /// Update the underlying file every `APPLIED_SEQ_SAVE_INTERVAL` updates.
    /// Always update memory representation
    pub fn update(&self, op_num: u64) -> CollectionResult<()> {
        // update in-memory
        self.op_num.store(op_num, Ordering::Relaxed);
        let prev_count = self.update_count.fetch_add(1, Ordering::Relaxed);
        if prev_count == 0 {
            return Ok(());
        }
        // update on disk according to interval to amortize fsync
        if prev_count.is_multiple_of(APPLIED_SEQ_SAVE_INTERVAL) {
            self.save(op_num)?;
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use std::io::Write;

    use tempfile::TempDir;

    use crate::update_workers::applied_seq::{APPLIED_SEQ_SAVE_INTERVAL, AppliedSeqHandler};

    #[test]
    fn nothing_persisted_on_init() {
        let dir = TempDir::with_prefix("applied_seq").unwrap();
        let handler = AppliedSeqHandler::load_or_init(dir.path(), 10);
        assert_eq!(handler.op_num(), Some(10));
        // nothing persisted yet because no updates observed
        assert!(!handler.path().exists());
    }

    #[test]
    fn persists_at_interval() {
        let dir = TempDir::with_prefix("applied_seq").unwrap();
        let handler = AppliedSeqHandler::load_or_init(dir.path(), 1);
        for i in 0..APPLIED_SEQ_SAVE_INTERVAL {
            handler.update(i).unwrap();
            assert_eq!(handler.op_num(), Some(i));
            // nothing persisted yet because less updates than interval
            assert!(!handler.path().exists(), "exists too early at {i}");
        }

        // one more update to reach APPLIED_SEQ_SAVE_INTERVAL
        let op_num = 12345;
        handler.update(op_num).unwrap();
        assert_eq!(handler.op_num(), Some(op_num));
        assert!(handler.path().exists());
        // ensure written to disk
        assert_eq!(handler.persisted_op_num(), op_num);
    }

    #[test]
    fn read_existing_value_on_init() {
        let dir = TempDir::with_prefix("applied_seq").unwrap();
        let handler = AppliedSeqHandler::load_or_init(dir.path(), 1);
        for i in 0..APPLIED_SEQ_SAVE_INTERVAL * 10 {
            handler.update(i).unwrap();
            assert_eq!(handler.op_num(), Some(i));
        }

        let op_num = 640;
        assert_eq!(APPLIED_SEQ_SAVE_INTERVAL * 10, op_num);

        handler.update(op_num).unwrap();
        assert_eq!(handler.op_num(), Some(op_num));
        assert!(handler.path().exists());
        // ensure written to disk
        assert_eq!(handler.persisted_op_num(), op_num);

        // drop and reload
        drop(handler);
        let handler = AppliedSeqHandler::load_or_init(dir.path(), 1000);
        assert_eq!(handler.op_num(), Some(op_num));
    }

    #[test]
    fn handles_file_corruption() {
        let dir = TempDir::with_prefix("applied_seq").unwrap();
        let handler = AppliedSeqHandler::load_or_init(dir.path(), 1);
        let path = handler.path.clone();

        for i in 0..APPLIED_SEQ_SAVE_INTERVAL * 10 {
            handler.update(i).unwrap();
            assert_eq!(handler.op_num(), Some(i));
        }

        let op_num = 640;
        assert_eq!(APPLIED_SEQ_SAVE_INTERVAL * 10, op_num);
        handler.update(op_num).unwrap();
        assert_eq!(handler.op_num(), Some(op_num));
        assert!(handler.path().exists());
        // ensure written to disk
        assert_eq!(handler.persisted_op_num(), op_num);

        // drop and reload
        drop(handler);

        // open in append mode
        let mut file = fs_err::OpenOptions::new()
            .read(true)
            .append(true)
            .open(path)
            .unwrap();

        // corrupt data on disk by appending random bytes
        file.write_all(&[0, 1, 0, 1, 0, 1]).unwrap();
        drop(file);

        // reopen handler without crashing
        let handler = AppliedSeqHandler::load_or_init(dir.path(), 650);

        // regenerate new file with correct WAL op_num
        assert!(handler.file.is_some());
        assert_eq!(handler.op_num(), Some(650));
    }
}
