// Copyright 2023 Greptime Team
//
// 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.

use std::collections::HashMap;
use std::sync::Arc;

use common_datasource::file_format::Format;
use object_store::ObjectStore;
use store_api::metadata::RegionMetadataRef;
use store_api::path_utils::region_name;
use store_api::region_request::{RegionCreateRequest, RegionOpenRequest};
use store_api::storage::RegionId;

use crate::FileOptions;
use crate::error::Result;
use crate::manifest::FileRegionManifest;

#[derive(Debug)]
pub struct FileRegion {
    pub(crate) region_dir: String,
    pub(crate) file_options: FileOptions,
    pub(crate) url: String,
    pub(crate) format: Format,
    pub(crate) options: HashMap<String, String>,
    pub(crate) metadata: RegionMetadataRef,
}

pub type FileRegionRef = Arc<FileRegion>;

impl FileRegion {
    pub async fn create(
        region_id: RegionId,
        request: RegionCreateRequest,
        object_store: &ObjectStore,
    ) -> Result<FileRegionRef> {
        let manifest = FileRegionManifest {
            region_id,
            column_metadatas: request.column_metadatas.clone(),
            primary_key: request.primary_key.clone(),
            options: request.options,
        };

        let region_dir = object_store::util::join_dir(
            &request.table_dir,
            &region_name(region_id.table_id(), region_id.region_sequence()),
        );
        let url = manifest.url()?;
        let file_options = manifest.file_options()?;
        let format = manifest.format()?;
        let options = manifest.options.clone();
        let metadata = manifest.metadata()?;

        manifest.store(&region_dir, object_store).await?;

        Ok(Arc::new(Self {
            region_dir,
            url,
            file_options,
            format,
            options,
            metadata,
        }))
    }

    pub async fn open(
        region_id: RegionId,
        request: RegionOpenRequest,
        object_store: &ObjectStore,
    ) -> Result<FileRegionRef> {
        let region_dir = object_store::util::join_dir(
            &request.table_dir,
            &region_name(region_id.table_id(), region_id.region_sequence()),
        );
        let manifest = FileRegionManifest::load(region_id, &region_dir, object_store).await?;

        Ok(Arc::new(Self {
            region_dir,
            url: manifest.url()?,
            file_options: manifest.file_options()?,
            format: manifest.format()?,
            metadata: manifest.metadata()?,
            options: manifest.options,
        }))
    }

    pub async fn drop(&self, object_store: &ObjectStore) -> Result<()> {
        FileRegionManifest::delete(self.metadata.region_id, &self.region_dir, object_store).await
    }

    pub fn metadata(&self) -> RegionMetadataRef {
        self.metadata.clone()
    }
}

#[cfg(test)]
mod tests {
    use std::assert_matches;

    use common_datasource::object_store::LocalFileAccess;
    use common_error::ext::{ErrorExt, RetryHint};
    use common_error::status_code::StatusCode;
    use store_api::region_request::PathType;
    use store_api::storage::ScanRequest;

    use super::*;
    use crate::error::Error;
    use crate::test_util::{new_test_column_metadata, new_test_object_store, new_test_options};

    #[tokio::test]
    async fn test_create_region() {
        let (_dir, object_store) = new_test_object_store("test_create_region");

        let request = RegionCreateRequest {
            engine: "file".to_string(),
            column_metadatas: new_test_column_metadata(),
            primary_key: vec![1],
            options: new_test_options(),
            table_dir: "create_region_dir/".to_string(),
            path_type: PathType::Bare,
            partition_expr_json: Some("".to_string()),
            requirements: Default::default(),
        };
        let region_id = RegionId::new(1, 0);

        let region = FileRegion::create(region_id, request.clone(), &object_store)
            .await
            .unwrap();

        assert_eq!(region.region_dir, "create_region_dir/1_0000000000/");
        assert_eq!(region.url, "test");
        assert_eq!(region.file_options.files, vec!["1.csv"]);
        assert_matches!(region.format, Format::Csv { .. });
        assert_eq!(region.options, new_test_options());
        assert_eq!(region.metadata.region_id, region_id);
        assert_eq!(region.metadata.primary_key, vec![1]);

        assert!(
            object_store
                .exists("create_region_dir/1_0000000000/manifest/_file_manifest")
                .await
                .unwrap()
        );

        // Object exists, should fail
        let err = FileRegion::create(region_id, request, &object_store)
            .await
            .unwrap_err();
        assert_matches!(err, Error::ManifestExists { .. });
    }

    #[tokio::test]
    async fn test_persisted_local_region_rejected_when_disabled() {
        let (_dir, object_store) = new_test_object_store("test_disabled_local_region");
        let request = RegionCreateRequest {
            engine: "file".to_string(),
            column_metadatas: new_test_column_metadata(),
            primary_key: vec![1],
            options: new_test_options(),
            table_dir: "disabled_local_region/".to_string(),
            path_type: PathType::Bare,
            partition_expr_json: Some("".to_string()),
            requirements: Default::default(),
        };
        let region = FileRegion::create(RegionId::new(1, 0), request, &object_store)
            .await
            .unwrap();

        let error = match region
            .query(ScanRequest::default(), &LocalFileAccess::Disabled)
            .await
        {
            Ok(_) => panic!("local file query must be rejected"),
            Err(error) => error,
        };
        assert_matches!(
            &error,
            Error::BuildBackend {
                source: common_datasource::error::Error::LocalFileAccessDisabled { .. },
                ..
            }
        );
        assert_eq!(error.status_code(), StatusCode::InvalidArguments);
        assert_eq!(error.retry_hint(), RetryHint::NonRetryable);
    }

    #[tokio::test]
    async fn test_open_region() {
        let (_dir, object_store) = new_test_object_store("test_open_region");

        let region_dir = "open_region_dir/".to_string();
        let request = RegionCreateRequest {
            engine: "file".to_string(),
            column_metadatas: new_test_column_metadata(),
            primary_key: vec![1],
            options: new_test_options(),
            table_dir: region_dir.clone(),
            path_type: PathType::Bare,
            partition_expr_json: Some("".to_string()),
            requirements: Default::default(),
        };
        let region_id = RegionId::new(1, 0);

        let _ = FileRegion::create(region_id, request.clone(), &object_store)
            .await
            .unwrap();

        let request = RegionOpenRequest {
            engine: "file".to_string(),
            table_dir: region_dir,
            path_type: PathType::Bare,
            options: HashMap::default(),
            skip_wal_replay: false,
            checkpoint: None,
            requirements: Default::default(),
        };

        let region = FileRegion::open(region_id, request, &object_store)
            .await
            .unwrap();

        assert_eq!(region.region_dir, "open_region_dir/1_0000000000/");
        assert_eq!(region.url, "test");
        assert_eq!(region.file_options.files, vec!["1.csv"]);
        assert_matches!(region.format, Format::Csv { .. });
        assert_eq!(region.options, new_test_options());
        assert_eq!(region.metadata.region_id, region_id);
        assert_eq!(region.metadata.primary_key, vec![1]);
    }

    #[tokio::test]
    async fn test_drop_region() {
        let (_dir, object_store) = new_test_object_store("test_drop_region");

        let region_dir = "drop_region_dir/".to_string();
        let request = RegionCreateRequest {
            engine: "file".to_string(),
            column_metadatas: new_test_column_metadata(),
            primary_key: vec![1],
            options: new_test_options(),
            table_dir: region_dir.clone(),
            path_type: PathType::Bare,
            partition_expr_json: Some("".to_string()),
            requirements: Default::default(),
        };
        let region_id = RegionId::new(1, 0);

        let region = FileRegion::create(region_id, request.clone(), &object_store)
            .await
            .unwrap();

        assert!(
            object_store
                .exists("drop_region_dir/1_0000000000/manifest/_file_manifest")
                .await
                .unwrap()
        );

        FileRegion::drop(&region, &object_store).await.unwrap();
        assert!(
            !object_store
                .exists("drop_region_dir/1_0000000000/manifest/_file_manifest")
                .await
                .unwrap()
        );

        let request = RegionOpenRequest {
            engine: "file".to_string(),
            table_dir: region_dir,
            path_type: PathType::Bare,
            options: HashMap::default(),
            skip_wal_replay: false,
            checkpoint: None,
            requirements: Default::default(),
        };
        let err = FileRegion::open(region_id, request, &object_store)
            .await
            .unwrap_err();
        assert_matches!(err, Error::LoadRegionManifest { .. });
    }
}
