// 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 "obj_storage_client.h"

#include <cpp/sync_point.h>

#include <glog/logging.h>

#include <algorithm>

#include <chrono>

namespace doris {

std::unique_ptr<ObjStorageListIterator> ObjStorageClient::list_objects(const ObjStoragePath& opts) {

    return std::make_unique<ObjStorageListIterator>(shared_from_this(), opts);

}

ObjStorageResponse ObjStorageClient::list_objects(const ObjStoragePath& opts,

                                                  std::vector<ObjectMeta>* objects) {

    objects->clear();

    auto iter = list_objects(opts);

    for (;;) {

        auto result = iter->next();

        if (!result.object.has_value()) {

            if (!result.resp.ok()) {

                objects->clear();

            }

            return result.resp;

        }

        objects->emplace_back(std::move(*result.object));

    }

}

ObjStorageResponse ObjStorageListIterator::has_next() {

    if (!is_valid_) {

        return {

                .status = {TStatusCode::INTERNAL_ERROR, "Iterator is invalid"},

                .http_code = 0,

        };

    }

    while (next_index_ == objects_.size()) {

        if (!has_more_) {

            return {

                    .status = {TStatusCode::END_OF_FILE, "No more results"},

                    .http_code = 200,

            };

        }

        auto page = client_->list_objects_page(opts_, continuation_token_);

        if (!page.resp.ok()) {

            is_valid_ = false;

            return page.resp;

        }

        objects_ = std::move(page.objects);

        next_index_ = 0;

        continuation_token_ = std::move(page.continuation_token);

        has_more_ = page.has_more;

    }

    return ObjStorageResponse::OK();

}

ObjStorageListResult ObjStorageListIterator::next() {

    auto response = has_next();

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

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

    }

    if (!response.ok()) {

        return {.resp = std::move(response), .object = {}};

    }

    return {

            .resp = ObjStorageResponse::OK(),

            .object = std::move(objects_[next_index_++]),

    };

}

ObjStorageResponse ObjStorageClient::delete_objects_recursively(

        const ObjStoragePath& path, const ObjStorageRecursiveDeleteOptions& delete_options) {

    return delete_objects_recursively_impl(path, delete_options);

}

ObjStorageResponse ObjStorageClient::delete_objects_recursively_impl(

        const ObjStoragePath& path, const ObjStorageRecursiveDeleteOptions& delete_options) {

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

    auto list_path = path;

    if (list_path.prefix.empty()) {

        list_path.prefix = list_path.key;

    }

    auto delete_batch_size = std::max<size_t>(1, capabilities().max_delete_batch);

    TEST_SYNC_POINT_CALLBACK("ObjStorageClient::delete_objects_recursively_", &delete_batch_size);

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

    const auto max_tasks_per_batch = std::max<size_t>(1, delete_options.max_tasks_per_batch);

    std::vector<std::string> keys;

    keys.reserve(delete_batch_size);

    size_t pending_tasks = 0;

    size_t total_batches = 0;

    size_t num_deleted = 0;

    size_t error_count = 0;

    auto first_error = ObjStorageResponse::OK();

    auto elapsed_milliseconds = [&]() {

        return std::chrono::duration_cast<std::chrono::milliseconds>(

                       std::chrono::steady_clock::now() - start_time)

                .count();

    };

    auto finish = [&](ObjStorageResponse response) {

        LOG(INFO) << "delete objects under " << list_path.bucket << "/" << list_path.prefix

                  << " finished, ret=" << response.status.code

                  << ", total_batches=" << total_batches << ", num_deleted=" << num_deleted

                  << ", error_count=" << error_count << ", cost=" << elapsed_milliseconds()

                  << " ms";

        return response;

    };

    auto record_error = [&](ObjStorageResponse response) {

        if (response.ok()) {

            return;

        }

        ++error_count;

        if (first_error.ok()) {

            first_error = std::move(response);

        }

    };

    auto wait_for_tasks = [&]() {

        if (pending_tasks == 0) {

            return ObjStorageResponse::OK();

        }

        const auto tasks_in_batch = pending_tasks;

        pending_tasks = 0;

        auto response = delete_options.executor ? delete_options.executor->wait()

                                                : ObjStorageResponse::OK();

        ++total_batches;

        LOG(INFO) << "delete objects under " << list_path.bucket << "/" << list_path.prefix

                  << " batch " << total_batches << " completed"

                  << ", tasks_in_batch=" << tasks_in_batch << ", total_deleted=" << num_deleted

                  << ", elapsed=" << elapsed_milliseconds() << " ms";

        return response;

    };

    auto submit_delete_task = [&]() -> bool {

        ObjStorageDeleteTask task = [this, bucket = path.bucket,

                                     batch = std::move(keys)]() mutable {

            return delete_objects(ObjStoragePath {.bucket = std::move(bucket)}, std::move(batch));

        };

        keys.clear();

        keys.reserve(delete_batch_size);

        ObjStorageResponse response;

        if (delete_options.executor) {

            response = delete_options.executor->submit(std::move(task));

        } else {

            response = task();

        }

        ++pending_tasks;

        if (!response.ok()) {

            record_error(std::move(response));

            record_error(wait_for_tasks());

            return false;

        }

        if (pending_tasks == max_tasks_per_batch) {

            record_error(wait_for_tasks());

        }

        // Match the pre-refactor Recycler behavior: do not scan the next task batch after the

        // current batch reports a submit, delete, or wait failure.

        return first_error.ok();

    };

    auto iter = list_objects(list_path);

    for (;;) {

        auto result = iter->next();

        if (!result.object.has_value()) {

            if (result.resp.ok()) {

                break;

            }

            if (!keys.empty()) {

                submit_delete_task();

            }

            record_error(wait_for_tasks());

            record_error(std::move(result.resp));

            return finish(std::move(first_error));

        }

        auto& object = *result.object;

        if (delete_options.expiration_time > 0 && object.mtime_s > delete_options.expiration_time) {

            continue;

        }

        ++num_deleted;

        keys.emplace_back(std::move(object.key));

        if (keys.size() == delete_batch_size && !submit_delete_task()) {

            return finish(std::move(first_error));

        }

    }

    if (!keys.empty() && !submit_delete_task()) {

        return finish(std::move(first_error));

    }

    record_error(wait_for_tasks());

    return finish(std::move(first_error));

}

} // namespace doris