// Copyright (C) 2013-2026 Internet Systems Consortium, Inc. ("ISC")

//

// This Source Code Form is subject to the terms of the Mozilla Public

// License, v. 2.0. If a copy of the MPL was not distributed with this

// file, You can obtain one at http://mozilla.org/MPL/2.0/.

#include <config.h>

#include <asiolink/asio_wrapper.h>

#include <dhcp_ddns/dhcp_ddns_log.h>

#include <dhcp_ddns/ncr_io.h>

#include <util/multi_threading_mgr.h>

#include <boost/algorithm/string/predicate.hpp>

#include <mutex>

namespace isc {

namespace dhcp_ddns {

using namespace isc::util;

using namespace std;

NameChangeProtocol stringToNcrProtocol(const std::string& protocol_str) {

    if (boost::iequals(protocol_str, "UDP")) {

        return (NCR_UDP);

    }

    if (boost::iequals(protocol_str, "TCP")) {

        return (NCR_TCP);

    }

    isc_throw(BadValue,

              "Invalid NameChangeRequest protocol: " << protocol_str);

}

std::string ncrProtocolToString(NameChangeProtocol protocol) {

    switch (protocol) {

    case NCR_UDP:

        return ("UDP");

    case NCR_TCP:

        return ("TCP");

    default:

        break;

    }

    std::ostringstream stream;

    stream  << "UNKNOWN(" << protocol << ")";

    return (stream.str());

}

//************************** NameChangeListener ***************************

NameChangeListener::NameChangeListener(RequestReceiveHandlerPtr recv_handler)

    : listening_(false), io_pending_(false), recv_handler_(recv_handler) {

};

void

NameChangeListener::startListening(const isc::asiolink::IOServicePtr& io_service) {

    if (amListening()) {

        // This amounts to a programmatic error.

        isc_throw(NcrListenerError, "NameChangeListener is already listening");

    }

    // Call implementation dependent open.

    try {

        open(io_service);

    } catch (const isc::Exception& ex) {

        stopListening();

        isc_throw(NcrListenerOpenError, "Open failed: " << ex.what());

    }

    // Set our status to listening.

    setListening(true);

    // Start the first asynchronous receive.

    try {

        receiveNext();

    } catch (const isc::Exception& ex) {

        stopListening();

        isc_throw(NcrListenerReceiveError, "doReceive failed: " << ex.what());

    }

}

void

NameChangeListener::receiveNext() {

    io_pending_ = true;

    doReceive();

}

void

NameChangeListener::stopListening() {

    try {

        // Call implementation dependent close.

        close();

    } catch (const isc::Exception &ex) {

        // Swallow exceptions. If we have some sort of error we'll log

        // it but we won't propagate the throw.

        LOG_ERROR(dhcp_ddns_logger, DHCP_DDNS_NCR_LISTEN_CLOSE_ERROR)

                  .arg(ex.what());

    }

    // Set it false, no matter what.  This allows us to at least try to

    // re-open via startListening().

    setListening(false);

}

void

NameChangeListener::invokeRecvHandler(const Result result,

                                      NameChangeRequestPtr& ncr) {

    // Call the registered application layer handler.

    // Surround the invocation with a try-catch. The invoked handler is

    // not supposed to throw, but in the event it does we will at least

    // report it.

    try {

        io_pending_ = false;

        (*recv_handler_)(result, ncr);

    } catch (const std::exception& ex) {

        LOG_ERROR(dhcp_ddns_logger, DHCP_DDNS_UNCAUGHT_NCR_RECV_HANDLER_ERROR)

                  .arg(ex.what());

    }

    // Start the next IO layer asynchronous receive.

    scheduleNextReceive();

}

void

NameChangeListener::scheduleNextReceive() {

    // In the event the application handler decided to stop listening

    // we need to check that first.

    if (!amListening()) {

        return;

    }

    try {

        receiveNext();

    } catch (const isc::Exception& isc_ex) {

        // It is possible though unlikely, for doReceive to fail without

        // scheduling the read. While, unlikely, it does mean the callback

        // will not get called with a failure. A throw here would surface

        // at the IOService::run (or run variant) invocation.  So we will

        // close the window by invoking the application handler with

        // a failed result, and let the application layer sort it out.

        LOG_ERROR(dhcp_ddns_logger, DHCP_DDNS_NCR_RECV_NEXT_ERROR)

                  .arg(isc_ex.what());

        // Call the registered application layer handler.

        // Surround the invocation with a try-catch. The invoked handler is

        // not supposed to throw, but in the event it does we will at least

        // report it.

        NameChangeRequestPtr empty;

        try {

            io_pending_ = false;

            (*recv_handler_)(ERROR, empty);

        } catch (const std::exception& std_ex) {

            LOG_ERROR(dhcp_ddns_logger,

                      DHCP_DDNS_UNCAUGHT_NCR_RECV_HANDLER_ERROR)

                      .arg(std_ex.what());

        }

    }

}

//************************* NameChangeSender ******************************

NameChangeSender::NameChangeSender(RequestSendHandlerPtr send_handler,

                                   size_t send_queue_max)

    : sending_(false), send_handler_(send_handler),

      send_queue_max_(send_queue_max), mutex_(new mutex()) {

    // Queue size must be big enough to hold at least 1 entry.

    setQueueMaxSize(send_queue_max);

}

void

NameChangeSender::startSending(const isc::asiolink::IOServicePtr& io_service) {

    if (amSending()) {

        // This amounts to a programmatic error.

        isc_throw(NcrSenderError, "NameChangeSender is already sending");

    }

    // Call implementation dependent open.

    try {

        if (MultiThreadingMgr::instance().getMode()) {

            lock_guard<mutex> lock(*mutex_);

            startSendingInternal(io_service);

        } else {

            startSendingInternal(io_service);

        }

    } catch (const isc::Exception& ex) {

        stopSending();

        isc_throw(NcrSenderOpenError, "Open failed: " << ex.what());

    }

}

void

NameChangeSender::startSendingInternal(const isc::asiolink::IOServicePtr& io_service) {

    // Clear send marker.

    ncr_to_send_.reset();

    // Remember io service we're given.

    io_service_ = io_service;

    open(io_service);

    // Set our status to sending.

    setSending(true);

    // If there's any queued already.. we'll start sending.

    sendNext();

}

void

NameChangeSender::stopSending() {

    // Set it send indicator to false, no matter what. This allows us to at

    // least try to re-open via startSending(). Also, setting it false now,

    // allows us to break sendNext() chain in invokeSendHandler.

    setSending(false);

    // If there is an outstanding IO to complete, attempt to process it.

    if (ioReady() && io_service_) {

        try {

            runReadyIO();

        } catch (const std::exception& ex) {

            // Swallow exceptions. If we have some sort of error we'll log

            // it but we won't propagate the throw.

            LOG_ERROR(dhcp_ddns_logger,

                  DHCP_DDNS_NCR_FLUSH_IO_ERROR).arg(ex.what());

        }

    }

    try {

        // Call implementation dependent close.

        close();

    } catch (const isc::Exception &ex) {

        // Swallow exceptions. If we have some sort of error we'll log

        // it but we won't propagate the throw.

        LOG_ERROR(dhcp_ddns_logger,

                  DHCP_DDNS_NCR_SEND_CLOSE_ERROR).arg(ex.what());

    }

    if (io_service_) {

        try {

            io_service_->stopAndPoll(false);

        } catch (const std::exception& ex) {

            // Swallow exceptions. If we have some sort of error we'll log

            // it but we won't propagate the throw.

            LOG_ERROR(dhcp_ddns_logger,

                      DHCP_DDNS_NCR_FLUSH_IO_ERROR).arg(ex.what());

        }

    }

    io_service_.reset();

}

void

NameChangeSender::sendRequest(NameChangeRequestPtr& ncr) {

    if (!amSending()) {

        isc_throw(NcrSenderError, "sender is not ready to send");

    }

    if (!ncr) {

        isc_throw(NcrSenderError, "request to send is empty");

    }

    if (MultiThreadingMgr::instance().getMode()) {

        lock_guard<mutex> lock(*mutex_);

        sendRequestInternal(ncr);

    } else {

        sendRequestInternal(ncr);

    }

}

void

NameChangeSender::sendRequestInternal(NameChangeRequestPtr& ncr) {

    if (send_queue_.size() >= send_queue_max_) {

        isc_throw(NcrSenderQueueFull,

                  "send queue has reached maximum capacity: "

                  << send_queue_max_);

    }

    // Put it on the queue.

    send_queue_.push_back(ncr);

    // Call sendNext to schedule the next one to go.

    sendNext();

}

void

NameChangeSender::sendNext() {

    if (ncr_to_send_) {

        // @todo Not sure if there is any risk of getting stuck here but

        // an interval timer to defend would be good.

        // In reality, the derivation should ensure they timeout themselves

        return;

    }

    // If queue isn't empty, then get one from the front. Note we leave

    // it on the front of the queue until we successfully send it.

    if (!send_queue_.empty()) {

        ncr_to_send_ = send_queue_.front();

       // @todo start defense timer

       // If a send were to hang and we timed it out, then timeout

       // handler need to cycle thru open/close ?

       // Call implementation dependent send.  If doSend throws before an

       // asynchronous send is started (for example because the serialized

       // NCR exceeds the UDP send buffer), clear the in-progress marker and

       // discard the request so the queue cannot permanently stall.

       try {

           doSend(ncr_to_send_);

       } catch (const std::exception& ex) {

           LOG_ERROR(dhcp_ddns_logger, DHCP_DDNS_NCR_SEND_NEXT_ERROR)

                     .arg(ex.what());

           send_queue_.pop_front();

           // Use the internal path: sendNext() may already run under lock.

           invokeSendHandlerInternal(ERROR);

       }

    }

}

void

NameChangeSender::invokeSendHandler(const NameChangeSender::Result result) {

    if (MultiThreadingMgr::instance().getMode()) {

        lock_guard<mutex> lock(*mutex_);

        invokeSendHandlerInternal(result);

    } else {

        invokeSendHandlerInternal(result);

    }

}

void

NameChangeSender::invokeSendHandlerInternal(const NameChangeSender::Result result) {

    // @todo reset defense timer

    if (result == SUCCESS) {

        // It shipped so pull it off the queue.

        send_queue_.pop_front();

    }

    // Invoke the completion handler passing in the result and a pointer

    // the request involved.

    // Surround the invocation with a try-catch. The invoked handler is

    // not supposed to throw, but in the event it does we will at least

    // report it.

    try {

        (*send_handler_)(result, ncr_to_send_);

    } catch (const std::exception& ex) {

        LOG_ERROR(dhcp_ddns_logger, DHCP_DDNS_UNCAUGHT_NCR_SEND_HANDLER_ERROR)

                  .arg(ex.what());

    }

    // Clear the pending ncr pointer.

    ncr_to_send_.reset();

    // Set up the next send

    try {

        if (amSending()) {

            sendNext();

        }

    } catch (const isc::Exception& isc_ex) {

        // It is possible though unlikely, for sendNext to fail without

        // scheduling the send. While, unlikely, it does mean the callback

        // will not get called with a failure. A throw here would surface

        // at the IOService::run (or run variant) invocation.  So we will

        // close the window by invoking the application handler with

        // a failed result, and let the application layer sort it out.

        LOG_ERROR(dhcp_ddns_logger, DHCP_DDNS_NCR_SEND_NEXT_ERROR)

                  .arg(isc_ex.what());

        // Invoke the completion handler passing in failed result.

        // Surround the invocation with a try-catch. The invoked handler is

        // not supposed to throw, but in the event it does we will at least

        // report it.

        try {

            (*send_handler_)(ERROR, ncr_to_send_);

        } catch (const std::exception& std_ex) {

            LOG_ERROR(dhcp_ddns_logger,

                      DHCP_DDNS_UNCAUGHT_NCR_SEND_HANDLER_ERROR).arg(std_ex.what());

        }

    }

}

void

NameChangeSender::skipNext() {

    if (MultiThreadingMgr::instance().getMode()) {

        lock_guard<mutex> lock(*mutex_);

        skipNextInternal();

    } else {

        skipNextInternal();

    }

}

void

NameChangeSender::skipNextInternal() {

    if (!send_queue_.empty()) {

        // Discards the request at the front of the queue.

        send_queue_.pop_front();

    }

}

void

NameChangeSender::clearSendQueue() {

    if (amSending()) {

        isc_throw(NcrSenderError, "Cannot clear queue while sending");

    }

    if (MultiThreadingMgr::instance().getMode()) {

        lock_guard<mutex> lock(*mutex_);

        send_queue_.clear();

    } else {

        send_queue_.clear();

    }

}

void

NameChangeSender::setQueueMaxSize(const size_t new_max) {

    if (new_max == 0) {

        isc_throw(NcrSenderError, "NameChangeSender:"

                  " queue size must be greater than zero");

    }

    send_queue_max_ = new_max;

}

size_t

NameChangeSender::getQueueSize() const {

    if (MultiThreadingMgr::instance().getMode()) {

        lock_guard<mutex> lock(*mutex_);

        return (getQueueSizeInternal());

    } else {

        return (getQueueSizeInternal());

    }

}

size_t

NameChangeSender::getQueueSizeInternal() const {

    return (send_queue_.size());

}

const NameChangeRequestPtr&

NameChangeSender::peekAt(const size_t index) const {

    if (MultiThreadingMgr::instance().getMode()) {

        lock_guard<mutex> lock(*mutex_);

        return (peekAtInternal(index));

    } else {

        return (peekAtInternal(index));

    }

}

const NameChangeRequestPtr&

NameChangeSender::peekAtInternal(const size_t index) const {

    auto size = getQueueSizeInternal();

    if (index >= size) {

        isc_throw(NcrSenderError,

                  "NameChangeSender::peekAt peek beyond end of queue attempted"

                  << " index: " << index << " queue size: " << size);

    }

    return (send_queue_.at(index));

}

bool

NameChangeSender::isSendInProgress() const {

    if (MultiThreadingMgr::instance().getMode()) {

        lock_guard<mutex> lock(*mutex_);

        return ((ncr_to_send_) ? true : false);

    } else {

        return ((ncr_to_send_) ? true : false);

    }

}

void

NameChangeSender::assumeQueue(NameChangeSender& source_sender) {

    if (source_sender.amSending()) {

        isc_throw(NcrSenderError, "Cannot assume queue:"

                  " source sender is actively sending");

    }

    if (amSending()) {

        isc_throw(NcrSenderError, "Cannot assume queue:"

                  " target sender is actively sending");

    }

    if (getQueueMaxSize() < source_sender.getQueueSize()) {

        isc_throw(NcrSenderError, "Cannot assume queue:"

                  " source queue count exceeds target queue max");

    }

    if (MultiThreadingMgr::instance().getMode()) {

        lock_guard<mutex> lock(*mutex_);

        assumeQueueInternal(source_sender);

    } else {

        assumeQueueInternal(source_sender);

    }

}

void

NameChangeSender::assumeQueueInternal(NameChangeSender& source_sender) {

    if (!send_queue_.empty()) {

        isc_throw(NcrSenderError, "Cannot assume queue:"

                  " target queue is not empty");

    }

    send_queue_.swap(source_sender.getSendQueue());

}

int

NameChangeSender::getSelectFd() {

    isc_throw(NotImplemented, "NameChangeSender::getSelectFd is not supported");

}

void

NameChangeSender::runReadyIO() {

    if (!io_service_) {

        isc_throw(NcrSenderError, "NameChangeSender::runReadyIO"

                  " sender io service is null");

    }

    // We shouldn't be here if IO isn't ready to execute.

    // By running poll we're guaranteed not to hang.

    io_service_->pollOne();

}

}  // namespace dhcp_ddns

}  // namespace isc