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

#include "s3_obj_storage_client.h"

#include <cpp/obj-client/obj_storage_client.h>

#include <gen_cpp/Status_types.h>

#include <algorithm>

#include <chrono>

#include "client_bvar.h"

#include "cpp/obj_retry_strategy.h"

namespace Aws::S3::Model {

class DeleteObjectRequest;

} // namespace Aws::S3::Model

using Aws::S3::Model::CompletedPart;

using Aws::S3::Model::CompletedMultipartUpload;

using Aws::S3::Model::CompleteMultipartUploadRequest;

using Aws::S3::Model::CreateMultipartUploadRequest;

using Aws::S3::Model::UploadPartRequest;

using Aws::S3::Model::UploadPartOutcome;

namespace doris {

using namespace Aws::S3::Model;

namespace {

constexpr int64_t S3_REQUEST_THRESHOLD_MS = 5000;

int64_t elapsed_time_milliseconds(std::chrono::steady_clock::time_point start) {

    return std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() -

                                                                 start)

            .count();

}

void record_s3_request_failed(const Aws::S3::S3Error& error) {

    record_object_request_failed(static_cast<int>(error.GetResponseCode()));

}

std::string object_identity(const ObjStoragePath& opts) {

    return opts.path.empty() ? opts.key : opts.path.native();

}

std::string s3_error_message(const Aws::S3::S3Error& error, std::string_view message) {

    return fmt::format("{}: {} {} code={}, type={}, request_id={}", message,

                       error.GetExceptionName(), error.GetMessage(),

                       static_cast<int>(error.GetResponseCode()),

                       static_cast<int>(error.GetErrorType()), error.GetRequestId());

}

} // namespace

ObjStorageStatus s3fs_error(const Aws::S3::S3Error& err, std::string_view msg) {

    return obj_storage_status_from_http_code(static_cast<int>(err.GetResponseCode()),

                                             s3_error_message(err, msg));

}

ObjStorageUploadResult S3ObjStorageClient::create_multipart_upload(const ObjStoragePath& opts) {

    CreateMultipartUploadRequest request;

    request.WithBucket(opts.bucket).WithKey(opts.key);

    request.SetContentType("application/octet-stream");

    const auto start = std::chrono::steady_clock::now();

    auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(

            [&]() {

                client_bvar::ScopedLatency scoped_latency(

                        client_bvar::s3_multi_part_upload_latency);

                return _client->CreateMultipartUpload(request);

            }(),

            "s3_file_writer::create_multi_part_upload", std::cref(request).get());

    SYNC_POINT_CALLBACK("s3_file_writer::_open", &outcome);

    const auto elapsed_ms = elapsed_time_milliseconds(start);

    const auto& request_id = outcome.IsSuccess() ? outcome.GetResult().GetRequestId()

                                                 : outcome.GetError().GetRequestId();

    LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS)

            << "CreateMultipartUpload cost=" << elapsed_ms << "ms"

            << ", request_id=" << request_id << ", bucket=" << opts.bucket << ", key=" << opts.key;

    if (!outcome.IsSuccess()) {

        record_s3_request_failed(outcome.GetError());

        auto st = s3fs_error(outcome.GetError(), fmt::format("failed to CreateMultipartUpload: {} ",

                                                             opts.path.native()));

        LOG(WARNING) << st.code << " request_id=" << request_id;

        return ObjStorageUploadResult {

                .resp = {.status = st,

                         .http_code = static_cast<int>(outcome.GetError().GetResponseCode()),

                         .request_id = outcome.GetError().GetRequestId()},

        };

    }

    return ObjStorageUploadResult {.resp = ObjStorageResponse::OK(),

                                   .upload_id {outcome.GetResult().GetUploadId()}};

}

ObjStorageResponse S3ObjStorageClient::put_object(const ObjStoragePath& opts,

                                                  std::string_view stream) {

    Aws::S3::Model::PutObjectRequest request;

    request.WithBucket(opts.bucket).WithKey(opts.key);

    auto string_view_stream = std::make_shared<StringViewStream>(stream.data(), stream.size());

    Aws::Utils::ByteBuffer part_md5(Aws::Utils::HashingUtils::CalculateMD5(*string_view_stream));

    request.SetContentMD5(Aws::Utils::HashingUtils::Base64Encode(part_md5));

    request.SetBody(string_view_stream);

    request.SetContentLength(stream.size());

    request.SetContentType("application/octet-stream");

    const auto start = std::chrono::steady_clock::now();

    auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(

            [&]() {

                client_bvar::ScopedLatency scoped_latency(client_bvar::s3_put_latency);

                return _client->PutObject(request);

            }(),

            "s3_file_writer::put_object", std::cref(request).get(), &stream);

    const auto elapsed_ms = elapsed_time_milliseconds(start);

    const auto& request_id = outcome.IsSuccess() ? outcome.GetResult().GetRequestId()

                                                 : outcome.GetError().GetRequestId();

    if (!outcome.IsSuccess()) {

        record_s3_request_failed(outcome.GetError());

        auto st = s3fs_error(outcome.GetError(),

                             fmt::format("failed to put object: {}", opts.path.native()));

        LOG(WARNING) << st.code << ", request_id=" << request_id;

        return ObjStorageResponse {

                .status = st,

                .http_code = static_cast<int>(outcome.GetError().GetResponseCode()),

                .request_id = outcome.GetError().GetRequestId()};

    }

    LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS)

            << "PutObject cost=" << elapsed_ms << "ms"

            << ", request_id=" << request_id << ", bucket=" << opts.bucket << ", key=" << opts.key;

    return ObjStorageResponse::OK();

}

ObjStorageUploadResult S3ObjStorageClient::upload_part(const ObjStoragePath& opts,

                                                       const std::string& upload_id,

                                                       std::string_view stream, int part_num) {

    UploadPartRequest request;

    request.WithBucket(opts.bucket)

            .WithKey(opts.key)

            .WithPartNumber(part_num)

            .WithUploadId(upload_id);

    auto string_view_stream = std::make_shared<StringViewStream>(stream.data(), stream.size());

    request.SetBody(string_view_stream);

    Aws::Utils::ByteBuffer part_md5(Aws::Utils::HashingUtils::CalculateMD5(*string_view_stream));

    request.SetContentMD5(Aws::Utils::HashingUtils::Base64Encode(part_md5));

    request.SetContentLength(stream.size());

    request.SetContentType("application/octet-stream");

    const auto start = std::chrono::steady_clock::now();

    auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(

            [&]() {

                client_bvar::ScopedLatency scoped_latency(

                        client_bvar::s3_multi_part_upload_latency);

                return _client->UploadPart(request);

            }(),

            "s3_file_writer::upload_part", std::cref(request).get(), &stream);

    const auto elapsed_ms = elapsed_time_milliseconds(start);

    const auto& request_id = outcome.IsSuccess() ? outcome.GetResult().GetRequestId()

                                                 : outcome.GetError().GetRequestId();

    TEST_SYNC_POINT_CALLBACK("S3FileWriter::_upload_one_part", &outcome);

    if (!outcome.IsSuccess()) {

        record_s3_request_failed(outcome.GetError());

        auto st = s3fs_error(outcome.GetError(),

                             fmt::format("failed to UploadPart: {}, part_num {}, upload_id={}",

                                         opts.path.native(), part_num, upload_id));

        LOG(WARNING) << st.code << ", request_id=" << request_id;

        return ObjStorageUploadResult {

                .resp = {.status = st,

                         .http_code = static_cast<int>(outcome.GetError().GetResponseCode()),

                         .request_id = outcome.GetError().GetRequestId()}};

    }

    LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS)

            << "UploadPart cost=" << elapsed_ms << "ms"

            << ", request_id=" << request_id << ", bucket=" << opts.bucket << ", key=" << opts.key

            << ", part_num=" << part_num << ", upload_id=" << upload_id;

    return ObjStorageUploadResult {.resp = ObjStorageResponse::OK(),

                                   .etag = outcome.GetResult().GetETag()};

}

ObjStorageResponse S3ObjStorageClient::complete_multipart_upload(

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

        const std::vector<ObjStorageCompletedPart>& completed_parts) {

    CompleteMultipartUploadRequest request;

    request.WithBucket(opts.bucket).WithKey(opts.key).WithUploadId(upload_id);

    CompletedMultipartUpload completed_upload;

    std::vector<CompletedPart> complete_parts;

    std::ranges::transform(completed_parts, std::back_inserter(complete_parts),

                           [](const ObjStorageCompletedPart& part_ptr) {

                               CompletedPart part;

                               part.SetPartNumber(part_ptr.part_num);

                               part.SetETag(part_ptr.etag);

                               return part;

                           });

    completed_upload.SetParts(std::move(complete_parts));

    request.WithMultipartUpload(completed_upload);

    TEST_SYNC_POINT_RETURN_WITH_VALUE("S3FileWriter::_complete:3", ObjStorageResponse(), this);

    const auto start = std::chrono::steady_clock::now();

    auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(

            [&]() {

                client_bvar::ScopedLatency scoped_latency(

                        client_bvar::s3_multi_part_upload_latency);

                return _client->CompleteMultipartUpload(request);

            }(),

            "s3_file_writer::complete_multi_part", std::cref(request).get());

    const auto elapsed_ms = elapsed_time_milliseconds(start);

    const auto& request_id = outcome.IsSuccess() ? outcome.GetResult().GetRequestId()

                                                 : outcome.GetError().GetRequestId();

    if (!outcome.IsSuccess()) {

        record_s3_request_failed(outcome.GetError());

        auto st = s3fs_error(outcome.GetError(),

                             fmt::format("failed to CompleteMultipartUpload: {}, upload_id={}",

                                         opts.path.native(), upload_id));

        LOG(WARNING) << st.code << ", request_id=" << request_id;

        return {.status = st,

                .http_code = static_cast<int>(outcome.GetError().GetResponseCode()),

                .request_id = outcome.GetError().GetRequestId()};

    }

    LOG_IF(INFO, elapsed_ms > S3_REQUEST_THRESHOLD_MS)

            << "CompleteMultipartUpload cost=" << elapsed_ms << "ms"

            << ", request_id=" << request_id << ", bucket=" << opts.bucket << ", key=" << opts.key

            << ", upload_id=" << upload_id;

    return ObjStorageResponse::OK();

}

ObjStorageHeadResult S3ObjStorageClient::head_object(const ObjStoragePath& opts) {

    Aws::S3::Model::HeadObjectRequest request;

    request.WithBucket(opts.bucket).WithKey(opts.key);

    auto outcome = SYNC_POINT_HOOK_RETURN_VALUE(

            [&]() {

                client_bvar::ScopedLatency scoped_latency(client_bvar::s3_head_latency);

                return _client->HeadObject(request);

            }(),

            "s3_file_system::head_object", std::ref(request).get());

    if (outcome.IsSuccess()) {

        return {.resp = ObjStorageResponse::OK(),

                .file_size = outcome.GetResult().GetContentLength()};

    } else if (outcome.GetError().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) {

        return {.resp = {.status = ObjStorageStatus::NOT_FOUND}, .file_size = 0};

    } else {

        record_s3_request_failed(outcome.GetError());

        LOG(WARNING) << "failed to head object"

                     << "bucket " << opts.bucket << " key " << opts.key << " responseCode "

                     << outcome.GetError() << " error " << outcome.GetError().GetMessage()

                     << " request_id " << outcome.GetError().GetRequestId();

        return {.resp = {.status = s3fs_error(

                                 outcome.GetError(),

                                 fmt::format("failed to head object: {}", object_identity(opts))),

                         .http_code = static_cast<int>(outcome.GetError().GetResponseCode()),

                         .request_id = outcome.GetError().GetRequestId()},

                .file_size = -1};

    }

}

ObjStorageResponse S3ObjStorageClient::get_object(const ObjStoragePath& opts, void* buffer,

                                                  size_t offset, size_t bytes_read,

                                                  size_t* size_return) {

    Aws::S3::Model::GetObjectRequest request;

    request.WithBucket(opts.bucket).WithKey(opts.key);

    request.SetRange(fmt::format("bytes={}-{}", offset, offset + bytes_read - 1));

    request.SetResponseStreamFactory(AwsWriteableStreamFactory(buffer, bytes_read));

    auto outcome = [&]() {

        client_bvar::ScopedLatency scoped_latency(client_bvar::s3_get_latency);

        return _client->GetObject(request);

    }();

    if (!outcome.IsSuccess()) {

        record_s3_request_failed(outcome.GetError());

        return ObjStorageResponse {

                .status = s3fs_error(outcome.GetError(), fmt::format("failed to get object: {}",

                                                                     object_identity(opts))),

                .http_code = static_cast<int>(outcome.GetError().GetResponseCode()),

                .request_id = outcome.GetError().GetRequestId(),

        };

    }

    *size_return = outcome.GetResult().GetContentLength();

    SYNC_POINT_CALLBACK("s3_obj_storage_client::get_object", size_return);

    if (*size_return != bytes_read) {

        const auto& request_id = outcome.GetResult().GetRequestId();

        return ObjStorageResponse {

                .status = {ObjStorageStatus::INTERNAL_ERROR,

                           fmt::format("incomplete read from {}, expect {}, got {}, request_id={}",

                                       object_identity(opts), bytes_read, *size_return,

                                       request_id)},

                .request_id = request_id};

    }

    return ObjStorageResponse::OK();

}

ObjStorageListPageResult S3ObjStorageClient::list_objects_page(

        const ObjStoragePath& opts, std::string_view continuation_token) {

    const auto& prefix = opts.prefix.empty() ? opts.key : opts.prefix;

    Aws::S3::Model::ListObjectsV2Request request;

    request.WithBucket(opts.bucket)

            .WithPrefix(prefix)

            .WithMaxKeys(static_cast<int>(capabilities().max_list_page));

    if (!continuation_token.empty()) {

        request.SetContinuationToken(std::string(continuation_token));

    }

    TEST_SYNC_POINT_CALLBACK("S3ObjStorageClient::list_objects", &request);

    auto outcome = [&]() {

        client_bvar::ScopedLatency scoped_latency(client_bvar::s3_list_latency);

        return _client->ListObjectsV2(request);

    }();

    const auto& request_id = outcome.IsSuccess() ? outcome.GetResult().GetRequestId()

                                                 : outcome.GetError().GetRequestId();

    if (!outcome.IsSuccess()) {

        // Some S3-compatible providers (for example TOS) return NoSuchKey instead of an empty page

        // when a prefix does not exist.

        if (outcome.GetError().GetErrorType() == Aws::S3::S3Errors::NO_SUCH_KEY) {

            LOG(INFO) << fmt::format(

                    "NoSuchKey when listing objects, treat as empty response, endpoint: {}, "

                    "bucket: {}, prefix: {}, request_id: {}",

                    _config.endpoint, request.GetBucket(), request.GetPrefix(), request_id);

            return {.resp = ObjStorageResponse::OK()};

        }

        record_object_request_failed(static_cast<int>(outcome.GetError().GetResponseCode()));

        const auto status = s3fs_error(outcome.GetError(),

                                       fmt::format("failed to list objects: {}, prefix: {}",

                                                   request.GetBucket(), request.GetPrefix()));

        LOG(WARNING) << fmt::format(

                "failed to list objects, endpoint: {}, bucket: {}, prefix: {}, responseCode: {}, "

                "error: {}, request_id: {}",

                _config.endpoint, request.GetBucket(), request.GetPrefix(),

                static_cast<int>(outcome.GetError().GetResponseCode()),

                outcome.GetError().GetMessage(), request_id);

        return {

                .resp = {.status = status,

                         .http_code = static_cast<int>(outcome.GetError().GetResponseCode()),

                         .request_id = request_id},

        };

    }

    const auto& result = outcome.GetResult();

    if (result.GetIsTruncated() && result.GetNextContinuationToken().empty()) {

        LOG(WARNING) << fmt::format(

                "failed to list objects, isTruncated but no continuation token, endpoint: {}, "

                "bucket: {}, prefix: {}, request_id: {}",

                _config.endpoint, request.GetBucket(), request.GetPrefix(), request_id);

        return {

                .resp = {.status = {ObjStorageStatus::INTERNAL_ERROR,

                                    fmt::format("failed to list objects: {}, prefix: {}",

                                                request.GetBucket(), request.GetPrefix())},

                         .http_code = 0,

                         .request_id = request_id},

        };

    }

    ObjStorageListPageResult page {

            .resp = ObjStorageResponse::OK(),

            .continuation_token = result.GetNextContinuationToken(),

            .has_more = result.GetIsTruncated(),

    };

    const auto& content = result.GetContents();

    page.objects.reserve(content.size());

    for (const auto& obj : content) {

        DCHECK(obj.GetKey().starts_with(request.GetPrefix()))

                << obj.GetKey() << ' ' << request.GetPrefix();

        page.objects.emplace_back(ObjectMeta {.key = obj.GetKey(),

                                              .size = obj.GetSize(),

                                              .mtime_s = obj.GetLastModified().Seconds()});

    }

    return page;

}

ObjStorageResponse S3ObjStorageClient::delete_objects(const ObjStoragePath& opts,

                                                      std::vector<std::string> objs) {

    size_t max_delete_batch = 1000;

    TEST_SYNC_POINT_CALLBACK("S3ObjClient::delete_objects", &max_delete_batch);

    TEST_SYNC_POINT_CALLBACK("S3ObjStorageClient::delete_objects", &max_delete_batch);

    max_delete_batch = std::max<size_t>(1, max_delete_batch);

    for (size_t begin = 0; begin < objs.size(); begin += max_delete_batch) {

        const size_t end = std::min(begin + max_delete_batch, objs.size());

        Aws::S3::Model::DeleteObjectsRequest delete_request;

        delete_request.SetBucket(opts.bucket);

        Aws::S3::Model::Delete del;

        Aws::Vector<Aws::S3::Model::ObjectIdentifier> objects;

        objects.reserve(end - begin);

        for (size_t i = begin; i < end; ++i) {

            Aws::S3::Model::ObjectIdentifier object;

            object.SetKey(std::move(objs[i]));

            objects.emplace_back(std::move(object));

        }

        del.WithObjects(std::move(objects)).SetQuiet(true);

        delete_request.SetDelete(std::move(del));

        auto delete_outcome = [&]() {

            client_bvar::ScopedLatency scoped_latency(client_bvar::s3_delete_objects_latency);

            return _client->DeleteObjects(delete_request);

        }();

        SYNC_POINT_CALLBACK("s3_obj_storage_client::delete_objects", &delete_outcome);

        SYNC_POINT_CALLBACK("s3_obj_storage_client::delete_objects_recursively", &delete_outcome);

        if (!delete_outcome.IsSuccess()) {

            record_s3_request_failed(delete_outcome.GetError());

            LOG(WARNING) << fmt::format(

                    "failed to delete objects, endpoint: {}, bucket: {}, key: {}, responseCode: "

                    "{}, error: {}, request_id: {}",

                    _config.endpoint, opts.bucket,

                    delete_request.GetDelete().GetObjects().front().GetKey(),

                    static_cast<int>(delete_outcome.GetError().GetResponseCode()),

                    delete_outcome.GetError().GetMessage(),

                    delete_outcome.GetError().GetRequestId());

            return ObjStorageResponse {

                    .status = s3fs_error(delete_outcome.GetError(),

                                         fmt::format("failed to delete dir {}", opts.key)),

                    .http_code = static_cast<int>(delete_outcome.GetError().GetResponseCode()),

                    .request_id = delete_outcome.GetError().GetRequestId()};

        }

        if (!delete_outcome.GetResult().GetErrors().empty()) {

            const auto& error = delete_outcome.GetResult().GetErrors().front();

            LOG(WARNING) << fmt::format(

                    "failed to delete object in batch, endpoint: {}, bucket: {}, key: {}, error "

                    "code: {}, error: {}, request_id: {}",

                    _config.endpoint, opts.bucket, error.GetKey(), error.GetCode(),

                    error.GetMessage(), delete_outcome.GetResult().GetRequestId());

            return ObjStorageResponse {

                    .status = {ObjStorageStatus::INTERNAL_ERROR,

                               fmt::format("failed to delete object {}: {}, request_id={}",

                                           error.GetKey(), error.GetMessage(),

                                           delete_outcome.GetResult().GetRequestId())},

                    .request_id = delete_outcome.GetResult().GetRequestId()};

        }

    }

    return ObjStorageResponse::OK();

}

ObjStorageResponse S3ObjStorageClient::delete_object(const ObjStoragePath& opts) {

    Aws::S3::Model::DeleteObjectRequest request;

    request.WithBucket(opts.bucket).WithKey(opts.key);

    auto outcome = [&]() {

        client_bvar::ScopedLatency scoped_latency(client_bvar::s3_delete_object_latency);

        return _client->DeleteObject(request);

    }();

    TEST_SYNC_POINT_CALLBACK("S3ObjClient::delete_object", &outcome);

    TEST_SYNC_POINT_CALLBACK("S3ObjStorageClient::delete_object", &outcome);

    if (outcome.IsSuccess()) {

        return ObjStorageResponse::OK();

    }

    ObjStorageResponse response {

            .status = s3fs_error(outcome.GetError(),

                                 fmt::format("failed to delete object {}", opts.key)),

            .http_code = static_cast<int>(outcome.GetError().GetResponseCode()),

            .request_id = outcome.GetError().GetRequestId()};

    if (response.status.code == ObjStorageStatus::NOT_FOUND) {

        return response;

    }

    record_s3_request_failed(outcome.GetError());

    LOG(WARNING) << fmt::format(

            "failed to delete object, endpoint: {}, bucket: {}, key: {}, responseCode: {}, "

            "error: {}, request_id: {}",

            _config.endpoint, opts.bucket, opts.key,

            static_cast<int>(outcome.GetError().GetResponseCode()), outcome.GetError().GetMessage(),

            outcome.GetError().GetRequestId());

    return response;

}

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

                                                       int64_t expiration_secs) {

    return _client->GeneratePresignedUrl(opts.bucket, opts.key, Aws::Http::HttpMethod::HTTP_GET,

                                         expiration_secs);

}

ObjStorageResponse S3ObjStorageClient::check_versioning(const std::string& bucket) {

    Aws::S3::Model::GetBucketVersioningRequest request;

    request.SetBucket(bucket);

    auto outcome = _client->GetBucketVersioning(request);

    if (outcome.IsSuccess()) {

        const auto& versioning_configuration = outcome.GetResult().GetStatus();

        if (versioning_configuration != Aws::S3::Model::BucketVersioningStatus::Enabled) {

            LOG(WARNING) << "Err for check interval: bucket doesn't enable bucket versioning"

                         << " endpoint=" << _config.endpoint << " bucket=" << bucket;

            return ObjStorageResponse {

                    .status = {ObjStorageStatus::INTERNAL_ERROR,

                               fmt::format("bucket versioning is not enabled: {}", bucket)}};

        }

    } else {

        record_s3_request_failed(outcome.GetError());

        LOG(WARNING) << "Err for check interval: failed to get status of bucket versioning"

                     << " endpoint=" << _config.endpoint << " bucket=" << bucket

                     << " responseCode=" << static_cast<int>(outcome.GetError().GetResponseCode())

                     << " error=" << outcome.GetError().GetMessage()

                     << " request_id=" << outcome.GetError().GetRequestId();

        return ObjStorageResponse {

                .status = s3fs_error(outcome.GetError(),

                                     fmt::format("failed to get bucket versioning: {}", bucket)),

                .http_code = static_cast<int>(outcome.GetError().GetResponseCode()),

                .request_id = outcome.GetError().GetRequestId()};

    }

    return ObjStorageResponse::OK();

}

ObjStorageResponse S3ObjStorageClient::abort_multipart_upload(const ObjStoragePath& opts,

                                                              const std::string& upload_id) {

    Aws::S3::Model::AbortMultipartUploadRequest request;

    request.WithBucket(opts.bucket).WithKey(opts.key).WithUploadId(upload_id);

    auto outcome = _client->AbortMultipartUpload(request);

    if (!outcome.IsSuccess()) {

        LOG(WARNING) << "failed to abort multipart upload"

                     << " endpoint=" << _config.endpoint << " bucket=" << opts.bucket

                     << " key=" << opts.key << " upload_id=" << upload_id

                     << " responseCode=" << static_cast<int>(outcome.GetError().GetResponseCode())

                     << " error=" << outcome.GetError().GetMessage()

                     << " request_id=" << outcome.GetError().GetRequestId();

        if (outcome.GetError().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) {

            return ObjStorageResponse::OK();

        }

        record_s3_request_failed(outcome.GetError());

        return ObjStorageResponse {

                .status =

                        s3fs_error(outcome.GetError(),

                                   fmt::format("failed to abort multipart upload: {}, upload_id={}",

                                               opts.path.native(), upload_id)),

                .http_code = static_cast<int>(outcome.GetError().GetResponseCode()),

                .request_id = outcome.GetError().GetRequestId(),

        };

    }

    return ObjStorageResponse::OK();

}

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

                                                     int64_t* expiration_days) {

    Aws::S3::Model::GetBucketLifecycleConfigurationRequest request;

    request.SetBucket(bucket);

    auto outcome = _client->GetBucketLifecycleConfiguration(request);

    bool has_lifecycle = false;

    if (outcome.IsSuccess()) {

        const auto& rules = outcome.GetResult().GetRules();

        for (const auto& rule : rules) {

            if (rule.NoncurrentVersionExpirationHasBeenSet()) {

                has_lifecycle = true;

                *expiration_days = rule.GetNoncurrentVersionExpiration().GetNoncurrentDays();

            }

        }

    } else {

        record_s3_request_failed(outcome.GetError());

        LOG(WARNING) << "Err for check interval: failed to get bucket lifecycle"

                     << " endpoint=" << _config.endpoint << " bucket=" << bucket

                     << " responseCode=" << static_cast<int>(outcome.GetError().GetResponseCode())

                     << " error=" << outcome.GetError().GetMessage()

                     << " request_id=" << outcome.GetError().GetRequestId();

        return ObjStorageResponse {

                .status = s3fs_error(outcome.GetError(),

                                     fmt::format("failed to get lift cycle: {}", bucket)),

                .http_code = static_cast<int>(outcome.GetError().GetResponseCode()),

                .request_id = outcome.GetError().GetRequestId()};

    }

    if (!has_lifecycle) {

        LOG(WARNING) << "Err for check interval: bucket doesn't have lifecycle configuration"

                     << " endpoint=" << _config.endpoint << " bucket=" << bucket;

        return ObjStorageResponse {

                .status = {ObjStorageStatus::NOT_FOUND,

                           fmt::format("bucket has no lifecycle configuration: {}", bucket)}};

    }

    return ObjStorageResponse::OK();

}

} // namespace doris