hyperfile 0.7.1

use std::io::{Result, Error, ErrorKind};
use log::debug;
use aws_sdk_s3::Client;
use crate::staging::{Staging, config::StagingConfig, s3::S3Staging, StagingIntercept};
use crate::Cno;
use crate::config::{HyperFileConfig, HyperFileConfigBuilder, HyperFileMetaConfig, HyperFileRuntimeConfig};
use crate::buffer::{AlignedDataBlockWrapper, BatchDataBlockWrapper};
use crate::BlockIndex;
use super::block::{BlockRef, BlockMut, BlockState};
use super::hyper::Hyper;
use super::flags::{HyperFileFlags, FileFlags};
use super::mode::{HyperFileMode, FileMode};
use crate::SegmentId;
use super::PlannedRead;

impl<'a: 'static> Hyper<'a> {
    pub async fn fs_create(client: &Client, uri: &str, flags: FileFlags, mode: FileMode) -> Result<Self>
    {
        debug!("fs_create - uri: {}, flags: {}", uri, flags);
        let staging_config = StagingConfig::new_s3_uri(uri, None);
        let file_config = HyperFileConfigBuilder::new()
                            .with_staging_config(&staging_config)
                            .build();
        let f = HyperFileFlags::from_flags(flags);
        let m = HyperFileMode::from_mode(mode);
        return Self::create(client.clone(), file_config, f, m).await;
    }

    pub async fn fs_create_with_interceptor(client: &Client, uri: &str, flags: FileFlags, mode: FileMode, interceptor: impl StagingIntercept<S3Staging> + 'static) -> Result<Self>
    {
        debug!("fs_create_with_interceptor - uri: {}, flags: {}", uri, flags);
        let staging_config = StagingConfig::new_s3_uri(uri, None);
        let file_config = HyperFileConfigBuilder::new()
                            .with_staging_config(&staging_config)
                            .build();
        let f = HyperFileFlags::from_flags(flags);
        let m = HyperFileMode::from_mode(mode);
        return Self::create_with_interceptor(client.clone(), file_config, f, m, interceptor).await;
    }

    pub async fn fs_create_opt(client: &Client, uri: &str, flags: FileFlags, mode: FileMode, meta_config: &HyperFileMetaConfig, runtime_config: &HyperFileRuntimeConfig) -> Result<Self>
    {
        debug!("fs_create_opt - uri: {}, flags: {}", uri, flags);
        let staging_config = StagingConfig::new_s3_uri(uri, None);
        let file_config = HyperFileConfigBuilder::new()
                            .with_meta_config(meta_config)
                            .with_staging_config(&staging_config)
                            .with_runtime_config(runtime_config)
                            .build();
        let f = HyperFileFlags::from_flags(flags);
        let m = HyperFileMode::from_mode(mode);
        return Self::create(client.clone(), file_config, f, m).await;
    }

    pub async fn fs_create_opt_with_interceptor(client: &Client, uri: &str, flags: FileFlags, mode: FileMode,
            meta_config: &HyperFileMetaConfig, runtime_config: &HyperFileRuntimeConfig,
            interceptor: impl StagingIntercept<S3Staging> + 'static) -> Result<Self>
    {
        debug!("fs_create_opt_with_interceptor - uri: {}, flags: {}", uri, flags);
        let staging_config = StagingConfig::new_s3_uri(uri, None);
        let file_config = HyperFileConfigBuilder::new()
                            .with_meta_config(meta_config)
                            .with_staging_config(&staging_config)
                            .with_runtime_config(runtime_config)
                            .build();
        let f = HyperFileFlags::from_flags(flags);
        let m = HyperFileMode::from_mode(mode);
        return Self::create_with_interceptor(client.clone(), file_config, f, m, interceptor).await;
    }

    pub async fn fs_open(client: &Client, uri: &str, flags: FileFlags) -> Result<Self>
    {
        debug!("fs_open - uri: {}, flags: {}", uri, flags);
        let staging_config = StagingConfig::new_s3_uri(uri, None);
        let file_config = HyperFileConfigBuilder::new()
                            .with_staging_config(&staging_config)
                            .build();
        let f = HyperFileFlags::from_flags(flags);
        return Self::open(client.clone(), file_config, f).await;
    }

    pub async fn fs_open_opt(client: &Client, uri: &str, flags: FileFlags, runtime_config: &HyperFileRuntimeConfig) -> Result<Self>
    {
        debug!("fs_open_opt - uri: {}, flags: {}", uri, flags);
        let staging_config = StagingConfig::new_s3_uri(uri, None);
        let file_config = HyperFileConfigBuilder::new()
                            .with_staging_config(&staging_config)
                            .with_runtime_config(runtime_config)
                            .build();
        let f = HyperFileFlags::from_flags(flags);
        return Self::open(client.clone(), file_config, f).await;
    }

    pub async fn fs_open_or_create_with_default_opt(client: &Client, uri: &str, flags: FileFlags, mode: FileMode) -> Result<Self>
    {
        debug!("fs_open_or_create - uri: {}, flags: {}", uri, flags);
        let staging_config = StagingConfig::new_s3_uri(uri, None);
        let file_config = HyperFileConfigBuilder::new()
                            .with_staging_config(&staging_config)
                            .build();
        let f = HyperFileFlags::from_flags(flags);
        let m = HyperFileMode::from_mode(mode);
        return Self::do_open_or_create(client.clone(), file_config, f, m, true).await;
    }

    pub async fn fs_open_or_create_with_config(client: &Client, config: HyperFileConfig, flags: FileFlags, mode: FileMode) -> Result<Self>
    {
        debug!("fs_open_or_create_with_config - flags: {}, mode: {}", flags, mode);
        let f = HyperFileFlags::from_flags(flags);
        let m = HyperFileMode::from_mode(mode);
        return Self::do_open_or_create(client.clone(), config, f, m, true).await;
    }

    pub async fn fs_unlink(client: &Client, uri: &str) -> Result<()>
    {
        debug!("fs_unlink - uri: {}", uri);
        let staging_config = StagingConfig::new_s3_uri(uri, None);
        let staging = S3Staging::from(client, staging_config, HyperFileRuntimeConfig::default()).await?;
        staging.unlink().await
    }

    pub async fn fs_unlink_with_interceptor(client: &Client, uri: &str, interceptor: impl StagingIntercept<S3Staging> + 'static) -> Result<()>
    {
        debug!("fs_unlink_with_interceptor - uri: {}", uri);
        let staging_config = StagingConfig::new_s3_uri(uri, None);
        let mut staging = S3Staging::from(client, staging_config, HyperFileRuntimeConfig::default()).await?;
        staging.interceptor(interceptor);
        staging.unlink().await
    }

    /// Rename the file at `src_uri` to `dst_uri`.
    ///
    /// **Not yet implemented.** This entry point is a placeholder
    /// for a future POSIX-`rename(2)`-equivalent operation. It
    /// will return [`std::io::ErrorKind::Unsupported`] until the
    /// implementation lands.
    ///
    /// # Why it isn't here yet
    ///
    /// Hyperfile stores a file as multiple S3 objects under a URI
    /// prefix (the inode object, segment objects, optional WAL
    /// chunks, optional local cache files). A POSIX-faithful
    /// rename has to:
    ///
    /// 1. Atomically move every object from the source prefix to
    ///    the destination prefix. S3 has no native cross-key
    ///    atomic-rename primitive; the natural implementation is
    ///    a copy-then-delete loop, which is not atomic across
    ///    multiple objects.
    /// 2. Decide what happens if `dst_uri` already exists. POSIX
    ///    requires atomic replacement; under copy-then-delete the
    ///    intermediate state is observable to a concurrent reader
    ///    on either prefix.
    /// 3. Coordinate with any open `Hyper` / `HyperFileHandler` /
    ///    `HyperFileTokio` instances pointing at either URI; an
    ///    in-flight write or flush must not race the rename.
    /// 4. Handle the WAL-enabled case (the WAL prefix lives at a
    ///    sibling URI and would need to move in lock-step) and
    ///    the local-cache case (cached blocks under one URI must
    ///    be invalidated on rename).
    ///
    /// None of those problems are unsolvable, but each requires
    /// design decisions (acceptable atomicity, retention of the
    /// destination URI's prior content if the rename half-fails,
    /// behavior under concurrent open handles) that haven't been
    /// made yet. See `docs/posix.md`'s "Rename" section for the
    /// status and the recommended workaround until then.
    pub async fn fs_rename(_client: &Client, src_uri: &str, dst_uri: &str) -> Result<()>
    {
        debug!("fs_rename - src: {}, dst: {} (not implemented)", src_uri, dst_uri);
        Err(Error::new(
            ErrorKind::Unsupported,
            "fs_rename is not yet implemented; see docs/posix.md for the rationale",
        ))
    }

    pub async fn fs_release(&mut self) -> Result<Cno>
    {
        debug!("fs_release - ");
        // The public surface answers in checkpoint numbers: that is what a caller
        // pins, logs, and hands back to `open_cno`. Inside, the same value is a
        // `SegmentId` so it cannot be mistaken for a length or an offset.
        Ok(self.inner.release().await?.as_cno())
    }

    /// Read `buf.len()` bytes from `off`, returning the number of
    /// bytes read. Reads stop at `i_size`.
    ///
    /// **Does not populate the data cache.** Reading the same bytes
    /// twice fetches them twice. A read does consult the cache, so it
    /// is served from there when the blocks happen to be resident —
    /// for instance after [`Self::fs_block`], after
    /// [`Self::fs_read_ahead`], or after a write that has since been
    /// flushed.
    ///
    /// `docs/block-api.md` has the full table of which entry points
    /// populate the cache.
    pub async fn fs_read(&mut self, off: usize, buf: &mut [u8]) -> Result<usize>
    {
        debug!("fs_read - offset: {}, size: {}", off, buf.len());
        self.inner.read(off, buf).await
    }

    /// Fetch a range into the data block cache without returning it.
    ///
    /// For a caller that knows what will be asked for next — a
    /// read-ahead. Nothing comes back but a count of the blocks
    /// installed; a later [`Self::fs_read`] of those bytes asks the
    /// ordinary way and finds them.
    ///
    /// This is the entry point that fills the cache for the byte path.
    /// `fs_read` consults the cache but does not fill it, so bytes
    /// fetched speculatively through `fs_read` have nowhere to live and
    /// the next read fetches them again. Only blocks brought in through
    /// here are cached, so ordinary reads still cannot evict what the
    /// write path is holding.
    ///
    /// Requests are coalesced the way a read of the same range would be,
    /// so warming a megabyte costs a handful of requests rather than one
    /// per block.
    ///
    /// The range is widened to whole blocks, since a block is what the
    /// cache stores, and clamped to `i_size`. Blocks already cached are
    /// left alone, and holes are skipped: they read as zeroes without a
    /// request, so caching them buys nothing.
    ///
    /// On failure nothing is installed and the reads that follow simply
    /// pay for their own fetches.
    pub async fn fs_read_ahead(&mut self, off: usize, len: usize) -> Result<usize>
    {
        debug!("fs_read_ahead - offset: {}, len: {}", off, len);
        self.inner.read_ahead(off, len).await
    }

    /// `lseek(SEEK_DATA)`: smallest offset >= `off` holding data, or `None`
    /// (caller maps to `ENXIO`) if none before EOF.
    pub async fn fs_seek_data(&self, off: usize) -> Result<Option<usize>>
    {
        debug!("fs_seek_data - offset: {}", off);
        self.inner.seek_data(off).await
    }

    /// `lseek(SEEK_HOLE)`: smallest offset >= `off` in a hole (EOF counts), or
    /// `None` (`ENXIO`) if `off` is at/past EOF.
    pub async fn fs_seek_hole(&self, off: usize) -> Result<Option<usize>>
    {
        debug!("fs_seek_hole - offset: {}", off);
        self.inner.seek_hole(off).await
    }

    pub async fn fs_write(&mut self, off: usize, buf: &[u8]) -> Result<usize>
    {
        debug!("fs_write - offset: {}, size: {}", off, buf.len());
        self.inner.write(off, buf).await
    }

    pub async fn fs_write_zero(&mut self, off: usize, len: usize) -> Result<usize>
    {
        debug!("fs_write_zero - offset: {}, len: {}", off, len);
        self.inner.write_zero(off, len).await
    }

    pub async fn fs_write_aligned_batch(&mut self, blocks: Vec<AlignedDataBlockWrapper>) -> Result<usize>
    {
        debug!("fs_write_aligned_batch - batch count: {}", blocks.len());
        self.inner.write_aligned_batch(blocks).await
    }

    /// Borrow block `idx` for reading. See [`HyperFile::block`].
    ///
    /// `Ok(None)` means the block is not backed by data — no bmap
    /// entry, or an explicit zero block. Use [`Self::fs_block_state`]
    /// to tell those apart.
    ///
    /// **Populates the data cache.** A block fetched here is retained,
    /// so a later borrow or byte read of it is served from memory.
    /// [`Self::fs_read`] does not do this.
    ///
    /// [`HyperFile::block`]: crate::file::file::HyperFile::block
    pub async fn fs_block(&mut self, idx: BlockIndex) -> Result<Option<BlockRef<'_>>>
    {
        debug!("fs_block - block index: {}", idx);
        self.inner.block(idx).await
    }

    /// Borrow block `idx` for in-place modification. See
    /// [`HyperFile::block_mut`].
    ///
    /// The block is dirty from acquisition; the next
    /// [`Self::fs_flush`] persists it. `create` decides whether a
    /// block with no data is materialized as zeros (`true`) or
    /// reported as `Ok(None)` (`false`). `i_size` is not changed.
    ///
    /// **Populates the data cache**, and the block stays cached after
    /// the flush that persists it.
    ///
    /// [`HyperFile::block_mut`]: crate::file::file::HyperFile::block_mut
    pub async fn fs_block_mut(&mut self, idx: BlockIndex, create: bool) -> Result<Option<BlockMut<'_>>>
    {
        debug!("fs_block_mut - block index: {}, create: {}", idx, create);
        self.inner.block_mut(idx, create).await
    }

    /// What reading `[off, off + len)` would cost, without reading it.
    ///
    /// See
    /// [`HyperFile::read_plan`](crate::file::file::HyperFile::read_plan) for
    /// what the entries mean, and
    /// [Seeing where blocks are](https://github.com/daiyy/hyperfile/blob/main/docs/placement.md)
    /// for when to use this rather than [`Self::fs_block_placement`]. The
    /// short of it: this follows the data cache, so a warm file looks free.
    ///
    /// Metadata only. No data request is issued; index reads land in
    /// `meta_gets`.
    pub async fn fs_read_plan(&mut self, off: u64, len: u64) -> Result<Vec<PlannedRead>>
    {
        debug!("fs_read_plan - offset: {}, len: {}", off, len);
        self.inner.read_plan(off, len).await
    }

    /// [`Self::fs_read_plan`] for several ranges, answered in the order the
    /// ranges were given.
    ///
    /// On the direct API this saves only the repeated call, since there is no
    /// channel to cross — unlike the handler surface, where batching is the
    /// point. Provided so that code written against one surface reads the
    /// same on the other.
    pub async fn fs_read_plan_many(&mut self, ranges: &[(u64, u64)]) -> Result<Vec<Vec<PlannedRead>>>
    {
        debug!("fs_read_plan_many - range count: {}", ranges.len());
        self.inner.read_plan_many(ranges).await
    }

    /// Where each of `n` blocks starting at `start` currently lives:
    /// which segment and where in it, or `None` for a hole or a block not
    /// yet flushed.
    ///
    /// See
    /// [`HyperFile::block_placement`](crate::file::file::HyperFile::block_placement).
    /// Unlike [`Self::fs_read_plan`] this does not consult the data cache, so
    /// it answers about placement alone — which is what to use when judging
    /// whether a file is worth rearranging.
    pub async fn fs_block_placement(&mut self, start: BlockIndex, n: usize)
        -> Result<Vec<Option<(SegmentId, u64)>>>
    {
        debug!("fs_block_placement - start: {}, n: {}", start, n);
        self.inner.block_placement(start, n).await
    }

    /// [`Self::fs_block_placement`] for several block ranges, answered in the
    /// order the ranges were given. Batched for the reason given on
    /// [`Self::fs_read_plan_many`].
    pub async fn fs_block_placement_many(&mut self, ranges: &[(BlockIndex, usize)])
        -> Result<Vec<Vec<Option<(SegmentId, u64)>>>>
    {
        debug!("fs_block_placement_many - range count: {}", ranges.len());
        self.inner.block_placement_many(ranges).await
    }

    /// Begin a transaction: the writes from here until [`Self::fs_commit_txn`]
    /// are one unit, and nothing is published in between. See
    /// [`HyperFile::begin_txn`](crate::file::file::HyperFile::begin_txn) for
    /// what a transaction here does and does not provide — in particular that
    /// it gives atomic publication and recovery but **no isolation**: writes
    /// inside one are visible to readers immediately.
    #[cfg(feature = "wal")]
    pub async fn fs_begin_txn(&mut self) -> Result<()>
    {
        debug!("fs_begin_txn - ");
        self.inner.begin_txn().await
    }

    /// Commit the open transaction, publishing it as one checkpoint. See
    /// [`HyperFile::commit_txn`](crate::file::file::HyperFile::commit_txn).
    #[cfg(feature = "wal")]
    pub async fn fs_commit_txn(&mut self) -> Result<Cno>
    {
        debug!("fs_commit_txn - ");
        // The public surface answers in checkpoint numbers: that is what a caller
        // pins, logs, and hands back to `open_cno`. Inside, the same value is a
        // `SegmentId` so it cannot be mistaken for a length or an offset.
        Ok(self.inner.commit_txn().await?.as_cno())
    }

    /// Abandon the open transaction, discarding its writes. See
    /// [`HyperFile::abort_txn`](crate::file::file::HyperFile::abort_txn).
    ///
    /// Not the same as never committing: that leaves the writes in memory until
    /// the file is closed, and only a reopen discards them.
    #[cfg(feature = "wal")]
    pub async fn fs_abort_txn(&mut self) -> Result<()>
    {
        debug!("fs_abort_txn - ");
        self.inner.abort_txn().await
    }

    /// Whether a transaction is open.
    #[cfg(feature = "wal")]
    pub fn fs_in_txn(&self) -> bool
    {
        self.inner.in_txn()
    }

    /// What recovery did when this container was opened. See
    /// [`HyperFile::wal_recovery_report`](crate::file::file::HyperFile::wal_recovery_report).
    #[cfg(feature = "wal")]
    pub fn wal_recovery_report(&self) -> crate::wal::WalRecoveryReport
    {
        self.inner.wal_recovery_report()
    }

    /// True when a write that has returned `Ok` is already recoverable with no
    /// further flush.
    ///
    /// See
    /// [`HyperFile::writes_durable_on_ack`](crate::file::file::HyperFile::writes_durable_on_ack)
    /// for exactly what is and is not promised. Ask it once after opening: a
    /// caller that may skip flushing on this basis should be reading the
    /// guarantee from here rather than inferring it from its own copy of the
    /// configuration, which cannot stay in step with a mechanism it does not
    /// own.
    pub fn writes_durable_on_ack(&self) -> bool
    {
        self.inner.writes_durable_on_ack()
    }

    /// How block `idx` is mapped. See [`BlockState`].
    ///
    /// One bmap lookup, no data transfer.
    pub async fn fs_block_state(&mut self, idx: BlockIndex) -> Result<BlockState>
    {
        debug!("fs_block_state - block index: {}", idx);
        self.inner.block_state(idx).await
    }

    pub async fn fs_write_batch(&mut self, blocks: Vec<BatchDataBlockWrapper>) -> Result<usize>
    {
        debug!("fs_write_batch - batch count: {}", blocks.len());
        self.inner.write_batch(blocks).await
    }

    pub async fn fs_flush(&mut self) -> Result<Cno>
    {
        // The public surface answers in checkpoint numbers: that is what a caller
        // pins, logs, and hands back to `open_cno`. Inside, the same value is a
        // `SegmentId`, so it cannot be mistaken for a length or an offset.
        Ok(self.inner.flush_with_rollback().await?.as_cno())
    }

    /// POSIX-`fdatasync` flavoured flush. Persists pending data
    /// and any metadata required for reads to be correct (i.e.
    /// the bmap and `i_size`), but skips writing the inode when
    /// only attribute fields (`atime` / `mtime` / `ctime` /
    /// `mode` / `uid` / `gid`) are dirty. See `docs/posix.md` for
    /// the full semantic.
    ///
    /// Equivalent to `fs_flush` when there's data or bmap dirt;
    /// strictly cheaper when only attrs are dirty (no S3 PUT
    /// happens).
    pub async fn fs_fdatasync(&mut self) -> Result<Cno>
    {
        debug!("fs_fdatasync - ");
        // The public surface answers in checkpoint numbers: that is what a caller
        // pins, logs, and hands back to `open_cno`. Inside, the same value is a
        // `SegmentId` so it cannot be mistaken for a length or an offset.
        Ok(self.inner.flush_data().await?.as_cno())
    }

    pub async fn fs_truncate(&mut self, offset: usize) -> Result<()>
    {
        debug!("fs_truncate - offset: {}", offset);
        self.inner.truncate(offset).await
    }

    pub fn fs_getattr(&self) -> Result<libc::stat>
    {
        debug!("fs_getattr -");
        Ok(self.inner.stat())
    }

    pub async fn fs_getattr_fast(client: &Client, uri: &str) -> Result<libc::stat>
    {
        debug!("fs_getattr_fast - uri: {}", uri);
        let staging_config = StagingConfig::new_s3_uri(uri, None);
        let file_config = HyperFileConfigBuilder::new()
                            .with_staging_config(&staging_config)
                            .build();
        return Self::stat_fast(client.clone(), file_config).await;
    }

    pub async fn fs_chmod(&mut self, mode: libc::mode_t) -> Result<libc::stat>
    {
        debug!("fs_chmod - mode: {:#o}", mode);
        let mut stat = self.inner.stat();
        // update permission part only, don't change file type part
        stat.st_mode = (stat.st_mode & libc::S_IFMT) | (mode & !libc::S_IFMT);
        self.inner.update_stat(&stat).await
    }

    pub async fn fs_chown(&mut self, uid: libc::uid_t, gid: libc::gid_t) -> Result<libc::stat> {
        debug!("fs_chown - uid: {}, gid: {}", uid, gid);
        let mut stat = self.inner.stat();
        stat.st_uid = uid;
        stat.st_gid = gid;
        self.inner.update_stat(&stat).await
    }

    pub async fn fs_setattr(&mut self, stat: &libc::stat) -> Result<libc::stat> {
        debug!("fs_setattr - mode: {}, uid: {}, gid: {}", stat.st_mode, stat.st_uid, stat.st_gid);
        self.inner.update_stat(stat).await
    }

    /// Opt this open file in/out of cross-mount open-but-unlinked
    /// ([`InodeRaw::FLAG_KEEP_OPEN`]); persists on flush.
    pub async fn fs_set_keep_open(&mut self, on: bool) -> Result<()> {
        debug!("fs_set_keep_open - {}", on);
        self.inner.set_keep_open(on).await
    }

    /// True if this open file opts in to cross-mount open-but-unlinked.
    pub fn fs_is_keep_open(&self) -> bool { self.inner.is_keep_open() }

    pub async fn fs_setattr_fast(client: &Client, uri: &str, stat: &libc::stat) -> Result<libc::stat>
    {
        debug!("fs_setattr_fast - uri: {}", uri);
        let staging_config = StagingConfig::new_s3_uri(uri, None);
        let file_config = HyperFileConfigBuilder::new()
                            .with_staging_config(&staging_config)
                            .build();
        return Self::update_stat_fast(client.clone(), file_config, &stat).await;
    }

    pub fn fs_last_cno(&self) -> Cno
    {
        debug!("fs_last_cno -");
        self.inner.last_cno()
    }
}