// 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.

#include "quiche/quic/moqt/moqt_session.h"

#include <algorithm>
#include <array>
#include <cstdint>
#include <memory>
#include <optional>
#include <string>
#include <utility>
#include <variant>
#include <vector>


#include "absl/algorithm/container.h"
#include "absl/base/nullability.h"
#include "absl/container/btree_map.h"
#include "absl/container/flat_hash_map.h"
#include "absl/container/flat_hash_set.h"
#include "absl/container/node_hash_map.h"
#include "absl/functional/bind_front.h"
#include "absl/memory/memory.h"
#include "absl/status/status.h"
#include "absl/strings/str_cat.h"
#include "absl/strings/string_view.h"
#include "absl/types/span.h"
#include "quiche/quic/core/quic_alarm_factory.h"
#include "quiche/quic/core/quic_time.h"
#include "quiche/quic/core/quic_types.h"
#include "quiche/quic/moqt/moqt_fetch_task.h"
#include "quiche/quic/moqt/moqt_framer.h"
#include "quiche/quic/moqt/moqt_messages.h"
#include "quiche/quic/moqt/moqt_object.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_callbacks.h"
#include "quiche/quic/moqt/moqt_session_interface.h"
#include "quiche/quic/moqt/moqt_subscribe_windows.h"
#include "quiche/quic/moqt/moqt_track.h"
#include "quiche/quic/platform/api/quic_logging.h"
#include "quiche/common/platform/api/quiche_bug_tracker.h"
#include "quiche/common/platform/api/quiche_logging.h"
#include "quiche/common/quiche_buffer_allocator.h"
#include "quiche/common/quiche_mem_slice.h"
#include "quiche/common/quiche_stream.h"
#include "quiche/common/quiche_weak_ptr.h"
#include "quiche/web_transport/web_transport.h"

#define ENDPOINT \
  (perspective() == Perspective::IS_SERVER ? "MoQT Server: " : "MoQT Client: ")

namespace moqt {

namespace {

using ::quic::Perspective;

// WebTransport lets applications split a session into multiple send groups
// that have equal weight for scheduling. We don't have a use for that, so the
// send group is always the same.
constexpr webtransport::SendGroupId kMoqtSendGroupId = 0;

std::optional<SubscribeWindow> SubscribeMessageToWindow(
    const MoqtSubscribe& subscribe) {
  if (!subscribe.forward ||
      subscribe.filter_type == MoqtFilterType::kLatestObject ||
      subscribe.filter_type == MoqtFilterType::kNextGroupStart) {
    return std::nullopt;
  }
  if (!subscribe.start.has_value()) {
    return std::nullopt;
  }
  return SubscribeWindow(*subscribe.start, subscribe.end_group);
}

class DefaultPublisher : public MoqtPublisher {
 public:
  static DefaultPublisher* GetInstance() {
    static DefaultPublisher* instance = new DefaultPublisher();
    return instance;
  }

  // MoqtPublisher implementation.
  absl_nullable std::shared_ptr<MoqtTrackPublisher> GetTrack(
      const FullTrackName& track_name) override {
    QUICHE_DCHECK(track_name.IsValid());
    return nullptr;
  }
};
}  // namespace

MoqtSession::MoqtSession(webtransport::Session* session,
                         MoqtSessionParameters parameters,
                         std::unique_ptr<quic::QuicAlarmFactory> alarm_factory,
                         MoqtSessionCallbacks callbacks)
    : session_(session),
      parameters_(parameters),
      callbacks_(std::move(callbacks)),
      framer_(parameters.using_webtrans),
      publisher_(DefaultPublisher::GetInstance()),
      local_max_request_id_(parameters.max_request_id),
      alarm_factory_(std::move(alarm_factory)),
      weak_ptr_factory_(this),
      liveness_token_(std::make_shared<Empty>()) {
  if (parameters_.using_webtrans) {
    session_->SetOnDraining([this]() {
      QUICHE_DLOG(INFO) << "WebTransport session is draining";
      received_goaway_ = true;
      if (callbacks_.goaway_received_callback != nullptr) {
        std::move(callbacks_.goaway_received_callback)(absl::string_view());
      }
    });
  }
  if (parameters_.perspective == Perspective::IS_SERVER) {
    next_request_id_ = 1;
  } else {
    next_incoming_request_id_ = 1;
  }
  QUICHE_DCHECK(parameters_.moqt_implementation.empty());
  parameters_.moqt_implementation = kVersionString;
}

MoqtSession::ControlStream* MoqtSession::GetControlStream() {
  if (!control_stream_.has_value()) {
    return nullptr;
  }
  webtransport::Stream* raw_stream = session_->GetStreamById(*control_stream_);
  if (raw_stream == nullptr) {
    return nullptr;
  }
  return static_cast<ControlStream*>(raw_stream->visitor());
}

void MoqtSession::SendControlMessage(quiche::QuicheBuffer message) {
  ControlStream* control_stream = GetControlStream();
  if (control_stream == nullptr) {
    QUICHE_LOG(DFATAL) << "Trying to send a message on the control stream "
                          "while it does not exist";
    return;
  }
  control_stream->SendOrBufferMessage(std::move(message));
}

void MoqtSession::OnSessionReady() {
  QUICHE_DLOG(INFO) << ENDPOINT << "Underlying session ready";
  if (parameters_.perspective == Perspective::IS_SERVER) {
    return;
  }

  webtransport::Stream* control_stream =
      session_->OpenOutgoingBidirectionalStream();
  if (control_stream == nullptr) {
    Error(MoqtError::kInternalError, "Unable to open a control stream");
    return;
  }
  control_stream->SetVisitor(
      std::make_unique<ControlStream>(this, control_stream));
  control_stream_ = control_stream->GetStreamId();
  MoqtClientSetup setup = MoqtClientSetup{
      .supported_versions = std::vector<MoqtVersion>{parameters_.version},
      .parameters = parameters_,
  };
  SendControlMessage(framer_.SerializeClientSetup(setup));
  QUIC_DLOG(INFO) << ENDPOINT << "Send the SETUP message";
}

void MoqtSession::OnSessionClosed(webtransport::SessionErrorCode,
                                  const std::string& error_message) {
  if (!error_.empty()) {
    // Avoid erroring out twice.
    return;
  }
  QUICHE_DLOG(INFO) << ENDPOINT << "Underlying session closed with message: "
                    << error_message;
  error_ = error_message;
  CleanUpState();
  std::move(callbacks_.session_terminated_callback)(error_message);
}

void MoqtSession::OnIncomingBidirectionalStreamAvailable() {
  while (webtransport::Stream* stream =
             session_->AcceptIncomingBidirectionalStream()) {
    if (control_stream_.has_value()) {
      Error(MoqtError::kProtocolViolation, "Bidirectional stream already open");
      return;
    }
    stream->SetVisitor(std::make_unique<ControlStream>(this, stream));
    stream->visitor()->OnCanRead();
  }
}
void MoqtSession::OnIncomingUnidirectionalStreamAvailable() {
  while (webtransport::Stream* stream =
             session_->AcceptIncomingUnidirectionalStream()) {
    stream->SetVisitor(std::make_unique<IncomingDataStream>(this, stream));
    stream->visitor()->OnCanRead();
  }
}

void MoqtSession::OnDatagramReceived(absl::string_view datagram) {
  MoqtObject message;
  std::optional<absl::string_view> payload = ParseDatagram(datagram, message);
  if (!payload.has_value()) {
    Error(MoqtError::kProtocolViolation, "Malformed datagram received");
    return;
  }
  QUICHE_DLOG(INFO) << ENDPOINT
                    << "Received OBJECT message in datagram for request_id "
                    << " for track alias " << message.track_alias
                    << " with sequence " << message.group_id << ":"
                    << message.object_id << " priority "
                    << message.publisher_priority << " length "
                    << payload->size();
  SubscribeRemoteTrack* track = RemoteTrackByAlias(message.track_alias);
  if (track == nullptr) {
    return;
  }
  if (!track->OnObject(/*is_datagram=*/true)) {
    OnMalformedTrack(track);
    return;
  }
  if (!track->InWindow(Location(message.group_id, message.object_id))) {
    // TODO(martinduke): a recent SUBSCRIBE_UPDATE could put us here, and it's
    // not an error.
    return;
  }
  QUICHE_CHECK(!track->is_fetch());
  SubscribeVisitor* visitor = track->visitor();
  if (visitor != nullptr) {
    // TODO(martinduke): Handle extension headers.
    PublishedObjectMetadata metadata;
    metadata.location = Location(message.group_id, message.object_id);
    metadata.subgroup = message.object_id;
    metadata.status = message.object_status;
    metadata.publisher_priority = message.publisher_priority;
    metadata.forwarding_preference = MoqtForwardingPreference::kDatagram;
    metadata.arrival_time = callbacks_.clock->Now();
    visitor->OnObjectFragment(track->full_track_name(), metadata, *payload,
                              true);
  }
}

void MoqtSession::Error(MoqtError code, absl::string_view error) {
  if (!error_.empty() || is_closing_) {
    // Avoid erroring out twice.
    return;
  }
  QUICHE_DLOG(INFO) << ENDPOINT << "MOQT session closed with code: "
                    << static_cast<int>(code) << " and message: " << error;
  error_ = std::string(error);
  session_->CloseSession(static_cast<uint64_t>(code), error);
  std::move(callbacks_.session_terminated_callback)(error);
  CleanUpState();
}

bool MoqtSession::SubscribeNamespace(
    TrackNamespace track_namespace,
    MoqtOutgoingSubscribeNamespaceCallback callback,
    VersionSpecificParameters parameters) {
  QUICHE_DCHECK(track_namespace.IsValid());
  if (received_goaway_ || sent_goaway_) {
    QUIC_DLOG(INFO) << ENDPOINT
                    << "Tried to send SUBSCRIBE_NAMESPACE after GOAWAY";
    return false;
  }
  if (next_request_id_ >= peer_max_request_id_) {
    if (!last_requests_blocked_sent_.has_value() ||
        peer_max_request_id_ > *last_requests_blocked_sent_) {
      MoqtRequestsBlocked requests_blocked;
      requests_blocked.max_request_id = peer_max_request_id_;
      SendControlMessage(framer_.SerializeRequestsBlocked(requests_blocked));
      last_requests_blocked_sent_ = peer_max_request_id_;
    }
    QUIC_DLOG(INFO) << ENDPOINT << "Tried to send SUBSCRIBE_NAMESPACE with ID "
                    << next_request_id_
                    << " which is greater than the maximum ID "
                    << peer_max_request_id_;
    return false;
  }
  if (outgoing_subscribe_namespaces_.contains(track_namespace)) {
    std::move(callback)(
        track_namespace, RequestErrorCode::kInternalError,
        "SUBSCRIBE_NAMESPACE already outstanding for namespace");
    return false;
  }
  MoqtSubscribeNamespace message;
  message.request_id = next_request_id_;
  next_request_id_ += 2;
  message.track_namespace = track_namespace;
  message.parameters = parameters;
  SendControlMessage(framer_.SerializeSubscribeNamespace(message));
  QUIC_DLOG(INFO) << ENDPOINT << "Sent SUBSCRIBE_NAMESPACE message for "
                  << message.track_namespace;
  pending_outgoing_subscribe_namespaces_[message.request_id] =
      PendingSubscribeNamespaceData{track_namespace, std::move(callback)};
  outgoing_subscribe_namespaces_.emplace(track_namespace);
  return true;
}

bool MoqtSession::UnsubscribeNamespace(TrackNamespace track_namespace) {
  QUICHE_DCHECK(track_namespace.IsValid());
  if (!outgoing_subscribe_namespaces_.contains(track_namespace)) {
    return false;
  }
  MoqtUnsubscribeNamespace message;
  message.track_namespace = track_namespace;
  SendControlMessage(framer_.SerializeUnsubscribeNamespace(message));
  QUIC_DLOG(INFO) << ENDPOINT << "Sent UNSUBSCRIBE_NAMESPACE message for "
                  << message.track_namespace;
  outgoing_subscribe_namespaces_.erase(track_namespace);
  return true;
}

void MoqtSession::PublishNamespace(
    TrackNamespace track_namespace,
    MoqtOutgoingPublishNamespaceCallback callback,
    VersionSpecificParameters parameters) {
  if (is_closing_) {
    return;
  }
  QUICHE_DCHECK(track_namespace.IsValid());
  if (outgoing_publish_namespaces_.contains(track_namespace)) {
    std::move(callback)(
        track_namespace,
        MoqtRequestError{
            RequestErrorCode::kInternalError,
            "PUBLISH_NAMESPACE already outstanding for namespace"});
    return;
  }
  if (next_request_id_ >= peer_max_request_id_) {
    if (!last_requests_blocked_sent_.has_value() ||
        peer_max_request_id_ > *last_requests_blocked_sent_) {
      MoqtRequestsBlocked requests_blocked;
      requests_blocked.max_request_id = peer_max_request_id_;
      SendControlMessage(framer_.SerializeRequestsBlocked(requests_blocked));
      last_requests_blocked_sent_ = peer_max_request_id_;
    }
    QUIC_DLOG(INFO) << ENDPOINT << "Tried to send PUBLISH_NAMESPACE with ID "
                    << next_request_id_
                    << " which is greater than the maximum ID "
                    << peer_max_request_id_;
    return;
  }
  if (received_goaway_ || sent_goaway_) {
    QUIC_DLOG(INFO) << ENDPOINT
                    << "Tried to send PUBLISH_NAMESPACE after GOAWAY";
    return;
  }
  MoqtPublishNamespace message;
  message.request_id = next_request_id_;
  next_request_id_ += 2;
  message.track_namespace = track_namespace;
  message.parameters = parameters;
  SendControlMessage(framer_.SerializePublishNamespace(message));
  QUIC_DLOG(INFO) << ENDPOINT << "Sent PUBLISH_NAMESPACE message for "
                  << message.track_namespace;
  pending_outgoing_publish_namespaces_[message.request_id] = track_namespace;
  outgoing_publish_namespaces_[track_namespace] = std::move(callback);
}

bool MoqtSession::PublishNamespaceDone(TrackNamespace track_namespace) {
  if (is_closing_) {
    return false;
  }
  QUICHE_DCHECK(track_namespace.IsValid());
  auto it = outgoing_publish_namespaces_.find(track_namespace);
  if (it == outgoing_publish_namespaces_.end()) {
    return false;  // Could have been destroyed by PUBLISH_NAMESPACE_CANCEL.
  }
  MoqtPublishNamespaceDone message;
  message.track_namespace = track_namespace;
  SendControlMessage(framer_.SerializePublishNamespaceDone(message));
  QUIC_DLOG(INFO) << ENDPOINT << "Sent PUBLISH_NAMESPACE_DONE message for "
                  << message.track_namespace;
  outgoing_publish_namespaces_.erase(it);
  return true;
}

void MoqtSession::CancelPublishNamespace(TrackNamespace track_namespace,
                                         RequestErrorCode code,
                                         absl::string_view reason) {
  QUICHE_DCHECK(track_namespace.IsValid());
  MoqtPublishNamespaceCancel message{track_namespace, code,
                                     std::string(reason)};

  incoming_publish_namespaces_.erase(track_namespace);
  SendControlMessage(framer_.SerializePublishNamespaceCancel(message));
  QUIC_DLOG(INFO) << ENDPOINT << "Sent PUBLISH_NAMESPACE_CANCEL message for "
                  << message.track_namespace << " with reason " << reason;
}

bool MoqtSession::SubscribeAbsolute(const FullTrackName& name,
                                    uint64_t start_group, uint64_t start_object,
                                    SubscribeVisitor* visitor,
                                    VersionSpecificParameters parameters) {
  QUICHE_DCHECK(name.IsValid());
  MoqtSubscribe message;
  message.full_track_name = name;
  message.subscriber_priority = kDefaultSubscriberPriority;
  message.group_order = std::nullopt;
  message.forward = true;
  message.filter_type = MoqtFilterType::kAbsoluteStart;
  message.start = Location(start_group, start_object);
  message.end_group = std::nullopt;
  message.parameters = parameters;
  return Subscribe(message, visitor);
}

bool MoqtSession::SubscribeAbsolute(const FullTrackName& name,
                                    uint64_t start_group, uint64_t start_object,
                                    uint64_t end_group,
                                    SubscribeVisitor* visitor,
                                    VersionSpecificParameters parameters) {
  QUICHE_DCHECK(name.IsValid());
  if (end_group < start_group) {
    QUIC_DLOG(ERROR) << "Subscription end is before beginning";
    return false;
  }
  MoqtSubscribe message;
  message.full_track_name = name;
  message.subscriber_priority = kDefaultSubscriberPriority;
  message.group_order = std::nullopt;
  message.forward = true;
  message.filter_type = MoqtFilterType::kAbsoluteRange;
  message.start = Location(start_group, start_object);
  message.end_group = end_group;
  message.parameters = parameters;
  return Subscribe(message, visitor);
}

bool MoqtSession::SubscribeCurrentObject(const FullTrackName& name,
                                         SubscribeVisitor* visitor,
                                         VersionSpecificParameters parameters) {
  QUICHE_DCHECK(name.IsValid());
  MoqtSubscribe message;
  message.full_track_name = name;
  message.subscriber_priority = kDefaultSubscriberPriority;
  message.group_order = std::nullopt;
  message.forward = true;
  message.filter_type = MoqtFilterType::kLatestObject;
  message.start = std::nullopt;
  message.end_group = std::nullopt;
  message.parameters = parameters;
  return Subscribe(message, visitor);
}

bool MoqtSession::SubscribeNextGroup(const FullTrackName& name,
                                     SubscribeVisitor* visitor,
                                     VersionSpecificParameters parameters) {
  QUICHE_DCHECK(name.IsValid());
  MoqtSubscribe message;
  message.full_track_name = name;
  message.subscriber_priority = kDefaultSubscriberPriority;
  message.group_order = std::nullopt;
  message.forward = true;
  message.filter_type = MoqtFilterType::kNextGroupStart;
  message.start = std::nullopt;
  message.end_group = std::nullopt;
  message.parameters = parameters;
  return Subscribe(message, visitor);
}

bool MoqtSession::SubscribeUpdate(
    const FullTrackName& name, std::optional<Location> start,
    std::optional<uint64_t> end_group,
    std::optional<MoqtPriority> subscriber_priority,
    std::optional<bool> forward, VersionSpecificParameters parameters) {
  QUICHE_DCHECK(name.IsValid());
  auto it = subscribe_by_name_.find(name);
  if (it == subscribe_by_name_.end()) {
    return false;
  }
  QUICHE_DCHECK(name.IsValid());
  SubscribeRemoteTrack* track = it->second;
  MoqtSubscribeUpdate subscribe_update;
  subscribe_update.request_id = track->request_id();
  subscribe_update.start = start.value_or(track->window().start());
  subscribe_update.end_group = end_group.value_or(track->window().end().group);
  if (subscribe_update.end_group == UINT64_MAX) {
    subscribe_update.end_group = std::nullopt;
  }
  subscribe_update.subscriber_priority =
      subscriber_priority.value_or(track->subscriber_priority());
  subscribe_update.forward = forward.value_or(track->forward());
  subscribe_update.parameters = parameters;
  if (subscribe_update.start < track->window().start() ||
      (subscribe_update.end_group.has_value() &&
       (*subscribe_update.end_group > track->window().end().group ||
        *subscribe_update.end_group < subscribe_update.start.group))) {
    // Invalid range.
    return false;
  }
  // Input is valid. Update subscription properties.
  track->TruncateStart(subscribe_update.start);
  if (subscribe_update.end_group.has_value()) {
    track->TruncateEnd(*subscribe_update.end_group);
  }
  track->set_subscriber_priority(subscribe_update.subscriber_priority);
  track->set_forward(subscribe_update.forward);
  SendControlMessage(framer_.SerializeSubscribeUpdate(subscribe_update));
  return true;
};

void MoqtSession::Unsubscribe(const FullTrackName& name) {
  if (is_closing_) {
    return;
  }
  QUICHE_DCHECK(name.IsValid());
  SubscribeRemoteTrack* track = RemoteTrackByName(name);
  if (track == nullptr) {
    return;
  }
  QUICHE_DCHECK(name.IsValid());
  QUIC_DLOG(INFO) << ENDPOINT << "Sent UNSUBSCRIBE message for " << name;
  MoqtUnsubscribe message;
  message.request_id = track->request_id();
  SendControlMessage(framer_.SerializeUnsubscribe(message));
  DestroySubscription(track);
}

bool MoqtSession::Fetch(const FullTrackName& name,
                        FetchResponseCallback callback, Location start,
                        uint64_t end_group, std::optional<uint64_t> end_object,
                        MoqtPriority priority,
                        std::optional<MoqtDeliveryOrder> delivery_order,
                        VersionSpecificParameters parameters) {
  QUICHE_DCHECK(name.IsValid());
  if (next_request_id_ >= peer_max_request_id_) {
    QUIC_DLOG(INFO) << ENDPOINT << "Tried to send FETCH with ID "
                    << next_request_id_
                    << " which is greater than the maximum ID "
                    << peer_max_request_id_;
    return false;
  }
  if (received_goaway_ || sent_goaway_) {
    QUIC_DLOG(INFO) << ENDPOINT << "Tried to send FETCH after GOAWAY";
    return false;
  }
  MoqtFetch message;
  Location end_location = end_object.has_value()
                              ? Location(end_group, *end_object)
                              : Location(end_group, kMaxObjectId);
  message.fetch = StandaloneFetch(name, start, end_location);
  message.request_id = next_request_id_;
  next_request_id_ += 2;
  message.subscriber_priority = priority;
  message.group_order = delivery_order;
  message.parameters = parameters;
  SendControlMessage(framer_.SerializeFetch(message));
  QUIC_DLOG(INFO) << ENDPOINT << "Sent FETCH message for " << name;
  auto fetch = std::make_unique<UpstreamFetch>(
      message, std::get<StandaloneFetch>(message.fetch), std::move(callback));
  upstream_by_id_.emplace(message.request_id, std::move(fetch));
  return true;
}

bool MoqtSession::RelativeJoiningFetch(const FullTrackName& name,
                                       SubscribeVisitor* visitor,
                                       uint64_t num_previous_groups,
                                       VersionSpecificParameters parameters) {
  QUICHE_DCHECK(name.IsValid());
  return RelativeJoiningFetch(
      name, visitor,
      [this, id = next_request_id_](std::unique_ptr<MoqtFetchTask> fetch_task) {
        // Move the fetch_task to the subscribe to plumb into its visitor.
        RemoteTrack* track = RemoteTrackById(id);
        if (track == nullptr || track->is_fetch()) {
          fetch_task.release();
          return;
        }
        auto* subscribe = static_cast<SubscribeRemoteTrack*>(track);
        RemoteTrackByName(track->full_track_name());
        subscribe->OnJoiningFetchReady(std::move(fetch_task));
      },
      num_previous_groups, kDefaultSubscriberPriority, std::nullopt,
      parameters);
}

bool MoqtSession::RelativeJoiningFetch(
    const FullTrackName& name, SubscribeVisitor* visitor,
    FetchResponseCallback callback, uint64_t num_previous_groups,
    MoqtPriority priority, std::optional<MoqtDeliveryOrder> delivery_order,
    VersionSpecificParameters parameters) {
  QUICHE_DCHECK(name.IsValid());
  if ((next_request_id_ + 2) >= peer_max_request_id_) {
    QUIC_DLOG(INFO) << ENDPOINT << "Tried to send JOINING_FETCH with ID "
                    << (next_request_id_ + 2)
                    << " which is greater than the maximum ID "
                    << peer_max_request_id_;
    return false;
  }
  MoqtSubscribe subscribe;
  subscribe.full_track_name = name;
  subscribe.subscriber_priority = priority;
  subscribe.group_order = delivery_order;
  subscribe.forward = true;
  subscribe.filter_type = MoqtFilterType::kLatestObject;
  subscribe.start = std::nullopt;
  subscribe.end_group = std::nullopt;
  subscribe.parameters = parameters;
  if (!Subscribe(subscribe, visitor)) {
    return false;
  }

  MoqtFetch fetch;
  fetch.request_id = next_request_id_;
  next_request_id_ += 2;
  fetch.subscriber_priority = priority;
  fetch.group_order = delivery_order;
  fetch.fetch = JoiningFetchRelative{subscribe.request_id, num_previous_groups};
  fetch.parameters = parameters;
  fetch.parameters.delivery_timeout = quic::QuicTimeDelta::Infinite();
  SendControlMessage(framer_.SerializeFetch(fetch));
  QUIC_DLOG(INFO) << ENDPOINT << "Sent Joining FETCH message for " << name;
  auto upstream_fetch =
      std::make_unique<UpstreamFetch>(fetch, name, std::move(callback));
  upstream_by_id_.emplace(fetch.request_id, std::move(upstream_fetch));
  return true;
}

void MoqtSession::GoAway(absl::string_view new_session_uri) {
  if (sent_goaway_) {
    QUIC_DLOG(INFO) << ENDPOINT << "Tried to send multiple GOAWAY";
    return;
  }
  if (!new_session_uri.empty() && !new_session_uri.empty()) {
    QUIC_DLOG(INFO) << ENDPOINT
                    << "Client tried to send GOAWAY with new session URI";
    return;
  }
  MoqtGoAway message;
  message.new_session_uri = std::string(new_session_uri);
  SendControlMessage(framer_.SerializeGoAway(message));
  sent_goaway_ = true;
  goaway_timeout_alarm_ = absl::WrapUnique(
      alarm_factory_->CreateAlarm(new GoAwayTimeoutDelegate(this)));
  goaway_timeout_alarm_->Set(callbacks_.clock->ApproximateNow() +
                             kDefaultGoAwayTimeout);
}

void MoqtSession::PublishedFetch::FetchStreamVisitor::OnCanWrite() {
  std::shared_ptr<PublishedFetch> fetch = fetch_.lock();
  if (fetch == nullptr) {
    return;
  }
  PublishedObject object;
  while (stream_->CanWrite()) {
    MoqtFetchTask::GetNextObjectResult result =
        fetch->fetch_task()->GetNextObject(object);
    switch (result) {
      case MoqtFetchTask::GetNextObjectResult::kSuccess:
        // Skip ObjectDoesNotExist in FETCH.
        if (object.metadata.status == MoqtObjectStatus::kObjectDoesNotExist) {
          QUIC_BUG(quic_bug_got_doesnotexist_in_fetch)
              << "Got ObjectDoesNotExist in FETCH";
          continue;
        }
        if (fetch->session_->WriteObjectToStream(
                stream_, fetch->request_id(), object.metadata,
                std::move(object.payload), MoqtDataStreamType::Fetch(),
                // last Object ID doesn't matter for FETCH, just use zero.
                stream_header_written_ ? std::optional<uint64_t>(0)
                                       : std::nullopt,
                /*fin=*/false)) {
          stream_header_written_ = true;
        }
        break;
      case MoqtFetchTask::GetNextObjectResult::kPending:
        return;
      case MoqtFetchTask::GetNextObjectResult::kEof:
        // TODO(martinduke): Either prefetch the next object, or alter the API
        // so that we're not sending FIN in a separate frame.
        if (!quiche::SendFinOnStream(*stream_).ok()) {
          QUIC_DVLOG(1) << "Sending FIN onStream " << stream_->GetStreamId()
                        << " failed";
        }
        return;
      case MoqtFetchTask::GetNextObjectResult::kError:
        stream_->ResetWithUserCode(static_cast<webtransport::StreamErrorCode>(
            fetch->fetch_task()->GetStatus().code()));
        return;
    }
  }
}

void MoqtSession::GoAwayTimeoutDelegate::OnAlarm() {
  session_->Error(MoqtError::kGoawayTimeout,
                  "Peer did not close session after GOAWAY");
}

bool MoqtSession::PublishIsDone(uint64_t request_id, PublishDoneCode code,
                                absl::string_view error_reason) {
  if (is_closing_) {
    return false;
  }
  auto it = published_subscriptions_.find(request_id);
  if (it == published_subscriptions_.end()) {
    return false;
  }

  PublishedSubscription& subscription = *it->second;
  std::vector<webtransport::StreamId> streams_to_reset =
      subscription.GetAllStreams();

  MoqtPublishDone publish_done;
  publish_done.request_id = request_id;
  publish_done.status_code = code;
  publish_done.stream_count = subscription.streams_opened();
  publish_done.error_reason = error_reason;
  SendControlMessage(framer_.SerializePublishDone(publish_done));
  QUIC_DLOG(INFO) << ENDPOINT << "Sent PUBLISH_DONE message for "
                  << subscription.publisher().GetTrackName();
  // Clean up the subscription
  published_subscriptions_.erase(it);
  for (webtransport::StreamId stream_id : streams_to_reset) {
    webtransport::Stream* stream = session_->GetStreamById(stream_id);
    if (stream == nullptr) {
      continue;
    }
    stream->ResetWithUserCode(kResetCodeCanceled);
  }
  return true;
}

void MoqtSession::MaybeDestroySubscription(SubscribeRemoteTrack* subscribe) {
  if (subscribe != nullptr && subscribe->all_streams_closed()) {
    DestroySubscription(subscribe);
  }
}

void MoqtSession::DestroySubscription(SubscribeRemoteTrack* subscribe) {
  if (subscribe->ErrorIsAllowed()) {
    subscribe->visitor()->OnReply(
        subscribe->full_track_name(),
        MoqtRequestError{RequestErrorCode::kNotSupported,
                         "Subscription closed"});
  } else {
    subscribe->visitor()->OnPublishDone(subscribe->full_track_name());
  }
  subscribe_by_name_.erase(subscribe->full_track_name());
  if (subscribe->track_alias().has_value()) {
    subscribe_by_alias_.erase(*subscribe->track_alias());
  }
  upstream_by_id_.erase(subscribe->request_id());
}

bool MoqtSession::Subscribe(MoqtSubscribe& message, SubscribeVisitor* visitor) {
  // TODO(martinduke): support authorization info
  if (next_request_id_ >= peer_max_request_id_) {
    if (!last_requests_blocked_sent_.has_value() ||
        peer_max_request_id_ > *last_requests_blocked_sent_) {
      MoqtRequestsBlocked requests_blocked;
      requests_blocked.max_request_id = peer_max_request_id_;
      SendControlMessage(framer_.SerializeRequestsBlocked(requests_blocked));
      last_requests_blocked_sent_ = peer_max_request_id_;
    }
    QUIC_DLOG(INFO) << ENDPOINT << "Tried to send SUBSCRIBE with ID "
                    << next_request_id_
                    << " which is greater than the maximum ID "
                    << peer_max_request_id_;
    return false;
  }
  if (subscribe_by_name_.contains(message.full_track_name)) {
    QUIC_DLOG(INFO) << ENDPOINT << "Tried to send SUBSCRIBE for track "
                    << message.full_track_name
                    << " which is already subscribed";
    return false;
  }
  if (received_goaway_ || sent_goaway_) {
    QUIC_DLOG(INFO) << ENDPOINT << "Tried to send SUBSCRIBE after GOAWAY";
    return false;
  }
  message.request_id = next_request_id_;
  next_request_id_ += 2;
  if (SupportsObjectAck() && visitor != nullptr) {
    // Since we do not expose subscribe IDs directly in the API, instead wrap
    // the session and subscribe ID in a callback.
    visitor->OnCanAckObjects(absl::bind_front(&MoqtSession::SendObjectAck, this,
                                              message.request_id));
  } else {
    QUICHE_DLOG_IF(WARNING, message.parameters.oack_window_size.has_value())
        << "Attempting to set object_ack_window on a connection that does not "
           "support it.";
    message.parameters.oack_window_size = std::nullopt;
  }
  SendControlMessage(framer_.SerializeSubscribe(message));
  QUIC_DLOG(INFO) << ENDPOINT << "Sent SUBSCRIBE message for "
                  << message.full_track_name;
  auto track = std::make_unique<SubscribeRemoteTrack>(message, visitor);
  subscribe_by_name_.emplace(message.full_track_name, track.get());
  upstream_by_id_.emplace(message.request_id, std::move(track));
  return true;
}

webtransport::Stream* MoqtSession::OpenOrQueueDataStream(
    uint64_t subscription_id, const NewStreamParameters& parameters) {
  auto it = published_subscriptions_.find(subscription_id);
  if (it == published_subscriptions_.end()) {
    // It is possible that the subscription has been discarded while the stream
    // was in the queue; discard those streams.
    return nullptr;
  }
  PublishedSubscription& subscription = *it->second;
  if (!session_->CanOpenNextOutgoingUnidirectionalStream()) {
    subscription.AddQueuedOutgoingDataStream(parameters);
    // The subscription will notify the session about how to update the
    // session's queue.
    // TODO: limit the number of streams in the queue.
    return nullptr;
  }
  return OpenDataStream(subscription, parameters);
}

webtransport::Stream* MoqtSession::OpenDataStream(
    PublishedSubscription& subscription,
    const NewStreamParameters& parameters) {
  webtransport::Stream* new_stream =
      session_->OpenOutgoingUnidirectionalStream();
  if (new_stream == nullptr) {
    QUICHE_BUG(MoqtSession_OpenDataStream_blocked)
        << "OpenDataStream called when creation of new streams is blocked.";
    return nullptr;
  }
  new_stream->SetVisitor(std::make_unique<OutgoingDataStream>(
      this, new_stream, subscription, parameters));
  subscription.OnDataStreamCreated(new_stream->GetStreamId(), parameters.index);
  return new_stream;
}

bool MoqtSession::OpenDataStream(std::shared_ptr<PublishedFetch> fetch,
                                 webtransport::SendOrder send_order) {
  webtransport::Stream* new_stream =
      session_->OpenOutgoingUnidirectionalStream();
  if (new_stream == nullptr) {
    QUICHE_BUG(MoqtSession_OpenDataStream_blocked)
        << "OpenDataStream called when creation of new streams is blocked.";
    return false;
  }
  fetch->SetStreamId(new_stream->GetStreamId());
  new_stream->SetPriority(webtransport::StreamPriority{
      /*send_group_id=*/kMoqtSendGroupId, send_order});
  // The line below will lead to updating ObjectsAvailableCallback in the
  // FetchTask to call OnCanWrite() on the stream. If there is an object
  // available, the callback will be invoked synchronously (i.e. before
  // SetVisitor() returns).
  new_stream->SetVisitor(
      std::make_unique<PublishedFetch::FetchStreamVisitor>(fetch, new_stream));
  return true;
}

SubscribeRemoteTrack* MoqtSession::RemoteTrackByAlias(uint64_t track_alias) {
  auto it = subscribe_by_alias_.find(track_alias);
  if (it == subscribe_by_alias_.end()) {
    return nullptr;
  }
  return it->second;
}

RemoteTrack* MoqtSession::RemoteTrackById(uint64_t request_id) {
  auto it = upstream_by_id_.find(request_id);
  if (it == upstream_by_id_.end()) {
    return nullptr;
  }
  return it->second.get();
}

SubscribeRemoteTrack* MoqtSession::RemoteTrackByName(
    const FullTrackName& name) {
  QUICHE_DCHECK(name.IsValid());
  auto it = subscribe_by_name_.find(name);
  if (it == subscribe_by_name_.end()) {
    return nullptr;
  }
  return it->second;
}

void MoqtSession::OnCanCreateNewOutgoingUnidirectionalStream() {
  while (!subscribes_with_queued_outgoing_data_streams_.empty() &&
         session_->CanOpenNextOutgoingUnidirectionalStream()) {
    auto next = subscribes_with_queued_outgoing_data_streams_.rbegin();
    auto subscription = published_subscriptions_.find(next->subscription_id);
    if (subscription == published_subscriptions_.end()) {
      auto fetch = incoming_fetches_.find(next->subscription_id);
      // Create the stream if the fetch still exists.
      if (fetch != incoming_fetches_.end() &&
          !OpenDataStream(fetch->second, next->send_order)) {
        return;  // A QUIC_BUG has fired because this shouldn't happen.
      }
      // FETCH needs only one stream, and can be deleted from the queue. Or,
      // there is no subscribe and no fetch; the entry in the queue is invalid.
      subscribes_with_queued_outgoing_data_streams_.erase((++next).base());
      continue;
    }
    // Pop the item from the subscription's queue, which might update
    // subscribes_with_queued_outgoing_data_streams_.
    NewStreamParameters next_queued_stream =
        subscription->second->NextQueuedOutgoingDataStream();
    // Check if Group is too old.
    if (next_queued_stream.index.group <
        subscription->second->first_active_group()) {
      // The stream is too old to be sent.
      continue;
    }
    // Open the stream.
    webtransport::Stream* stream =
        OpenDataStream(*subscription->second, next_queued_stream);
    if (stream != nullptr) {
      stream->visitor()->OnCanWrite();
    }
  }
}

void MoqtSession::UpdateQueuedSendOrder(
    uint64_t request_id, std::optional<webtransport::SendOrder> old_send_order,
    std::optional<webtransport::SendOrder> new_send_order) {
  if (old_send_order == new_send_order) {
    return;
  }
  if (old_send_order.has_value()) {
    subscribes_with_queued_outgoing_data_streams_.erase(
        SubscriptionWithQueuedStream{*old_send_order, request_id});
  }
  if (new_send_order.has_value()) {
    subscribes_with_queued_outgoing_data_streams_.emplace(*new_send_order,
                                                          request_id);
  }
}

void MoqtSession::GrantMoreRequests(uint64_t num_requests) {
  local_max_request_id_ += (num_requests * 2);
  MoqtMaxRequestId message;
  message.max_request_id = local_max_request_id_;
  SendControlMessage(framer_.SerializeMaxRequestId(message));
}

bool MoqtSession::ValidateRequestId(uint64_t request_id) {
  if (request_id >= local_max_request_id_) {
    QUIC_DLOG(INFO) << ENDPOINT << "Received request with too large ID";
    Error(MoqtError::kTooManyRequests, "Received request with too large ID");
    return false;
  }
  if (request_id != next_incoming_request_id_) {
    QUIC_DLOG(INFO) << ENDPOINT << "Request ID not monotonically increasing";
    Error(MoqtError::kInvalidRequestId,
          "Request ID not monotonically increasing");
    return false;
  }
  next_incoming_request_id_ = request_id + 2;
  return true;
}

MoqtSession::ControlStream::ControlStream(MoqtSession* session,
                                          webtransport::Stream* stream)
    : session_(session),
      stream_(stream),
      parser_(session->parameters_.using_webtrans, stream, *this) {
  stream_->SetPriority(
      webtransport::StreamPriority{/*send_group_id=*/kMoqtSendGroupId,
                                   /*send_order=*/kMoqtControlStreamSendOrder});
  session->trace_recorder_.RecordControlStreamCreated(stream->GetStreamId());
}

void MoqtSession::ControlStream::OnCanRead() {
  parser_.ReadAndDispatchMessages();
}
void MoqtSession::ControlStream::OnCanWrite() {
  // We buffer serialized control frames unconditionally, thus OnCanWrite()
  // requires no handling for control streams.
}

void MoqtSession::ControlStream::OnResetStreamReceived(
    webtransport::StreamErrorCode error) {
  session_->Error(MoqtError::kProtocolViolation,
                  absl::StrCat("Control stream reset with error code ", error));
}
void MoqtSession::ControlStream::OnStopSendingReceived(
    webtransport::StreamErrorCode error) {
  session_->Error(MoqtError::kProtocolViolation,
                  absl::StrCat("Control stream reset with error code ", error));
}

void MoqtSession::ControlStream::OnClientSetupMessage(
    const MoqtClientSetup& message) {
  session_->control_stream_ = stream_->GetStreamId();
  if (perspective() == Perspective::IS_CLIENT) {
    session_->Error(MoqtError::kProtocolViolation,
                    "Received CLIENT_SETUP from server");
    return;
  }
  if (absl::c_find(message.supported_versions, session_->parameters_.version) ==
      message.supported_versions.end()) {
    // TODO(martinduke): Is this the right error code? See issue #346.
    session_->Error(MoqtError::kVersionNegotiationFailed,
                    absl::StrCat("Version mismatch: expected 0x",
                                 absl::Hex(session_->parameters_.version)));
    return;
  }
  session_->peer_supports_object_ack_ = message.parameters.support_object_acks;
  QUICHE_DLOG(INFO) << ENDPOINT << "Received the SETUP message";
  if (session_->parameters_.perspective == Perspective::IS_SERVER) {
    MoqtServerSetup response;
    response.parameters = session_->parameters_;
    response.selected_version = session_->parameters_.version;
    SendOrBufferMessage(session_->framer_.SerializeServerSetup(response));
    QUIC_DLOG(INFO) << ENDPOINT << "Sent the SETUP message";
  }
  // TODO: handle path.
  session_->peer_max_request_id_ = message.parameters.max_request_id;
  std::move(session_->callbacks_.session_established_callback)();
}

void MoqtSession::ControlStream::OnServerSetupMessage(
    const MoqtServerSetup& message) {
  if (perspective() == Perspective::IS_SERVER) {
    session_->Error(MoqtError::kProtocolViolation,
                    "Received SERVER_SETUP from client");
    return;
  }
  if (message.selected_version != session_->parameters_.version) {
    // TODO(martinduke): Is this the right error code? See issue #346.
    session_->Error(MoqtError::kProtocolViolation,
                    absl::StrCat("Version mismatch: expected 0x",
                                 absl::Hex(session_->parameters_.version)));
    return;
  }
  session_->peer_supports_object_ack_ = message.parameters.support_object_acks;
  QUIC_DLOG(INFO) << ENDPOINT << "Received the SETUP message";
  // TODO: handle path.
  session_->peer_max_request_id_ = message.parameters.max_request_id;
  std::move(session_->callbacks_.session_established_callback)();
}

void MoqtSession::ControlStream::SendSubscribeError(
    uint64_t request_id, RequestErrorCode error_code,
    absl::string_view reason_phrase) {
  MoqtSubscribeError subscribe_error;
  subscribe_error.request_id = request_id;
  subscribe_error.error_code = error_code;
  subscribe_error.reason_phrase = reason_phrase;
  SendOrBufferMessage(
      session_->framer_.SerializeSubscribeError(subscribe_error));
}

void MoqtSession::ControlStream::SendFetchError(
    uint64_t request_id, RequestErrorCode error_code,
    absl::string_view error_reason) {
  MoqtFetchError fetch_error;
  fetch_error.request_id = request_id;
  fetch_error.error_code = error_code;
  fetch_error.error_reason = error_reason;
  SendOrBufferMessage(session_->framer_.SerializeFetchError(fetch_error));
}

void MoqtSession::ControlStream::OnSubscribeMessage(
    const MoqtSubscribe& message) {
  if (!session_->ValidateRequestId(message.request_id)) {
    return;
  }
  QUIC_DLOG(INFO) << ENDPOINT << "Received a SUBSCRIBE for "
                  << message.full_track_name;
  if (session_->sent_goaway_) {
    QUIC_DLOG(INFO) << ENDPOINT << "Received a SUBSCRIBE after GOAWAY";
    SendSubscribeError(message.request_id, RequestErrorCode::kUnauthorized,
                       "SUBSCRIBE after GOAWAY");
    return;
  }
  if (session_->subscribed_track_names_.contains(message.full_track_name)) {
    session_->Error(MoqtError::kProtocolViolation,
                    "Duplicate subscribe for track");
    return;
  }
  const FullTrackName& track_name = message.full_track_name;
  std::shared_ptr<MoqtTrackPublisher> track_publisher =
      session_->publisher_->GetTrack(track_name);
  if (track_publisher == nullptr) {
    QUIC_DLOG(INFO) << ENDPOINT << "SUBSCRIBE for " << track_name
                    << " rejected by the application: does not exist";
    SendSubscribeError(message.request_id, RequestErrorCode::kTrackDoesNotExist,
                       "not found");
    return;
  }

  MoqtPublishingMonitorInterface* monitoring = nullptr;
  auto monitoring_it =
      session_->monitoring_interfaces_for_published_tracks_.find(track_name);
  if (monitoring_it !=
      session_->monitoring_interfaces_for_published_tracks_.end()) {
    monitoring = monitoring_it->second;
    session_->monitoring_interfaces_for_published_tracks_.erase(monitoring_it);
  }

  MoqtTrackPublisher* track_publisher_ptr = track_publisher.get();
  auto subscription = std::make_unique<PublishedSubscription>(
      session_, track_publisher, message, session_->next_local_track_alias_++,
      monitoring);
  subscription->set_delivery_timeout(message.parameters.delivery_timeout);
  PublishedSubscription* subscription_ptr = subscription.get();
  auto [it, success] = session_->published_subscriptions_.emplace(
      message.request_id, std::move(subscription));
  if (!success) {
    QUICHE_NOTREACHED();  // ValidateRequestId() should have caught this.
  }
  track_publisher_ptr->AddObjectListener(subscription_ptr);
}

void MoqtSession::ControlStream::OnSubscribeOkMessage(
    const MoqtSubscribeOk& message) {
  RemoteTrack* track = session_->RemoteTrackById(message.request_id);
  if (track == nullptr) {
    QUIC_DLOG(INFO) << ENDPOINT << "Received the SUBSCRIBE_OK for "
                    << "request_id = " << message.request_id
                    << " but no track exists";
    // Subscription state might have been destroyed for internal reasons.
    return;
  }
  if (track->is_fetch()) {
    session_->Error(MoqtError::kProtocolViolation,
                    "Received SUBSCRIBE_OK for a FETCH");
    return;
  }
  if (message.largest_location.has_value()) {
    QUIC_DLOG(INFO) << ENDPOINT << "Received the SUBSCRIBE_OK for "
                    << "request_id = " << message.request_id << " "
                    << track->full_track_name()
                    << " largest_id = " << *message.largest_location;
  } else {
    QUIC_DLOG(INFO) << ENDPOINT << "Received the SUBSCRIBE_OK for "
                    << "request_id = " << message.request_id << " "
                    << track->full_track_name();
  }
  SubscribeRemoteTrack* subscribe = static_cast<SubscribeRemoteTrack*>(track);
  subscribe->OnObjectOrOk();
  auto [it, success] =
      session_->subscribe_by_alias_.try_emplace(message.track_alias, subscribe);
  if (!success) {
    session_->Error(MoqtError::kDuplicateTrackAlias, "");
    return;
  }
  subscribe->set_track_alias(message.track_alias);
  // TODO(martinduke): Handle expires field.
  if (message.largest_location.has_value()) {
    subscribe->TruncateStart(message.largest_location->Next());
  }
  if (subscribe->visitor() != nullptr) {
    subscribe->visitor()->OnReply(
        track->full_track_name(),
        SubscribeOkData{message.expires, message.group_order,
                        message.largest_location, message.parameters});
  }
}

void MoqtSession::ControlStream::OnSubscribeErrorMessage(
    const MoqtSubscribeError& message) {
  RemoteTrack* track = session_->RemoteTrackById(message.request_id);
  if (track == nullptr) {
    QUIC_DLOG(INFO) << ENDPOINT << "Received the SUBSCRIBE_ERROR for "
                    << "request_id = " << message.request_id
                    << " but no track exists";
    // Subscription state might have been destroyed for internal reasons.
    return;
  }
  if (track->is_fetch()) {
    session_->Error(MoqtError::kProtocolViolation,
                    "Received SUBSCRIBE_ERROR for a FETCH");
    return;
  }
  if (!track->ErrorIsAllowed()) {
    session_->Error(MoqtError::kProtocolViolation,
                    "Received SUBSCRIBE_ERROR after SUBSCRIBE_OK or objects");
    return;
  }
  QUIC_DLOG(INFO) << ENDPOINT << "Received the SUBSCRIBE_ERROR for "
                  << "request_id = " << message.request_id << " ("
                  << track->full_track_name() << ")"
                  << ", error = " << static_cast<int>(message.error_code)
                  << " (" << message.reason_phrase << ")";
  SubscribeRemoteTrack* subscribe = static_cast<SubscribeRemoteTrack*>(track);
  // Delete the by-name entry at this point prevents Subscribe() from throwing
  // an error due to a duplicate track name. The other entries for this
  // subscribe will be deleted after calling Subscribe().
  session_->subscribe_by_name_.erase(subscribe->full_track_name());
  if (subscribe->visitor() != nullptr) {
    subscribe->visitor()->OnReply(
        subscribe->full_track_name(),
        MoqtRequestError{message.error_code, message.reason_phrase});
  }
  if (!session_->is_closing_) {
    // The visitor might have closed the session.
    session_->upstream_by_id_.erase(subscribe->request_id());
  }
}

void MoqtSession::ControlStream::OnUnsubscribeMessage(
    const MoqtUnsubscribe& message) {
  auto it = session_->published_subscriptions_.find(message.request_id);
  if (it == session_->published_subscriptions_.end()) {
    return;
  }
  QUIC_DLOG(INFO) << ENDPOINT << "Received an UNSUBSCRIBE for "
                  << it->second->publisher().GetTrackName();
  session_->published_subscriptions_.erase(it);
}

void MoqtSession::ControlStream::OnPublishDoneMessage(
    const MoqtPublishDone& message) {
  auto it = session_->upstream_by_id_.find(message.request_id);
  if (it == session_->upstream_by_id_.end()) {
    return;
  }
  auto* subscribe = static_cast<SubscribeRemoteTrack*>(it->second.get());
  QUIC_DLOG(INFO) << ENDPOINT << "Received a PUBLISH_DONE for "
                  << it->second->full_track_name();
  subscribe->OnPublishDone(
      message.stream_count, session_->callbacks_.clock,
      absl::WrapUnique(session_->alarm_factory_->CreateAlarm(
          new PublishDoneDelegate(session_, subscribe))));
  session_->MaybeDestroySubscription(subscribe);
}

void MoqtSession::ControlStream::OnSubscribeUpdateMessage(
    const MoqtSubscribeUpdate& message) {
  auto it = session_->published_subscriptions_.find(message.request_id);
  if (it == session_->published_subscriptions_.end()) {
    return;
  }
  it->second->Update(message.start, message.end_group,
                     message.subscriber_priority);
  it->second->set_delivery_timeout(message.parameters.delivery_timeout);
}

void MoqtSession::ControlStream::OnPublishNamespaceMessage(
    const MoqtPublishNamespace& message) {
  if (!session_->ValidateRequestId(message.request_id)) {
    return;
  }
  if (session_->sent_goaway_) {
    QUIC_DLOG(INFO) << ENDPOINT << "Received a PUBLISH_NAMESPACE after GOAWAY";
    MoqtPublishNamespaceError error;
    error.request_id = message.request_id;
    error.error_code = RequestErrorCode::kUnauthorized;
    error.error_reason = "PUBLISH_NAMESPACE after GOAWAY";
    SendOrBufferMessage(
        session_->framer_.SerializePublishNamespaceError(error));
    return;
  }
  QUIC_DLOG(INFO) << ENDPOINT << "Received a PUBLISH_NAMESPACE for "
                  << message.track_namespace;
  quiche::QuicheWeakPtr<MoqtSessionInterface> session_weakptr =
      session_->GetWeakPtr();
  session_->callbacks_.incoming_publish_namespace_callback(
      message.track_namespace, message.parameters,
      [&](std::optional<MoqtRequestError> error) {
        MoqtSession* session =
            static_cast<MoqtSession*>(session_weakptr.GetIfAvailable());
        if (session == nullptr) {
          return;
        }
        if (error.has_value()) {
          MoqtPublishNamespaceError reply;
          reply.request_id = message.request_id;
          reply.error_code = error->error_code;
          reply.error_reason = error->reason_phrase;
          SendOrBufferMessage(
              session->framer_.SerializePublishNamespaceError(reply));
        } else {
          MoqtPublishNamespaceOk ok;
          ok.request_id = message.request_id;
          SendOrBufferMessage(session->framer_.SerializePublishNamespaceOk(ok));
          session->incoming_publish_namespaces_.insert(message.track_namespace);
        }
      });
}

// Do not enforce that there is only one of OK or ERROR per PUBLISH_NAMESPACE.
// Upon ERROR, we immediately destroy the state.
void MoqtSession::ControlStream::OnPublishNamespaceOkMessage(
    const MoqtPublishNamespaceOk& message) {
  auto it =
      session_->pending_outgoing_publish_namespaces_.find(message.request_id);
  if (it == session_->pending_outgoing_publish_namespaces_.end()) {
    session_->Error(MoqtError::kProtocolViolation,
                    "Received PUBLISH_NAMESPACE_OK for unknown request_id");
    return;
  }
  TrackNamespace track_namespace = it->second;
  session_->pending_outgoing_publish_namespaces_.erase(it);
  auto callback_it =
      session_->outgoing_publish_namespaces_.find(track_namespace);
  if (callback_it == session_->outgoing_publish_namespaces_.end()) {
    // It might have already been destroyed due to PUBLISH_NAMESPACE_DONE.
    return;
  }
  std::move(callback_it->second)(track_namespace, std::nullopt);
}

void MoqtSession::ControlStream::OnPublishNamespaceErrorMessage(
    const MoqtPublishNamespaceError& message) {
  auto it =
      session_->pending_outgoing_publish_namespaces_.find(message.request_id);
  if (it == session_->pending_outgoing_publish_namespaces_.end()) {
    session_->Error(MoqtError::kProtocolViolation,
                    "Received PUBLISH_NAMESPACE_ERROR for unknown request_id");
    return;
  }
  TrackNamespace track_namespace = it->second;
  session_->pending_outgoing_publish_namespaces_.erase(it);
  auto it2 = session_->outgoing_publish_namespaces_.find(track_namespace);
  if (it2 == session_->outgoing_publish_namespaces_.end()) {
    return;  // State might have been destroyed due to PUBLISH_NAMESPACE_DONE.
  }
  std::move(it2->second)(
      track_namespace,
      MoqtRequestError{message.error_code, std::string(message.error_reason)});
  session_->outgoing_publish_namespaces_.erase(it2);
}

void MoqtSession::ControlStream::OnPublishNamespaceDoneMessage(
    const MoqtPublishNamespaceDone& message) {
  session_->incoming_publish_namespaces_.erase(message.track_namespace);
  session_->callbacks_.incoming_publish_namespace_callback(
      message.track_namespace, std::nullopt, nullptr);
}

void MoqtSession::ControlStream::OnPublishNamespaceCancelMessage(
    const MoqtPublishNamespaceCancel& message) {
  // The spec currently says that if a later SUBSCRIBE arrives for this
  // namespace, that SHOULD be a session error. I'm hoping that via Issue #557,
  // this will go away. Regardless, a SHOULD will not compel the session to keep
  // state forever, so there is no support for this requirement.
  auto it =
      session_->outgoing_publish_namespaces_.find(message.track_namespace);
  if (it == session_->outgoing_publish_namespaces_.end()) {
    return;  // State might have been destroyed due to PUBLISH_NAMESPACE_DONE.
  }
  std::move(it->second)(
      message.track_namespace,
      MoqtRequestError{message.error_code, std::string(message.error_reason)});
  session_->outgoing_publish_namespaces_.erase(it);
}

void MoqtSession::ControlStream::OnTrackStatusMessage(
    const MoqtTrackStatus& message) {
  if (!session_->ValidateRequestId(message.request_id)) {
    return;
  }
  if (session_->sent_goaway_) {
    QUIC_DLOG(INFO) << ENDPOINT
                    << "Received a TRACK_STATUS_REQUEST after GOAWAY";
    MoqtTrackStatusError error;
    error.request_id = message.request_id;
    error.error_code = RequestErrorCode::kUnauthorized;
    error.reason_phrase = "TRACK_STATUS_REQUEST after GOAWAY";
    SendOrBufferMessage(session_->framer_.SerializeTrackStatusError(error));
    return;
  }
  // TODO(martinduke): Handle authentication.
  std::shared_ptr<MoqtTrackPublisher> track =
      session_->publisher_->GetTrack(message.full_track_name);
  if (track == nullptr) {
    MoqtTrackStatusError error;
    error.request_id = message.request_id;
    error.error_code = RequestErrorCode::kTrackDoesNotExist;
    error.reason_phrase = "Track does not exist";
    SendOrBufferMessage(session_->framer_.SerializeTrackStatusError(error));
    return;
  }
  auto [it, inserted] = session_->incoming_track_status_.emplace(
      message.request_id, std::make_unique<DownstreamTrackStatus>(
                              message.request_id, session_, track.get()));
  track->AddObjectListener(it->second.get());
}

void MoqtSession::ControlStream::OnGoAwayMessage(const MoqtGoAway& message) {
  if (!message.new_session_uri.empty() &&
      perspective() == quic::Perspective::IS_SERVER) {
    session_->Error(MoqtError::kProtocolViolation,
                    "Received GOAWAY with new_session_uri on the server");
    return;
  }
  if (session_->received_goaway_) {
    session_->Error(MoqtError::kProtocolViolation,
                    "Received multiple GOAWAY messages");
    return;
  }
  session_->received_goaway_ = true;
  if (session_->callbacks_.goaway_received_callback != nullptr) {
    std::move(session_->callbacks_.goaway_received_callback)(
        message.new_session_uri);
  }
}

void MoqtSession::ControlStream::OnSubscribeNamespaceMessage(
    const MoqtSubscribeNamespace& message) {
  if (!session_->ValidateRequestId(message.request_id)) {
    return;
  }
  // TODO(martinduke): Handle authentication.
  if (session_->sent_goaway_) {
    QUIC_DLOG(INFO) << ENDPOINT
                    << "Received a SUBSCRIBE_NAMESPACE after GOAWAY";
    MoqtSubscribeNamespaceError error;
    error.request_id = message.request_id;
    error.error_code = RequestErrorCode::kUnauthorized;
    error.error_reason = "SUBSCRIBE_NAMESPACE after GOAWAY";
    SendOrBufferMessage(
        session_->framer_.SerializeSubscribeNamespaceError(error));
    return;
  }
  if (!session_->incoming_subscribe_namespace_.SubscribeNamespace(
          message.track_namespace)) {
    QUIC_DLOG(INFO) << ENDPOINT << "Received a SUBSCRIBE_NAMESPACE for "
                    << message.track_namespace
                    << " that is already subscribed to";
    MoqtSubscribeNamespaceError error;
    error.request_id = message.request_id;
    error.error_code = RequestErrorCode::kNamespacePrefixOverlap;
    error.error_reason = "SUBSCRIBE_NAMESPACE for similar subscribed namespace";
    SendOrBufferMessage(
        session_->framer_.SerializeSubscribeNamespaceError(error));
    return;
  }
  (session_->callbacks_.incoming_subscribe_namespace_callback)(
      message.track_namespace, message.parameters,
      [&](std::optional<MoqtRequestError> error) {
        if (error.has_value()) {
          MoqtSubscribeNamespaceError reply;
          reply.request_id = message.request_id;
          reply.error_code = error->error_code;
          reply.error_reason = error->reason_phrase;
          SendOrBufferMessage(
              session_->framer_.SerializeSubscribeNamespaceError(reply));
          session_->incoming_subscribe_namespace_.UnsubscribeNamespace(
              message.track_namespace);
        } else {
          MoqtSubscribeNamespaceOk ok;
          ok.request_id = message.request_id;
          SendOrBufferMessage(
              session_->framer_.SerializeSubscribeNamespaceOk(ok));
        }
      });
}

void MoqtSession::ControlStream::OnSubscribeNamespaceOkMessage(
    const MoqtSubscribeNamespaceOk& message) {
  auto it =
      session_->pending_outgoing_subscribe_namespaces_.find(message.request_id);
  if (it == session_->pending_outgoing_subscribe_namespaces_.end()) {
    session_->Error(MoqtError::kProtocolViolation,
                    "Received SUBSCRIBE_NAMESPACE_OK for unknown request_id");
    return;  // UNSUBSCRIBE_NAMESPACE may already have deleted the entry.
  }
  std::move(it->second.callback)(it->second.track_namespace, std::nullopt, "");
  session_->pending_outgoing_subscribe_namespaces_.erase(it);
}

void MoqtSession::ControlStream::OnSubscribeNamespaceErrorMessage(
    const MoqtSubscribeNamespaceError& message) {
  auto it =
      session_->pending_outgoing_subscribe_namespaces_.find(message.request_id);
  if (it == session_->pending_outgoing_subscribe_namespaces_.end()) {
    session_->Error(
        MoqtError::kProtocolViolation,
        "Received SUBSCRIBE_NAMESPACE_ERROR for unknown request_id");
    return;  // UNSUBSCRIBE_NAMESPACE may already have deleted the entry.
  }
  std::move(it->second.callback)(it->second.track_namespace, message.error_code,
                                 absl::string_view(message.error_reason));
  session_->outgoing_subscribe_namespaces_.erase(it->second.track_namespace);
  session_->pending_outgoing_subscribe_namespaces_.erase(it);
}

void MoqtSession::ControlStream::OnUnsubscribeNamespaceMessage(
    const MoqtUnsubscribeNamespace& message) {
  // MoqtSession keeps no state here, so just tell the application.
  session_->incoming_subscribe_namespace_.UnsubscribeNamespace(
      message.track_namespace);
  session_->callbacks_.incoming_subscribe_namespace_callback(
      message.track_namespace, std::nullopt, nullptr);
}

void MoqtSession::ControlStream::OnMaxRequestIdMessage(
    const MoqtMaxRequestId& message) {
  if (message.max_request_id < session_->peer_max_request_id_) {
    QUIC_DLOG(INFO) << ENDPOINT
                    << "Peer sent MAX_REQUEST_ID message with "
                       "lower value than previous";
    session_->Error(MoqtError::kProtocolViolation,
                    "MAX_REQUEST_ID has lower value than previous");
    return;
  }
  session_->peer_max_request_id_ = message.max_request_id;
}

void MoqtSession::ControlStream::OnFetchMessage(const MoqtFetch& message) {
  if (!session_->ValidateRequestId(message.request_id)) {
    return;
  }
  if (session_->sent_goaway_) {
    QUIC_DLOG(INFO) << ENDPOINT << "Received a FETCH after GOAWAY";
    SendFetchError(message.request_id, RequestErrorCode::kUnauthorized,
                   "FETCH after GOAWAY");
    return;
  }
  std::unique_ptr<MoqtFetchTask> fetch;
  FullTrackName track_name;
  if (std::holds_alternative<StandaloneFetch>(message.fetch)) {
    const StandaloneFetch& standalone_fetch =
        std::get<StandaloneFetch>(message.fetch);
    track_name = standalone_fetch.full_track_name;
    std::shared_ptr<MoqtTrackPublisher> track_publisher =
        session_->publisher_->GetTrack(track_name);
    if (track_publisher == nullptr) {
      QUIC_DLOG(INFO) << ENDPOINT << "FETCH for " << track_name
                      << " rejected by the application: not found";
      SendFetchError(message.request_id, RequestErrorCode::kTrackDoesNotExist,
                     "not found");
    }
    QUIC_DLOG(INFO) << ENDPOINT << "Received a StandaloneFETCH for "
                    << track_name;
    // The check for end_object < start_object is done in
    // MoqtTrackPublisher::Fetch().
    fetch = track_publisher->StandaloneFetch(standalone_fetch.start_location,
                                             standalone_fetch.end_location,
                                             message.group_order);
  } else {
    // Joining Fetch processing.
    uint64_t joining_request_id =
        std::holds_alternative<JoiningFetchRelative>(message.fetch)
            ? std::get<struct JoiningFetchRelative>(message.fetch)
                  .joining_request_id
            : std::get<JoiningFetchAbsolute>(message.fetch).joining_request_id;
    auto it = session_->published_subscriptions_.find(joining_request_id);
    if (it == session_->published_subscriptions_.end()) {
      QUIC_DLOG(INFO) << ENDPOINT << "Received a JOINING_FETCH for "
                      << "request_id " << joining_request_id
                      << " that does not exist";
      SendFetchError(message.request_id,
                     RequestErrorCode::kInvalidJoiningRequestId,
                     "Joining Fetch for non-existent request");
      return;
    }
    if (it->second->filter_type() != MoqtFilterType::kLatestObject) {
      // Current state variables do not allow us to distinguish between
      // LatestObject and AbsoluteStart with object ID > 0, but accept
      // JoiningFetch for AbsoluteStart.
      QUIC_DLOG(INFO) << ENDPOINT << "Received a JOINING_FETCH for "
                      << "joining_request_id " << joining_request_id
                      << " that is not a LatestObject";
      session_->Error(MoqtError::kProtocolViolation,
                      "Joining Fetch for non-LatestObject subscribe");
      return;
    }
    track_name = it->second->publisher().GetTrackName();
    if (std::holds_alternative<JoiningFetchRelative>(message.fetch)) {
      const JoiningFetchRelative& relative_fetch =
          std::get<JoiningFetchRelative>(message.fetch);
      QUIC_DLOG(INFO) << ENDPOINT << "Received a Relative Joining FETCH for "
                      << track_name;
      fetch = it->second->publisher().RelativeFetch(
          relative_fetch.joining_start, message.group_order);
    } else {
      const JoiningFetchAbsolute& absolute_fetch =
          std::get<JoiningFetchAbsolute>(message.fetch);
      QUIC_DLOG(INFO) << ENDPOINT << "Received a Absolute Joining FETCH for "
                      << track_name;
      fetch = it->second->publisher().AbsoluteFetch(
          absolute_fetch.joining_start, message.group_order);
    }
  }
  if (!fetch->GetStatus().ok()) {
    QUIC_DLOG(INFO) << ENDPOINT << "FETCH for " << track_name
                    << " could not initialize the task";
    SendFetchError(message.request_id, RequestErrorCode::kInvalidRange,
                   fetch->GetStatus().message());
    return;
  }
  auto published_fetch = std::make_unique<PublishedFetch>(
      message.request_id, session_, std::move(fetch));
  auto result = session_->incoming_fetches_.emplace(message.request_id,
                                                    std::move(published_fetch));
  if (!result.second) {  // Emplace failed.
    QUIC_DLOG(INFO) << ENDPOINT << "FETCH for " << track_name
                    << " could not be added to the session";
    SendFetchError(message.request_id, RequestErrorCode::kInternalError,
                   "Could not initialize FETCH state");
  }
  MoqtFetchTask* fetch_task = result.first->second->fetch_task();
  fetch_task->SetFetchResponseCallback(
      [this, request_id = message.request_id](
          std::variant<MoqtFetchOk, MoqtFetchError> message) {
        if (!session_->incoming_fetches_.contains(request_id)) {
          return;  // FETCH was cancelled.
        }
        if (std::holds_alternative<MoqtFetchOk>(message)) {
          MoqtFetchOk& fetch_ok = std::get<MoqtFetchOk>(message);
          fetch_ok.request_id = request_id;
          SendOrBufferMessage(session_->framer_.SerializeFetchOk(fetch_ok));
          return;
        }
        MoqtFetchError& fetch_error = std::get<MoqtFetchError>(message);
        fetch_error.request_id = request_id;
        SendOrBufferMessage(session_->framer_.SerializeFetchError(fetch_error));
      });
  // Set a temporary new-object callback that creates a data stream. When
  // created, the stream visitor will replace this callback.
  fetch_task->SetObjectAvailableCallback(
      [this, send_order = SendOrderForFetch(message.subscriber_priority),
       request_id = message.request_id]() {
        auto it = session_->incoming_fetches_.find(request_id);
        if (it == session_->incoming_fetches_.end()) {
          return;
        }
        if (!session_->session()->CanOpenNextOutgoingUnidirectionalStream() ||
            !session_->OpenDataStream(it->second, send_order)) {
          if (!session_->subscribes_with_queued_outgoing_data_streams_.contains(
                  SubscriptionWithQueuedStream(request_id, send_order))) {
            // Put the FETCH in the queue for a new stream unless it has already
            // done so.
            session_->UpdateQueuedSendOrder(request_id, std::nullopt,
                                            send_order);
          }
        }
      });
}

void MoqtSession::ControlStream::OnFetchOkMessage(const MoqtFetchOk& message) {
  RemoteTrack* track = session_->RemoteTrackById(message.request_id);
  if (track == nullptr) {
    QUIC_DLOG(INFO) << ENDPOINT << "Received the FETCH_OK for "
                    << "request_id = " << message.request_id
                    << " but no track exists";
    // Subscription state might have been destroyed for internal reasons.
    return;
  }
  if (!track->is_fetch()) {
    session_->Error(MoqtError::kProtocolViolation,
                    "Received FETCH_OK for a SUBSCRIBE");
    return;
  }
  QUIC_DLOG(INFO) << ENDPOINT << "Received the FETCH_OK for request_id = "
                  << message.request_id << " " << track->full_track_name();
  UpstreamFetch* fetch = static_cast<UpstreamFetch*>(track);
  fetch->OnFetchResult(
      message.end_location, message.group_order, absl::OkStatus(),
      [=, session = session_]() { session->CancelFetch(message.request_id); });
}

void MoqtSession::ControlStream::OnFetchErrorMessage(
    const MoqtFetchError& message) {
  RemoteTrack* track = session_->RemoteTrackById(message.request_id);
  if (track == nullptr) {
    QUIC_DLOG(INFO) << ENDPOINT << "Received the FETCH_ERROR for "
                    << "request_id = " << message.request_id
                    << " but no track exists";
    // Subscription state might have been destroyed for internal reasons.
    return;
  }
  if (!track->is_fetch()) {
    session_->Error(MoqtError::kProtocolViolation,
                    "Received FETCH_ERROR for a SUBSCRIBE");
    return;
  }
  if (!track->ErrorIsAllowed()) {
    session_->Error(MoqtError::kProtocolViolation,
                    "Received FETCH_ERROR after FETCH_OK or objects");
    return;
  }
  QUIC_DLOG(INFO) << ENDPOINT << "Received the FETCH_ERROR for "
                  << "request_id = " << message.request_id << " ("
                  << track->full_track_name() << ")"
                  << ", error = " << static_cast<int>(message.error_code)
                  << " (" << message.error_reason << ")";
  UpstreamFetch* fetch = static_cast<UpstreamFetch*>(track);
  absl::Status status =
      RequestErrorCodeToStatus(message.error_code, message.error_reason);
  fetch->OnFetchResult(Location(0, 0), MoqtDeliveryOrder::kAscending, status,
                       nullptr);
  session_->upstream_by_id_.erase(message.request_id);
}

void MoqtSession::ControlStream::OnRequestsBlockedMessage(
    const MoqtRequestsBlocked& message) {
  // TODO(martinduke): Derive logic for granting more subscribes.
}

void MoqtSession::ControlStream::OnPublishMessage(const MoqtPublish& message) {
  if (!session_->ValidateRequestId(message.request_id)) {
    return;
  }
  MoqtPublishError publish_error = {
      .request_id = message.request_id,
      .error_code = RequestErrorCode::kNotSupported,
      .error_reason = "PUBLISH is not supported",
  };
  if (session_->sent_goaway_) {
    QUIC_DLOG(INFO) << ENDPOINT << "Received a PUBLISH after GOAWAY";
    publish_error.error_code = RequestErrorCode::kUnauthorized;
    publish_error.error_reason = "Received a PUBLISH after GOAWAY";
  }
  // TODO(martinduke): Process these messages.
  SendOrBufferMessage(session_->framer_.SerializePublishError(publish_error));
}

void MoqtSession::ControlStream::OnParsingError(MoqtError error_code,
                                                absl::string_view reason) {
  session_->Error(error_code, absl::StrCat("Parse error: ", reason));
}

void MoqtSession::ControlStream::SendOrBufferMessage(
    quiche::QuicheBuffer message, bool fin) {
  quiche::StreamWriteOptions options;
  options.set_send_fin(fin);
  // TODO: while we buffer unconditionally, we should still at some point tear
  // down the connection if we've buffered too many control messages; otherwise,
  // there is potential for memory exhaustion attacks.
  options.set_buffer_unconditionally(true);
  std::array write_vector = {quiche::QuicheMemSlice(std::move(message))};
  absl::Status success = stream_->Writev(absl::MakeSpan(write_vector), options);
  if (!success.ok()) {
    session_->Error(MoqtError::kInternalError,
                    "Failed to write a control message");
  }
}

void MoqtSession::IncomingDataStream::OnObjectMessage(const MoqtObject& message,
                                                      absl::string_view payload,
                                                      bool end_of_message) {
  QUICHE_DVLOG(1) << ENDPOINT << "Received OBJECT message on stream "
                  << stream_->GetStreamId() << " for track alias "
                  << message.track_alias << " with sequence "
                  << message.group_id << ":" << message.object_id
                  << " priority " << message.publisher_priority << " length "
                  << payload.size() << " length " << message.payload_length
                  << (end_of_message ? "F" : "");
  if (!index_.has_value()) {
    index_ = DataStreamIndex(message.group_id, message.subgroup_id);
  }
  if (!session_->parameters_.deliver_partial_objects) {
    if (!end_of_message) {  // Buffer partial object.
      if (partial_object_.empty()) {
        // Avoid redundant allocations by reserving the appropriate amount of
        // memory if known.
        partial_object_.reserve(message.payload_length);
      }
      absl::StrAppend(&partial_object_, payload);
      return;
    }
    if (!partial_object_.empty()) {  // Completes the object
      absl::StrAppend(&partial_object_, payload);
      payload = absl::string_view(partial_object_);
    }
  }
  if (!parser_.stream_type().has_value()) {
    QUICHE_BUG(quic_bug_object_with_no_stream_type)
        << "Object delivered without a stream type";
    return;
  }
  // Get a pointer to the upstream state.
  RemoteTrack* track = track_.GetIfAvailable();
  if (track == nullptr) {
    track = (parser_.stream_type()->IsFetch())
                // message.track_alias is actually a fetch ID for fetches.
                ? session_->RemoteTrackById(message.track_alias)
                : session_->RemoteTrackByAlias(message.track_alias);
    if (track == nullptr) {
      stream_->SendStopSending(kResetCodeCanceled);
      // Received object for nonexistent track.
      return;
    }
    track_ = track->weak_ptr();
  }
  if (!track->CheckDataStreamType(*parser_.stream_type())) {
    session_->Error(MoqtError::kProtocolViolation,
                    "Received object for a track with a different stream type");
    return;
  }
  if (!track->InWindow(Location(message.group_id, message.object_id))) {
    // This is not an error. It can be the result of a recent SUBSCRIBE_UPDATE.
    return;
  }
  if (!track->is_fetch()) {
    if (no_more_objects_) {
      // Already got a stream-ending object.
      session_->OnMalformedTrack(track);
      return;
    }
    if (end_of_message) {
      next_object_id_ = message.object_id + 1;
      if (message.object_status == MoqtObjectStatus::kEndOfTrack ||
          message.object_status == MoqtObjectStatus::kEndOfGroup) {
        no_more_objects_ = true;
      }
    }
    SubscribeRemoteTrack* subscribe = static_cast<SubscribeRemoteTrack*>(track);
    if (!subscribe->OnObject(/*is_datagram=*/false)) {
      session_->OnMalformedTrack(track);
      return;
    }
    if (subscribe->visitor() != nullptr) {
      PublishedObjectMetadata metadata;
      metadata.location = Location(message.group_id, message.object_id);
      metadata.subgroup = message.subgroup_id;
      metadata.extensions = message.extension_headers;
      metadata.status = message.object_status;
      metadata.publisher_priority = message.publisher_priority;
      metadata.forwarding_preference = MoqtForwardingPreference::kSubgroup;
      metadata.arrival_time = session_->callbacks_.clock->Now();
      subscribe->visitor()->OnObjectFragment(track->full_track_name(), metadata,
                                             payload, end_of_message);
    }
  } else {  // FETCH
    track->OnObjectOrOk();
    UpstreamFetch* fetch = static_cast<UpstreamFetch*>(track);
    if (!fetch->LocationIsValid(Location(message.group_id, message.object_id),
                                message.object_status, end_of_message)) {
      session_->OnMalformedTrack(track);
      return;
    }
    UpstreamFetch::UpstreamFetchTask* task = fetch->task();
    if (task == nullptr) {
      // The application killed the FETCH.
      stream_->SendStopSending(kResetCodeCanceled);
      return;
    }
    if (!task->HasObject()) {
      task->NewObject(message);
    }
    if (task->NeedsMorePayload() && !payload.empty()) {
      task->AppendPayloadToObject(payload);
    }
  }
  partial_object_.clear();
}

MoqtSession::IncomingDataStream::~IncomingDataStream() {
  QUICHE_DVLOG(1) << ENDPOINT << "Destroying incoming data stream "
                  << stream_->GetStreamId();
  if (!parser_.track_alias().has_value()) {
    QUIC_DVLOG(1) << ENDPOINT
                  << "Destroying incoming data stream before "
                     "learning track alias";
    return;
  }
  if (!track_.IsValid()) {
    return;
  }
  if (parser_.stream_type().has_value() && parser_.stream_type()->IsFetch()) {
    session_->upstream_by_id_.erase(*parser_.track_alias());
    return;
  }
  if (session_->is_closing_) {
    return;
  }
  // It's a subscribe.
  SubscribeRemoteTrack* subscribe =
      static_cast<SubscribeRemoteTrack*>(track_.GetIfAvailable());
  if (subscribe == nullptr) {
    return;
  }
  subscribe->OnStreamClosed(fin_received_, index_);
  session_->MaybeDestroySubscription(subscribe);
}

void MoqtSession::IncomingDataStream::MaybeReadOneObject() {
  if (!parser_.track_alias().has_value() ||
      !parser_.stream_type().has_value() || !parser_.stream_type()->IsFetch()) {
    QUICHE_BUG(quic_bug_read_one_object_parser_unexpected_state)
        << "Requesting object, parser in unexpected state";
  }
  RemoteTrack* track = session_->RemoteTrackById(*parser_.track_alias());
  if (track == nullptr || !track->is_fetch()) {
    QUICHE_BUG(quic_bug_read_one_object_track_unexpected_state)
        << "Requesting object, track in unexpected state";
    return;
  }
  UpstreamFetch* fetch = static_cast<UpstreamFetch*>(track);
  UpstreamFetch::UpstreamFetchTask* task = fetch->task();
  if (task == nullptr) {
    return;
  }
  if (task->HasObject() && !task->NeedsMorePayload()) {
    return;
  }
  parser_.ReadAtMostOneObject();
  // If it read an object, it called OnObjectMessage and may have altered the
  // task's object state.
  if (task->HasObject() && !task->NeedsMorePayload()) {
    task->NotifyNewObject();
  }
}

void MoqtSession::IncomingDataStream::OnCanRead() {
  if (!parser_.stream_type().has_value()) {
    parser_.ReadStreamType();
    if (!parser_.stream_type().has_value()) {
      return;
    }
  }
  if (parser_.stream_type()->IsPadding()) {
    (void)stream_->SkipBytes(stream_->ReadableBytes());
    return;
  }
  bool knew_track_alias = parser_.track_alias().has_value();
  if (parser_.stream_type()->IsSubgroup()) {
    parser_.ReadAllData();
  } else if (!knew_track_alias) {
    parser_.ReadTrackAlias();
  }
  if (!parser_.track_alias().has_value()) {
    return;
  }
  if (parser_.stream_type()->IsSubgroup()) {
    if (knew_track_alias) {
      return;
    }
    // This is a new stream for a subscribe. Notify the subscription.
    auto it = session_->subscribe_by_alias_.find(*parser_.track_alias());
    if (it == session_->subscribe_by_alias_.end()) {
      QUIC_DLOG(INFO) << ENDPOINT
                      << "Received object for a track with no SUBSCRIBE";
      // This is a not a session error because there might be an UNSUBSCRIBE or
      // SUBSCRIBE_OK (containing the track alias) in flight.
      stream_->SendStopSending(kResetCodeCanceled);
      return;
    }
    it->second->OnStreamOpened();
    return;
  }
  auto it = session_->upstream_by_id_.find(*parser_.track_alias());
  if (it == session_->upstream_by_id_.end()) {
    QUIC_DLOG(INFO) << ENDPOINT << "Received object for a track with no FETCH";
    // This is a not a session error because there might be an UNSUBSCRIBE in
    // flight.
    stream_->SendStopSending(kResetCodeCanceled);
    return;
  }
  if (it->second == nullptr) {
    QUICHE_BUG(quiche_bug_moqt_fetch_pointer_is_null)
        << "Fetch pointer is null";
    return;
  }
  UpstreamFetch* fetch = static_cast<UpstreamFetch*>(it->second.get());
  if (!knew_track_alias) {
    // If the task already exists (FETCH_OK has arrived), the callback will
    // immediately execute to read the first object. Otherwise, it will only
    // execute when the task is created or a cached object is read.
    fetch->OnStreamOpened([this]() { MaybeReadOneObject(); });
    return;
  }
  MaybeReadOneObject();
}

void MoqtSession::IncomingDataStream::OnControlMessageReceived() {
  session_->Error(MoqtError::kProtocolViolation,
                  "Received a control message on a data stream");
}

void MoqtSession::IncomingDataStream::OnParsingError(MoqtError error_code,
                                                     absl::string_view reason) {
  session_->Error(error_code, absl::StrCat("Parse error: ", reason));
}

MoqtSession::PublishedSubscription::PublishedSubscription(
    MoqtSession* session, std::shared_ptr<MoqtTrackPublisher> track_publisher,
    const MoqtSubscribe& subscribe, uint64_t track_alias,
    MoqtPublishingMonitorInterface* monitoring_interface)
    : session_(session),
      track_publisher_(track_publisher),
      request_id_(subscribe.request_id),
      track_alias_(track_alias),
      filter_type_(subscribe.filter_type),
      forward_(subscribe.forward),
      window_(SubscribeMessageToWindow(subscribe)),
      subscriber_priority_(subscribe.subscriber_priority),
      subscriber_delivery_order_(subscribe.group_order),
      monitoring_interface_(monitoring_interface) {
  if (monitoring_interface_ != nullptr) {
    monitoring_interface_->OnObjectAckSupportKnown(
        subscribe.parameters.oack_window_size);
  }
  QUIC_DLOG(INFO) << ENDPOINT << "Created subscription for "
                  << subscribe.full_track_name;
  session_->subscribed_track_names_.insert(subscribe.full_track_name);
}

MoqtSession::PublishedSubscription::~PublishedSubscription() {
  session_->subscribed_track_names_.erase(track_publisher_->GetTrackName());
  track_publisher_->RemoveObjectListener(this);
}

SendStreamMap& MoqtSession::PublishedSubscription::stream_map() {
  // The stream map is lazily initialized, since initializing it requires
  // knowing the forwarding preference in advance, and it might not be known
  // when the subscription is first created.
  if (!lazily_initialized_stream_map_.has_value()) {
    QUICHE_DCHECK(track_publisher_->largest_location().has_value());
    lazily_initialized_stream_map_.emplace();
  }
  return *lazily_initialized_stream_map_;
}

void MoqtSession::PublishedSubscription::Update(
    Location start, std::optional<uint64_t> end_group,
    MoqtPriority subscriber_priority) {
  subscriber_priority_ = subscriber_priority;
  if (!window_.has_value()) {
    window_ = SubscribeWindow(start, end_group);
    return;
  }
  window_->TruncateStart(start);
  if (end_group.has_value()) {
    window_->TruncateEnd(*end_group);
  }
  // TODO: update priority of all data streams that are currently open.
  // TODO: update delivery timeout.
  // TODO: update forward and subscribe filter.

  // TODO: reset streams that are no longer in-window.
  // TODO: send PUBLISH_DONE if required.
  // TODO: send an error for invalid updates now that it's a part of draft-05.
}

void MoqtSession::PublishedSubscription::set_subscriber_priority(
    MoqtPriority priority) {
  if (priority == subscriber_priority_) {
    return;
  }
  if (queued_outgoing_data_streams_.empty()) {
    subscriber_priority_ = priority;
    return;
  }
  webtransport::SendOrder old_send_order =
      FinalizeSendOrder(queued_outgoing_data_streams_.rbegin()->first);
  subscriber_priority_ = priority;
  session_->UpdateQueuedSendOrder(request_id_, old_send_order,
                                  FinalizeSendOrder(old_send_order));
};

void MoqtSession::PublishedSubscription::OnSubscribeAccepted() {
  std::optional<Location> largest_location;
  ControlStream* stream = session_->GetControlStream();
  largest_location = track_publisher_->largest_location();
  if (largest_location.has_value()) {
    if (forward_) {
      switch (filter_type_) {
        case MoqtFilterType::kLatestObject:
          window_ = SubscribeWindow(largest_location->Next());
          break;
        case MoqtFilterType::kNextGroupStart:
          window_ = SubscribeWindow(Location(largest_location->group + 1, 0));
          break;
        default:
          break;
      }
    }
  } else if (filter_type_ == MoqtFilterType::kLatestObject ||
             filter_type_ == MoqtFilterType::kNextGroupStart) {
    // No data yet. All objects will be in-window.
    window_ = SubscribeWindow(Location(0, 0));
  }
  MoqtSubscribeOk subscribe_ok;
  subscribe_ok.request_id = request_id_;
  subscribe_ok.track_alias = track_alias_;
  QUICHE_BUG_IF(quic_bug_subscribe_ok_no_expiration,
                !track_publisher_->expiration().has_value())
      << "Request accepted without expiration";
  subscribe_ok.expires =
      track_publisher_->expiration().value_or(quic::QuicTimeDelta::Zero());
  QUICHE_BUG_IF(quic_bug_subscribe_ok_no_delivery_order,
                !track_publisher_->delivery_order().has_value())
      << "Request accepted without delivery order";
  subscribe_ok.group_order = track_publisher_->delivery_order().value_or(
      MoqtDeliveryOrder::kAscending);
  subscribe_ok.largest_location = largest_location;
  // TODO(martinduke): Support sending DELIVERY_TIMEOUT parameter as the
  // publisher.
  stream->SendOrBufferMessage(
      session_->framer_.SerializeSubscribeOk(subscribe_ok));
  // TODO(martinduke): If we buffer objects that arrived previously, the arrival
  // of the track alias disambiguates what subscription they belong to. Send
  // them.
}

void MoqtSession::PublishedSubscription::OnSubscribeRejected(
    MoqtSubscribeErrorReason reason) {
  session_->GetControlStream()->SendSubscribeError(
      request_id_, reason.error_code, reason.reason_phrase);
  session_->published_subscriptions_.erase(request_id_);
  // No class access below this line!
}

void MoqtSession::PublishedSubscription::OnNewObjectAvailable(
    Location location, uint64_t subgroup, MoqtPriority publisher_priority,
    MoqtForwardingPreference forwarding_preference) {
  if (!InWindow(location)) {
    return;
  }

  if (monitoring_interface_ != nullptr) {
    // Notify the monitoring interface about all newly published normal objects.
    // Objects with other statuses are not guaranteed to be acknowledged, thus
    // passing them into the monitoring interface can lead to confusion.
    std::optional<PublishedObject> object = track_publisher_->GetCachedObject(
        location.group, subgroup, location.object);
    QUICHE_DCHECK(object.has_value())
        << "Object " << absl::StrCat(location) << " on track "
        << track_publisher_->GetTrackName().ToString()
        << " does not exist, despite OnNewObjectAvailable being called";
    if (object.has_value() && object->metadata.location == location &&
        object->metadata.status == MoqtObjectStatus::kNormal) {
      monitoring_interface_->OnNewObjectEnqueued(location);
    }
  }

  session_->trace_recorder_.RecordNewObjectAvaliable(
      track_alias_, *track_publisher_, location, subgroup, publisher_priority);

  DataStreamIndex index(location.group, subgroup);
  if (reset_subgroups_.contains(index)) {
    // This subgroup has already been reset, ignore.
    return;
  }
  if (session_->alternate_delivery_timeout_ &&
      !delivery_timeout_.IsInfinite() && largest_sent_.has_value() &&
      location.group >= largest_sent_->group) {
    // Start the delivery timeout timer on all previous groups.
    for (uint64_t group = first_active_group_; group < location.group;
         ++group) {
      for (webtransport::StreamId stream_id :
           stream_map().GetStreamsForGroup(group)) {
        webtransport::Stream* raw_stream =
            session_->session_->GetStreamById(stream_id);
        if (raw_stream == nullptr) {
          continue;
        }
        OutgoingDataStream* stream =
            static_cast<OutgoingDataStream*>(raw_stream->visitor());
        stream->CreateAndSetAlarm(session_->callbacks_.clock->ApproximateNow() +
                                  delivery_timeout_);
      }
    }
  }
  QUICHE_DCHECK_GE(location.group, first_active_group_);

  if (forwarding_preference == MoqtForwardingPreference::kDatagram) {
    SendDatagram(location);
    return;
  }

  std::optional<webtransport::StreamId> stream_id =
      stream_map().GetStreamFor(index);
  webtransport::Stream* raw_stream = nullptr;
  if (stream_id.has_value()) {
    raw_stream = session_->session_->GetStreamById(*stream_id);
  } else {
    raw_stream = session_->OpenOrQueueDataStream(
        request_id_, NewStreamParameters(location.group, subgroup,
                                         location.object, publisher_priority));
  }
  if (raw_stream == nullptr) {
    return;
  }

  OutgoingDataStream* stream =
      static_cast<OutgoingDataStream*>(raw_stream->visitor());
  stream->SendObjects(*this);
}

void MoqtSession::PublishedSubscription::OnTrackPublisherGone() {
  session_->PublishIsDone(request_id_, PublishDoneCode::kGoingAway,
                          "Publisher is gone");
}

// TODO(martinduke): Revise to check if the last object has been delivered.
void MoqtSession::PublishedSubscription::OnNewFinAvailable(Location location,
                                                           uint64_t subgroup) {
  if (!GroupInWindow(location.group)) {
    return;
  }
  DataStreamIndex index(location.group, subgroup);
  if (reset_subgroups_.contains(index)) {
    // This subgroup has already been reset, ignore.
    return;
  }
  QUICHE_DCHECK_GE(location.group, first_active_group_);
  std::optional<webtransport::StreamId> stream_id =
      stream_map().GetStreamFor(index);
  if (!stream_id.has_value()) {
    return;
  }
  webtransport::Stream* raw_stream =
      session_->session_->GetStreamById(*stream_id);
  if (raw_stream == nullptr) {
    return;
  }
  OutgoingDataStream* stream =
      static_cast<OutgoingDataStream*>(raw_stream->visitor());
  stream->Fin(location);
}

void MoqtSession::PublishedSubscription::OnSubgroupAbandoned(
    uint64_t group, uint64_t subgroup,
    webtransport::StreamErrorCode error_code) {
  if (session_->is_closing_) {
    return;
  }
  if (!GroupInWindow(group)) {
    return;
  }
  DataStreamIndex index(group, subgroup);
  if (reset_subgroups_.contains(index)) {
    // This subgroup has already been reset, ignore.
    return;
  }
  QUICHE_DCHECK_GE(group, first_active_group_);
  std::optional<webtransport::StreamId> stream_id =
      stream_map().GetStreamFor(index);
  if (!stream_id.has_value()) {
    return;
  }
  webtransport::Stream* raw_stream =
      session_->session_->GetStreamById(*stream_id);
  if (raw_stream == nullptr) {
    return;
  }
  raw_stream->ResetWithUserCode(error_code);
}

void MoqtSession::PublishedSubscription::OnGroupAbandoned(uint64_t group_id) {
  if (session_->is_closing_) {
    return;
  }
  if (!window_.has_value() || window_->end().group < group_id ||
      window_->start().group > group_id) {
    // The group is not in the window, ignore.
    return;
  }
  std::vector<webtransport::StreamId> streams =
      stream_map().GetStreamsForGroup(group_id);
  if (delivery_timeout_.IsInfinite() && largest_sent_.has_value() &&
      largest_sent_->group <= group_id) {
    session_->PublishIsDone(request_id_, PublishDoneCode::kTooFarBehind, "");
    // No class access below this line!
    return;
  }
  for (webtransport::StreamId stream_id : streams) {
    webtransport::Stream* raw_stream =
        session_->session_->GetStreamById(stream_id);
    if (raw_stream == nullptr) {
      continue;
    }
    raw_stream->ResetWithUserCode(kResetCodeDeliveryTimeout);
    // Sending the Reset will call the destructor for OutgoingDataStream, which
    // will erase it from the SendStreamMap.
  }
  first_active_group_ = std::max(first_active_group_, group_id + 1);
  absl::erase_if(reset_subgroups_, [&](const DataStreamIndex& index) {
    return index.group < first_active_group_;
  });
}

std::vector<webtransport::StreamId>
MoqtSession::PublishedSubscription::GetAllStreams() const {
  if (!lazily_initialized_stream_map_.has_value()) {
    return {};
  }
  return lazily_initialized_stream_map_->GetAllStreams();
}

webtransport::SendOrder MoqtSession::PublishedSubscription::GetSendOrder(
    Location sequence, uint64_t subgroup,
    MoqtPriority publisher_priority) const {
  MoqtForwardingPreference forwarding_preference =
      track_publisher_->forwarding_preference().value_or(
          MoqtForwardingPreference::kSubgroup);
  QUICHE_BUG_IF(GetSendOrder_no_delivery_order,
                !track_publisher_->delivery_order().has_value())
      << "No delivery order";
  MoqtDeliveryOrder delivery_order =
      track_publisher_->delivery_order().value_or(
          MoqtDeliveryOrder::kAscending);
  if (forwarding_preference == MoqtForwardingPreference::kDatagram) {
    return SendOrderForDatagram(subscriber_priority_, publisher_priority,
                                sequence.group, sequence.object,
                                delivery_order);
  }
  return SendOrderForStream(subscriber_priority_, publisher_priority,
                            sequence.group, subgroup, delivery_order);
}

// Returns the highest send order in the subscription.
void MoqtSession::PublishedSubscription::AddQueuedOutgoingDataStream(
    const NewStreamParameters& parameters) {
  std::optional<webtransport::SendOrder> start_send_order =
      queued_outgoing_data_streams_.empty()
          ? std::optional<webtransport::SendOrder>()
          : queued_outgoing_data_streams_.rbegin()->first;
  webtransport::SendOrder send_order =
      GetSendOrder(Location(parameters.index.group, parameters.first_object),
                   parameters.index.subgroup, parameters.publisher_priority);
  // Zero out the subscriber priority bits, since these will be added when
  // updating the session.
  queued_outgoing_data_streams_.emplace(
      UpdateSendOrderForSubscriberPriority(send_order, 0), parameters);
  if (!start_send_order.has_value()) {
    session_->UpdateQueuedSendOrder(request_id_, std::nullopt, send_order);
  } else if (*start_send_order < send_order) {
    session_->UpdateQueuedSendOrder(
        request_id_, FinalizeSendOrder(*start_send_order), send_order);
  }
}

MoqtSession::NewStreamParameters
MoqtSession::PublishedSubscription::NextQueuedOutgoingDataStream() {
  QUICHE_DCHECK(!queued_outgoing_data_streams_.empty());
  if (queued_outgoing_data_streams_.empty()) {
    QUICHE_BUG(NextQueuedOutgoingDataStream_no_stream)
        << "NextQueuedOutgoingDataStream called when there are no streams "
           "pending.";
    return NewStreamParameters(0, 0, 0, 0);
  }
  auto it = queued_outgoing_data_streams_.rbegin();
  webtransport::SendOrder old_send_order = FinalizeSendOrder(it->first);
  NewStreamParameters first_stream = it->second;
  // converting a reverse iterator to an iterator involves incrementing it and
  // then taking base().
  queued_outgoing_data_streams_.erase((++it).base());
  if (queued_outgoing_data_streams_.empty()) {
    session_->UpdateQueuedSendOrder(request_id_, old_send_order, std::nullopt);
  } else {
    webtransport::SendOrder new_send_order =
        FinalizeSendOrder(queued_outgoing_data_streams_.rbegin()->first);
    if (old_send_order != new_send_order) {
      session_->UpdateQueuedSendOrder(request_id_, old_send_order,
                                      new_send_order);
    }
  }
  return first_stream;
}

void MoqtSession::PublishedSubscription::OnDataStreamCreated(
    webtransport::StreamId id, DataStreamIndex start_sequence) {
  ++streams_opened_;
  stream_map().AddStream(start_sequence, id);
}
void MoqtSession::PublishedSubscription::OnDataStreamDestroyed(
    webtransport::StreamId id, DataStreamIndex end_sequence) {
  stream_map().RemoveStream(end_sequence);
}

void MoqtSession::PublishedSubscription::OnObjectSent(Location sequence) {
  if (largest_sent_.has_value()) {
    largest_sent_ = std::max(*largest_sent_, sequence);
  } else {
    largest_sent_ = sequence;
  }
  // TODO: send PUBLISH_DONE if the subscription is done.
}

MoqtSession::OutgoingDataStream::OutgoingDataStream(
    MoqtSession* session, webtransport::Stream* stream,
    PublishedSubscription& subscription, const NewStreamParameters& parameters)
    : session_(session),
      stream_(stream),
      subscription_id_(subscription.request_id()),
      index_(parameters.index),
      publisher_priority_(parameters.publisher_priority),
      // Always include extension header length, because it's difficult to know
      // a priori if they're going to appear on a stream.
      stream_type_(MoqtDataStreamType::Subgroup(
          index_.subgroup, parameters.first_object, false, false)),
      next_object_(parameters.first_object),
      session_liveness_(session->liveness_token_) {
  UpdateSendOrder(subscription);
  session->trace_recorder_.RecordSubgroupStreamCreated(
      stream->GetStreamId(), subscription.track_alias(), parameters.index);
}

MoqtSession::OutgoingDataStream::~OutgoingDataStream() {
  // Though it might seem intuitive that the session object has to outlive the
  // connection object (and this is indeed how something like QuicSession and
  // QuicStream works), this is not the true for WebTransport visitors: the
  // session getting destroyed will inevitably lead to all related streams being
  // destroyed, but the actual order of destruction is not guaranteed.  Thus, we
  // need to check if the session still exists while accessing it in a stream
  // destructor.
  if (session_liveness_.expired()) {
    return;
  }
  if (delivery_timeout_alarm_ != nullptr) {
    delivery_timeout_alarm_->PermanentCancel();
  }
  auto it = session_->published_subscriptions_.find(subscription_id_);
  if (it != session_->published_subscriptions_.end()) {
    it->second->OnDataStreamDestroyed(stream_->GetStreamId(), index_);
  }
}

void MoqtSession::OutgoingDataStream::OnCanWrite() {
  PublishedSubscription* subscription = GetSubscriptionIfValid();
  if (subscription == nullptr) {
    return;
  }
  SendObjects(*subscription);
}

void MoqtSession::OutgoingDataStream::DeliveryTimeoutDelegate::OnAlarm() {
  auto it = stream_->session_->published_subscriptions_.find(
      stream_->subscription_id_);
  if (it != stream_->session_->published_subscriptions_.end()) {
    it->second->OnStreamTimeout(stream_->index());
  }
  stream_->stream_->ResetWithUserCode(kResetCodeDeliveryTimeout);
}

MoqtSession::PublishedSubscription*
MoqtSession::OutgoingDataStream::GetSubscriptionIfValid() {
  auto it = session_->published_subscriptions_.find(subscription_id_);
  if (it == session_->published_subscriptions_.end()) {
    stream_->ResetWithUserCode(kResetCodeCanceled);
    return nullptr;
  }

  PublishedSubscription* subscription = it->second.get();
  if (!subscription->publisher().largest_location().has_value()) {
    QUICHE_BUG(GetSubscriptionIfValid_InvalidTrackStatusOk)
        << "The track publisher returned a status indicating that no objects "
           "are available, but a stream for those objects exists.";
    session_->Error(MoqtError::kInternalError,
                    "Invalid track state provided by application");
    return nullptr;
  }
  return subscription;
}

void MoqtSession::OutgoingDataStream::SendObjects(
    PublishedSubscription& subscription) {
  while (stream_->CanWrite()) {
    std::optional<PublishedObject> object =
        subscription.publisher().GetCachedObject(index_.group, index_.subgroup,
                                                 next_object_);
    if (!object.has_value()) {
      break;
    }

    QUICHE_DCHECK_EQ(object->metadata.location.group, index_.group);
    QUICHE_DCHECK(object->metadata.subgroup == index_.subgroup);
    if (!subscription.InWindow(object->metadata.location)) {
      // It is possible that the next object became irrelevant due to a
      // SUBSCRIBE_UPDATE.  Close the stream if so.
      bool success = stream_->SendFin();
      QUICHE_BUG_IF(OutgoingDataStream_fin_due_to_update, !success)
          << "Writing FIN failed despite CanWrite() being true.";
      return;
    }

    quic::QuicTimeDelta delivery_timeout = subscription.delivery_timeout();
    if (!session_->alternate_delivery_timeout_ &&
        session_->callbacks_.clock->ApproximateNow() -
                object->metadata.arrival_time >
            delivery_timeout) {
      subscription.OnStreamTimeout(index_);
      stream_->ResetWithUserCode(kResetCodeDeliveryTimeout);
      return;
    }
    if (!session_->WriteObjectToStream(
            stream_, subscription.track_alias(), object->metadata,
            std::move(object->payload), stream_type_, last_object_id_,
            object->fin_after_this)) {
      // WriteObjectToStream() closes the connection on error, meaning that
      // there is no need to process the stream any further.
      return;
    }
    last_object_id_ = object->metadata.location.object;
    next_object_ = *last_object_id_ + 1;
    subscription.OnObjectSent(object->metadata.location);

    if (object->fin_after_this && !delivery_timeout.IsInfinite() &&
        !session_->alternate_delivery_timeout_) {
      CreateAndSetAlarm(object->metadata.arrival_time + delivery_timeout);
    }
  }
}

void MoqtSession::OutgoingDataStream::Fin(Location last_object) {
  QUICHE_DCHECK_EQ(last_object.group, index_.group);
  if (next_object_ <= last_object.object) {
    // There is still data to send, do nothing.
    return;
  }
  // All data has already been sent; send a pure FIN.
  bool success = stream_->SendFin();
  QUICHE_BUG_IF(OutgoingDataStream_fin_failed, !success)
      << "Writing pure FIN failed.";
  auto it = session_->published_subscriptions_.find(subscription_id_);
  if (it == session_->published_subscriptions_.end()) {
    return;
  }
  quic::QuicTimeDelta delivery_timeout = it->second->delivery_timeout();
  if (!delivery_timeout.IsInfinite()) {
    CreateAndSetAlarm(session_->callbacks_.clock->ApproximateNow() +
                      delivery_timeout);
  }
}

bool MoqtSession::WriteObjectToStream(webtransport::Stream* stream, uint64_t id,
                                      const PublishedObjectMetadata& metadata,
                                      quiche::QuicheMemSlice payload,
                                      MoqtDataStreamType type,
                                      std::optional<uint64_t> last_id,
                                      bool fin) {
  QUICHE_DCHECK(stream->CanWrite());
  MoqtObject header;
  header.track_alias = id;
  header.group_id = metadata.location.group;
  header.subgroup_id = metadata.subgroup;
  header.object_id = metadata.location.object;
  header.publisher_priority = metadata.publisher_priority;
  header.extension_headers = metadata.extensions;
  header.object_status = metadata.status;
  header.payload_length = payload.length();

  quiche::QuicheBuffer serialized_header =
      framer_.SerializeObjectHeader(header, type, last_id);
  // TODO(vasilvv): add a version of WebTransport write API that accepts
  // memslices so that we can avoid a copy here.
  std::array write_vector = {
      quiche::QuicheMemSlice(std::move(serialized_header)), std::move(payload)};
  quiche::StreamWriteOptions options;
  options.set_send_fin(fin);
  absl::Status write_status =
      stream->Writev(absl::MakeSpan(write_vector), options);
  if (!write_status.ok()) {
    QUICHE_BUG(MoqtSession_WriteObjectToStream_write_failed)
        << "Writing into MoQT stream failed despite CanWrite() being true "
           "before; status: "
        << write_status;
    Error(MoqtError::kInternalError, "Data stream write error");
    return false;
  }

  QUIC_DVLOG(1) << "Stream " << stream->GetStreamId() << " successfully wrote "
                << metadata.location << ", fin = " << fin;
  return true;
}

void MoqtSession::OnMalformedTrack(RemoteTrack* track) {
  if (!track->is_fetch()) {
    static_cast<SubscribeRemoteTrack*>(track)->visitor()->OnMalformedTrack(
        track->full_track_name());
    Unsubscribe(track->full_track_name());
    return;
  }
  UpstreamFetch::UpstreamFetchTask* task =
      static_cast<UpstreamFetch*>(track)->task();
  if (task != nullptr) {
    task->OnStreamAndFetchClosed(kResetCodeMalformedTrack,
                                 "Malformed track received");
  }
  CancelFetch(track->request_id());
}

void MoqtSession::CleanUpState() {
  if (is_closing_) {
    return;
  }
  is_closing_ = true;
  if (goaway_timeout_alarm_ != nullptr) {
    goaway_timeout_alarm_->PermanentCancel();
  }
  // When the session closes, report to the application implied receipt of
  // UNSUBSCRIBE_NAMESPACE, PUBLISH_NAMESPACE_DONE, PUBLISH_NAMESPACE_CANCEL,
  // PUBLISH_DONE, and UNSUBSCRIBE.
  for (const TrackNamespace& track_namespace :
       incoming_subscribe_namespace_.GetSubscribedNamespaces()) {
    callbacks_.incoming_subscribe_namespace_callback(track_namespace,
                                                     std::nullopt, nullptr);
  }
  published_subscriptions_.clear();
  for (const TrackNamespace& track_namespace : incoming_publish_namespaces_) {
    callbacks_.incoming_publish_namespace_callback(track_namespace,
                                                   std::nullopt, nullptr);
  }
  for (auto& [track_namespace, callback] : outgoing_publish_namespaces_) {
    callback(track_namespace, MoqtRequestError{RequestErrorCode::kUninterested,
                                               "Session closed"});
  }
  while (!upstream_by_id_.empty()) {
    auto upstream = upstream_by_id_.begin();
    if (upstream->second->is_fetch()) {
      upstream_by_id_.erase(upstream);
      continue;
    }
    DestroySubscription(
        static_cast<SubscribeRemoteTrack*>(upstream->second.get()));
  }
}

void MoqtSession::CancelFetch(uint64_t request_id) {
  if (is_closing_) {
    return;
  }
  // This is only called from the callback where UpstreamFetchTask has been
  // destroyed, so there is no need to notify the application.
  upstream_by_id_.erase(request_id);
  ControlStream* stream = GetControlStream();
  if (stream == nullptr) {
    return;
  }
  MoqtFetchCancel message;
  message.request_id = request_id;
  stream->SendOrBufferMessage(framer_.SerializeFetchCancel(message));
  // The FETCH_CANCEL will cause a RESET_STREAM to return, which would be the
  // same as a STOP_SENDING. However, a FETCH_CANCEL works even if the stream
  // hasn't opened yet.
}

void MoqtSession::PublishedSubscription::SendDatagram(Location sequence) {
  std::optional<PublishedObject> object = track_publisher_->GetCachedObject(
      sequence.group, sequence.object, sequence.object);
  if (!object.has_value()) {
    QUICHE_BUG(PublishedSubscription_SendDatagram_object_not_in_cache)
        << "Got notification about an object that is not in the cache";
    return;
  }
  MoqtObject header;
  header.track_alias = track_alias_;
  header.group_id = object->metadata.location.group;
  header.object_id = object->metadata.location.object;
  header.publisher_priority = object->metadata.publisher_priority;
  header.extension_headers = object->metadata.extensions;
  header.object_status = object->metadata.status;
  header.subgroup_id = header.object_id;
  header.payload_length = object->payload.length();
  quiche::QuicheBuffer datagram = session_->framer_.SerializeObjectDatagram(
      header, object->payload.AsStringView());
  session_->session_->SendOrQueueDatagram(datagram.AsStringView());
  OnObjectSent(object->metadata.location);
}

void MoqtSession::OutgoingDataStream::UpdateSendOrder(
    PublishedSubscription& subscription) {
  stream_->SetPriority(webtransport::StreamPriority{
      /*send_group_id=*/kMoqtSendGroupId,
      subscription.GetSendOrder(Location(index_.group, next_object_),
                                index_.subgroup, publisher_priority_)});
}

void MoqtSession::OutgoingDataStream::CreateAndSetAlarm(
    quic::QuicTime deadline) {
  if (delivery_timeout_alarm_ != nullptr) {
    return;
  }
  delivery_timeout_alarm_ = absl::WrapUnique(
      session_->alarm_factory_->CreateAlarm(new DeliveryTimeoutDelegate(this)));
  delivery_timeout_alarm_->Set(deadline);
}

MoqtSession::PublishedFetch::FetchStreamVisitor::FetchStreamVisitor(
    std::shared_ptr<PublishedFetch> fetch, webtransport::Stream* stream)
    : fetch_(fetch), stream_(stream) {
  fetch->fetch_task()->SetObjectAvailableCallback(
      [this]() { this->OnCanWrite(); });
  fetch->session()->trace_recorder_.RecordFetchStreamCreated(
      stream->GetStreamId());
}

void MoqtSession::PublishedSubscription::ProcessObjectAck(
    const MoqtObjectAck& message) {
  session_->trace_recorder_.RecordObjectAck(
      track_alias_, Location(message.group_id, message.object_id),
      message.delta_from_deadline);

  if (monitoring_interface_ == nullptr) {
    return;
  }
  monitoring_interface_->OnObjectAckReceived(
      Location(message.group_id, message.object_id),
      message.delta_from_deadline);
}

}  // namespace moqt