use crate::error::AedbError;
use parking_lot::{Mutex, RwLock};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, VecDeque};
use std::fs::{File, OpenOptions};
use std::io::{ErrorKind, Read, Seek, SeekFrom, Write};
#[cfg(unix)]
use std::os::unix::fs::FileExt;
#[cfg(windows)]
use std::os::windows::fs::FileExt;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};

const PAGE_STORE_MAGIC: &[u8; 8] = b"AEDBPG1\0";
const PAGE_FRAME_HEADER_BYTES: usize = 48;
const MIN_PAGE_BYTES: usize = 256;
const MAX_PAGE_BYTES: usize = 16 * 1024 * 1024;
const PAGE_STORE_WRITE_BUFFER_BYTES: usize = 4 * 1024 * 1024;
const PAGE_STORE_ZERO_BUFFER_BYTES: usize = 16 * 1024;

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct PageId(pub u64);

#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct PageRef {
    pub page_id: PageId,
    pub len: u32,
    pub blake3_hash: [u8; 32],
}

#[derive(Debug)]
pub struct PagedStore {
    path: PathBuf,
    file: Mutex<File>,
    read_file: File,
    page_size: usize,
    page_stride: u64,
    cache_capacity_pages: usize,
    page_count: AtomicU64,
    dirty: AtomicBool,
    cache: RwLock<PageCache>,
}

impl PagedStore {
    pub fn open(
        data_dir: &Path,
        file_name: &str,
        page_size: usize,
        cache_capacity_pages: usize,
    ) -> Result<Self, AedbError> {
        if !(MIN_PAGE_BYTES..=MAX_PAGE_BYTES).contains(&page_size) {
            return Err(AedbError::InvalidConfig {
                message: format!(
                    "page_size must be between {MIN_PAGE_BYTES} and {MAX_PAGE_BYTES} bytes"
                ),
            });
        }
        if file_name.contains('/') || file_name.contains('\\') || file_name.is_empty() {
            return Err(AedbError::InvalidConfig {
                message: "page store file_name must be a single path segment".into(),
            });
        }

        std::fs::create_dir_all(data_dir)?;
        let path = data_dir.join(file_name);
        let mut file = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(&path)?;
        if file.metadata()?.len() == 0 {
            file.write_all(PAGE_STORE_MAGIC)?;
            file.flush()?;
            file.sync_all()?;
        } else {
            let mut magic = [0u8; PAGE_STORE_MAGIC.len()];
            file.seek(SeekFrom::Start(0))?;
            file.read_exact(&mut magic)?;
            if &magic != PAGE_STORE_MAGIC {
                return Err(AedbError::IntegrityError {
                    message: "page store magic mismatch".into(),
                });
            }
        }

        let page_stride = u64::try_from(PAGE_FRAME_HEADER_BYTES.saturating_add(page_size))
            .map_err(|_| AedbError::InvalidConfig {
                message: "page store stride does not fit u64".into(),
            })?;
        let file_len = file.metadata()?.len();
        let payload_len = file_len
            .checked_sub(PAGE_STORE_MAGIC.len() as u64)
            .ok_or_else(|| AedbError::IntegrityError {
                message: "page store file shorter than magic header".into(),
            })?;
        if payload_len % page_stride != 0 {
            return Err(AedbError::IntegrityError {
                message: "page store contains a partial page frame".into(),
            });
        }
        let page_count = payload_len / page_stride;
        let read_file = file.try_clone()?;
        Ok(Self {
            path,
            file: Mutex::new(file),
            read_file,
            page_size,
            page_stride,
            cache_capacity_pages,
            page_count: AtomicU64::new(page_count),
            dirty: AtomicBool::new(false),
            cache: RwLock::new(PageCache::new(cache_capacity_pages)),
        })
    }

    pub fn append_page(&self, payload: &[u8]) -> Result<PageRef, AedbError> {
        let page_len = self.validate_payload(payload)?;
        let page_ref = {
            let mut file = self.file.lock();
            let page_id = PageId(self.page_count.load(Ordering::Acquire));
            let blake3_hash = page_hash(page_id, payload);

            let mut header = [0u8; PAGE_FRAME_HEADER_BYTES];
            header[0..8].copy_from_slice(&page_id.0.to_le_bytes());
            header[8..12].copy_from_slice(&page_len.to_le_bytes());
            header[16..48].copy_from_slice(&blake3_hash);

            file.seek(SeekFrom::End(0))?;
            file.write_all(&header)?;
            file.write_all(payload)?;
            write_zero_padding(&mut *file, self.page_size.saturating_sub(payload.len()))?;
            file.flush()?;
            let next_page_count =
                page_id
                    .0
                    .checked_add(1)
                    .ok_or_else(|| AedbError::IntegrityError {
                        message: "page store page id overflow".into(),
                    })?;
            self.page_count.store(next_page_count, Ordering::Release);
            PageRef {
                page_id,
                len: page_len,
                blake3_hash,
            }
        };
        self.dirty.store(true, Ordering::Release);
        if self.cache_capacity_pages > 0 {
            self.cache
                .write()
                .insert(page_ref.page_id, page_ref.blake3_hash, payload);
        }
        Ok(page_ref)
    }

    pub fn append_pages(&self, payloads: &[&[u8]]) -> Result<Vec<PageRef>, AedbError> {
        if payloads.is_empty() {
            return Ok(Vec::new());
        }
        if let [payload] = payloads {
            return self.append_page(payload).map(|page_ref| vec![page_ref]);
        }
        let mut page_lens = Vec::with_capacity(payloads.len());
        for payload in payloads {
            page_lens.push(self.validate_payload(payload)?);
        }

        let mut refs = Vec::with_capacity(payloads.len());
        {
            let mut file = self.file.lock();
            let mut next_page_id = self.page_count.load(Ordering::Acquire);
            let write_buf_capacity = PAGE_STORE_WRITE_BUFFER_BYTES
                .min(
                    PAGE_FRAME_HEADER_BYTES
                        .saturating_add(self.page_size)
                        .saturating_mul(payloads.len()),
                )
                .max(PAGE_FRAME_HEADER_BYTES);
            let mut write_buf = Vec::with_capacity(write_buf_capacity);
            file.seek(SeekFrom::End(0))?;
            for (payload, page_len) in payloads.iter().zip(page_lens) {
                let page_id = PageId(next_page_id);
                let blake3_hash = page_hash(page_id, payload);
                let mut header = [0u8; PAGE_FRAME_HEADER_BYTES];
                header[0..8].copy_from_slice(&page_id.0.to_le_bytes());
                header[8..12].copy_from_slice(&page_len.to_le_bytes());
                header[16..48].copy_from_slice(&blake3_hash);
                append_bytes_to_writer(&mut *file, &mut write_buf, &header)?;
                append_bytes_to_writer(&mut *file, &mut write_buf, payload)?;
                append_zero_padding_to_writer(
                    &mut *file,
                    &mut write_buf,
                    self.page_size.saturating_sub(payload.len()),
                )?;
                refs.push(PageRef {
                    page_id,
                    len: page_len,
                    blake3_hash,
                });
                next_page_id =
                    next_page_id
                        .checked_add(1)
                        .ok_or_else(|| AedbError::IntegrityError {
                            message: "page store page id overflow".into(),
                        })?;
            }
            flush_write_buffer(&mut *file, &mut write_buf)?;
            file.flush()?;
            self.page_count.store(next_page_id, Ordering::Release);
        }
        self.dirty.store(true, Ordering::Release);
        if self.cache_capacity_pages > 0 {
            let cache_start = refs.len().saturating_sub(self.cache_capacity_pages);
            let mut cache = self.cache.write();
            for (page_ref, payload) in refs[cache_start..]
                .iter()
                .zip(payloads[cache_start..].iter())
            {
                cache.insert(page_ref.page_id, page_ref.blake3_hash, payload);
            }
        }
        Ok(refs)
    }

    fn validate_payload(&self, payload: &[u8]) -> Result<u32, AedbError> {
        if payload.len() > self.page_size {
            return Err(AedbError::Validation(format!(
                "page payload exceeds page_size: payload_bytes={}, page_size={}",
                payload.len(),
                self.page_size
            )));
        }
        page_len_u32(payload)
    }

    pub fn read_page(&self, page_ref: &PageRef) -> Result<Vec<u8>, AedbError> {
        if self.cache_capacity_pages > 0
            && let Some((page, cached_hash)) = self.cache.read().get(&page_ref.page_id)
        {
            if cached_hash != page_ref.blake3_hash {
                return Err(AedbError::IntegrityError {
                    message: "cached page hash mismatch".into(),
                });
            }
            return Ok(page.as_ref().to_vec());
        }

        let current_count = self.page_count.load(Ordering::Acquire);
        if page_ref.page_id.0 >= current_count {
            return Err(AedbError::IntegrityError {
                message: "page reference outside page store".into(),
            });
        }
        let expected_len =
            usize::try_from(page_ref.len).map_err(|_| AedbError::IntegrityError {
                message: "page reference length does not fit usize".into(),
            })?;
        if expected_len > self.page_size {
            return Err(AedbError::IntegrityError {
                message: "page reference length exceeds page size".into(),
            });
        }
        let frame_offset = frame_offset(page_ref.page_id, self.page_stride)?;
        let mut header = [0u8; PAGE_FRAME_HEADER_BYTES];
        read_exact_at(&self.read_file, &mut header, frame_offset)?;
        let stored_page_id =
            u64::from_le_bytes(
                header[0..8]
                    .try_into()
                    .map_err(|_| AedbError::IntegrityError {
                        message: "invalid page id header".into(),
                    })?,
            );
        if stored_page_id != page_ref.page_id.0 {
            return Err(AedbError::IntegrityError {
                message: "page id header mismatch".into(),
            });
        }
        let stored_len = u32::from_le_bytes(header[8..12].try_into().map_err(|_| {
            AedbError::IntegrityError {
                message: "invalid page length header".into(),
            }
        })?);
        if stored_len != page_ref.len || stored_len as usize > self.page_size {
            return Err(AedbError::IntegrityError {
                message: "page length header mismatch".into(),
            });
        }
        let mut stored_hash = [0u8; 32];
        stored_hash.copy_from_slice(&header[16..48]);
        if stored_hash != page_ref.blake3_hash {
            return Err(AedbError::IntegrityError {
                message: "page hash header mismatch".into(),
            });
        }
        let mut payload = vec![0u8; expected_len];
        read_exact_at(
            &self.read_file,
            &mut payload,
            frame_offset + PAGE_FRAME_HEADER_BYTES as u64,
        )?;
        if page_hash(page_ref.page_id, &payload) != page_ref.blake3_hash {
            return Err(AedbError::IntegrityError {
                message: "page payload hash mismatch".into(),
            });
        }
        if self.cache_capacity_pages > 0 {
            self.cache
                .write()
                .insert(page_ref.page_id, page_ref.blake3_hash, &payload);
        }
        Ok(payload)
    }

    pub fn sync_all(&self) -> Result<(), AedbError> {
        if !self.dirty.load(Ordering::Acquire) {
            return Ok(());
        }
        let file = self.file.lock();
        if !self.dirty.load(Ordering::Acquire) {
            return Ok(());
        }
        file.sync_all()?;
        self.dirty.store(false, Ordering::Release);
        Ok(())
    }

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

    pub fn page_count(&self) -> u64 {
        self.page_count.load(Ordering::Acquire)
    }

    pub fn len_bytes(&self) -> u64 {
        PAGE_STORE_MAGIC.len() as u64 + self.page_count() * self.page_stride
    }

    pub fn cache_resident_pages(&self) -> usize {
        self.cache.read().resident_pages()
    }

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

    pub fn path(&self) -> &Path {
        &self.path
    }
}

#[cfg(unix)]
fn read_exact_at(file: &File, mut buf: &mut [u8], mut offset: u64) -> Result<(), AedbError> {
    while !buf.is_empty() {
        match file.read_at(buf, offset) {
            Ok(0) => {
                return Err(AedbError::Io(std::io::Error::new(
                    ErrorKind::UnexpectedEof,
                    "failed to fill whole page buffer",
                )));
            }
            Ok(bytes_read) => {
                offset = offset.checked_add(bytes_read as u64).ok_or_else(|| {
                    AedbError::IntegrityError {
                        message: "page read offset overflow".into(),
                    }
                })?;
                let remaining = buf;
                buf = &mut remaining[bytes_read..];
            }
            Err(err) if err.kind() == ErrorKind::Interrupted => {}
            Err(err) => return Err(AedbError::Io(err)),
        }
    }
    Ok(())
}

#[cfg(windows)]
fn read_exact_at(file: &File, mut buf: &mut [u8], mut offset: u64) -> Result<(), AedbError> {
    while !buf.is_empty() {
        match file.seek_read(buf, offset) {
            Ok(0) => {
                return Err(AedbError::Io(std::io::Error::new(
                    ErrorKind::UnexpectedEof,
                    "failed to fill whole page buffer",
                )));
            }
            Ok(bytes_read) => {
                offset = offset.checked_add(bytes_read as u64).ok_or_else(|| {
                    AedbError::IntegrityError {
                        message: "page read offset overflow".into(),
                    }
                })?;
                let remaining = buf;
                buf = &mut remaining[bytes_read..];
            }
            Err(err) if err.kind() == ErrorKind::Interrupted => {}
            Err(err) => return Err(AedbError::Io(err)),
        }
    }
    Ok(())
}

#[cfg(not(any(unix, windows)))]
compile_error!("PagedStore requires positional file reads for concurrent cold read safety");

fn append_bytes_to_writer<W: Write>(
    writer: &mut W,
    write_buf: &mut Vec<u8>,
    bytes: &[u8],
) -> Result<(), AedbError> {
    if bytes.len() > PAGE_STORE_WRITE_BUFFER_BYTES {
        flush_write_buffer(writer, write_buf)?;
        writer.write_all(bytes)?;
        return Ok(());
    }
    if write_buf.len() + bytes.len() > PAGE_STORE_WRITE_BUFFER_BYTES {
        flush_write_buffer(writer, write_buf)?;
    }
    write_buf.extend_from_slice(bytes);
    Ok(())
}

fn append_zero_padding_to_writer<W: Write>(
    writer: &mut W,
    write_buf: &mut Vec<u8>,
    mut byte_count: usize,
) -> Result<(), AedbError> {
    let zeros = [0u8; PAGE_STORE_ZERO_BUFFER_BYTES];
    while byte_count > 0 {
        let write_count = byte_count.min(zeros.len());
        append_bytes_to_writer(writer, write_buf, &zeros[..write_count])?;
        byte_count -= write_count;
    }
    Ok(())
}

fn flush_write_buffer<W: Write>(writer: &mut W, write_buf: &mut Vec<u8>) -> Result<(), AedbError> {
    if !write_buf.is_empty() {
        writer.write_all(write_buf)?;
        write_buf.clear();
    }
    Ok(())
}

fn write_zero_padding<W: Write>(writer: &mut W, mut byte_count: usize) -> Result<(), AedbError> {
    let zeros = [0u8; PAGE_STORE_ZERO_BUFFER_BYTES];
    while byte_count > 0 {
        let write_count = byte_count.min(zeros.len());
        writer.write_all(&zeros[..write_count])?;
        byte_count -= write_count;
    }
    Ok(())
}

fn frame_offset(page_id: PageId, page_stride: u64) -> Result<u64, AedbError> {
    page_id
        .0
        .checked_mul(page_stride)
        .and_then(|offset| offset.checked_add(PAGE_STORE_MAGIC.len() as u64))
        .ok_or_else(|| AedbError::IntegrityError {
            message: "page frame offset overflow".into(),
        })
}

fn page_len_u32(payload: &[u8]) -> Result<u32, AedbError> {
    u32::try_from(payload.len())
        .map_err(|_| AedbError::Validation("page payload length does not fit u32".into()))
}

fn page_hash(page_id: PageId, payload: &[u8]) -> [u8; 32] {
    let mut hasher = blake3::Hasher::new();
    hasher.update(&page_id.0.to_le_bytes());
    hasher.update(&(payload.len() as u64).to_le_bytes());
    hasher.update(payload);
    *hasher.finalize().as_bytes()
}

#[derive(Debug)]
struct PageCache {
    capacity_pages: usize,
    pages: HashMap<PageId, CachedPage>,
    insertion_order: VecDeque<PageId>,
}

#[derive(Debug)]
struct CachedPage {
    payload: Arc<[u8]>,
    blake3_hash: [u8; 32],
}

impl PageCache {
    fn new(capacity_pages: usize) -> Self {
        Self {
            capacity_pages,
            pages: HashMap::with_capacity(capacity_pages),
            insertion_order: VecDeque::with_capacity(capacity_pages),
        }
    }

    fn resident_pages(&self) -> usize {
        self.pages.len()
    }

    fn get(&self, page_id: &PageId) -> Option<(Arc<[u8]>, [u8; 32])> {
        let cached = self.pages.get(page_id)?;
        Some((Arc::clone(&cached.payload), cached.blake3_hash))
    }

    fn insert(&mut self, page_id: PageId, blake3_hash: [u8; 32], payload: &[u8]) {
        if self.capacity_pages == 0 {
            return;
        }
        if let std::collections::hash_map::Entry::Occupied(mut cached) = self.pages.entry(page_id) {
            cached.insert(CachedPage {
                payload: Arc::from(payload),
                blake3_hash,
            });
            return;
        }
        self.insertion_order.push_back(page_id);
        self.pages.insert(
            page_id,
            CachedPage {
                payload: Arc::from(payload),
                blake3_hash,
            },
        );
        self.evict_to_capacity();
    }

    fn evict_to_capacity(&mut self) {
        while self.pages.len() > self.capacity_pages {
            let Some(page_id) = self.insertion_order.pop_front() else {
                break;
            };
            self.pages.remove(&page_id);
        }
    }
}

#[cfg(test)]
mod tests;
