#include <eosio/net_plugin/net_plugin_impl.hpp>
#include <eosio/net_plugin/dispatch_manager.hpp>
#include <eosio/net_plugin/connection.hpp>

#include <eosio/chain_plugin/chain_plugin.hpp>
#include <eosio/producer_plugin/producer_plugin.hpp>

using boost::asio::ip::tcp;
using namespace eosio::chain;

namespace eosio { namespace p2p {

std::shared_ptr<net_plugin_impl> net_plugin_impl::my_impl;

void net_plugin_impl::destroy() {
   my_impl.reset();
}
void net_plugin_impl::create_instance() {
   EOS_ASSERT(!my_impl, fc::exception, "net_plugin_impl instance already exists");
   my_impl.reset( new net_plugin_impl );
}

void net_plugin_impl::handle_sighup() {
   fc::logger::update( "net_plugin_impl", get_logger() );
}

void net_plugin_impl::start_listen_loop() {
   connection_ptr new_connection = std::make_shared<connection>();
   new_connection->connecting = true;
   new_connection->strand.post( [this, new_connection = std::move( new_connection )](){
      acceptor->async_accept( *new_connection->socket,
         boost::asio::bind_executor( new_connection->strand, [new_connection, socket=new_connection->socket, this]( boost::system::error_code ec ) {
         if( !ec ) {
            uint32_t visitors = 0;
            uint32_t from_addr = 0;
            boost::system::error_code rec;
            const auto& paddr_add = socket->remote_endpoint( rec ).address();
            string paddr_str;
            if( rec ) {
               fc_elog( get_logger(), "Error getting remote endpoint: {m}", ("m", rec.message()));
            } else {
               paddr_str = paddr_add.to_string();
               for_each_connection( [&visitors, &from_addr, &paddr_str]( auto& conn ) {
                  if( conn->socket_is_open()) {
                     if( conn->peer_address().empty()) {
                        ++visitors;
                        std::lock_guard<std::mutex> g_conn( conn->conn_mtx );
                        if( paddr_str == conn->remote_endpoint_ip ) {
                           ++from_addr;
                        }
                     }
                  }
                  return true;
               } );
               if( from_addr < max_nodes_per_host && (max_client_count == 0 || visitors < max_client_count)) {
                  fc_ilog( get_logger(), "Accepted new connection: " + paddr_str );
                  new_connection->set_heartbeat_timeout( heartbeat_timeout );
                  if( new_connection->start_session()) {
                     std::unique_lock<std::shared_mutex> lock( connections_mtx );
                     connections.insert( new_connection );
                  }

               } else {
                  if( from_addr >= max_nodes_per_host ) {
                     fc_dlog( get_logger(), "Number of connections ({n}) from {ra} exceeds limit {l}",
                              ("n", from_addr + 1)( "ra", paddr_str )( "l", max_nodes_per_host ));
                  } else {
                     fc_dlog( get_logger(), "max_client_count {m} exceeded", ("m", max_client_count));
                  }
                  // new_connection never added to connections and start_session not called, lifetime will end
                  boost::system::error_code ec;
                  socket->shutdown( tcp::socket::shutdown_both, ec );
                  socket->close( ec );
               }
            }
         } else {
            fc_elog( get_logger(), "Error accepting connection: {m}", ("m", ec.message()));
            // For the listed error codes below, recall start_listen_loop()
            switch (ec.value()) {
               case ECONNABORTED:
               case EMFILE:
               case ENFILE:
               case ENOBUFS:
               case ENOMEM:
               case EPROTO:
                  break;
               default:
                  return;
            }
         }
         start_listen_loop();
      }));
   } );
}

// call only from main application thread
void net_plugin_impl::update_chain_info() {
   controller& cc = chain_plug->chain();
   std::lock_guard<std::mutex> g( chain_info_mtx );
   chain_lib_num = cc.last_irreversible_block_num();
   chain_lib_id = cc.last_irreversible_block_id();
   chain_head_blk_num = cc.head_block_num();
   chain_head_blk_id = cc.head_block_id();
   chain_fork_head_blk_num = cc.fork_db_pending_head_block_num();
   chain_fork_head_blk_id = cc.fork_db_pending_head_block_id();
   fc_dlog( get_logger(), "updating chain info lib {lib}, head {head}, fork {fork}",
            ("lib", chain_lib_num)("head", chain_head_blk_num)("fork", chain_fork_head_blk_num) );
}

//         lib_num, head_blk_num, fork_head_blk_num, lib_id, head_blk_id, fork_head_blk_id
std::tuple<uint32_t, uint32_t, uint32_t, block_id_type, block_id_type, block_id_type>
net_plugin_impl::get_chain_info() const {
   std::lock_guard<std::mutex> g( chain_info_mtx );
   return std::make_tuple(
         chain_lib_num, chain_head_blk_num, chain_fork_head_blk_num,
         chain_lib_id, chain_head_blk_id, chain_fork_head_blk_id );
}

// called from any thread
void net_plugin_impl::start_conn_timer(boost::asio::steady_timer::duration du, std::weak_ptr<connection> from_connection) {
   if( in_shutdown ) return;
   std::lock_guard<std::mutex> g( connector_check_timer_mtx );
   ++connector_checks_in_flight;
   connector_check_timer->expires_from_now( du );
   connector_check_timer->async_wait( [my = get(), from_connection](boost::system::error_code ec) {
         std::unique_lock<std::mutex> g( my->connector_check_timer_mtx );
         int num_in_flight = --my->connector_checks_in_flight;
         g.unlock();
         if( !ec ) {
            my->connection_monitor(from_connection, num_in_flight == 0 );
         } else {
            if( num_in_flight == 0 ) {
               if( my->in_shutdown ) return;
               fc_elog( get_logger(), "Error from connection check monitor: {m}", ("m", ec.message()));
               my->start_conn_timer( my->connector_period, std::weak_ptr<connection>() );
            }
         }
   });
}

// thread safe
void net_plugin_impl::start_expire_timer() {
   if( in_shutdown ) return;
   std::lock_guard<std::mutex> g( expire_timer_mtx );
   expire_timer->expires_from_now( txn_exp_period);
   expire_timer->async_wait( [my = get()]( boost::system::error_code ec ) {
      if( !ec ) {
         my->expire();
      } else {
         if( my->in_shutdown ) return;
         fc_elog( get_logger(), "Error from transaction check monitor: {m}", ("m", ec.message()) );
         my->start_expire_timer();
      }
   } );
}

// thread safe
void net_plugin_impl::ticker() {
   if( in_shutdown ) return;
   std::lock_guard<std::mutex> g( keepalive_timer_mtx );
   keepalive_timer->expires_from_now(keepalive_interval);
   keepalive_timer->async_wait([my = get()]( boost::system::error_code ec ) {
         my->ticker();
         if( ec ) {
            if( my->in_shutdown ) return;
            fc_wlog( get_logger(), "Peer keepalive ticked sooner than expected: {m}", ("m", ec.message()) );
         }

         tstamp current_time = connection::get_time();
         my->for_each_connection( [current_time]( auto& c ) {
            if( c->socket_is_open() ) {
               c->strand.post([c, current_time]() {
                  c->check_heartbeat(current_time);
               } );
            }
            return true;
         } );
      } );
}

void net_plugin_impl::start_monitors() {
   {
      std::lock_guard<std::mutex> g( connector_check_timer_mtx );
      connector_check_timer.reset(new boost::asio::steady_timer( my_impl->thread_pool->get_executor() ));
   }
   {
      std::lock_guard<std::mutex> g( expire_timer_mtx );
      expire_timer.reset( new boost::asio::steady_timer( my_impl->thread_pool->get_executor() ) );
   }
   start_conn_timer(connector_period, std::weak_ptr<connection>());
   start_expire_timer();
}

void net_plugin_impl::expire() {
   auto now = time_point::now();
   uint32_t lib = 0;
   std::tie( lib, std::ignore, std::ignore, std::ignore, std::ignore, std::ignore ) = get_chain_info();
   dispatcher->expire_blocks( lib );
   dispatcher->expire_txns( lib );
   fc_dlog( get_logger(), "expire_txns {n}us", ("n", time_point::now() - now) );

   start_expire_timer();
}

// called from any thread
void net_plugin_impl::connection_monitor(std::weak_ptr<connection> from_connection, bool reschedule ) {
   auto max_time = fc::time_point::now();
   max_time += fc::milliseconds(max_cleanup_time_ms);
   auto from = from_connection.lock();
   std::unique_lock<std::shared_mutex> lock( connections_mtx );
   auto it = (from ? connections.find(from) : connections.begin());
   if (it == connections.end()) it = connections.begin();
   size_t num_rm = 0, num_clients = 0, num_peers = 0;
   while (it != connections.end()) {
      if (fc::time_point::now() >= max_time) {
         connection_wptr wit = *it;
         lock.unlock();
         fc_dlog( get_logger(), "Exiting connection monitor early, ran out of time: {t}", ("t", max_time - fc::time_point::now()) );
         fc_ilog( get_logger(), "p2p client connections: {num}/{max}, peer connections: {pnum}/{pmax}",
                  ("num", num_clients)("max", max_client_count)("pnum", num_peers)("pmax", supplied_peers.size()) );
         if( reschedule ) {
            start_conn_timer( std::chrono::milliseconds( 1 ), wit ); // avoid exhausting
         }
         return;
      }
      (*it)->peer_address().empty() ? ++num_clients : ++num_peers;
      if( !(*it)->socket_is_open() && !(*it)->connecting) {
         if( !(*it)->peer_address().empty() ) {
            if( !(*it)->resolve_and_connect() ) {
               it = connections.erase(it);
               --num_peers; ++num_rm;
               continue;
            }
         } else {
            --num_clients; ++num_rm;
            it = connections.erase(it);
            continue;
         }
      }
      ++it;
   }
   lock.unlock();
   if( num_clients > 0 || num_peers > 0 )
      fc_ilog( get_logger(), "p2p client connections: {num}/{max}, peer connections: {pnum}/{pmax}",
               ("num", num_clients)("max", max_client_count)("pnum", num_peers)("pmax", supplied_peers.size()) );
   fc_dlog( get_logger(), "connection monitor, removed {n} connections", ("n", num_rm) );
   if( reschedule ) {
      start_conn_timer( connector_period, std::weak_ptr<connection>());
   }
}

// called from application thread
void net_plugin_impl::on_accepted_block(const block_state_ptr& bs) {
   update_chain_info();
   controller& cc = chain_plug->chain();
   dispatcher->strand.post( [this, bs]() {
      fc_dlog( get_logger(), "signaled accepted_block, blk num = {num}, id = {id}", ("num", bs->block_num)("id", bs->id) );
      dispatcher->bcast_block( bs->block, bs->id );
   });
}

// called from application thread
void net_plugin_impl::on_pre_accepted_block(const signed_block_ptr& block) {
   update_chain_info();
   controller& cc = chain_plug->chain();
   if( cc.is_trusted_producer(block->producer) ) {
      dispatcher->strand.post( [this, block]() {
         auto id = block->calculate_id();
         fc_dlog( get_logger(), "signaled pre_accepted_block, blk num = {num}, id = {id}", ("num", block->block_num())("id", id) );

         dispatcher->bcast_block( block, id );
      });
   }
}

// called from application thread
void net_plugin_impl::on_irreversible_block( const block_state_ptr& block) {
   fc_dlog( get_logger(), "on_irreversible_block, blk num = {num}, id = {id}", ("num", block->block_num)("id", block->id) );
   update_chain_info();
}

// called from application thread
void net_plugin_impl::transaction_ack(const std::pair<fc::exception_ptr, transaction_metadata_ptr>& results) {
   dispatcher->strand.post( [this, results]() {
      const auto& id = results.second->id();
      if (results.first) {
         fc_dlog( get_logger(), "signaled NACK, trx-id = {id} : {why}", ("id", id)( "why", results.first->to_detail_string() ) );

         uint32_t head_blk_num = 0;
         std::tie( std::ignore, head_blk_num, std::ignore, std::ignore, std::ignore, std::ignore ) = get_chain_info();
         dispatcher->rejected_transaction(results.second->packed_trx(), head_blk_num);
      } else {
         fc_dlog( get_logger(), "signaled ACK, trx-id = {id}", ("id", id) );
         dispatcher->bcast_transaction(results.second->packed_trx());
      }
   });
}

bool net_plugin_impl::authenticate_peer(const handshake_message& msg) const {
   if(allowed_connections == None)
      return false;

   if(allowed_connections == Any)
      return true;

   if(allowed_connections & (Producers | Specified)) {
      auto allowed_it = std::find(allowed_peers.begin(), allowed_peers.end(), msg.key);
      auto private_it = private_keys.find(msg.key);
      bool found_producer_key = false;
      if(producer_plug != nullptr)
         found_producer_key = producer_plug->is_producer_key(msg.key);
      if( allowed_it == allowed_peers.end() && private_it == private_keys.end() && !found_producer_key) {
         fc_elog( get_logger(), "Peer {peer} sent a handshake with an unauthorized key: {key}.",
                  ("peer", msg.p2p_address)("key", msg.key.to_string()) );
         return false;
      }
   }

   if(msg.sig != chain::signature_type() && msg.token != sha256()) {
      sha256 hash = fc::sha256::hash(msg.time);
      if(hash != msg.token) {
         fc_elog( get_logger(), "Peer {peer} sent a handshake with an invalid token.", ("peer", msg.p2p_address) );
         return false;
      }
      chain::public_key_type peer_key;
      try {
         peer_key = crypto::public_key(msg.sig, msg.token, true);
      }
      catch (const std::exception& /*e*/) {
         fc_elog( get_logger(), "Peer {peer} sent a handshake with an unrecoverable key.", ("peer", msg.p2p_address) );
         return false;
      }
      if((allowed_connections & (Producers | Specified)) && peer_key != msg.key) {
         fc_elog( get_logger(), "Peer {peer} sent a handshake with an unauthenticated key.", ("peer", msg.p2p_address) );
         return false;
      }
   }
   else if(allowed_connections & (Producers | Specified)) {
      fc_dlog( get_logger(), "Peer sent a handshake with blank signature and token, but this node accepts only authenticated connections." );
      return false;
   }
   return true;
}

chain::public_key_type net_plugin_impl::get_authentication_key() const {
   if(!private_keys.empty())
      return private_keys.begin()->first;
   /*producer_plugin* pp = app().find_plugin<producer_plugin>();
   if(pp != nullptr && pp->get_state() == abstract_plugin::started)
      return pp->first_producer_public_key();*/
   return chain::public_key_type();
}

chain::signature_type net_plugin_impl::sign_compact(const chain::public_key_type& signer, const fc::sha256& digest) const
{
   auto private_key_itr = private_keys.find(signer);
   if(private_key_itr != private_keys.end())
      return private_key_itr->second.sign(digest);
   if(producer_plug != nullptr && producer_plug->get_state() == abstract_plugin::started)
      return producer_plug->sign_compact(signer, digest);
   return chain::signature_type();
}

}} //eosio::p2p