//! Storage backends for high-availability replication.
//!
//! Provides the StorageBackend trait for persisting write-ahead log entries.

use crate::wal::SagaEntry;
use async_trait::async_trait;
use bincode;
use mimirs_core::MimirError;

/// Trait for pluggable storage backends.
#[async_trait]
pub trait StorageBackend: Send + Sync {
    /// Writes a single entry to the backend.
    async fn write_entry(&self, lsn: u64, entry: &SagaEntry) -> Result<(), MimirError>;

    /// Reads entries from the backend starting at a given LSN.
    async fn read_entries(&self, start_lsn: u64) -> Result<Vec<(u64, SagaEntry)>, MimirError>;

    /// Gets the last LSN persisted in the backend.
    async fn last_lsn(&self) -> Result<u64, MimirError>;
}

/// A file-based storage backend for the HA WAL.
pub struct FileStorageBackend {
    path: std::path::PathBuf,
    last_lsn_cache: std::sync::atomic::AtomicU64,
}

impl FileStorageBackend {
    /// Creates a new FileStorageBackend at the specified path.
    pub fn new(path: impl Into<std::path::PathBuf>) -> Self {
        let path = path.into();
        let last = if path.exists() {
            // Quick scan for last LSN could be done here, but for now we'll start at 0
            // and it will be updated on first write/read.
            0
        } else {
            0
        };
        Self {
            path,
            last_lsn_cache: std::sync::atomic::AtomicU64::new(last),
        }
    }
}

#[async_trait]
impl StorageBackend for FileStorageBackend {
    async fn write_entry(&self, lsn: u64, entry: &SagaEntry) -> Result<(), MimirError> {
        use std::io::Write;
        let mut file = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&self.path)
            .map_err(|e| MimirError::Storage(e.to_string()))?;

        let bytes = bincode::serialize(&(lsn, entry))
            .map_err(|e| MimirError::Serialization(e.to_string()))?;

        file.write_all(&(bytes.len() as u64).to_le_bytes())
            .map_err(|e| MimirError::Storage(e.to_string()))?;
        file.write_all(&bytes)
            .map_err(|e| MimirError::Storage(e.to_string()))?;

        self.last_lsn_cache
            .store(lsn, std::sync::atomic::Ordering::SeqCst);

        Ok(())
    }

    async fn read_entries(&self, start_lsn: u64) -> Result<Vec<(u64, SagaEntry)>, MimirError> {
        use std::io::Read;
        if !self.path.exists() {
            return Ok(Vec::new());
        }

        let mut file =
            std::fs::File::open(&self.path).map_err(|e| MimirError::Storage(e.to_string()))?;

        let mut entries = Vec::new();
        loop {
            let mut len_bytes = [0u8; 8];
            if file.read_exact(&mut len_bytes).is_err() {
                break;
            }
            let len = u64::from_le_bytes(len_bytes);
            let mut bytes = vec![0u8; len as usize];
            file.read_exact(&mut bytes)
                .map_err(|e| MimirError::Storage(e.to_string()))?;

            let (lsn, entry): (u64, SagaEntry) = bincode::deserialize(&bytes)
                .map_err(|e| MimirError::Serialization(e.to_string()))?;

            if lsn >= start_lsn {
                entries.push((lsn, entry));
            }
        }

        Ok(entries)
    }

    async fn last_lsn(&self) -> Result<u64, MimirError> {
        let cached = self
            .last_lsn_cache
            .load(std::sync::atomic::Ordering::SeqCst);
        if cached > 0 {
            return Ok(cached);
        }

        let entries = self.read_entries(0).await?;
        let last = entries.last().map(|(lsn, _)| *lsn).unwrap_or(0);
        self.last_lsn_cache
            .store(last, std::sync::atomic::Ordering::SeqCst);
        Ok(last)
    }
}
