//! The open file every object of a `File` shares.
//!
//! Every read goes through [`Handle::with`], which releases the GIL and
//! parses through `File::storage()`, so the same code serves a local file
//! (memory-mapped), a remote one (`clawhdf5-remote`: range requests through
//! a block cache, so a network read never holds the GIL) and a file open
//! for editing.
//!
//! A file opened with `'r+'` also holds a [`FileEditor`]. An edit takes the
//! file's write lock, so no read runs while the file changes underneath it,
//! and reopens the file afterwards, through the editor's own open file
//! rather than its path: reads after an edit see the new bytes
//! (a grown file, a new dataspace), never a stale mapping or chunk cache.
//! Objects that cache something an edit can change compare
//! [`Handle::generation`] with the value they cached it at.
//!
//! Lock discipline (no deadlock with the GIL): the file lock is only taken
//! with the GIL released, and code that holds it never touches Python.

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, PoisonError, RwLock};

use clawhdf5_rs::{File, FileEditor};
use pyo3::exceptions::PyOSError;
use pyo3::prelude::*;

use crate::{panic_text, to_py_err};

/// Where the file's bytes come from.
pub(crate) enum Source {
    /// A local file (memory-mapped). Its path is not kept: nothing reopens
    /// it by path (see `open_editable`).
    Local,
    /// A URL, read through `clawhdf5-remote`'s block cache.
    Remote {
        url: String,
        storage: Arc<clawhdf5_remote::RemoteStorage>,
    },
}

pub(crate) struct Handle {
    /// The file as last opened; `None` if reopening it after an edit failed
    /// (every read is then an error rather than a read of stale bytes).
    file: RwLock<Option<File>>,
    /// For `'r+'`: the editor, until the file is closed.
    editor: Option<Mutex<Option<FileEditor>>>,
    source: Source,
    /// Bumped by every edit.
    generation: AtomicU64,
    pub offset_size: u8,
    pub length_size: u8,
    pub root: u64,
}

fn closed_after_failed_reopen() -> PyErr {
    PyOSError::new_err("the file could not be reopened after an edit; open it again")
}

impl Handle {
    fn new(file: File, source: Source, editor: Option<FileEditor>) -> Arc<Self> {
        let sb = file.superblock();
        let (offset_size, length_size, root) =
            (sb.offset_size, sb.length_size, sb.root_group_address);
        Arc::new(Self {
            file: RwLock::new(Some(file)),
            editor: editor.map(|e| Mutex::new(Some(e))),
            source,
            generation: AtomicU64::new(0),
            offset_size,
            length_size,
            root,
        })
    }

    /// A local file, read-only.
    pub(crate) fn open_local(py: Python<'_>, path: &str) -> PyResult<Arc<Self>> {
        let file = py.detach(|| crate::no_panic(|| File::open(path).map_err(to_py_err)))?;
        Ok(Self::new(file, Source::Local, None))
    }

    /// A local file, open for in-place editing (`'r+'`): the editor takes
    /// the file's exclusive lock and checks that it can edit the file, then
    /// the file is read through the editor's own file — never by path
    /// again, so a later `os.chdir` or a rename or replacement of the path
    /// cannot make reads (or the editor's plans) come from another file.
    pub(crate) fn open_editable(py: Python<'_>, path: &str) -> PyResult<Arc<Self>> {
        let (file, editor) = py.detach(|| {
            crate::no_panic(|| {
                let editor = FileEditor::open(path).map_err(to_py_err)?;
                let file = editor.reader().map_err(to_py_err)?;
                Ok((file, editor))
            })
        })?;
        Ok(Self::new(file, Source::Local, Some(editor)))
    }

    /// A remote file (`http(s)://`, `s3://`, ...).
    pub(crate) fn open_url(
        py: Python<'_>,
        url: &str,
        options: &clawhdf5_remote::Options,
    ) -> PyResult<Arc<Self>> {
        let (file, storage) = py.detach(|| {
            crate::no_panic(|| {
                let storage = clawhdf5_remote::storage_for_url(url, options).map_err(remote_err)?;
                let file = File::open_storage(storage.clone()).map_err(to_py_err)?;
                Ok((file, storage))
            })
        })?;
        Ok(Self::new(
            file,
            Source::Remote {
                url: url.to_string(),
                storage,
            },
            None,
        ))
    }

    /// Run `f` on the file with the GIL released (a remote read may wait
    /// on the network; other Python threads run meanwhile). `f` must not
    /// touch Python.
    pub(crate) fn with<R: Send>(
        &self,
        py: Python<'_>,
        f: impl FnOnce(&File) -> PyResult<R> + Send,
    ) -> PyResult<R> {
        py.detach(|| self.with_detached(f))
    }

    /// [`with`](Self::with) for code that already runs without the GIL.
    pub(crate) fn with_detached<R>(&self, f: impl FnOnce(&File) -> PyResult<R>) -> PyResult<R> {
        crate::no_panic(|| {
            let guard = self.file.read().unwrap_or_else(PoisonError::into_inner);
            let file = guard.as_ref().ok_or_else(closed_after_failed_reopen)?;
            f(file)
        })
    }

    /// Edits so far: objects that cache something an edit can change (a
    /// dataset's shape, an object's attributes) re-read it when this moved.
    pub(crate) fn generation(&self) -> u64 {
        self.generation.load(Ordering::Acquire)
    }

    /// Whether the file was opened for editing (`'r+'`), even if closed since.
    pub(crate) fn is_writable(&self) -> bool {
        self.editor.is_some()
    }

    /// The remote file's block cache.
    pub(crate) fn remote_storage(&self) -> Option<&clawhdf5_remote::RemoteStorage> {
        match &self.source {
            Source::Remote { storage, .. } => Some(storage),
            Source::Local => None,
        }
    }

    /// The URL of a remote file, credentials and query values redacted.
    pub(crate) fn redacted_url(&self) -> Option<String> {
        match &self.source {
            Source::Remote { url, .. } => Some(clawhdf5_remote::redact_url(url)),
            Source::Local => None,
        }
    }

    /// Release the editor, and with it the file's lock. Objects still
    /// open keep reading the file as it was last written; an edit through
    /// them is an error.
    pub(crate) fn close(&self) {
        if let Some(ed) = &self.editor {
            ed.lock().unwrap_or_else(PoisonError::into_inner).take();
        }
    }

    /// Apply one edit with the GIL released. No read runs while it writes,
    /// and the file is reopened afterwards — also after a failed edit, since
    /// a commit that failed part-way may have changed the file.
    pub(crate) fn edit<R: Send>(
        &self,
        py: Python<'_>,
        f: impl FnOnce(&mut FileEditor) -> Result<R, clawhdf5_rs::Error> + Send,
    ) -> PyResult<R> {
        let Some(editor) = &self.editor else {
            return Err(PyOSError::new_err(match self.source {
                Source::Remote { .. } => "remote files are read-only",
                Source::Local => "the file is open read-only; open it with mode 'r+' to change it",
            }));
        };
        if matches!(self.source, Source::Remote { .. }) {
            return Err(PyOSError::new_err("remote files are read-only"));
        }
        py.detach(|| {
            let mut ed = editor.lock().unwrap_or_else(PoisonError::into_inner);
            let ed = ed
                .as_mut()
                .ok_or_else(|| PyOSError::new_err("the file is closed"))?;
            let mut file = self.file.write().unwrap_or_else(PoisonError::into_inner);
            let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(ed)));
            // Drop the old mapping and its chunk cache before reopening.
            *file = None;
            // Through the editor's file, not the path (see `open_editable`).
            let reopened = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| ed.reader()));
            self.generation.fetch_add(1, Ordering::AcqRel);
            match reopened {
                Ok(Ok(f)) => *file = Some(f),
                Ok(Err(e)) => return Err(to_py_err(e)),
                Err(_) => return Err(closed_after_failed_reopen()),
            }
            drop(file);
            match result {
                Ok(r) => r.map_err(to_py_err),
                Err(p) => Err(crate::InternalError::new_err(format!(
                    "clawhdf5 internal error (please report it): {}",
                    panic_text(&*p)
                ))),
            }
        })
    }
}

/// A `clawhdf5_remote::Error` as a Python exception: the network side
/// (unreachable, a status, no range support, a changed file) is `OSError`,
/// a file that is not HDF5 is what `to_py_err` makes of it.
pub(crate) fn remote_err(e: clawhdf5_remote::Error) -> PyErr {
    match e {
        clawhdf5_remote::Error::Hdf5(e) => to_py_err(e),
        other => PyOSError::new_err(other.to_string()),
    }
}