// Licensed to the Apache Software Foundation (ASF) under one

// or more contributor license agreements.  See the NOTICE file

// distributed with this work for additional information

// regarding copyright ownership.  The ASF licenses this file

// to you 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.

#pragma once

#include <gen_cpp/Status_types.h>

#include <cstdint>

#include <filesystem>

#include <functional>

#include <memory>

#include <optional>

#include <string>

#include <string_view>

#include <utility>

#include <vector>

namespace doris {

// Names are in lexico order.

enum class ObjStorageProvider : uint8_t {

    UNKNOWN = 0,

    AWS = 1,

    AZURE = 2,

    BOS = 3,

    COS = 4,

    OSS = 5,

    OBS = 6,

    GCP = 7,

    TOS = 8,

};

/// eg:

///     s3://bucket1/path/to/file.txt

/// path:   s3://bucket1/path/to/file.txt

/// bucket: bucket1

/// key:    path/to/file.txt

struct ObjStoragePath {

    std::filesystem::path path = "";

    std::string bucket {}; // blob container in azure

    std::string key {};    // blob name in azure

    std::string prefix {}; // for list and recursive delete

    // Empty means recursive listing; "/" lists direct children.

    std::string delimiter {}; // NOLINT(readability-redundant-member-init)

};

struct ObjStorageEndpointInfo {

    std::string endpoint {};

    std::string ak {};

    std::string sk {};

    std::string tls_debug_context {};

};

struct ObjectMeta {

    std::string key {};

    int64_t size {0};

    int64_t mtime_s {0};

    std::optional<std::string> etag {};

};

struct ObjStorageCompletedPart {

    int part_num = 0;

    std::string etag {};

};

struct ObjStorageStatus {

    enum Code : int {

        OK = TStatusCode::OK,

        INTERNAL_ERROR = TStatusCode::INTERNAL_ERROR,

        LIMIT_REACH = TStatusCode::LIMIT_REACH,

        NOT_FOUND = TStatusCode::NOT_FOUND,

        END_OF_FILE = TStatusCode::END_OF_FILE,

        IO_ERROR = TStatusCode::IO_ERROR,

        NETWORK_ERROR = TStatusCode::NETWORK_ERROR,

        // Keep the legacy BE error code for access-denied object storage responses.

        PERMISSION_DENIED = -256,

    };

    ObjStorageStatus(int r = OK, std::string msg = "") : code(r), msg(std::move(msg)) {}

    // clang-format off

    int code {OK}; // To unify the error handle logic with BE, we'd better use the same error code as BE

    // clang-format on

    std::string msg;

};

// We only store error code along with err_msg instead of Status to unify BE and recycler's error handle logic

struct ObjStorageResponse {

    ObjStorageStatus status {ObjStorageStatus::OK, ""};

    int http_code {200};

    std::string request_id {};

    static ObjStorageResponse OK() {

        // clang-format off

        return {

                .status = ObjStorageStatus{ObjStorageStatus::OK, ""},

                .http_code = 200,

                .request_id = ""

        };

        // clang-format on

    }

    static ObjStorageResponse rate_limit(int status_code, int http_code, std::string message) {

        return {

                .status = ObjStorageStatus {status_code, std::move(message)},

                .http_code = http_code,

        };

    }

    bool ok() const { return status.code == ObjStorageStatus::OK; }

};

// Convert a provider HTTP response code into the object-storage status domain. A non-positive

// response code means that no HTTP response was received.

ObjStorageStatus obj_storage_status_from_http_code(int http_code, std::string message);

struct ObjStorageUploadResult {

    ObjStorageResponse resp = ObjStorageResponse::OK();

    std::optional<std::string> upload_id = std::nullopt;

    std::optional<std::string> etag = std::nullopt;

};

struct ObjStorageHeadResult {

    ObjStorageResponse resp = ObjStorageResponse::OK();

    long long file_size {0};

    std::optional<std::string> content_type {};

    std::optional<std::string> etag {};

};

struct ObjStorageListResult {

    ObjStorageResponse resp = ObjStorageResponse::OK();

    std::optional<ObjectMeta> object = std::nullopt;

};

struct ObjStorageListPageResult {

    ObjStorageResponse resp = ObjStorageResponse::OK();

    std::vector<ObjectMeta> objects {};

    std::string continuation_token {};

    bool has_more = false;

};

struct ObjStorageCapabilities {

    size_t max_delete_batch = 1;

    size_t max_list_page = 1000;

};

using ObjStorageDeleteTask = std::function<ObjStorageResponse()>;

// Implementations must enqueue tasks immediately and remain reusable after wait() completes.

// Recycler implements this in cloud/src/recycler/s3_accessor.cpp; BE uses the synchronous fallback.

class ObjStorageDeleteExecutor {

public:

    virtual ~ObjStorageDeleteExecutor() = default;

    virtual ObjStorageResponse submit(ObjStorageDeleteTask task) = 0;

    virtual ObjStorageResponse wait() = 0;

};

struct ObjStorageRecursiveDeleteOptions {

    int64_t expiration_time = 0;

    size_t max_tasks_per_batch = 1;

    std::shared_ptr<ObjStorageDeleteExecutor> executor {};

};

class ObjStorageListIterator;

class RateLimitedObjStorageClient;

// Provider implementations are in common/cpp/obj-client/s3_obj_storage_client.cpp and

// common/cpp/obj-client/azure_obj_storage_client.cpp.

// Clients are shared-owned so a lazy list iterator can keep the complete decorator chain alive.

class ObjStorageClient : public std::enable_shared_from_this<ObjStorageClient> {

public:

    ObjStorageClient() = default;

    virtual ~ObjStorageClient() = default;

    ObjStorageClient(const ObjStorageClient&) = delete;

    ObjStorageClient& operator=(const ObjStorageClient&) = delete;

    // Create a multipart upload request. The returned token may be provider-issued or local and

    // identifies this writer's parts.

    // The input parameters should include the bucket and key for the object storage.

    virtual ObjStorageUploadResult create_multipart_upload(const ObjStoragePath& opts) = 0;

    // To directly upload a piece of data to object storage and generate a user-visible file.

    // You need to clearly specify the bucket and key

    virtual ObjStorageResponse put_object(const ObjStoragePath& opts, std::string_view stream) = 0;

    // Upload one part of a large object without making the object visible to users. upload_id is

    // the provider-issued or local writer token returned by create_multipart_upload.

    // Reusing the same bucket, key, upload_id, and part_num replaces that writer's staged part.

    virtual ObjStorageUploadResult upload_part(const ObjStoragePath& opts,

                                               const std::string& upload_id,

                                               std::string_view stream, int part_num) = 0;

    // Combine the parts belonging to upload_id into the key passed in opts. After this succeeds,

    // the complete object is visible in object storage.

    virtual ObjStorageResponse complete_multipart_upload(

            const ObjStoragePath& opts, const std::string& upload_id,

            const std::vector<ObjStorageCompletedPart>& completed_parts) = 0;

    // According to the passed bucket and key, it will access whether the corresponding file exists in the object storage.

    // If it exists, it will return the corresponding file size

    virtual ObjStorageHeadResult head_object(const ObjStoragePath& opts) = 0;

    // According to the bucket and key, it finds the corresponding file in the object storage

    // and starting from the offset, it reads bytes_read bytes into the buffer, with size_return recording the actual number of bytes read

    virtual ObjStorageResponse get_object(const ObjStoragePath& opts, void* buffer, size_t offset,

                                          size_t bytes_read, size_t* size_return) = 0;

    // Return a lazy iterator that fetches one provider page at a time.

    virtual std::unique_ptr<ObjStorageListIterator> list_objects(const ObjStoragePath& opts);

    // Collect all objects by consuming the lazy iterator. This preserves the eager BE API.

    // **Notice**: The files returned by this function contain the full key in object storage.

    virtual ObjStorageResponse list_objects(const ObjStoragePath& opts,

                                            std::vector<ObjectMeta>* objects);

    // According to the bucket and prefix specified by the user, it performs batch deletion based on the object names in the object array.

    virtual ObjStorageResponse delete_objects(const ObjStoragePath& opts,

                                              std::vector<std::string> objs) = 0;

    // Delete the file named key in the object storage bucket.

    virtual ObjStorageResponse delete_object(const ObjStoragePath& opts) = 0;

    virtual ObjStorageCapabilities capabilities() const = 0;

    // Return a presigned URL for users to access the object

    virtual std::string generate_presigned_url(const ObjStoragePath& opts,

                                               int64_t expiration_secs) = 0;

    // Get the objects' expiration time on the bucket

    virtual ObjStorageResponse get_lifecycle(const std::string& bucket,

                                             int64_t* expiration_days) = 0;

    // Check if the objects' versioning is on or off

    // returns 0 when versioning is on, otherwise versioning is off or check failed

    virtual ObjStorageResponse check_versioning(const std::string& bucket) = 0;

    virtual ObjStorageResponse abort_multipart_upload(const ObjStoragePath& path,

                                                      const std::string& upload_id) = 0;

    // Fetch at most one page. One call corresponds to exactly one provider request.

    virtual ObjStorageListPageResult list_objects_page(const ObjStoragePath& opts,

                                                       std::string_view continuation_token) = 0;

private:

    friend class ObjStorageListIterator;

    friend class RateLimitedObjStorageClient;

};

// A client-side iterator returned by ObjStorageClient. It requests one fixed-size page at a time.

class ObjStorageListIterator {

public:

    ObjStorageListIterator(std::shared_ptr<ObjStorageClient> client, ObjStoragePath opts)

            : client_(std::move(client)), opts_(std::move(opts)) {}

    bool is_valid() const { return is_valid_; }

    ObjStorageResponse has_next();

    ObjStorageListResult next();

private:

    std::shared_ptr<ObjStorageClient> client_;

    ObjStoragePath opts_;

    std::vector<ObjectMeta> objects_;

    size_t next_index_ = 0;

    std::string continuation_token_;

    bool has_more_ = true;

    bool is_valid_ = true;

};

// Provider-independent recursive deletion shared by concrete clients. Passing the complete client

// decorator chain keeps it alive for asynchronous delete tasks and applies policy per list page and

// delete batch.

ObjStorageResponse delete_objects_recursively(std::shared_ptr<ObjStorageClient> client,

                                              const ObjStoragePath& path,

                                              const ObjStorageRecursiveDeleteOptions& options = {});

} // namespace doris

// Keep the BE namespace spelling source-compatible while the implementation is

// shared with Recycler in `doris`.

namespace doris::io {

using ::doris::ObjStorageCapabilities;

using ::doris::ObjStorageClient;

using ::doris::ObjStorageCompletedPart;

using ::doris::ObjStorageDeleteExecutor;

using ::doris::ObjStorageDeleteTask;

using ::doris::ObjStorageEndpointInfo;

using ::doris::ObjStorageHeadResult;

using ::doris::ObjStorageListIterator;

using ::doris::ObjStorageListPageResult;

using ::doris::ObjStorageListResult;

using ::doris::ObjStoragePath;

using ::doris::ObjStorageProvider;

using ::doris::ObjStorageRecursiveDeleteOptions;

using ::doris::ObjStorageResponse;

using ::doris::ObjStorageStatus;

using ::doris::ObjStorageUploadResult;

using ::doris::ObjectMeta;

using ::doris::delete_objects_recursively;

} // namespace doris::io