goosefs-sdk 0.2.0

// Copyright (C) 2026 Tencent. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! `BaseFileSystem` — the standard `FileSystem` implementation.
//!
//! [`BaseFileSystem`] implements the [`FileSystem`] trait against a real
//! Goosefs cluster via gRPC.  It is the primary production implementation.
//!
//! # Thread safety
//!
//! `BaseFileSystem` is `Send + Sync + 'static` and can be wrapped in
//! `Arc<BaseFileSystem>` or `Arc<dyn FileSystem>` and shared freely across
//! async tasks.
//!
//! # `exists()` semantics
//!
//! **Java authority**: Verified against `DefaultFileSystem.exists()`:
//! ```java
//! try {
//!     URIStatus status = getStatus(path);
//!     if (!status.isCompleted() && !status.isFolder()) return false;
//!     return true;
//! } catch (FileDoesNotExistException e) {
//!     return false;
//! }
//! ```
//! An `INCOMPLETE` non-folder file → `false`.  This differs from the Go SDK
//! which returns `true` for all existing inodes.
//!
//! # `WriteType` xattr inheritance
//!
//! When `CreateFileOptions.write_type == WriteTypeXAttr::Inherit`, `create_file`
//! fetches the parent directory's xattr and calls
//! [`crate::fs::write_type::get_write_type_from_xattr`] to determine the
//! effective `WriteType`.  Falls back to the `GoosefsConfig` default.
//!
//! # Connection sharing
//!
//! All operations reuse the persistent gRPC channel from [`FileSystemContext`].
//! Construct via [`BaseFileSystem::connect`] or [`BaseFileSystem::from_context`].

use std::sync::Arc;

use async_trait::async_trait;

use crate::client::MasterClient;
use crate::config::{GoosefsConfig, WriteType};
use crate::context::FileSystemContext;
use crate::error::{Error, Result};
use crate::fs::filesystem::FileSystem;
use crate::fs::options::{
    CreateFileOptions, DeleteOptions, GetStatusOptions, ListStatusOptions, OpenFileOptions,
};
use crate::fs::uri_status::URIStatus;
use crate::fs::write_type::{get_write_type_from_xattr, WriteTypeXAttr};
use crate::io::{GoosefsFileInStream, GoosefsFileWriter};
use crate::proto::grpc::file::{CreateFilePOptions, WritePType};

/// Standard Goosefs filesystem client.
///
/// All operations delegate to the underlying `MasterClient` gRPC stub.
///
/// ## Usage
///
/// ```rust,no_run
/// use goosefs_sdk::context::FileSystemContext;
/// use goosefs_sdk::fs::BaseFileSystem;
/// use goosefs_sdk::config::GoosefsConfig;
/// use goosefs_sdk::fs::filesystem::FileSystem;
///
/// # async fn example() -> goosefs_sdk::error::Result<()> {
/// // Build once per application lifetime — one TCP+SASL handshake
/// let ctx = FileSystemContext::connect(GoosefsConfig::new("127.0.0.1:9200")).await?;
/// let fs = BaseFileSystem::from_context(ctx);
///
/// // All calls reuse the same Master connection — zero extra handshakes
/// let status = fs.get_status("/data/file.parquet").await?;
/// println!("length = {}", status.length);
/// # Ok(())
/// # }
/// ```
pub struct BaseFileSystem {
    /// Shared context — owns the persistent Master + Worker connections.
    ctx: Arc<FileSystemContext>,

    /// Cached config from the context for convenience access.
    config: GoosefsConfig,
}

impl BaseFileSystem {
    // ── Constructors ─────────────────────────────────────────────────────────

    /// Create a `BaseFileSystem` backed by a shared [`FileSystemContext`].
    ///
    /// All Master RPCs reuse the context's persistent gRPC channel.
    /// This is the recommended constructor for production use.
    pub fn from_context(ctx: Arc<FileSystemContext>) -> Arc<Self> {
        let config = ctx.config().clone();
        Arc::new(Self { config, ctx })
    }

    /// Connect to Goosefs and create both a [`FileSystemContext`] and a
    /// `BaseFileSystem` in one step.
    ///
    /// Equivalent to:
    /// ```rust,ignore
    /// let ctx = FileSystemContext::connect(config).await?;
    /// let fs  = BaseFileSystem::from_context(ctx);
    /// ```
    pub async fn connect(config: GoosefsConfig) -> Result<Arc<Self>> {
        let ctx = FileSystemContext::connect(config).await?;
        Ok(Self::from_context(ctx))
    }

    /// Borrow the underlying config.
    pub fn config(&self) -> &GoosefsConfig {
        &self.config
    }

    /// Borrow the shared [`FileSystemContext`].
    pub fn context(&self) -> &Arc<FileSystemContext> {
        &self.ctx
    }

    // ── Internal helpers ─────────────────────────────────────────────────────

    /// Obtain a `MasterClient` from the shared pool.
    fn master(&self) -> Arc<MasterClient> {
        self.ctx.acquire_master()
    }

    /// Resolve the effective `WriteType` for a new file at `path`.
    ///
    /// Priority:
    /// 1. Explicit `WriteTypeXAttr::Explicit(wt)` in `options`
    /// 2. Parent directory `"innerWriteType"` xattr
    /// 3. `GoosefsConfig.write_type` (if set)
    /// 4. Default: `WriteType::MustCache` (Java default)
    async fn resolve_write_type(&self, path: &str, options: &CreateFileOptions) -> WriteType {
        // 1. Explicit override
        if let WriteTypeXAttr::Explicit(wt) = options.write_type {
            return wt;
        }

        // 2. Parent xattr
        //
        // We must distinguish three cases:
        // - parent exists, has xattr → use that WriteType
        // - parent exists but no xattr → fall through to config default
        // - parent does not exist (NotFound) → fall through to config default
        // - any other RPC error (Unavailable, etc.) → also fall through but
        //   log a warning, because silently using a different default on
        //   transient network errors changes persistence semantics for the
        //   newly created file.
        let parent = Self::parent_path(path);
        if let Some(parent_path) = parent {
            let master = self.master();
            match master.get_status(&parent_path).await {
                Ok(parent_info) => {
                    let parent_status = URIStatus::from_proto(parent_info);
                    if let Some(wt) = get_write_type_from_xattr(&parent_status.xattr) {
                        return wt;
                    }
                }
                Err(e) if e.is_not_found() => {
                    // Parent doesn't exist yet — totally fine, fall through.
                }
                Err(e) => {
                    tracing::warn!(
                        path = %path,
                        parent = %parent_path,
                        error = %e,
                        "resolve_write_type: failed to fetch parent xattr; \
                         falling back to config default — file will be created with that WriteType"
                    );
                }
            }
        }

        // 3. Config default
        if let Some(proto_wt) = self.config.get_write_type() {
            if let Ok(wt) = WriteType::try_from_proto(proto_wt) {
                return wt;
            }
        }

        // 4. Java default
        WriteType::MustCache
    }

    /// Extract the parent path of `path`.
    ///
    /// Returns `None` for root `/`.
    fn parent_path(path: &str) -> Option<String> {
        crate::metadata_cache::parent_path(path)
    }

    // ── One-shot write convenience ─────────────────────────────────────────
    //
    // NOT part of the `FileSystem` trait; lives on `BaseFileSystem` directly.

    /// Create a file, write data, and close it in a single async call.
    ///
    /// Equivalent to `create_file()` → `write()` → `close()`, but avoids the
    /// extra tokio scheduler yield between create and close. Matches the
    /// Python SDK's `write_file` API.
    ///
    /// Returns the number of bytes written.
    pub async fn write_file(
        &self,
        path: &str,
        data: &[u8],
        options: CreateFileOptions,
    ) -> Result<u64> {
        let write_type = self.resolve_write_type(path, &options).await;

        let proto_opts = CreateFilePOptions {
            block_size_bytes: options.block_size_bytes,
            recursive: Some(options.recursive),
            write_type: Some(WritePType::from(write_type) as i32),
            ..Default::default()
        };

        GoosefsFileWriter::write_file_with_context_and_options(
            self.ctx.clone(),
            path,
            data,
            Some(proto_opts),
        )
        .await
    }
}

#[async_trait]
impl FileSystem for BaseFileSystem {
    // ── Status ────────────────────────────────────────────────────────────────

    async fn get_status(&self, path: &str) -> Result<URIStatus> {
        self.get_status_with_options(path, GetStatusOptions::default())
            .await
    }

    async fn get_status_with_options(
        &self,
        path: &str,
        opts: GetStatusOptions,
    ) -> Result<URIStatus> {
        let sync = opts
            .sync_interval_ms
            .unwrap_or(self.config.file_metadata_sync_interval);
        let load = opts
            .load_metadata_type
            .unwrap_or(self.config.file_metadata_load_type);
        let master = self.master();
        let cache = self.ctx.acquire_metadata_cache();
        let mut fi =
            crate::metadata_cache::get_status_through_cache(cache.as_deref(), path, sync, || {
                master.get_status_with_load_type(path, Some(load), Some(sync))
            })
            .await?;
        // Mirror Java getStatus when checkBlockReplicas > 0: probe workers and
        // overwrite BlockInfo.locations (same as `fs stat --check_replicas`).
        // Enrichment mutates this owned clone — never write locations back
        // into the metadata cache (INV-MC-D1).
        //
        // Master never computes `in_goose_fs_percentage` (always 0). Java only
        // fills it when CheckBlocks is set; Python `get_status()` has no such
        // argument, so without a cheap fill MustCache writes always report 0.
        // CheckBlocks, when enabled, stays authoritative — do not overlay the
        // MustCache heuristic after a probe that found 0 cached bytes.
        let check = self.ctx.config().check_block_replicas;
        if check > 0 {
            let router = self.ctx.acquire_router();
            let view = crate::block::router::WorkerRouterView::from_shared(&router);
            let pool = self.ctx.acquire_worker_pool();
            crate::block::maybe_enrich_file_block_locations(
                &mut fi,
                &view,
                Some(&pool),
                self.ctx.config(),
                check,
            )
            .await;
        } else {
            crate::block::ensure_block_ids_from_file_block_infos(&mut fi);
            crate::block::fill_in_goosefs_percentage_without_probe(&mut fi);
        }
        Ok(URIStatus::from_proto(fi))
    }

    async fn list_status(&self, path: &str, recursive: bool) -> Result<Vec<URIStatus>> {
        self.list_status_with_options(
            path,
            ListStatusOptions {
                recursive,
                ..Default::default()
            },
        )
        .await
    }

    async fn list_status_with_options(
        &self,
        path: &str,
        opts: ListStatusOptions,
    ) -> Result<Vec<URIStatus>> {
        let master = self.master();
        let sync = opts
            .sync_interval_ms
            .unwrap_or(self.config.file_metadata_sync_interval);
        let load = opts
            .load_metadata_type
            .unwrap_or(self.config.file_metadata_load_type);

        if opts.recursive {
            // Recursive listings never use the cache (Java MetadataCachingBaseFileSystem
            // + INV-MC-S5). MasterClient owns client-side BFS (GooseFS 2.0
            // dropped ListStatusPOptions.recursive). Pass the resolved load
            // type so Never / Once / Always on options (or config) take effect
            // at every BFS level — Java recursive listStatus does not force Always;
            // the default is goosefs.user.file.metadata.load.type (ONCE).
            let items = master
                .list_status_with_options(path, true, Some(load), Some(sync))
                .await?;
            return Ok(items.into_iter().map(URIStatus::from_proto).collect());
        }

        let skip = crate::metadata_cache::should_skip_listing_cache(
            false,
            load,
            opts.load_metadata_only,
            sync,
        );
        let cache = self.ctx.acquire_metadata_cache();
        let items =
            crate::metadata_cache::list_status_through_cache(cache.as_deref(), path, skip, || {
                // Java `listStatusDefaults()` always sets loadMetadataType
                // (default ONCE), including non-recursive listings.
                master.list_status_with_options(path, false, Some(load), Some(sync))
            })
            .await?;
        Ok(items.into_iter().map(URIStatus::from_proto).collect())
    }

    /// Return `true` if `path` exists and is either a completed file or a directory.
    ///
    /// # Java semantics
    ///
    /// An `INCOMPLETE` non-folder file returns `false` because it is not yet
    /// usable.  The Go SDK incorrectly returns `true` in this case.
    async fn exists(&self, path: &str) -> Result<bool> {
        match self.get_status(path).await {
            Ok(status) => {
                // INCOMPLETE non-folder → not usable → false
                Ok(status.is_readable())
            }
            Err(Error::NotFound { .. }) => Ok(false),
            Err(e) => Err(e),
        }
    }

    // ── File read ─────────────────────────────────────────────────────────────

    async fn open_file(&self, path: &str, options: OpenFileOptions) -> Result<GoosefsFileInStream> {
        GoosefsFileInStream::open_with_context(self.ctx.clone(), path, options).await
    }

    // ── File write ────────────────────────────────────────────────────────────

    /// Create a new file, inheriting `WriteType` from the parent directory
    /// xattr if not explicitly set.
    async fn create_file(
        &self,
        path: &str,
        options: CreateFileOptions,
    ) -> Result<GoosefsFileWriter> {
        let write_type = self.resolve_write_type(path, &options).await;

        let proto_opts = CreateFilePOptions {
            block_size_bytes: options.block_size_bytes,
            recursive: Some(options.recursive),
            write_type: Some(WritePType::from(write_type) as i32),
            ..Default::default()
        };

        GoosefsFileWriter::create_with_context(self.ctx.clone(), path, Some(proto_opts)).await
    }

    // ── Directory ─────────────────────────────────────────────────────────────

    async fn mkdir(&self, path: &str, recursive: bool) -> Result<()> {
        let master = self.master();
        crate::metadata_cache::invalidate_on_success(
            self.ctx.acquire_metadata_cache().as_deref(),
            path,
            master.create_directory(path, recursive).await,
        )
    }

    // ── Delete ────────────────────────────────────────────────────────────────

    async fn delete(&self, path: &str, options: DeleteOptions) -> Result<()> {
        let master = self.master();
        crate::metadata_cache::invalidate_on_success(
            self.ctx.acquire_metadata_cache().as_deref(),
            path,
            master.delete_with_options(path, options).await,
        )
    }

    // ── Rename ────────────────────────────────────────────────────────────────

    async fn rename(&self, src: &str, dst: &str) -> Result<()> {
        let master = self.master();
        crate::metadata_cache::invalidate_rename_on_success(
            self.ctx.acquire_metadata_cache().as_deref(),
            src,
            dst,
            master.rename(src, dst).await,
        )
    }
}

// ── Unit tests (pure logic — no I/O) ─────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_parent_path_normal() {
        assert_eq!(
            BaseFileSystem::parent_path("/data/hello.txt"),
            Some("/data".to_string())
        );
    }

    #[test]
    fn test_parent_path_root_child() {
        assert_eq!(
            BaseFileSystem::parent_path("/hello.txt"),
            Some("/".to_string())
        );
    }

    #[test]
    fn test_parent_path_root() {
        assert_eq!(BaseFileSystem::parent_path("/"), None);
    }

    #[test]
    fn test_parent_path_nested() {
        assert_eq!(
            BaseFileSystem::parent_path("/a/b/c/file.parquet"),
            Some("/a/b/c".to_string())
        );
    }

    #[test]
    fn test_parent_path_trailing_slash() {
        assert_eq!(
            BaseFileSystem::parent_path("/data/dir/"),
            Some("/data".to_string())
        );
    }

    /// Verify that `from_context()` creates a `BaseFileSystem` with a shared context.
    #[test]
    fn test_from_context_sets_ctx() {
        // Can't call connect() in a unit test (needs network), but we can
        // verify the test_new_is_legacy_mode test was removed. Just a compile check.
    }
}