// The MIT License (MIT)
//
// Copyright (c) 11/27/18 nick. <nbatkins@gmail.com>
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.#include "consumer.hpp"

#include <cpp_redis/core/consumer.hpp>

#include <functional>

using std::bind;
using namespace std::placeholders;

namespace cpp_redis {

consumer::client_container::client_container() : ack_client(), poll_client() {}

consumer::consumer(string_t stream, string_t consumer,
                   size_t max_concurrency)
    : m_callbacks(), m_stream(std::move(stream)), m_name(std::move(consumer)),
      m_read_id("0"), m_block_sec(-1), m_max_concurrency(max_concurrency),
      m_read_count(static_cast<int>(max_concurrency)) {
  // Supply the dispatch queue a callback to notify the queue when it is at max
  // capacity
  m_dispatch_queue = dispatch_queue_ptr_t(new dispatch_queue(
      stream, [&](size_t size) { dispatch_changed_handler(size); },
      max_concurrency));
  m_client = consumer::client_container_ptr_t(new consumer::client_container());
}

consumer_t &consumer::subscribe(
    const string_t &group, const consumer_callback_t &consumer_callback,
    const acknowledgement_callback_t &acknowledgement_callback) {
  m_callbacks.insert({group, {consumer_callback, acknowledgement_callback}});
  return *this;
}

void consumer::dispatch_changed_handler(size_t size) {
  if (size >= m_max_concurrency) {
    dispatch_queue_full.store(true);
    dispatch_queue_changed.notify_all();
  }
}

void consumer::connect(const string_t &host, size_t port,
                       const connect_callback_t &connect_callback,
                       int timeout_ms, int max_reconnects,
                       int reconnect_interval_ms) {
  m_client->ack_client.connect(host, port, connect_callback, timeout_ms,
                               max_reconnects, reconnect_interval_ms);
  m_client->poll_client.connect(host, port, connect_callback, timeout_ms,
                                max_reconnects, reconnect_interval_ms);
}

void consumer::auth(const string_t &password,
                    const reply_callback_t &reply_callback) {
  m_client->ack_client.auth(password, reply_callback);
  m_client->poll_client.auth(password, reply_callback);
}

consumer_t &consumer::commit() {
  while (!is_ready) {
    if (!is_ready) {
      mutex_lock_t dispatch_lock(dispatch_queue_changed_mutex);
      dispatch_queue_changed.wait(
          dispatch_lock, [&]() { return !dispatch_queue_full.load(); });
      m_read_count =
          static_cast<int>(m_max_concurrency - m_dispatch_queue->size());
      poll();
    }
  }
  return *this;
}

void consumer::dispatch(
    const xmessage_t &message,
    const pair<string_t, consumer_callback_container_t> &cb) {
  m_dispatch_queue->dispatch(message, [&](const message_type &message) {
    auto response = cb.second.consumer_callback(message);

    // add results to result stream
    m_client->ack_client.xadd(m_stream + ":results", CPP_REDIS_WILD_CARD,
                              response);

    // acknowledge task completion
    m_client->ack_client
        .xack(m_stream, cb.first, {message.get_id()},
              [&](const reply_t &r) {
                if (r.is_integer()) {
                  auto ret_int = r.as_integer();
                  cb.second.acknowledgement_callback(ret_int);
                }
              })
        .sync_commit();
    return response;
  });
}

void consumer::poll() {
  for (auto &cb : m_callbacks) {
    m_client->poll_client
        .xreadgroup(
            {cb.first,
             m_name,
             {{m_stream}, {m_read_id}},
             m_read_count,
             m_block_sec,
             false},
            [&](reply_t &reply) {
              // The reply is an array if valid
              xstream_reply s_reply(reply);
              if (!s_reply.is_null()) {
                __CPP_REDIS_LOG(2, "Stream " << s_reply)
                for (const auto &stream : s_reply) {
                  for (auto &m : stream.Messages) {
                    if (m_should_read_pending.load())
                      m_read_id = m.get_id();
                    try {
                      dispatch(m, cb);
                    } catch (std::exception &exc) {
                      __CPP_REDIS_LOG(
                          1, "Processing failed for message id: " + m.get_id() +
                                 "\nDetails: " + exc.what());
                      throw exc;
                    }
                  }
                }
              } else {
                check_for_pending();
              }
              return;
            })
        .sync_commit();
  }
}
} // namespace cpp_redis