/*
 *  Copyright (c) 2022 The WebRTC project authors. All Rights Reserved.
 *
 *  Use of this source code is governed by a BSD-style license
 *  that can be found in the LICENSE file in the root of the source
 *  tree. An additional intellectual property rights grant can be found
 *  in the file PATENTS.  All contributing project authors may
 *  be found in the AUTHORS file in the root of the source tree.
 */

#include "video/frame_buffer_proxy.h"

#include <algorithm>
#include <memory>
#include <utility>

#include "absl/functional/bind_front.h"
#include "api/sequence_checker.h"
#include "modules/video_coding/frame_buffer2.h"
#include "modules/video_coding/frame_buffer3.h"
#include "modules/video_coding/frame_helpers.h"
#include "rtc_base/logging.h"
#include "rtc_base/thread_annotations.h"
#include "system_wrappers/include/field_trial.h"
#include "video/frame_decode_scheduler.h"
#include "video/frame_decode_timing.h"
#include "video/video_receive_stream_timeout_tracker.h"

namespace webrtc {

class FrameBuffer2Proxy : public FrameBufferProxy {
 public:
  FrameBuffer2Proxy(Clock* clock,
                    VCMTiming* timing,
                    VCMReceiveStatisticsCallback* stats_proxy,
                    rtc::TaskQueue* decode_queue,
                    FrameSchedulingReceiver* receiver,
                    TimeDelta max_wait_for_keyframe,
                    TimeDelta max_wait_for_frame)
      : max_wait_for_keyframe_(max_wait_for_keyframe),
        max_wait_for_frame_(max_wait_for_frame),
        frame_buffer_(clock, timing, stats_proxy),
        decode_queue_(decode_queue),
        stats_proxy_(stats_proxy),
        receiver_(receiver) {
    RTC_DCHECK(decode_queue_);
    RTC_DCHECK(stats_proxy_);
    RTC_DCHECK(receiver_);
  }

  void StopOnWorker() override {
    RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
    decode_queue_->PostTask([this] {
      frame_buffer_.Stop();
      decode_safety_->SetNotAlive();
    });
  }

  void SetProtectionMode(VCMVideoProtection protection_mode) override {
    RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
    frame_buffer_.SetProtectionMode(kProtectionNackFEC);
  }

  void Clear() override {
    RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
    frame_buffer_.Clear();
  }

  absl::optional<int64_t> InsertFrame(
      std::unique_ptr<EncodedFrame> frame) override {
    RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
    int64_t last_continuous_pid = frame_buffer_.InsertFrame(std::move(frame));
    if (last_continuous_pid != -1)
      return last_continuous_pid;
    return absl::nullopt;
  }

  void UpdateRtt(int64_t max_rtt_ms) override {
    RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
    frame_buffer_.UpdateRtt(max_rtt_ms);
  }

  void StartNextDecode(bool keyframe_required) override {
    if (!decode_queue_->IsCurrent()) {
      decode_queue_->PostTask(ToQueuedTask(
          decode_safety_,
          [this, keyframe_required] { StartNextDecode(keyframe_required); }));
      return;
    }
    RTC_DCHECK_RUN_ON(decode_queue_);

    frame_buffer_.NextFrame(
        MaxWait(keyframe_required).ms(), keyframe_required, decode_queue_,
        /* encoded frame handler */
        [this, keyframe_required](std::unique_ptr<EncodedFrame> frame) {
          RTC_DCHECK_RUN_ON(decode_queue_);
          if (!decode_safety_->alive())
            return;
          if (frame) {
            receiver_->OnEncodedFrame(std::move(frame));
          } else {
            receiver_->OnDecodableFrameTimeout(MaxWait(keyframe_required));
          }
        });
  }

  int Size() override {
    RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
    return frame_buffer_.Size();
  }

 private:
  TimeDelta MaxWait(bool keyframe_required) const {
    return keyframe_required ? max_wait_for_keyframe_ : max_wait_for_frame_;
  }

  RTC_NO_UNIQUE_ADDRESS SequenceChecker worker_sequence_checker_;
  const TimeDelta max_wait_for_keyframe_;
  const TimeDelta max_wait_for_frame_;
  video_coding::FrameBuffer frame_buffer_;
  rtc::TaskQueue* const decode_queue_;
  VCMReceiveStatisticsCallback* const stats_proxy_;
  FrameSchedulingReceiver* const receiver_;
  rtc::scoped_refptr<PendingTaskSafetyFlag> decode_safety_ =
      PendingTaskSafetyFlag::CreateDetached();
};

// Max number of frames the buffer will hold.
static constexpr size_t kMaxFramesBuffered = 800;
// Max number of decoded frame info that will be saved.
static constexpr int kMaxFramesHistory = 1 << 13;

// Default value for the maximum decode queue size that is used when the
// low-latency renderer is used.
static constexpr size_t kZeroPlayoutDelayDefaultMaxDecodeQueueSize = 8;

// Encapsulates use of the new frame buffer for use in VideoReceiveStream. This
// behaves the same as the FrameBuffer2Proxy but uses frame_buffer3 instead.
// Responsiblities from frame_buffer2, like stats, jitter and frame timing
// accounting are moved into this pro
class FrameBuffer3Proxy : public FrameBufferProxy {
 public:
  FrameBuffer3Proxy(Clock* clock,
                    TaskQueueBase* worker_queue,
                    VCMTiming* timing,
                    VCMReceiveStatisticsCallback* stats_proxy,
                    rtc::TaskQueue* decode_queue,
                    FrameSchedulingReceiver* receiver,
                    TimeDelta max_wait_for_keyframe,
                    TimeDelta max_wait_for_frame)
      : max_wait_for_keyframe_(max_wait_for_keyframe),
        max_wait_for_frame_(max_wait_for_frame),
        clock_(clock),
        worker_queue_(worker_queue),
        decode_queue_(decode_queue),
        stats_proxy_(stats_proxy),
        receiver_(receiver),
        timing_(timing),
        jitter_estimator_(clock_),
        inter_frame_delay_(clock_->TimeInMilliseconds()),
        buffer_(std::make_unique<FrameBuffer>(kMaxFramesBuffered,
                                              kMaxFramesHistory)),
        frame_decode_scheduler_(
            clock_,
            worker_queue,
            absl::bind_front(&FrameBuffer3Proxy::OnFrameReadyForExtraction,
                             this)),
        decode_timing_(clock_, timing_),
        timeout_tracker_(clock_,
                         worker_queue_,
                         VideoReceiveStreamTimeoutTracker::Timeouts{
                             .max_wait_for_keyframe = max_wait_for_keyframe,
                             .max_wait_for_frame = max_wait_for_frame},
                         absl::bind_front(&FrameBuffer3Proxy::OnTimeout, this)),
        zero_playout_delay_max_decode_queue_size_(
            "max_decode_queue_size",
            kZeroPlayoutDelayDefaultMaxDecodeQueueSize) {
    RTC_DCHECK(decode_queue_);
    RTC_DCHECK(stats_proxy_);
    RTC_DCHECK(receiver_);
    RTC_DCHECK(timing_);
    RTC_DCHECK(worker_queue_);
    RTC_DCHECK(clock_);
    RTC_LOG(LS_WARNING) << "Using FrameBuffer3";

    ParseFieldTrial({&zero_playout_delay_max_decode_queue_size_},
                    field_trial::FindFullName("WebRTC-ZeroPlayoutDelay"));
  }

  // FrameBufferProxy implementation.
  void StopOnWorker() override {
    RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
    timeout_tracker_.Stop();
    frame_decode_scheduler_.CancelOutstanding();
    decoder_ready_for_new_frame_ = false;
    decode_queue_->PostTask([this] {
      RTC_DCHECK_RUN_ON(decode_queue_);
      decode_safety_->SetNotAlive();
    });
  }

  void SetProtectionMode(VCMVideoProtection protection_mode) override {
    RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
    protection_mode_ = kProtectionNackFEC;
  }

  void Clear() override {
    RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
    stats_proxy_->OnDroppedFrames(buffer_->CurrentSize());
    buffer_ =
        std::make_unique<FrameBuffer>(kMaxFramesBuffered, kMaxFramesHistory);
    frame_decode_scheduler_.CancelOutstanding();
  }

  absl::optional<int64_t> InsertFrame(
      std::unique_ptr<EncodedFrame> frame) override {
    RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
    if (frame->is_last_spatial_layer)
      stats_proxy_->OnCompleteFrame(frame->is_keyframe(), frame->size(),
                                    frame->contentType());
    if (!frame->delayed_by_retransmission())
      timing_->IncomingTimestamp(frame->Timestamp(), frame->ReceivedTime());

    buffer_->InsertFrame(std::move(frame));
    MaybeScheduleFrameForRelease();

    return buffer_->LastContinuousFrameId();
  }

  void UpdateRtt(int64_t max_rtt_ms) override {
    RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
    jitter_estimator_.UpdateRtt(max_rtt_ms);
  }

  void StartNextDecode(bool keyframe_required) override {
    if (!worker_queue_->IsCurrent()) {
      worker_queue_->PostTask(ToQueuedTask(
          worker_safety_,
          [this, keyframe_required] { StartNextDecode(keyframe_required); }));
      return;
    }

    RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
    if (!timeout_tracker_.Running())
      timeout_tracker_.Start(keyframe_required);
    keyframe_required_ = keyframe_required;
    if (keyframe_required_) {
      timeout_tracker_.SetWaitingForKeyframe();
    }
    decoder_ready_for_new_frame_ = true;
    MaybeScheduleFrameForRelease();
  }

  int Size() override {
    RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
    return buffer_->CurrentSize();
  }

  void OnFrameReady(
      absl::InlinedVector<std::unique_ptr<EncodedFrame>, 4> frames,
      Timestamp render_time) {
    RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
    RTC_DCHECK(!frames.empty());

    timeout_tracker_.OnEncodedFrameReleased();

    int64_t now_ms = clock_->TimeInMilliseconds();
    bool superframe_delayed_by_retransmission = false;
    size_t superframe_size = 0;
    const EncodedFrame& first_frame = *frames.front();
    int64_t receive_time_ms = first_frame.ReceivedTime();

    if (first_frame.is_keyframe())
      keyframe_required_ = false;

    // Gracefully handle bad RTP timestamps and render time issues.
    if (FrameHasBadRenderTiming(render_time.ms(), now_ms,
                                timing_->TargetVideoDelay())) {
      jitter_estimator_.Reset();
      timing_->Reset();
      render_time = Timestamp::Millis(
          timing_->RenderTimeMs(first_frame.Timestamp(), now_ms));
    }

    for (std::unique_ptr<EncodedFrame>& frame : frames) {
      frame->SetRenderTime(render_time.ms());

      superframe_delayed_by_retransmission |=
          frame->delayed_by_retransmission();
      receive_time_ms = std::max(receive_time_ms, frame->ReceivedTime());
      superframe_size += frame->size();
    }

    if (!superframe_delayed_by_retransmission) {
      int64_t frame_delay;

      if (inter_frame_delay_.CalculateDelay(first_frame.Timestamp(),
                                            &frame_delay, receive_time_ms)) {
        jitter_estimator_.UpdateEstimate(frame_delay, superframe_size);
      }

      float rtt_mult = protection_mode_ == kProtectionNackFEC ? 0.0 : 1.0;
      absl::optional<float> rtt_mult_add_cap_ms = absl::nullopt;
      if (rtt_mult_settings_.has_value()) {
        rtt_mult = rtt_mult_settings_->rtt_mult_setting;
        rtt_mult_add_cap_ms = rtt_mult_settings_->rtt_mult_add_cap_ms;
      }
      timing_->SetJitterDelay(
          jitter_estimator_.GetJitterEstimate(rtt_mult, rtt_mult_add_cap_ms));
      timing_->UpdateCurrentDelay(render_time.ms(), now_ms);
    } else if (RttMultExperiment::RttMultEnabled()) {
      jitter_estimator_.FrameNacked();
    }

    // Update stats.
    UpdateDroppedFrames();
    UpdateJitterDelay();
    UpdateTimingFrameInfo();

    std::unique_ptr<EncodedFrame> frame =
        CombineAndDeleteFrames(std::move(frames));

    decoder_ready_for_new_frame_ = false;
    // VideoReceiveStream2 wants frames on the decoder thread.
    decode_queue_->PostTask(ToQueuedTask(
        decode_safety_, [this, frame = std::move(frame)]() mutable {
          receiver_->OnEncodedFrame(std::move(frame));
        }));
  }

  void OnTimeout() {
    RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
    // If the stream is paused then ignore the timeout.
    if (!decoder_ready_for_new_frame_) {
      timeout_tracker_.Stop();
      return;
    }
    receiver_->OnDecodableFrameTimeout(MaxWait());
    // Stop sending timeouts until receive starts waiting for a new frame.
    timeout_tracker_.Stop();
    decoder_ready_for_new_frame_ = false;
  }

 private:
  void OnFrameReadyForExtraction(uint32_t rtp_timestamp,
                                 Timestamp render_time) {
    RTC_DCHECK_RUN_ON(&worker_sequence_checker_);
    RTC_DCHECK(buffer_->NextDecodableTemporalUnitRtpTimestamp() ==
               rtp_timestamp)
        << "Frame buffer's next decodable frame was not the one sent for "
           "extraction rtp="
        << rtp_timestamp << " next="
        << buffer_->NextDecodableTemporalUnitRtpTimestamp().value_or(-1);
    auto frames = buffer_->ExtractNextDecodableTemporalUnit();
    OnFrameReady(std::move(frames), render_time);
  }

  TimeDelta MaxWait() const RTC_RUN_ON(&worker_sequence_checker_) {
    return keyframe_required_ ? max_wait_for_keyframe_ : max_wait_for_frame_;
  }

  void UpdateDroppedFrames() RTC_RUN_ON(&worker_sequence_checker_) {
    const int dropped_frames = buffer_->GetTotalNumberOfDroppedFrames() -
                               frames_dropped_before_last_new_frame_;
    if (dropped_frames > 0)
      stats_proxy_->OnDroppedFrames(dropped_frames);
    frames_dropped_before_last_new_frame_ =
        buffer_->GetTotalNumberOfDroppedFrames();
  }

  void UpdateJitterDelay() {
    int max_decode_ms;
    int current_delay_ms;
    int target_delay_ms;
    int jitter_buffer_ms;
    int min_playout_delay_ms;
    int render_delay_ms;
    if (timing_->GetTimings(&max_decode_ms, &current_delay_ms, &target_delay_ms,
                            &jitter_buffer_ms, &min_playout_delay_ms,
                            &render_delay_ms)) {
      stats_proxy_->OnFrameBufferTimingsUpdated(
          max_decode_ms, current_delay_ms, target_delay_ms, jitter_buffer_ms,
          min_playout_delay_ms, render_delay_ms);
    }
  }

  void UpdateTimingFrameInfo() {
    absl::optional<TimingFrameInfo> info = timing_->GetTimingFrameInfo();
    if (info)
      stats_proxy_->OnTimingFrameInfoUpdated(*info);
  }

  bool IsTooManyFramesQueued() const RTC_RUN_ON(&worker_sequence_checker_) {
    return buffer_->CurrentSize() > zero_playout_delay_max_decode_queue_size_;
  }

  void ForceKeyFrameReleaseImmediately() RTC_RUN_ON(&worker_sequence_checker_) {
    RTC_DCHECK(keyframe_required_);
    // Iterate through the frame buffer until there is a complete keyframe and
    // release this right away.
    while (buffer_->NextDecodableTemporalUnitRtpTimestamp()) {
      auto next_frame = buffer_->ExtractNextDecodableTemporalUnit();
      if (next_frame.empty()) {
        RTC_DCHECK_NOTREACHED()
            << "Frame buffer should always return at least 1 frame.";
        continue;
      }
      // Found keyframe - decode right away.
      if (next_frame.front()->is_keyframe()) {
        auto render_time = Timestamp::Millis(timing_->RenderTimeMs(
            next_frame.front()->Timestamp(), clock_->TimeInMilliseconds()));
        OnFrameReady(std::move(next_frame), render_time);
        return;
      }
    }
  }

  void MaybeScheduleFrameForRelease() RTC_RUN_ON(&worker_sequence_checker_) {
    if (!decoder_ready_for_new_frame_ ||
        !buffer_->NextDecodableTemporalUnitRtpTimestamp())
      return;

    if (keyframe_required_) {
      return ForceKeyFrameReleaseImmediately();
    }

    // TODO(https://bugs.webrtc.org/13343): Make [next,last] decodable returned
    // as an optional pair and remove this check.
    RTC_CHECK(buffer_->LastDecodableTemporalUnitRtpTimestamp());
    auto last_rtp = *buffer_->LastDecodableTemporalUnitRtpTimestamp();

    // If already scheduled then abort.
    if (frame_decode_scheduler_.scheduled_rtp() ==
        buffer_->NextDecodableTemporalUnitRtpTimestamp())
      return;

    absl::optional<FrameDecodeTiming::FrameSchedule> schedule;
    while (buffer_->NextDecodableTemporalUnitRtpTimestamp()) {
      auto next_rtp = *buffer_->NextDecodableTemporalUnitRtpTimestamp();
      schedule = decode_timing_.OnFrameBufferUpdated(next_rtp, last_rtp,
                                                     IsTooManyFramesQueued());
      if (schedule) {
        // Don't schedule if already waiting for the same frame.
        if (frame_decode_scheduler_.scheduled_rtp() != next_rtp) {
          frame_decode_scheduler_.CancelOutstanding();
          frame_decode_scheduler_.ScheduleFrame(next_rtp, *schedule);
        }
        return;
      }
      // If no schedule for current rtp, drop and try again.
      buffer_->DropNextDecodableTemporalUnit();
    }
  }

  RTC_NO_UNIQUE_ADDRESS SequenceChecker worker_sequence_checker_;
  const TimeDelta max_wait_for_keyframe_;
  const TimeDelta max_wait_for_frame_;
  const absl::optional<RttMultExperiment::Settings> rtt_mult_settings_ =
      RttMultExperiment::GetRttMultValue();
  Clock* const clock_;
  TaskQueueBase* const worker_queue_;
  rtc::TaskQueue* const decode_queue_;
  VCMReceiveStatisticsCallback* const stats_proxy_;
  FrameSchedulingReceiver* const receiver_;
  VCMTiming* const timing_;

  VCMJitterEstimator jitter_estimator_
      RTC_GUARDED_BY(&worker_sequence_checker_);
  VCMInterFrameDelay inter_frame_delay_
      RTC_GUARDED_BY(&worker_sequence_checker_);
  bool keyframe_required_ RTC_GUARDED_BY(&worker_sequence_checker_) = false;
  std::unique_ptr<FrameBuffer> buffer_
      RTC_GUARDED_BY(&worker_sequence_checker_);
  FrameDecodeScheduler frame_decode_scheduler_
      RTC_GUARDED_BY(&worker_sequence_checker_);
  FrameDecodeTiming decode_timing_ RTC_GUARDED_BY(&worker_sequence_checker_);
  VideoReceiveStreamTimeoutTracker timeout_tracker_
      RTC_GUARDED_BY(&worker_sequence_checker_);
  int frames_dropped_before_last_new_frame_
      RTC_GUARDED_BY(&worker_sequence_checker_) = 0;
  VCMVideoProtection protection_mode_
      RTC_GUARDED_BY(&worker_sequence_checker_) = kProtectionNack;

  // This flag guards frames from queuing in front of the decoder. Without this
  // guard, encoded frames will not wait for the decoder to finish decoding a
  // frame and just queue up, meaning frames will not be dropped or
  // fast-forwarded when the decoder is slow or hangs.
  bool decoder_ready_for_new_frame_ RTC_GUARDED_BY(&worker_sequence_checker_) =
      false;

  // Maximum number of frames in the decode queue to allow pacing. If the
  // queue grows beyond the max limit, pacing will be disabled and frames will
  // be pushed to the decoder as soon as possible. This only has an effect
  // when the low-latency rendering path is active, which is indicated by
  // the frame's render time == 0.
  FieldTrialParameter<unsigned> zero_playout_delay_max_decode_queue_size_;

  rtc::scoped_refptr<PendingTaskSafetyFlag> decode_safety_ =
      PendingTaskSafetyFlag::CreateDetached();
  ScopedTaskSafety worker_safety_;
};

std::unique_ptr<FrameBufferProxy> FrameBufferProxy::CreateFromFieldTrial(
    Clock* clock,
    TaskQueueBase* worker_queue,
    VCMTiming* timing,
    VCMReceiveStatisticsCallback* stats_proxy,
    rtc::TaskQueue* decode_queue,
    FrameSchedulingReceiver* receiver,
    TimeDelta max_wait_for_keyframe,
    TimeDelta max_wait_for_frame) {
  if (field_trial::IsEnabled("WebRTC-FrameBuffer3"))
    return std::make_unique<FrameBuffer3Proxy>(
        clock, worker_queue, timing, stats_proxy, decode_queue, receiver,
        max_wait_for_keyframe, max_wait_for_frame);
  return std::make_unique<FrameBuffer2Proxy>(
      clock, timing, stats_proxy, decode_queue, receiver, max_wait_for_keyframe,
      max_wait_for_frame);
}

}  // namespace webrtc