/**
 * 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.
 */
#pragma once

#include <memory>
#include <string>
#include <condition_variable>
#include <thread>
#include <mutex>
#include <atomic>

#include "minifi-cpp/io/BaseStream.h"
#include "core/logging/LoggerFactory.h"
#include "core/state/nodes/StateMonitor.h"
#include "core/controller/ControllerServiceProvider.h"
#include "ControllerSocketReporter.h"
#include "utils/MinifiConcurrentQueue.h"
#include "asio/ip/tcp.hpp"
#include "asio/ssl/context.hpp"
#include "utils/net/AsioCoro.h"
#include "minifi-cpp/controllers/SSLContextServiceInterface.h"

namespace org::apache::nifi::minifi::c2 {

/**
 * Purpose: Creates a reporter that can handle basic c2 operations for a localized environment
 * through a simple TCP socket.
 */
class ControllerSocketProtocol {
 public:
  ControllerSocketProtocol(state::StateMonitor& update_sink, std::shared_ptr<Configure> configuration, const std::shared_ptr<ControllerSocketReporter>& controller_socket_reporter);
  ~ControllerSocketProtocol();
  void initialize();
  void setRoot(core::ProcessGroup* root);

 private:
  void handleStart(io::BaseStream &stream);
  void handleStop(io::BaseStream &stream);
  void handleClear(io::BaseStream &stream);
  void handleUpdate(io::BaseStream &stream);
  void handleTransfer(io::BaseStream &stream);
  void writeQueueSizesResponse(io::BaseStream &stream);
  void writeComponentsResponse(io::BaseStream &stream);
  void writeConnectionsResponse(io::BaseStream &stream);
  void writeGetFullResponse(io::BaseStream &stream);
  void writeManifestResponse(io::BaseStream &stream);
  void writeJstackResponse(io::BaseStream &stream);
  void writeFlowStatusResponse(io::BaseStream &stream);
  void writeDebugBundleResponse(io::BaseStream &stream);
  void handleDescribe(io::BaseStream &stream);
  asio::awaitable<void> handleCommand(std::unique_ptr<io::BaseStream> stream);
  asio::awaitable<void> handshakeAndHandleCommand(asio::ip::tcp::socket&& socket, std::shared_ptr<minifi::controllers::SSLContextServiceInterface> ssl_context_service);
  std::string getJstack();
  asio::awaitable<void> startAccept();
  asio::awaitable<void> startAcceptSsl(std::shared_ptr<minifi::controllers::SSLContextServiceInterface> ssl_context_service);
  void stopListener();

  state::StateMonitor& update_sink_;

  asio::io_context io_context_;
  std::unique_ptr<asio::ip::tcp::acceptor> acceptor_;
  std::thread server_thread_;

  std::weak_ptr<ControllerSocketReporter> controller_socket_reporter_;
  std::shared_ptr<Configure> configuration_;
  std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<ControllerSocketProtocol>::getLogger();
  std::mutex initialization_mutex_;

  // Some commands need to restart the controller socket to reinitialize the socket with new data for example new SSL data in case of a flow update
  // These commands are handled on a separate thread, and while these commands are handled other incoming commands are dropped
  class SocketRestartCommandProcessor {
   public:
    explicit SocketRestartCommandProcessor(state::StateMonitor& update_sink_, const std::shared_ptr<core::logging::Logger>& logger);
    ~SocketRestartCommandProcessor();

    enum class Command {
      FLOW_UPDATE,
      START
    };

    struct CommandData {
      Command command;
      std::string data;
    };

    void enqueue(const CommandData& command_data) {
      is_socket_restarting_ = true;
      command_queue_.enqueue(command_data);
    }

    bool isSocketRestarting() const {
      return is_socket_restarting_;
    }

   private:
    mutable std::atomic_bool is_socket_restarting_ = false;
    state::StateMonitor& update_sink_;
    std::thread command_processor_thread_;
    std::atomic_bool running_ = true;
    utils::ConditionConcurrentQueue<CommandData> command_queue_;
    std::shared_ptr<core::logging::Logger> logger_;
  };

  SocketRestartCommandProcessor socket_restart_processor_;
};

}  // namespace org::apache::nifi::minifi::c2