rustfs/ecstore/src/disk/remote.rs

use std::path::PathBuf;

use bytes::Bytes;
use common::error::{Error, Result};
use futures::lock::Mutex;
use protos::{
    node_service_time_out_client,
    proto_gen::node_service::{
        DeleteRequest, DeleteVersionsRequest, DeleteVolumeRequest, ListDirRequest, ListVolumesRequest, MakeVolumeRequest,
        MakeVolumesRequest, ReadAllRequest, ReadMultipleRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest,
        RenameFileRequst, StatVolumeRequest, WalkDirRequest, WriteAllRequest, WriteMetadataRequest,
    },
};
use tonic::Request;
use tracing::info;
use uuid::Uuid;

use crate::{
    disk::error::DiskError,
    store_api::{FileInfo, RawFileInfo},
};

use super::{
    endpoint::Endpoint, DeleteOptions, DiskAPI, DiskOption, FileInfoVersions, FileReader, FileWriter, MetaCacheEntry,
    ReadMultipleReq, ReadMultipleResp, ReadOptions, RemoteFileReader, RemoteFileWriter, RenameDataResp, VolumeInfo,
    WalkDirOptions,
};

#[derive(Debug)]
pub struct RemoteDisk {
    pub id: Mutex<Option<Uuid>>,
    pub addr: String,
    pub url: url::Url,
    pub root: PathBuf,
}

impl RemoteDisk {
    pub async fn new(ep: &Endpoint, _opt: &DiskOption) -> Result<Self> {
        // let root = fs::canonicalize(ep.url.path()).await?;
        let root = PathBuf::from(ep.url.path());
        let addr = format!("{}://{}:{}", ep.url.scheme(), ep.url.host_str().unwrap(), ep.url.port().unwrap());
        Ok(Self {
            id: Mutex::new(None),
            addr,
            url: ep.url.clone(),
            root,
        })
    }
}

// TODO: all api need to handle errors
#[async_trait::async_trait]
impl DiskAPI for RemoteDisk {
    fn is_local(&self) -> bool {
        false
    }
    async fn close(&self) -> Result<()> {
        Ok(())
    }
    fn path(&self) -> PathBuf {
        self.root.clone()
    }

    async fn get_disk_id(&self) -> Option<Uuid> {
        self.id.lock().await.clone()
    }
    async fn set_disk_id(&self, id: Option<Uuid>) -> Result<()> {
        let mut lock = self.id.lock().await;
        *lock = id;

        Ok(())
    }

    async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes> {
        info!("read_all");
        let mut client = node_service_time_out_client(&self.addr)
            .await
            .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
        let request = Request::new(ReadAllRequest {
            disk: self.root.to_string_lossy().to_string(),
            volume: volume.to_string(),
            path: path.to_string(),
        });

        let response = client.read_all(request).await?.into_inner();

        info!("read_all success");

        if !response.success {
            return Err(DiskError::FileNotFound.into());
        }

        Ok(Bytes::from(response.data))
    }

    async fn write_all(&self, volume: &str, path: &str, data: Vec<u8>) -> Result<()> {
        info!("write_all");
        let mut client = node_service_time_out_client(&self.addr)
            .await
            .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
        let request = Request::new(WriteAllRequest {
            disk: self.root.to_string_lossy().to_string(),
            volume: volume.to_string(),
            path: path.to_string(),
            data,
        });

        let response = client.write_all(request).await?.into_inner();

        if !response.success {
            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
        }

        Ok(())
    }

    async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> {
        info!("delete");
        let options = serde_json::to_string(&opt)?;
        let mut client = node_service_time_out_client(&self.addr)
            .await
            .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
        let request = Request::new(DeleteRequest {
            disk: self.root.to_string_lossy().to_string(),
            volume: volume.to_string(),
            path: path.to_string(),
            options,
        });

        let response = client.delete(request).await?.into_inner();

        if !response.success {
            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
        }

        Ok(())
    }

    async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> {
        info!("rename_file");
        let mut client = node_service_time_out_client(&self.addr)
            .await
            .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
        let request = Request::new(RenameFileRequst {
            disk: self.root.to_string_lossy().to_string(),
            src_volume: src_volume.to_string(),
            src_path: src_path.to_string(),
            dst_volume: dst_volume.to_string(),
            dst_path: dst_path.to_string(),
        });

        let response = client.rename_file(request).await?.into_inner();

        if !response.success {
            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
        }

        Ok(())
    }

    async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, _file_size: usize) -> Result<FileWriter> {
        info!("create_file");
        Ok(FileWriter::Remote(
            RemoteFileWriter::new(
                self.root.clone(),
                volume.to_string(),
                path.to_string(),
                false,
                node_service_time_out_client(&self.addr)
                    .await
                    .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?,
            )
            .await?,
        ))
    }

    async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter> {
        info!("append_file");
        Ok(FileWriter::Remote(
            RemoteFileWriter::new(
                self.root.clone(),
                volume.to_string(),
                path.to_string(),
                true,
                node_service_time_out_client(&self.addr)
                    .await
                    .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?,
            )
            .await?,
        ))
    }

    async fn read_file(&self, volume: &str, path: &str) -> Result<FileReader> {
        info!("read_file");
        Ok(FileReader::Remote(
            RemoteFileReader::new(
                self.root.clone(),
                volume.to_string(),
                path.to_string(),
                node_service_time_out_client(&self.addr)
                    .await
                    .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?,
            )
            .await?,
        ))
    }

    async fn list_dir(&self, _origvolume: &str, volume: &str, _dir_path: &str, _count: i32) -> Result<Vec<String>> {
        info!("list_dir");
        let mut client = node_service_time_out_client(&self.addr)
            .await
            .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
        let request = Request::new(ListDirRequest {
            disk: self.root.to_string_lossy().to_string(),
            volume: volume.to_string(),
        });

        let response = client.list_dir(request).await?.into_inner();

        if !response.success {
            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
        }

        Ok(response.volumes)
    }

    async fn walk_dir(&self, opts: WalkDirOptions) -> Result<Vec<MetaCacheEntry>> {
        info!("walk_dir");
        let walk_dir_options = serde_json::to_string(&opts)?;
        let mut client = node_service_time_out_client(&self.addr)
            .await
            .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
        let request = Request::new(WalkDirRequest {
            disk: self.root.to_string_lossy().to_string(),
            walk_dir_options,
        });

        let response = client.walk_dir(request).await?.into_inner();

        if !response.success {
            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
        }

        let entries = response
            .meta_cache_entry
            .into_iter()
            .filter_map(|json_str| serde_json::from_str::<MetaCacheEntry>(&json_str).ok())
            .collect();

        Ok(entries)
    }

    async fn rename_data(
        &self,
        src_volume: &str,
        src_path: &str,
        fi: FileInfo,
        dst_volume: &str,
        dst_path: &str,
    ) -> Result<RenameDataResp> {
        info!("rename_data");
        let file_info = serde_json::to_string(&fi)?;
        let mut client = node_service_time_out_client(&self.addr)
            .await
            .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
        let request = Request::new(RenameDataRequest {
            disk: self.root.to_string_lossy().to_string(),
            src_volume: src_volume.to_string(),
            src_path: src_path.to_string(),
            file_info,
            dst_volume: dst_volume.to_string(),
            dst_path: dst_path.to_string(),
        });

        let response = client.rename_data(request).await?.into_inner();

        if !response.success {
            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
        }

        let rename_data_resp = serde_json::from_str::<RenameDataResp>(&response.rename_data_resp)?;

        Ok(rename_data_resp)
    }

    async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> {
        info!("make_volumes");
        let mut client = node_service_time_out_client(&self.addr)
            .await
            .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
        let request = Request::new(MakeVolumesRequest {
            disk: self.root.to_string_lossy().to_string(),
            volumes: volumes.iter().map(|s| (*s).to_string()).collect(),
        });

        let response = client.make_volumes(request).await?.into_inner();

        if !response.success {
            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
        }

        Ok(())
    }

    async fn make_volume(&self, volume: &str) -> Result<()> {
        info!("make_volume");
        let mut client = node_service_time_out_client(&self.addr)
            .await
            .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
        let request = Request::new(MakeVolumeRequest {
            disk: self.root.to_string_lossy().to_string(),
            volume: volume.to_string(),
        });

        let response = client.make_volume(request).await?.into_inner();

        if !response.success {
            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
        }

        Ok(())
    }

    async fn list_volumes(&self) -> Result<Vec<VolumeInfo>> {
        info!("list_volumes");
        let mut client = node_service_time_out_client(&self.addr)
            .await
            .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
        let request = Request::new(ListVolumesRequest {
            disk: self.root.to_string_lossy().to_string(),
        });

        let response = client.list_volumes(request).await?.into_inner();

        if !response.success {
            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
        }

        let infos = response
            .volume_infos
            .into_iter()
            .filter_map(|json_str| serde_json::from_str::<VolumeInfo>(&json_str).ok())
            .collect();

        Ok(infos)
    }

    async fn stat_volume(&self, volume: &str) -> Result<VolumeInfo> {
        info!("stat_volume");
        let mut client = node_service_time_out_client(&self.addr)
            .await
            .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
        let request = Request::new(StatVolumeRequest {
            disk: self.root.to_string_lossy().to_string(),
            volume: volume.to_string(),
        });

        let response = client.stat_volume(request).await?.into_inner();

        if !response.success {
            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
        }

        let volume_info = serde_json::from_str::<VolumeInfo>(&response.volume_info)?;

        Ok(volume_info)
    }

    async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> {
        info!("write_metadata");
        let file_info = serde_json::to_string(&fi)?;
        let mut client = node_service_time_out_client(&self.addr)
            .await
            .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
        let request = Request::new(WriteMetadataRequest {
            disk: self.root.to_string_lossy().to_string(),
            volume: volume.to_string(),
            path: path.to_string(),
            file_info,
        });

        let response = client.write_metadata(request).await?.into_inner();

        if !response.success {
            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
        }

        Ok(())
    }

    async fn read_version(
        &self,
        _org_volume: &str,
        volume: &str,
        path: &str,
        version_id: &str,
        opts: &ReadOptions,
    ) -> Result<FileInfo> {
        info!("read_version");
        let opts = serde_json::to_string(opts)?;
        let mut client = node_service_time_out_client(&self.addr)
            .await
            .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
        let request = Request::new(ReadVersionRequest {
            disk: self.root.to_string_lossy().to_string(),
            volume: volume.to_string(),
            path: path.to_string(),
            version_id: version_id.to_string(),
            opts,
        });

        let response = client.read_version(request).await?.into_inner();

        if !response.success {
            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
        }

        let file_info = serde_json::from_str::<FileInfo>(&response.file_info)?;

        Ok(file_info)
    }

    async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result<RawFileInfo> {
        info!("read_xl");
        let mut client = node_service_time_out_client(&self.addr)
            .await
            .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
        let request = Request::new(ReadXlRequest {
            disk: self.root.to_string_lossy().to_string(),
            volume: volume.to_string(),
            path: path.to_string(),
            read_data,
        });

        let response = client.read_xl(request).await?.into_inner();

        if !response.success {
            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
        }

        let raw_file_info = serde_json::from_str::<RawFileInfo>(&response.raw_file_info)?;

        Ok(raw_file_info)
    }

    async fn delete_versions(
        &self,
        volume: &str,
        versions: Vec<FileInfoVersions>,
        opts: DeleteOptions,
    ) -> Result<Vec<Option<Error>>> {
        info!("delete_versions");
        let opts = serde_json::to_string(&opts)?;
        let mut versions_str = Vec::with_capacity(versions.len());
        for file_info_versions in versions.iter() {
            versions_str.push(serde_json::to_string(file_info_versions)?);
        }
        let mut client = node_service_time_out_client(&self.addr)
            .await
            .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
        let request = Request::new(DeleteVersionsRequest {
            disk: self.root.to_string_lossy().to_string(),
            volume: volume.to_string(),
            versions: versions_str,
            opts,
        });

        let response = client.delete_versions(request).await?.into_inner();
        if !response.success {
            return Err(Error::from_string(format!(
                "delete versions remote err: {}",
                response.error_info.unwrap_or("None".to_string())
            )));
        }
        let errors = response
            .errors
            .iter()
            .map(|error| {
                if error.is_empty() {
                    None
                } else {
                    Some(Error::from_string(error))
                }
            })
            .collect();

        Ok(errors)
    }

    async fn read_multiple(&self, req: ReadMultipleReq) -> Result<Vec<ReadMultipleResp>> {
        info!("read_multiple");
        let read_multiple_req = serde_json::to_string(&req)?;
        let mut client = node_service_time_out_client(&self.addr)
            .await
            .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
        let request = Request::new(ReadMultipleRequest {
            disk: self.root.to_string_lossy().to_string(),
            read_multiple_req,
        });

        let response = client.read_multiple(request).await?.into_inner();

        if !response.success {
            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
        }

        let read_multiple_resps = response
            .read_multiple_resps
            .into_iter()
            .filter_map(|json_str| serde_json::from_str::<ReadMultipleResp>(&json_str).ok())
            .collect();

        Ok(read_multiple_resps)
    }

    async fn delete_volume(&self, volume: &str) -> Result<()> {
        info!("delete_volume");
        let mut client = node_service_time_out_client(&self.addr)
            .await
            .map_err(|err| Error::from_string(format!("can not get client, err: {}", err.to_string())))?;
        let request = Request::new(DeleteVolumeRequest {
            disk: self.root.to_string_lossy().to_string(),
            volume: volume.to_string(),
        });

        let response = client.delete_volume(request).await?.into_inner();

        if !response.success {
            return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
        }

        Ok(())
    }
}