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

#ifndef QUICHE_QUIC_MOQT_TEST_TOOLS_MOQT_SESSION_PEER_H_
#define QUICHE_QUIC_MOQT_TEST_TOOLS_MOQT_SESSION_PEER_H_

#include <cstddef>
#include <cstdint>
#include <memory>
#include <optional>
#include <string>
#include <utility>

#include "absl/base/casts.h"
#include "absl/base/nullability.h"
#include "absl/container/flat_hash_set.h"
#include "absl/memory/memory.h"
#include "absl/status/status.h"
#include "absl/strings/string_view.h"
#include "quiche/quic/core/quic_alarm.h"
#include "quiche/quic/core/quic_alarm_factory.h"
#include "quiche/quic/core/quic_time.h"
#include "quiche/quic/moqt/moqt_bidi_stream.h"
#include "quiche/quic/moqt/moqt_fetch_task.h"
#include "quiche/quic/moqt/moqt_key_value_pair.h"
#include "quiche/quic/moqt/moqt_messages.h"
#include "quiche/quic/moqt/moqt_names.h"
#include "quiche/quic/moqt/moqt_parser.h"
#include "quiche/quic/moqt/moqt_priority.h"
#include "quiche/quic/moqt/moqt_publisher.h"
#include "quiche/quic/moqt/moqt_session.h"
#include "quiche/quic/moqt/moqt_session_interface.h"
#include "quiche/quic/moqt/moqt_subscription.h"
#include "quiche/quic/moqt/moqt_track.h"
#include "quiche/quic/moqt/moqt_types.h"
#include "quiche/quic/moqt/moqt_uni_stream.h"
#include "quiche/quic/moqt/test_tools/moqt_framer_utils.h"
#include "quiche/common/platform/api/quiche_logging.h"
#include "quiche/common/platform/api/quiche_test.h"
#include "quiche/common/quiche_data_reader.h"
#include "quiche/web_transport/test_tools/mock_web_transport.h"
#include "quiche/web_transport/web_transport.h"

namespace moqt::test {

class MoqtDataParserPeer {
 public:
  static void SetType(MoqtDataParser* parser, MoqtDataStreamType type) {
    parser->type_ = type;
    parser->next_input_ = MoqtDataParser::NextInput::kTrackAlias;
  }
};

// TODO(martinduke): When subscription-specific tests are removed from,
// MoqtSessionTest, much of this file can be deleted (including
// SubscriptionPublisherPeer).

class SubscriptionPublisherPeer {
 public:
  static size_t num_open_streams(SubscriptionPublisher* publisher) {
    return publisher->stream_map_.GetAllStreams().size();
  }
  static std::optional<Location> largest_sent(
      const SubscriptionPublisher* publisher) {
    return publisher->largest_sent_;
  }
  static const absl::flat_hash_set<DataStreamIndex>& reset_subgroups(
      const SubscriptionPublisher* publisher) {
    return publisher->reset_subgroups_;
  }
};

// Helper class to interact with MOQT bidi streams in tests.
class MoqtBidiStreamTestWrapper {
 public:
  explicit MoqtBidiStreamTestWrapper(
      std::unique_ptr<MoqtBidiStreamBase> absl_nonnull stream)
      : stream_(std::move(stream)) {}

  MoqtBidiStreamBase& stream() { return *stream_; }

  // Simulates receiving the specified control message on the bidi stream.
  void ReceiveMessage(const AnyMoqtControlMessage& message) {
    std::string serialized = SerializeGenericMessage(message);
    quiche::QuicheDataReader reader(serialized);
    uint64_t raw_type;
    ASSERT_TRUE(reader.ReadVarInt62(&raw_type));
    ASSERT_TRUE(reader.Seek(2));
    absl::Status status = stream_->OnRawControlMessage(MoqtRawControlMessage{
        .type = static_cast<MoqtMessageType>(raw_type),
        .payload = std::string(reader.ReadRemainingPayload())});
    stream_->CheckStatus(status);
  }

 private:
  std::unique_ptr<MoqtBidiStreamBase> absl_nonnull stream_;
};

class MoqtSessionPeer {
 public:
  static constexpr webtransport::StreamId kControlStreamId = 4;

  static std::unique_ptr<MoqtBidiStreamTestWrapper> CreateControlStream(
      MoqtSession* session, webtransport::test::MockStream* stream) {
    auto new_stream = std::make_unique<MoqtSession::ControlStream>(session);
    session->control_stream_ = new_stream->GetWeakPtr();
    new_stream->BindStream(stream);
    ON_CALL(*stream, visitor())
        .WillByDefault(::testing::Return(new_stream.get()));
    ON_CALL(*stream, CanWrite).WillByDefault(::testing::Return(true));
    return std::make_unique<MoqtBidiStreamTestWrapper>(std::move(new_stream));
  }

  static std::unique_ptr<MoqtDataParserVisitor> CreateIncomingDataStream(
      MoqtSession* session, webtransport::Stream* stream,
      MoqtDataStreamType type) {
    auto new_stream =
        std::make_unique<MoqtSession::IncomingDataStream>(session, stream);
    MoqtDataParserPeer::SetType(&new_stream->parser_, type);
    return new_stream;
  }

  static std::unique_ptr<webtransport::StreamVisitor>
  CreateIncomingStreamVisitor(MoqtSession* session,
                              webtransport::Stream* stream) {
    auto new_stream =
        std::make_unique<MoqtSession::IncomingDataStream>(session, stream);
    return new_stream;
  }

  // In the test OnSessionReady, the session creates a stream and then passes
  // its unique_ptr to the mock webtransport stream. This function casts
  // that unique_ptr into a MoqtSession::Stream*, which is a private class of
  // MoqtSession, and then casts again into MoqtParserVisitor so that the test
  // can inject packets into that stream.
  // This function is useful for any test that wants to inject packets on a
  // stream created by the MoqtSession.
  static std::unique_ptr<MoqtBidiStreamTestWrapper>
  FetchParserVisitorFromWebtransportStreamVisitor(
      std::unique_ptr<webtransport::StreamVisitor> visitor) {
    return std::make_unique<MoqtBidiStreamTestWrapper>(absl::WrapUnique(
        absl::down_cast<MoqtSession::ControlStream*>(visitor.release())));
  }

  static void CreateRemoteTrack(MoqtSession* session,
                                const MoqtSubscribe& subscribe,
                                const std::optional<uint64_t> track_alias,
                                SubscribeVisitor* visitor) {
    auto track = std::make_unique<SubscribeRemoteTrack>(subscribe, visitor);
    if (track_alias.has_value()) {
      track->set_track_alias(*track_alias);
      session->subscribe_by_alias_.try_emplace(*track_alias, track.get());
    }
    session->subscribe_by_name_.try_emplace(subscribe.full_track_name,
                                            track.get());
    session->upstream_by_id_.try_emplace(subscribe.request_id,
                                         std::move(track));
  }

  static MoqtObjectListener* AddSubscription(
      MoqtSession* session, std::shared_ptr<MoqtTrackPublisher> publisher,
      uint64_t subscribe_id, uint64_t track_alias, uint64_t start_group,
      uint64_t start_object) {
    MessageParameters parameters;
    parameters.subscription_filter.emplace(Location(start_group, start_object));
    MoqtSubscribe subscribe(subscribe_id, publisher->GetTrackName(),
                            parameters);
    subscribe.parameters.set_forward(true);
    subscribe.parameters.subscription_filter.emplace(
        Location(start_group, start_object));
    subscribe.parameters.subscriber_priority = 0x80;
    subscribe.parameters.group_order = MoqtDeliveryOrder::kAscending;
    session->published_subscriptions_.emplace(
        subscribe_id,
        std::make_unique<SubscriptionPublisher>(
            session->framer_, std::move(publisher), session->GetControlStream(),
            subscribe_id, track_alias, subscribe.parameters, session,
            /*monitoring_interface=*/nullptr, session->callbacks_.clock,
            session->trace_recorder_));
    return session->published_subscriptions_[subscribe_id].get();
  }

  static bool InSubscriptionWindow(MoqtObjectListener* subscription,
                                   Location sequence) {
    std::optional<SubscriptionFilter> filter =
        absl::down_cast<SubscriptionPublisher*>(subscription)
            ->parameters()
            .subscription_filter;
    return (!filter.has_value() || filter->InWindow(sequence));
  }

  static MoqtObjectListener* GetSubscription(MoqtSession* session,
                                             uint64_t subscribe_id) {
    auto it = session->published_subscriptions_.find(subscribe_id);
    if (it == session->published_subscriptions_.end()) {
      return nullptr;
    }
    return it->second.get();
  }

  static void DeleteSubscription(MoqtSession* session, uint64_t subscribe_id) {
    session->published_subscriptions_.erase(subscribe_id);
  }

  static void UpdateSubscriberPriority(MoqtSession* session,
                                       uint64_t subscribe_id,
                                       MoqtPriority priority) {
    MessageParameters parameters;
    parameters.subscriber_priority = priority;
    session->published_subscriptions_[subscribe_id]->Update(parameters);
  }

  static SubscribeRemoteTrack* remote_track(MoqtSession* session,
                                            uint64_t track_alias) {
    return session->RemoteTrackByAlias(track_alias);
  }

  static void set_next_request_id(MoqtSession* session, uint64_t id) {
    session->next_request_id_ = id;
  }

  static void set_peer_max_request_id(MoqtSession* session, uint64_t id) {
    session->peer_max_request_id_ = id;
  }

  static MoqtSession::PublishedFetch* GetFetch(MoqtSession* session,
                                               uint64_t fetch_id) {
    auto it = session->incoming_fetches_.find(fetch_id);
    if (it == session->incoming_fetches_.end()) {
      return nullptr;
    }
    return it->second.get();
  }

  static void ValidateRequestId(MoqtSession* session, uint64_t id) {
    session->ValidateRequestId(id);
  }

  static std::optional<Location> LargestSentForSubscription(
      MoqtSession* session, uint64_t subscribe_id) {
    return SubscriptionPublisherPeer::largest_sent(
        session->published_subscriptions_[subscribe_id].get());
  }

  // Adds an upstream fetch and a stream ready to receive data.
  static std::unique_ptr<MoqtFetchTask> CreateUpstreamFetch(
      MoqtSession* session, webtransport::Stream* stream) {
    MoqtFetch fetch_message = {
        0,
        StandaloneFetch(FullTrackName{"foo", "bar"}, Location{0, 0},
                        Location{4, kMaxObjectId}),
        MessageParameters(),
    };
    std::unique_ptr<MoqtFetchTask> task;
    auto [it, success] = session->upstream_by_id_.try_emplace(
        0, std::make_unique<UpstreamFetch>(
               fetch_message, std::get<StandaloneFetch>(fetch_message.fetch),
               [&](std::unique_ptr<MoqtFetchTask> fetch_task) {
                 task = std::move(fetch_task);
               }));
    QUICHE_DCHECK(success);
    UpstreamFetch* fetch = absl::down_cast<UpstreamFetch*>(it->second.get());
    // Initialize the fetch task
    fetch->OnFetchResult(
        Location{4, 10}, absl::OkStatus(),
        [=, session_ptr = session, request_id = fetch_message.request_id]() {
          session_ptr->CancelFetch(request_id);
        });
    ;
    auto mock_session =
        absl::down_cast<webtransport::test::MockSession*>(session->session());
    EXPECT_CALL(*mock_session, AcceptIncomingUnidirectionalStream())
        .WillOnce(testing::Return(stream))
        .WillOnce(testing::Return(nullptr));
    session->OnIncomingUnidirectionalStreamAvailable();
    return task;
  }

  static quic::QuicAlarmFactory* GetAlarmFactory(MoqtSession* session) {
    return session->alarm_factory_.get();
  }

  static quic::QuicTime Now(MoqtSession* session) {
    return session->callbacks_.clock->ApproximateNow();
  }

  static quic::QuicAlarm* GetAlarm(webtransport::StreamVisitor* visitor) {
    return absl::down_cast<OutgoingSubgroupStream*>(visitor)
        ->delivery_timeout_alarm_.get();
  }

  static quic::QuicAlarm* GetPublishDoneAlarm(
      SubscribeRemoteTrack* subscription) {
    return subscription->publish_done_alarm_.get();
  }

  static quic::QuicAlarm* GetGoAwayTimeoutAlarm(MoqtSession* session) {
    return session->goaway_timeout_alarm_.get();
  }

  static quic::QuicTimeDelta GetDeliveryTimeout(
      MoqtObjectListener* subscription) {
    return absl::down_cast<SubscriptionPublisher*>(subscription)
        ->delivery_timeout();
  }
  static void SetDeliveryTimeout(MoqtObjectListener* subscription,
                                 quic::QuicTimeDelta timeout) {
    absl::down_cast<SubscriptionPublisher*>(subscription)
        ->parameters()
        .delivery_timeout = timeout;
  }

  static bool SubgroupHasBeenReset(MoqtObjectListener* subscription,
                                   DataStreamIndex index) {
    return SubscriptionPublisherPeer::reset_subgroups(
               absl::down_cast<SubscriptionPublisher*>(subscription))
        .contains(index);
  }

  static absl::string_view GetImplementationString(MoqtSession* session) {
    return session->parameters_.moqt_implementation;
  }

  static MoqtSession::ControlStream* GetControlStream(MoqtSession* session) {
    return session->control_stream_.GetIfAvailable();
  }

  static const MoqtSessionParameters& GetParameters(MoqtSession* session) {
    return session->parameters_;
  }
};

}  // namespace moqt::test

#endif  // QUICHE_QUIC_MOQT_TEST_TOOLS_MOQT_SESSION_PEER_H_