/*
 *  Copyright (c) 2021 The WebRTC project authors. All Rights Reserved.
 *
 *  Use of this source code is governed by a BSD-style license
 *  that can be found in the LICENSE file in the root of the source
 *  tree. An additional intellectual property rights grant can be found
 *  in the file PATENTS.  All contributing project authors may
 *  be found in the AUTHORS file in the root of the source tree.
 */
#include <cstdint>
#include <deque>
#include <memory>
#include <string>
#include <utility>
#include <vector>

#include "absl/memory/memory.h"
#include "absl/strings/string_view.h"
#include "absl/types/optional.h"
#include "api/array_view.h"
#include "api/task_queue/pending_task_safety_flag.h"
#include "api/task_queue/task_queue_base.h"
#include "api/test/create_network_emulation_manager.h"
#include "api/test/network_emulation_manager.h"
#include "api/units/data_rate.h"
#include "api/units/time_delta.h"
#include "net/dcsctp/public/dcsctp_options.h"
#include "net/dcsctp/public/dcsctp_socket.h"
#include "net/dcsctp/public/types.h"
#include "net/dcsctp/socket/dcsctp_socket.h"
#include "net/dcsctp/testing/testing_macros.h"
#include "net/dcsctp/timer/task_queue_timeout.h"
#include "rtc_base/copy_on_write_buffer.h"
#include "rtc_base/gunit.h"
#include "rtc_base/logging.h"
#include "rtc_base/socket_address.h"
#include "rtc_base/strings/string_format.h"
#include "rtc_base/time_utils.h"
#include "test/gmock.h"

#if !defined(WEBRTC_ANDROID) && defined(NDEBUG) && \
    !defined(THREAD_SANITIZER) && !defined(MEMORY_SANITIZER)
#define DCSCTP_NDEBUG_TEST(t) t
#else
// In debug mode, and when MSAN or TSAN sanitizers are enabled, these tests are
// too expensive to run due to extensive consistency checks that iterate on all
// outstanding chunks. Same with low-end Android devices, which have
// difficulties with these tests.
#define DCSCTP_NDEBUG_TEST(t) DISABLED_##t
#endif

namespace dcsctp {
namespace {
using ::testing::AllOf;
using ::testing::Ge;
using ::testing::Le;
using ::testing::SizeIs;
using ::webrtc::DataRate;
using ::webrtc::TimeDelta;
using ::webrtc::Timestamp;

constexpr StreamID kStreamId(1);
constexpr PPID kPpid(53);
constexpr size_t kSmallPayloadSize = 10;
constexpr size_t kLargePayloadSize = 10000;
constexpr size_t kHugePayloadSize = 262144;
constexpr size_t kBufferedAmountLowThreshold = kLargePayloadSize * 2;
constexpr webrtc::TimeDelta kPrintBandwidthDuration =
    webrtc::TimeDelta::Seconds(1);
constexpr webrtc::TimeDelta kBenchmarkRuntime(webrtc::TimeDelta::Seconds(10));
constexpr webrtc::TimeDelta kAWhile(webrtc::TimeDelta::Seconds(1));

inline int GetUniqueSeed() {
  static int seed = 0;
  return ++seed;
}

DcSctpOptions MakeOptionsForTest() {
  DcSctpOptions options;

  // Throughput numbers are affected by the MTU. Ensure it's constant.
  options.mtu = 1200;

  // By disabling the heartbeat interval, there will no timers at all running
  // when the socket is idle, which makes it easy to just continue the test
  // until there are no more scheduled tasks. Note that it _will_ run for longer
  // than necessary as timers aren't cancelled when they are stopped (as that's
  // not supported), but it's still simulated time and passes quickly.
  options.heartbeat_interval = DurationMs(0);
  return options;
}

// When doing throughput tests, knowing what each actor should do.
enum class ActorMode {
  kAtRest,
  kThroughputSender,
  kThroughputReceiver,
  kLimitedRetransmissionSender,
};

// An abstraction around EmulatedEndpoint, representing a bound socket that
// will send its packet to a given destination.
class BoundSocket : public webrtc::EmulatedNetworkReceiverInterface {
 public:
  void Bind(webrtc::EmulatedEndpoint* endpoint) {
    endpoint_ = endpoint;
    uint16_t port = endpoint->BindReceiver(0, this).value();
    source_address_ =
        rtc::SocketAddress(endpoint_->GetPeerLocalAddress(), port);
  }

  void SetDestination(const BoundSocket& socket) {
    dest_address_ = socket.source_address_;
  }

  void SetReceiver(std::function<void(rtc::CopyOnWriteBuffer)> receiver) {
    receiver_ = std::move(receiver);
  }

  void SendPacket(rtc::ArrayView<const uint8_t> data) {
    endpoint_->SendPacket(source_address_, dest_address_,
                          rtc::CopyOnWriteBuffer(data.data(), data.size()));
  }

 private:
  // Implementation of `webrtc::EmulatedNetworkReceiverInterface`.
  void OnPacketReceived(webrtc::EmulatedIpPacket packet) override {
    receiver_(std::move(packet.data));
  }

  std::function<void(rtc::CopyOnWriteBuffer)> receiver_;
  webrtc::EmulatedEndpoint* endpoint_ = nullptr;
  rtc::SocketAddress source_address_;
  rtc::SocketAddress dest_address_;
};

// Sends at a constant rate but with random packet sizes.
class SctpActor : public DcSctpSocketCallbacks {
 public:
  SctpActor(absl::string_view name,
            BoundSocket& emulated_socket,
            const DcSctpOptions& sctp_options)
      : log_prefix_(std::string(name) + ": "),
        thread_(rtc::Thread::Current()),
        emulated_socket_(emulated_socket),
        timeout_factory_(
            *thread_,
            [this]() { return TimeMs(Now().ms()); },
            [this](dcsctp::TimeoutID timeout_id) {
              sctp_socket_.HandleTimeout(timeout_id);
            }),
        random_(GetUniqueSeed()),
        sctp_socket_(name, *this, nullptr, sctp_options),
        last_bandwidth_printout_(Now()) {
    emulated_socket.SetReceiver([this](rtc::CopyOnWriteBuffer buf) {
      // The receiver will be executed on the NetworkEmulation task queue, but
      // the dcSCTP socket is owned by `thread_` and is not thread-safe.
      thread_->PostTask([this, buf] { this->sctp_socket_.ReceivePacket(buf); });
    });
  }

  void PrintBandwidth() {
    Timestamp now = Now();
    TimeDelta duration = now - last_bandwidth_printout_;

    double bitrate_mbps =
        static_cast<double>(received_bytes_ * 8) / duration.ms() / 1000;
    RTC_LOG(LS_INFO) << log_prefix()
                     << rtc::StringFormat("Received %0.2f Mbps", bitrate_mbps);

    received_bitrate_mbps_.push_back(bitrate_mbps);
    received_bytes_ = 0;
    last_bandwidth_printout_ = now;
    // Print again in a second.
    if (mode_ == ActorMode::kThroughputReceiver) {
      thread_->PostDelayedTask(
          SafeTask(safety_.flag(), [this] { PrintBandwidth(); }),
          kPrintBandwidthDuration);
    }
  }

  void SendPacket(rtc::ArrayView<const uint8_t> data) override {
    emulated_socket_.SendPacket(data);
  }

  std::unique_ptr<Timeout> CreateTimeout(
      webrtc::TaskQueueBase::DelayPrecision precision) override {
    return timeout_factory_.CreateTimeout(precision);
  }

  Timestamp Now() override { return Timestamp::Millis(rtc::TimeMillis()); }

  uint32_t GetRandomInt(uint32_t low, uint32_t high) override {
    return random_.Rand(low, high);
  }

  void OnMessageReceived(DcSctpMessage message) override {
    received_bytes_ += message.payload().size();
    last_received_message_ = std::move(message);
  }

  void OnError(ErrorKind error, absl::string_view message) override {
    RTC_LOG(LS_WARNING) << log_prefix() << "Socket error: " << ToString(error)
                        << "; " << message;
  }

  void OnAborted(ErrorKind error, absl::string_view message) override {
    RTC_LOG(LS_ERROR) << log_prefix() << "Socket abort: " << ToString(error)
                      << "; " << message;
  }

  void OnConnected() override {}

  void OnClosed() override {}

  void OnConnectionRestarted() override {}

  void OnStreamsResetFailed(rtc::ArrayView<const StreamID> outgoing_streams,
                            absl::string_view reason) override {}

  void OnStreamsResetPerformed(
      rtc::ArrayView<const StreamID> outgoing_streams) override {}

  void OnIncomingStreamsReset(
      rtc::ArrayView<const StreamID> incoming_streams) override {}

  void NotifyOutgoingMessageBufferEmpty() override {}

  void OnBufferedAmountLow(StreamID stream_id) override {
    if (mode_ == ActorMode::kThroughputSender) {
      std::vector<uint8_t> payload(kHugePayloadSize);
      sctp_socket_.Send(DcSctpMessage(kStreamId, kPpid, std::move(payload)),
                        SendOptions());

    } else if (mode_ == ActorMode::kLimitedRetransmissionSender) {
      while (sctp_socket_.buffered_amount(kStreamId) <
             kBufferedAmountLowThreshold * 2) {
        SendOptions send_options;
        send_options.max_retransmissions = 0;
        sctp_socket_.Send(
            DcSctpMessage(kStreamId, kPpid,
                          std::vector<uint8_t>(kLargePayloadSize)),
            send_options);

        send_options.max_retransmissions = absl::nullopt;
        sctp_socket_.Send(
            DcSctpMessage(kStreamId, kPpid,
                          std::vector<uint8_t>(kSmallPayloadSize)),
            send_options);
      }
    }
  }

  absl::optional<DcSctpMessage> ConsumeReceivedMessage() {
    if (!last_received_message_.has_value()) {
      return absl::nullopt;
    }
    DcSctpMessage ret = *std::move(last_received_message_);
    last_received_message_ = absl::nullopt;
    return ret;
  }

  DcSctpSocket& sctp_socket() { return sctp_socket_; }

  void SetActorMode(ActorMode mode) {
    mode_ = mode;
    if (mode_ == ActorMode::kThroughputSender) {
      sctp_socket_.SetBufferedAmountLowThreshold(kStreamId,
                                                 kBufferedAmountLowThreshold);
      std::vector<uint8_t> payload(kHugePayloadSize);
      sctp_socket_.Send(DcSctpMessage(kStreamId, kPpid, std::move(payload)),
                        SendOptions());

    } else if (mode_ == ActorMode::kLimitedRetransmissionSender) {
      sctp_socket_.SetBufferedAmountLowThreshold(kStreamId,
                                                 kBufferedAmountLowThreshold);
      std::vector<uint8_t> payload(kHugePayloadSize);
      sctp_socket_.Send(DcSctpMessage(kStreamId, kPpid, std::move(payload)),
                        SendOptions());

    } else if (mode == ActorMode::kThroughputReceiver) {
      thread_->PostDelayedTask(
          SafeTask(safety_.flag(), [this] { PrintBandwidth(); }),
          kPrintBandwidthDuration);
    }
  }

  // Returns the average bitrate, stripping the first `remove_first_n` that
  // represent the time it took to ramp up the congestion control algorithm.
  double avg_received_bitrate_mbps(size_t remove_first_n = 3) const {
    std::vector<double> bitrates = received_bitrate_mbps_;
    bitrates.erase(bitrates.begin(), bitrates.begin() + remove_first_n);

    double sum = 0;
    for (double bitrate : bitrates) {
      sum += bitrate;
    }

    return sum / bitrates.size();
  }

 private:
  std::string log_prefix() const {
    rtc::StringBuilder sb;
    sb << log_prefix_;
    sb << rtc::TimeMillis();
    sb << ": ";
    return sb.Release();
  }

  ActorMode mode_ = ActorMode::kAtRest;
  const std::string log_prefix_;
  rtc::Thread* thread_;
  BoundSocket& emulated_socket_;
  TaskQueueTimeoutFactory timeout_factory_;
  webrtc::Random random_;
  DcSctpSocket sctp_socket_;
  size_t received_bytes_ = 0;
  absl::optional<DcSctpMessage> last_received_message_;
  Timestamp last_bandwidth_printout_;
  // Per-second received bitrates, in Mbps
  std::vector<double> received_bitrate_mbps_;
  webrtc::ScopedTaskSafety safety_;
};

class DcSctpSocketNetworkTest : public testing::Test {
 protected:
  DcSctpSocketNetworkTest()
      : options_(MakeOptionsForTest()),
        emulation_(webrtc::CreateNetworkEmulationManager(
            {.time_mode = webrtc::TimeMode::kSimulated})) {}

  void MakeNetwork(const webrtc::BuiltInNetworkBehaviorConfig& config) {
    webrtc::EmulatedEndpoint* endpoint_a =
        emulation_->CreateEndpoint(webrtc::EmulatedEndpointConfig());
    webrtc::EmulatedEndpoint* endpoint_z =
        emulation_->CreateEndpoint(webrtc::EmulatedEndpointConfig());

    webrtc::EmulatedNetworkNode* node1 = emulation_->CreateEmulatedNode(config);
    webrtc::EmulatedNetworkNode* node2 = emulation_->CreateEmulatedNode(config);

    emulation_->CreateRoute(endpoint_a, {node1}, endpoint_z);
    emulation_->CreateRoute(endpoint_z, {node2}, endpoint_a);

    emulated_socket_a_.Bind(endpoint_a);
    emulated_socket_z_.Bind(endpoint_z);

    emulated_socket_a_.SetDestination(emulated_socket_z_);
    emulated_socket_z_.SetDestination(emulated_socket_a_);
  }

  void Sleep(webrtc::TimeDelta duration) {
    // Sleep in one-millisecond increments, to let timers expire when expected.
    for (int i = 0; i < duration.ms(); ++i) {
      emulation_->time_controller()->AdvanceTime(webrtc::TimeDelta::Millis(1));
    }
  }

  DcSctpOptions options_;
  std::unique_ptr<webrtc::NetworkEmulationManager> emulation_;
  BoundSocket emulated_socket_a_;
  BoundSocket emulated_socket_z_;
};

TEST_F(DcSctpSocketNetworkTest, CanConnectAndShutdown) {
  webrtc::BuiltInNetworkBehaviorConfig pipe_config;
  MakeNetwork(pipe_config);

  SctpActor sender("A", emulated_socket_a_, options_);
  SctpActor receiver("Z", emulated_socket_z_, options_);
  EXPECT_THAT(sender.sctp_socket().state(), SocketState::kClosed);

  sender.sctp_socket().Connect();
  Sleep(kAWhile);
  EXPECT_THAT(sender.sctp_socket().state(), SocketState::kConnected);

  sender.sctp_socket().Shutdown();
  Sleep(kAWhile);
  EXPECT_THAT(sender.sctp_socket().state(), SocketState::kClosed);
}

TEST_F(DcSctpSocketNetworkTest, CanSendLargeMessage) {
  webrtc::BuiltInNetworkBehaviorConfig pipe_config;
  pipe_config.queue_delay_ms = 30;
  MakeNetwork(pipe_config);

  SctpActor sender("A", emulated_socket_a_, options_);
  SctpActor receiver("Z", emulated_socket_z_, options_);
  sender.sctp_socket().Connect();

  constexpr size_t kPayloadSize = 100 * 1024;

  std::vector<uint8_t> payload(kPayloadSize);
  sender.sctp_socket().Send(DcSctpMessage(kStreamId, kPpid, payload),
                            SendOptions());

  Sleep(kAWhile);

  ASSERT_HAS_VALUE_AND_ASSIGN(DcSctpMessage message,
                              receiver.ConsumeReceivedMessage());

  EXPECT_THAT(message.payload(), SizeIs(kPayloadSize));

  sender.sctp_socket().Shutdown();
  Sleep(kAWhile);
}

TEST_F(DcSctpSocketNetworkTest, CanSendMessagesReliablyWithLowBandwidth) {
  webrtc::BuiltInNetworkBehaviorConfig pipe_config;
  pipe_config.queue_delay_ms = 30;
  pipe_config.link_capacity = DataRate::KilobitsPerSec(1000);
  MakeNetwork(pipe_config);

  SctpActor sender("A", emulated_socket_a_, options_);
  SctpActor receiver("Z", emulated_socket_z_, options_);
  sender.sctp_socket().Connect();

  sender.SetActorMode(ActorMode::kThroughputSender);
  receiver.SetActorMode(ActorMode::kThroughputReceiver);

  Sleep(kBenchmarkRuntime);
  sender.SetActorMode(ActorMode::kAtRest);
  receiver.SetActorMode(ActorMode::kAtRest);

  Sleep(kAWhile);

  sender.sctp_socket().Shutdown();

  Sleep(kAWhile);

  // Verify that the bitrates are in the range of 0.5-1.0 Mbps.
  double bitrate = receiver.avg_received_bitrate_mbps();
  EXPECT_THAT(bitrate, AllOf(Ge(0.5), Le(1.0)));
}

TEST_F(DcSctpSocketNetworkTest,
       DCSCTP_NDEBUG_TEST(CanSendMessagesReliablyWithMediumBandwidth)) {
  webrtc::BuiltInNetworkBehaviorConfig pipe_config;
  pipe_config.queue_delay_ms = 30;
  pipe_config.link_capacity = DataRate::KilobitsPerSec(18000);
  MakeNetwork(pipe_config);

  SctpActor sender("A", emulated_socket_a_, options_);
  SctpActor receiver("Z", emulated_socket_z_, options_);
  sender.sctp_socket().Connect();

  sender.SetActorMode(ActorMode::kThroughputSender);
  receiver.SetActorMode(ActorMode::kThroughputReceiver);

  Sleep(kBenchmarkRuntime);
  sender.SetActorMode(ActorMode::kAtRest);
  receiver.SetActorMode(ActorMode::kAtRest);

  Sleep(kAWhile);

  sender.sctp_socket().Shutdown();

  Sleep(kAWhile);

  // Verify that the bitrates are in the range of 16-18 Mbps.
  double bitrate = receiver.avg_received_bitrate_mbps();
  EXPECT_THAT(bitrate, AllOf(Ge(16), Le(18)));
}

TEST_F(DcSctpSocketNetworkTest, CanSendMessagesReliablyWithMuchPacketLoss) {
  webrtc::BuiltInNetworkBehaviorConfig config;
  config.queue_delay_ms = 30;
  config.loss_percent = 1;
  MakeNetwork(config);

  SctpActor sender("A", emulated_socket_a_, options_);
  SctpActor receiver("Z", emulated_socket_z_, options_);
  sender.sctp_socket().Connect();

  sender.SetActorMode(ActorMode::kThroughputSender);
  receiver.SetActorMode(ActorMode::kThroughputReceiver);

  Sleep(kBenchmarkRuntime);
  sender.SetActorMode(ActorMode::kAtRest);
  receiver.SetActorMode(ActorMode::kAtRest);

  Sleep(kAWhile);

  sender.sctp_socket().Shutdown();

  Sleep(kAWhile);

  // TCP calculator gives: 1200 MTU, 60ms RTT and 1% packet loss -> 1.6Mbps.
  // This test is doing slightly better (doesn't have any additional header
  // overhead etc). Verify that the bitrates are in the range of 1.5-2.5 Mbps.
  double bitrate = receiver.avg_received_bitrate_mbps();
  EXPECT_THAT(bitrate, AllOf(Ge(1.5), Le(2.5)));
}

TEST_F(DcSctpSocketNetworkTest, DCSCTP_NDEBUG_TEST(HasHighBandwidth)) {
  webrtc::BuiltInNetworkBehaviorConfig pipe_config;
  pipe_config.queue_delay_ms = 30;
  MakeNetwork(pipe_config);

  SctpActor sender("A", emulated_socket_a_, options_);
  SctpActor receiver("Z", emulated_socket_z_, options_);
  sender.sctp_socket().Connect();

  sender.SetActorMode(ActorMode::kThroughputSender);
  receiver.SetActorMode(ActorMode::kThroughputReceiver);

  Sleep(kBenchmarkRuntime);

  sender.SetActorMode(ActorMode::kAtRest);
  receiver.SetActorMode(ActorMode::kAtRest);
  Sleep(kAWhile);

  sender.sctp_socket().Shutdown();
  Sleep(kAWhile);

  // Verify that the bitrate is in the range of 540-640 Mbps
  double bitrate = receiver.avg_received_bitrate_mbps();
  EXPECT_THAT(bitrate, AllOf(Ge(520), Le(640)));
}
}  // namespace
}  // namespace dcsctp