/**
 * 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 "ProducerConfig.h"
#include <map>

static const std::string CFG_TOPIC = "topic";
static const std::string CFG_PRODUCER_NAME = "producerName";
static const std::string CFG_SEND_TIMEOUT = "sendTimeoutMs";
static const std::string CFG_INIT_SEQUENCE_ID = "initialSequenceId";
static const std::string CFG_MAX_PENDING = "maxPendingMessages";
static const std::string CFG_MAX_PENDING_ACROSS_PARTITIONS = "maxPendingMessagesAcrossPartitions";
static const std::string CFG_BLOCK_IF_QUEUE_FULL = "blockIfQueueFull";
static const std::string CFG_ROUTING_MODE = "messageRoutingMode";
static const std::string CFG_HASH_SCHEME = "hashingScheme";
static const std::string CFG_COMPRESS_TYPE = "compressionType";
static const std::string CFG_BATCH_ENABLED = "batchingEnabled";
static const std::string CFG_BATCH_MAX_DELAY = "batchingMaxPublishDelayMs";
static const std::string CFG_BATCH_MAX_MSG = "batchingMaxMessages";
static const std::string CFG_PROPS = "properties";

static const std::map<std::string, pulsar_partitions_routing_mode> MESSAGE_ROUTING_MODE = {
    {"UseSinglePartition", pulsar_UseSinglePartition},
    {"RoundRobinDistribution", pulsar_RoundRobinDistribution},
    {"CustomPartition", pulsar_CustomPartition}};

static const std::map<std::string, pulsar_hashing_scheme> HASHING_SCHEME = {
    {"Murmur3_32Hash", pulsar_Murmur3_32Hash},
    {"BoostHash", pulsar_BoostHash},
    {"JavaStringHash", pulsar_JavaStringHash},
};

static std::map<std::string, pulsar_compression_type> COMPRESSION_TYPE = {{"Zlib", pulsar_CompressionZLib},
                                                                          {"LZ4", pulsar_CompressionLZ4}};

ProducerConfig::ProducerConfig(const Napi::Object& producerConfig) : topic("") {
  this->cProducerConfig = pulsar_producer_configuration_create();

  if (producerConfig.Has(CFG_TOPIC) && producerConfig.Get(CFG_TOPIC).IsString()) {
    this->topic = producerConfig.Get(CFG_TOPIC).ToString().Utf8Value();
  }

  if (producerConfig.Has(CFG_PRODUCER_NAME) && producerConfig.Get(CFG_PRODUCER_NAME).IsString()) {
    std::string producerName = producerConfig.Get(CFG_PRODUCER_NAME).ToString().Utf8Value();
    if (!producerName.empty())
      pulsar_producer_configuration_set_producer_name(this->cProducerConfig, producerName.c_str());
  }

  if (producerConfig.Has(CFG_SEND_TIMEOUT) && producerConfig.Get(CFG_SEND_TIMEOUT).IsNumber()) {
    int32_t sendTimeoutMs = producerConfig.Get(CFG_SEND_TIMEOUT).ToNumber().Int32Value();
    if (sendTimeoutMs > 0) {
      pulsar_producer_configuration_set_send_timeout(this->cProducerConfig, sendTimeoutMs);
    }
  }

  if (producerConfig.Has(CFG_INIT_SEQUENCE_ID) && producerConfig.Get(CFG_INIT_SEQUENCE_ID).IsNumber()) {
    int64_t initialSequenceId = producerConfig.Get(CFG_INIT_SEQUENCE_ID).ToNumber().Int64Value();
    pulsar_producer_configuration_set_initial_sequence_id(this->cProducerConfig, initialSequenceId);
  }

  if (producerConfig.Has(CFG_MAX_PENDING) && producerConfig.Get(CFG_MAX_PENDING).IsNumber()) {
    int32_t maxPendingMessages = producerConfig.Get(CFG_MAX_PENDING).ToNumber().Int32Value();
    if (maxPendingMessages > 0) {
      pulsar_producer_configuration_set_max_pending_messages(this->cProducerConfig, maxPendingMessages);
    }
  }

  if (producerConfig.Has(CFG_MAX_PENDING_ACROSS_PARTITIONS) &&
      producerConfig.Get(CFG_MAX_PENDING_ACROSS_PARTITIONS).IsNumber()) {
    int32_t maxPendingMessagesAcrossPartitions =
        producerConfig.Get(CFG_MAX_PENDING_ACROSS_PARTITIONS).ToNumber().Int32Value();
    if (maxPendingMessagesAcrossPartitions > 0) {
      pulsar_producer_configuration_set_max_pending_messages(this->cProducerConfig,
                                                             maxPendingMessagesAcrossPartitions);
    }
  }

  if (producerConfig.Has(CFG_BLOCK_IF_QUEUE_FULL) &&
      producerConfig.Get(CFG_BLOCK_IF_QUEUE_FULL).IsBoolean()) {
    bool blockIfQueueFull = producerConfig.Get(CFG_BLOCK_IF_QUEUE_FULL).ToBoolean().Value();
    pulsar_producer_configuration_set_block_if_queue_full(this->cProducerConfig, blockIfQueueFull);
  }

  if (producerConfig.Has(CFG_ROUTING_MODE) && producerConfig.Get(CFG_ROUTING_MODE).IsString()) {
    std::string messageRoutingMode = producerConfig.Get(CFG_ROUTING_MODE).ToString().Utf8Value();
    if (MESSAGE_ROUTING_MODE.count(messageRoutingMode))
      pulsar_producer_configuration_set_partitions_routing_mode(this->cProducerConfig,
                                                                MESSAGE_ROUTING_MODE.at(messageRoutingMode));
  }

  if (producerConfig.Has(CFG_HASH_SCHEME) && producerConfig.Get(CFG_HASH_SCHEME).IsString()) {
    std::string hashingScheme = producerConfig.Get(CFG_HASH_SCHEME).ToString().Utf8Value();
    if (HASHING_SCHEME.count(hashingScheme))
      pulsar_producer_configuration_set_hashing_scheme(this->cProducerConfig,
                                                       HASHING_SCHEME.at(hashingScheme));
  }

  if (producerConfig.Has(CFG_COMPRESS_TYPE) && producerConfig.Get(CFG_COMPRESS_TYPE).IsString()) {
    std::string compressionType = producerConfig.Get(CFG_COMPRESS_TYPE).ToString().Utf8Value();
    if (COMPRESSION_TYPE.count(compressionType))
      pulsar_producer_configuration_set_compression_type(this->cProducerConfig,
                                                         COMPRESSION_TYPE.at(compressionType));
  }

  if (producerConfig.Has(CFG_BATCH_ENABLED) && producerConfig.Get(CFG_BATCH_ENABLED).IsBoolean()) {
    bool batchingEnabled = producerConfig.Get(CFG_BATCH_ENABLED).ToBoolean().Value();
    pulsar_producer_configuration_set_batching_enabled(this->cProducerConfig, batchingEnabled);
  }

  if (producerConfig.Has(CFG_BATCH_MAX_DELAY) && producerConfig.Get(CFG_BATCH_MAX_DELAY).IsNumber()) {
    int64_t batchingMaxPublishDelayMs = producerConfig.Get(CFG_BATCH_MAX_DELAY).ToNumber().Int64Value();
    if (batchingMaxPublishDelayMs > 0) {
      pulsar_producer_configuration_set_batching_max_publish_delay_ms(this->cProducerConfig,
                                                                      (long)batchingMaxPublishDelayMs);
    }
  }

  if (producerConfig.Has(CFG_BATCH_MAX_MSG) && producerConfig.Get(CFG_BATCH_MAX_MSG).IsNumber()) {
    uint32_t batchingMaxMessages = producerConfig.Get(CFG_BATCH_MAX_MSG).ToNumber().Uint32Value();
    if (batchingMaxMessages > 0) {
      pulsar_producer_configuration_set_batching_max_messages(this->cProducerConfig, batchingMaxMessages);
    }
  }

  if (producerConfig.Has(CFG_PROPS) && producerConfig.Get(CFG_PROPS).IsObject()) {
    Napi::Object propObj = producerConfig.Get(CFG_PROPS).ToObject();
    Napi::Array arr = propObj.GetPropertyNames();
    int size = arr.Length();
    for (int i = 0; i < size; i++) {
      Napi::String key = arr.Get(i).ToString();
      Napi::String value = propObj.Get(key).ToString();
      pulsar_producer_configuration_set_property(this->cProducerConfig, key.Utf8Value().c_str(),
                                                 value.Utf8Value().c_str());
    }
  }
}

ProducerConfig::~ProducerConfig() { pulsar_producer_configuration_free(this->cProducerConfig); }

pulsar_producer_configuration_t* ProducerConfig::GetCProducerConfig() { return this->cProducerConfig; }

std::string ProducerConfig::GetTopic() { return this->topic; }