// Copyright 2018 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "src/connectivity/overnet/lib/routing/router.h"

#include <iostream>

#include "garnet/public/lib/fostr/fidl/fuchsia/overnet/protocol/formatting.h"
#include "src/connectivity/overnet/lib/protocol/fidl.h"

namespace overnet {

static constexpr TimeDelta kPollLinkChangeTimeout =
    TimeDelta::FromMilliseconds(100);

Router::Router(Timer* timer, NodeId node_id, bool allow_non_determinism)
    : timer_(timer),
      node_id_(node_id),
      rng_(allow_non_determinism ? std::random_device()() : 0),
      routing_table_(node_id, timer, allow_non_determinism),
      own_node_status_{node_id.as_fidl(), 1} {
  std::vector<fuchsia::overnet::protocol::NodeStatus> node_status;
  node_status.emplace_back(fidl::Clone(own_node_status_));
  UpdateRoutingTable({fidl::Clone(own_node_status_)}, {}, false);
}

Router::~Router() { shutting_down_ = true; }

void Router::Close(Callback<void> quiesced) {
  ScopedModule<Router> scoped_module(this);
  OVERNET_TRACE(DEBUG) << node_id_ << " Close";
  shutting_down_ = true;
  poll_link_changes_timeout_.Reset();
  flush_old_nodes_timeout_.Reset();
  links_.clear();
  CloseLinks(std::move(quiesced));
}

void Router::CloseLinks(Callback<void> quiesced) {
  OVERNET_TRACE(DEBUG) << node_id_
                       << " CloseLinks remaining=" << owned_links_.size();
  if (owned_links_.empty()) {
    CloseStreams(std::move(quiesced));
    return;
  }
  auto it = owned_links_.begin();
  auto p = it->second.release();
  owned_links_.erase(it);
  p->Close(Callback<void>(ALLOCATED_CALLBACK,
                          [this, p, quiesced = std::move(quiesced)]() mutable {
                            ScopedModule<Router> scoped_module(this);
                            delete p;
                            CloseLinks(std::move(quiesced));
                          }));
}

void Router::CloseStreams(Callback<void> quiesced) {
  if (streams_.empty()) {
    OVERNET_TRACE(DEBUG) << "Closed";
    return;
  }
  auto it = streams_.begin();
  auto id = it->first;
  OVERNET_TRACE(DEBUG) << node_id_
                       << " CloseStreams remaining=" << streams_.size()
                       << " next=" << id.peer << "/" << id.stream_id;
  it->second.Close(Callback<void>(
      ALLOCATED_CALLBACK, [this, id, quiesced = std::move(quiesced)]() mutable {
        ScopedModule<Router> scoped_module(this);
        auto it = streams_.find(id);
        if (it != streams_.end()) {
          assert(!it->second.has_handler());
          streams_.erase(it);
        }
        assert(streams_.count(id) == 0);
        CloseStreams(std::move(quiesced));
      }));
}

void Router::Forward(Message message) {
  ScopedModule<Router> scoped_module(this);
  OVERNET_TRACE(DEBUG) << "Forward " << message.header
                       << " shutting_down=" << shutting_down_;
  if (shutting_down_) {
    return;
  }
  assert(!message.make_payload.empty());
  // There are three primary cases we care about here, that can be discriminated
  // based on the destination count of the message:
  // 1. If there are zero destinations, this is a malformed message (fail).
  // 2. If there is one destination, forward the message on.
  // 3. If there are multiple destinations, broadcast this message to all
  //    destinations.
  // We separate 2 & 3 as the single forwarding case can be made
  // (much) more efficient.
  switch (message.header.destinations().size()) {
    case 0:
      // Malformed message, bail
      // TODO(ctiller): Log error: Routing header must have at least one
      // destination
      break;
    case 1: {
      // Single destination... it could be either a local stream or need to be
      // forwarded to a remote node over some link.
      const RoutableMessage::Destination& dst =
          message.header.destinations()[0];
      if (dst.dst() == node_id_) {
        if (!stream_holder(message.header.src(), dst.stream_id())
                 ->HandleMessage(dst.seq(), message.received,
                                 message.make_payload(LazySliceArgs{
                                     Border::None(),
                                     std::numeric_limits<uint32_t>::max()}))) {
          OnUnknownStream(message.header.src(), dst.stream_id());
        }
      } else {
        link_holder(dst.dst())->Forward(std::move(message));
      }
    } break;
    default: {
      // Multiple destination:
      // - Handle local streams directly.
      // - For remote forwarding:
      //   1. If we know the next hop, and that next hop is used for multiple of
      //      our destinations, keep the multicast group together for that set.
      //   2. Separate the multicast if next hops are different.
      //   3. Separate the multicast if we do not know about next hops yet.
      std::unordered_map<Link*, std::vector<RoutableMessage::Destination>>
          group_forward;
      std::vector<std::pair<RoutableMessage::Destination, LinkHolder*>>
          disconnected_holders;
      Optional<std::pair<RoutableMessage::Destination, StreamHolder*>>
          handle_locally;
      uint32_t overall_mss = std::numeric_limits<uint32_t>::max();
      for (const auto& dst : message.header.destinations()) {
        if (dst.dst() == node_id_) {
          // Locally handled stream
          if (!handle_locally.has_value()) {
            handle_locally = std::make_pair(
                dst, stream_holder(message.header.src(), dst.stream_id()));
          }
        } else {
          // Remote destination
          LinkHolder* h = link_holder(dst.dst());
          if (h->link() == nullptr) {
            // We don't know the next link, ask the LinkHolder to forward (which
            // will continue forwarding the message when we know the next hop).
            disconnected_holders.push_back(std::make_pair(dst, h));
          } else {
            // We know the next link: gather destinations together by link so
            // that we can (hopefully) keep multicast groups together
            group_forward[h->link()].emplace_back(dst);
            overall_mss = std::min(overall_mss, h->path_mss());
          }
        }
      }
      const auto max_header_length = message.header.MaxHeaderLength();
      if (overall_mss < max_header_length) {
        return;
      }
      overall_mss -= max_header_length;
      Slice payload =
          message.make_payload(LazySliceArgs{Border::None(), overall_mss});
      // Forward any grouped messages now that we've examined all destinations
      for (auto& grp : group_forward) {
        grp.first->Forward(Message::SimpleForwarder(
            message.header.WithDestinations(std::move(grp.second)), payload,
            message.received));
      }
      for (auto& lh : disconnected_holders) {
        lh.second->Forward(Message::SimpleForwarder(
            message.header.WithDestinations({lh.first}), payload,
            message.received));
      }
      if (handle_locally.has_value()) {
        if (!handle_locally->second->HandleMessage(handle_locally->first.seq(),
                                                   message.received,
                                                   std::move(payload))) {
          OnUnknownStream(message.header.src(),
                          handle_locally->first.stream_id());
        }
      }
    } break;
  }
}

void Router::UpdateRoutingTable(
    std::initializer_list<fuchsia::overnet::protocol::NodeStatus> node_updates,
    std::initializer_list<fuchsia::overnet::protocol::LinkStatus> link_updates,
    bool flush_old_nodes) {
  ScopedModule<Router> scoped_module(this);
  routing_table_.ProcessUpdate(std::move(node_updates), std::move(link_updates),
                               flush_old_nodes);
  MaybeStartPollingLinkChanges();
}

void Router::MaybeStartPollingLinkChanges() {
  ScopedModule<Router> scoped_module(this);
  if (shutting_down_ || poll_link_changes_timeout_) {
    return;
  }
  poll_link_changes_timeout_.Reset(
      timer_, timer_->Now() + kPollLinkChangeTimeout,
      [this](const Status& status) {
        if (status.is_ok()) {
          poll_link_changes_timeout_.Reset();
          const bool keep_polling = !routing_table_.PollLinkUpdates(
              [this](const RoutingTable::SelectedLinks& selected_links) {
                // Clear routing information for now unreachable links.
                for (auto& lnk : links_) {
                  if (selected_links.count(lnk.first) == 0) {
                    lnk.second.SetLink(nullptr, 0, false);
                  }
                }
                // Set routing information for other links.
                for (const auto& sl : selected_links) {
                  OVERNET_TRACE(DEBUG)
                      << node_id() << " Select: dest=" << sl.first << " link"
                      << sl.second.target_node << "#" << sl.second.link_id
                      << " (route_mss=" << sl.second.route_mss << ")";
                  auto it = owned_links_.find(
                      OwnedLabel{sl.second.target_node, sl.second.link_id});
                  auto* link =
                      it == owned_links_.end() ? nullptr : it->second.get();
                  link_holder(sl.first)->SetLink(
                      link, sl.second.route_mss,
                      link ? link->GetLinkStatus().to == sl.first : false);
                }
                MaybeStartFlushingOldEntries();
              });
          if (keep_polling) {
            MaybeStartPollingLinkChanges();
          }
        }
      });
}

void Router::MaybeStartFlushingOldEntries() {
  ScopedModule<Router> scoped_module(this);
  if (flush_old_nodes_timeout_)
    return;
  flush_old_nodes_timeout_.Reset(timer_,
                                 timer_->Now() + routing_table_.EntryExpiry(),
                                 [this](const Status& status) {
                                   if (status.is_ok()) {
                                     flush_old_nodes_timeout_.Reset();
                                     UpdateRoutingTable({}, {}, true);
                                   }
                                 });
}

Status Router::RegisterStream(NodeId peer, StreamId stream_id,
                              StreamHandler* stream_handler) {
  ScopedModule<Router> scoped_module(this);
  OVERNET_TRACE(DEBUG) << "RegisterStream: " << peer << "/" << stream_id
                       << " at " << stream_handler
                       << " shutting_down=" << shutting_down_;
  if (shutting_down_) {
    return Status(StatusCode::FAILED_PRECONDITION, "Router shutting down");
  }
  return stream_holder(peer, stream_id)->SetHandler(stream_handler);
}

Status Router::UnregisterStream(NodeId peer, StreamId stream_id,
                                StreamHandler* stream_handler) {
  ScopedModule<Router> scoped_module(this);
  OVERNET_TRACE(DEBUG) << "UnregisterStream: " << peer << "/" << stream_id
                       << " at " << stream_handler
                       << " shutting_down=" << shutting_down_;
  auto it = streams_.find(LocalStreamId{peer, stream_id});
  if (it == streams_.end()) {
    return Status(StatusCode::FAILED_PRECONDITION, "Stream not registered");
  }
  Status status = it->second.ClearHandler(stream_handler);
  streams_.erase(it);
  return status;
}

Optional<NodeId> Router::SelectGossipPeer() {
  ScopedModule<Router> scoped_module(this);
  const uint64_t gossip_version = routing_table_.gossip_version();
  std::vector<NodeId> eligible_nodes;
  for (const auto& peer : links_) {
    if (peer.second.has_direct_link() &&
        peer.second.last_gossip_version() < gossip_version) {
      eligible_nodes.push_back(peer.first);
    }
  }
  if (eligible_nodes.empty()) {
    return Nothing;
  }
  std::uniform_int_distribution<> dis(0, eligible_nodes.size() - 1);
  return eligible_nodes[dis(rng_)];
}

void Router::SendGossipUpdate(fuchsia::overnet::protocol::Peer_Proxy* peer,
                              NodeId target) {
  ScopedModule<Router> scoped_module(this);
  link_holder(target)->set_last_gossip_version(
      routing_table_.SendUpdate(peer, target));
}

namespace {
template <class T>
std::vector<T> TakeVector(std::vector<T>* vec) {
  if (vec == nullptr) {
    return {};
  }
  return std::move(*vec);
}
}  // namespace

void Router::RegisterLink(LinkPtr<> link) {
  ScopedModule<Router> scoped_module(this);
  auto status = link->GetLinkStatus();
  OVERNET_TRACE(DEBUG) << node_id() << " RegisterLink: " << status;
  assert(status.from == node_id());
  owned_links_.emplace(OwnedLabel{status.to, status.local_id}, std::move(link));
  auto target = status.to;
  UpdateRoutingTable({{target, 0}}, {std::move(status)}, false);
}

bool Router::StreamHolder::HandleMessage(SeqNum seq, TimeStamp received,
                                         Slice payload) {
  if (handler_ == nullptr) {
    if (!buffered_ || buffered_->seq.Reconstruct(1) < seq.Reconstruct(1)) {
      OVERNET_TRACE(DEBUG) << "Buffer message: peer=" << peer_
                           << " stream=" << stream_ << " seq=" << seq
                           << " message=" << payload;
      buffered_.reset(new BufferedPacket{seq, received, std::move(payload)});
    } else {
      OVERNET_TRACE(DEBUG) << "Drop message: peer=" << peer_
                           << " stream=" << stream_ << " seq=" << seq
                           << " message=" << payload;
    }
    return false;
  } else {
    handler_->HandleMessage(seq, received, std::move(payload));
    return true;
  }
}

Status Router::StreamHolder::SetHandler(StreamHandler* handler) {
  if (handler_ != nullptr) {
    return Status(StatusCode::FAILED_PRECONDITION, "Handler already set");
  }
  handler_ = handler;
  if (buffered_) {
    auto pkt = std::move(buffered_);
    handler_->HandleMessage(pkt->seq, pkt->received, std::move(pkt->payload));
  }
  return Status::Ok();
}

Status Router::StreamHolder::ClearHandler(StreamHandler* handler) {
  if (handler_ != handler) {
    return Status(StatusCode::FAILED_PRECONDITION, "Invalid clear handler");
  }
  handler_ = nullptr;
  return Status::Ok();
}

void Router::LinkHolder::Forward(Message message) {
  if (link_ == nullptr) {
    OVERNET_TRACE(DEBUG) << "Queue: " << message.header;
    pending_.emplace_back(std::move(message));
  } else {
    message.mss = std::min(message.mss, path_mss_);
    link_->Forward(std::move(message));
  }
}

void Router::LinkHolder::SetLink(Link* link, uint32_t path_mss,
                                 bool is_direct) {
  link_ = link;
  is_direct_ = is_direct;
  path_mss_ = path_mss;
  std::vector<Message> pending;
  pending.swap(pending_);
  for (auto& p : pending) {
    Forward(std::move(p));
  }
}

}  // namespace overnet