// Copyright (C) 2019-2020 Zilliz. All rights reserved.
//
// Licensed 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 <string.h>
#include <exception>
#include <iosfwd>
#include <memory>
#include <string>
#include <unordered_map>
#include <utility>

#include "analyze_c.h"
#include "bitset/common.h"
#include "clustering/KmeansClustering.h"
#include "common/EasyAssert.h"
#include "common/Types.h"
#include "fmt/core.h"
#include "monitor/scope_metric.h"
#include "pb/clustering.pb.h"
#include "pb/schema.pb.h"
#include "storage/FileManager.h"
#include "storage/StorageV2FSCache.h"
#include "storage/Types.h"
#include "storage/Util.h"
#include "type_c.h"

using namespace milvus;

milvus::storage::StorageConfig
get_storage_config(const milvus::proto::clustering::StorageConfig& config) {
    auto storage_config = milvus::storage::StorageConfig();
    storage_config.address = config.address();
    storage_config.bucket_name = config.bucket_name();
    storage_config.access_key_id = config.access_keyid();
    storage_config.access_key_value = config.secret_access_key();
    storage_config.root_path = config.root_path();
    storage_config.storage_type = config.storage_type();
    storage_config.cloud_provider = config.cloud_provider();
    storage_config.iam_endpoint = config.iamendpoint();
    storage_config.useSSL = config.usessl();
    storage_config.sslCACert = config.sslcacert();
    storage_config.useIAM = config.useiam();
    storage_config.region = config.region();
    storage_config.useVirtualHost = config.use_virtual_host();
    storage_config.requestTimeoutMs = config.request_timeout_ms();
    storage_config.gcp_credential_json = config.gcpcredentialjson();
    storage_config.max_connections = config.max_connections();
    storage_config.tls_min_version = config.ssl_tls_min_version();
    storage_config.use_crc32c_checksum = config.use_crc32c_checksum();

    return storage_config;
}

CStatus
Analyze(CAnalyze* res_analyze,
        const uint8_t* serialized_analyze_info,
        const uint64_t len) {
    SCOPE_CGO_CALL_METRIC();

    try {
        auto analyze_info =
            std::make_unique<milvus::proto::clustering::AnalyzeInfo>();
        auto res = analyze_info->ParseFromArray(serialized_analyze_info, len);
        AssertInfo(res, "Unmarshal analyze info failed");
        auto field_type =
            static_cast<DataType>(analyze_info->field_schema().data_type());
        auto field_id = analyze_info->field_schema().fieldid();

        // init file manager
        milvus::storage::FieldDataMeta field_meta{analyze_info->collectionid(),
                                                  analyze_info->partitionid(),
                                                  0,
                                                  field_id};

        milvus::storage::IndexMeta index_meta{
            0, field_id, analyze_info->buildid(), analyze_info->version()};
        auto storage_config =
            get_storage_config(analyze_info->storage_config());
        auto chunk_manager =
            milvus::storage::CreateChunkManager(storage_config);
        auto fs = milvus::storage::StorageV2FSCache::Instance().Get({
            storage_config.address,
            storage_config.bucket_name,
            storage_config.access_key_id,
            storage_config.access_key_value,
            storage_config.root_path,
            storage_config.storage_type,
            storage_config.cloud_provider,
            storage_config.iam_endpoint,
            storage_config.log_level,
            storage_config.region,
            storage_config.useSSL,
            storage_config.sslCACert,
            storage_config.useIAM,
            storage_config.useVirtualHost,
            storage_config.requestTimeoutMs,
            false,
            storage_config.gcp_credential_json,
            false,
            storage_config.max_connections,
            storage_config.tls_min_version,
            storage_config.use_crc32c_checksum,
        });

        milvus::storage::FileManagerContext fileManagerContext(
            field_meta, index_meta, chunk_manager, fs);

        if (field_type != DataType::VECTOR_FLOAT) {
            throw SegcoreError(
                DataTypeInvalid,
                fmt::format("invalid data type for clustering is {}",
                            int(field_type)));
        }
        auto clusteringJob =
            std::make_unique<milvus::clustering::KmeansClustering>(
                fileManagerContext);

        clusteringJob->Run<float>(*analyze_info);
        *res_analyze = clusteringJob.release();
        auto status = CStatus();
        status.error_code = Success;
        status.error_msg = "";
        return status;
    } catch (SegcoreError& e) {
        auto status = CStatus();
        status.error_code = e.get_error_code();
        status.error_msg = strdup(e.what());
        return status;
    } catch (std::exception& e) {
        auto status = CStatus();
        status.error_code = UnexpectedError;
        status.error_msg = strdup(e.what());
        return status;
    }
}

CStatus
DeleteAnalyze(CAnalyze analyze) {
    SCOPE_CGO_CALL_METRIC();

    auto status = CStatus();
    try {
        AssertInfo(analyze, "failed to delete analyze, passed index was null");
        auto real_analyze =
            reinterpret_cast<milvus::clustering::KmeansClustering*>(analyze);
        delete real_analyze;
        status.error_code = Success;
        status.error_msg = "";
    } catch (std::exception& e) {
        status.error_code = UnexpectedError;
        status.error_msg = strdup(e.what());
    }
    return status;
}

CStatus
GetAnalyzeResultMeta(CAnalyze analyze,
                     const char** centroid_path,
                     int64_t* centroid_file_size,
                     void* id_mapping_paths,
                     int64_t* id_mapping_sizes) {
    SCOPE_CGO_CALL_METRIC();

    auto status = CStatus();
    try {
        AssertInfo(analyze,
                   "failed to serialize analyze to binary set, passed index "
                   "was null");
        auto real_analyze =
            reinterpret_cast<milvus::clustering::KmeansClustering*>(analyze);
        // Bind by reference: the pointers handed out below point into this
        // struct's std::string members, so they must reference the object owned
        // by real_analyze rather than a local copy that dies on return.
        const auto& res = real_analyze->GetClusteringResultMeta();
        *centroid_path = res.centroid_path.data();
        *centroid_file_size = res.centroid_file_size;

        const auto& map_ = res.id_mappings;
        const char** id_mapping_paths_ = (const char**)id_mapping_paths;
        size_t i = 0;
        for (auto it = map_.begin(); it != map_.end(); ++it, i++) {
            id_mapping_paths_[i] = it->first.data();
            id_mapping_sizes[i] = it->second;
        }
        status.error_code = Success;
        status.error_msg = "";
    } catch (std::exception& e) {
        status.error_code = UnexpectedError;
        status.error_msg = strdup(e.what());
    }
    return status;
}
