// Licensed to the LF AI & Data foundation 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
// export CPLUS_INCLUDE_PATH=/opt/homebrew/Cellar/boost/1.81.0_1/include/

#pragma once

#include <stdint.h>
#include <stdlib.h>
#include <atomic>
#include <exception>
#include <iostream>
#include <memory>
#include <mutex>
#include <string>
#include <utility>
#include <vector>

#include "AzureBlobChunkManager.h"
#include "azure/core/context.hpp"
#include "azure/core/diagnostics/logger.hpp"
#include "azure/core/exception.hpp"
#include "azure/core/http/http.hpp"
#include "azure/core/http/http_status_code.hpp"
#include "common/EasyAssert.h"
#include "glog/logging.h"
#include "log/Log.h"
#include "storage/ChunkManager.h"
#include "storage/Types.h"

namespace milvus {
namespace storage {

inline ErrorCode
AzureHttpStatusToErrorCode(Azure::Core::Http::HttpStatusCode status_code) {
    if (status_code == Azure::Core::Http::HttpStatusCode::NotFound) {
        return ErrorCode::ObjectNotExist;
    }
    if (status_code == Azure::Core::Http::HttpStatusCode::RequestTimeout ||
        status_code == Azure::Core::Http::HttpStatusCode::TooManyRequests ||
        status_code == Azure::Core::Http::HttpStatusCode::InternalServerError ||
        status_code == Azure::Core::Http::HttpStatusCode::BadGateway ||
        status_code == Azure::Core::Http::HttpStatusCode::ServiceUnavailable ||
        status_code == Azure::Core::Http::HttpStatusCode::GatewayTimeout) {
        return ErrorCode::S3Error;
    }
    return ErrorCode::UnexpectedError;
}

inline ErrorCode
AzureExceptionToErrorCode(const std::exception& error) {
    if (const auto* segcore_error = dynamic_cast<const SegcoreError*>(&error)) {
        return segcore_error->get_error_code();
    }
    if (dynamic_cast<const Azure::Core::OperationCancelledException*>(&error) !=
            nullptr ||
        dynamic_cast<const Azure::Core::Http::TransportException*>(&error) !=
            nullptr) {
        return ErrorCode::S3Error;
    }
    if (const auto* request_error =
            dynamic_cast<const Azure::Core::RequestFailedException*>(&error)) {
        return AzureHttpStatusToErrorCode(request_error->StatusCode);
    }
    return ErrorCode::UnexpectedError;
}

template <typename... Args>
static std::string
AzureErrorMessage(const std::string& func,
                  const std::exception& err,
                  const std::string& fmt_string,
                  Args&&... args) {
    std::ostringstream oss;
    const auto& message =
        fmt::format(fmt::runtime(fmt_string), std::forward<Args>(args)...);
    oss << "Error in " << func << "[exception:" << err.what()
        << ", params:" << message << "]";
    return oss.str();
}

template <typename... Args>
static SegcoreError
ThrowAzureError(const std::string& func,
                const std::exception& err,
                const std::string& fmt_string,
                Args&&... args) {
    std::string error_message =
        AzureErrorMessage(func, err, fmt_string, args...);
    LOG_WARN("{}", error_message);
    throw SegcoreError(AzureExceptionToErrorCode(err), error_message);
}

void
AzureLogger(Azure::Core::Diagnostics::Logger::Level level,
            std::string const& msg);

/**
 * @brief This AzureChunkManager is responsible for read and write file in blob.
   */
class AzureChunkManager : public ChunkManager {
 public:
    explicit AzureChunkManager(const StorageConfig& storage_config);

    AzureChunkManager(const AzureChunkManager&);
    AzureChunkManager&
    operator=(const AzureChunkManager&);

 public:
    virtual ~AzureChunkManager();

    virtual bool
    Exist(const std::string& filepath);

    virtual uint64_t
    Size(const std::string& filepath);

    virtual uint64_t
    Read(const std::string& filepath,
         uint64_t offset,
         void* buf,
         uint64_t len) {
        ThrowInfo(NotImplemented, GetName() + "Read with offset not implement");
    }

    virtual void
    Write(const std::string& filepath,
          uint64_t offset,
          void* buf,
          uint64_t len) {
        ThrowInfo(NotImplemented,
                  GetName() + "Write with offset not implement");
    }

    virtual uint64_t
    Read(const std::string& filepath, void* buf, uint64_t len);

    virtual void
    Write(const std::string& filepath, void* buf, uint64_t len);

    virtual std::vector<std::string>
    ListWithPrefix(const std::string& filepath);

    virtual void
    Remove(const std::string& filepath);

    virtual std::string
    GetName() const {
        return "AzureChunkManager";
    }

    virtual std::string
    GetRootPath() const {
        return path_prefix_;
    }

    virtual std::string
    GetBucketName() const {
        return default_bucket_name_;
    }

    inline void
    SetBucketName(const std::string& bucket_name) {
        default_bucket_name_ = bucket_name;
    }

    bool
    BucketExists(const std::string& bucket_name);

    bool
    CreateBucket(const std::string& bucket_name);

    bool
    DeleteBucket(const std::string& bucket_name);

    std::vector<std::string>
    ListBuckets();

 public:
    bool
    ObjectExists(const std::string& bucket_name,
                 const std::string& object_name);
    uint64_t
    GetObjectSize(const std::string& bucket_name,
                  const std::string& object_name);
    bool
    DeleteObject(const std::string& bucket_name,
                 const std::string& object_name);
    bool
    PutObjectBuffer(const std::string& bucket_name,
                    const std::string& object_name,
                    void* buf,
                    uint64_t size);
    uint64_t
    GetObjectBuffer(const std::string& bucket_name,
                    const std::string& object_name,
                    void* buf,
                    uint64_t size);
    std::vector<std::string>
    ListObjects(const std::string& bucket_name,
                const std::string& prefix = nullptr);

 private:
    static std::atomic<size_t> init_count_;
    static std::mutex client_mutex_;
    std::shared_ptr<azure::AzureBlobChunkManager> client_;
    std::string default_bucket_name_;
    std::string path_prefix_;
};

using AzureChunkManagerPtr = std::unique_ptr<AzureChunkManager>;

}  // namespace storage
}  // namespace milvus
