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

#include <cstddef>
#include <optional>
#include <tuple>
#ifdef UNSAFE_BUFFERS_BUILD
// TODO(crbug.com/40285824): Remove this and convert code to safer constructs.
#pragma allow_unsafe_buffers
#endif

#include "cronet_bidirectional_stream_adapter.h"

#include <string>
#include <utility>
#include <vector>

#include "base/functional/bind.h"
#include "base/location.h"
#include "base/logging.h"
#include "base/numerics/safe_conversions.h"
#include "base/strings/string_number_conversions.h"
#include "components/cronet/android/cronet_context_adapter.h"
#include "components/cronet/android/io_buffer_with_byte_buffer.h"
#include "components/cronet/android/url_request_close_source.h"
#include "components/cronet/android/url_request_error.h"
#include "components/cronet/metrics_util.h"
#include "net/base/http_user_agent_settings.h"
#include "net/base/net_errors.h"
#include "net/base/request_priority.h"
#include "net/http/bidirectional_stream_request_info.h"
#include "net/http/http_network_session.h"
#include "net/http/http_response_headers.h"
#include "net/http/http_status_code.h"
#include "net/http/http_transaction_factory.h"
#include "net/http/http_util.h"
#include "net/ssl/ssl_info.h"
#include "net/third_party/quiche/src/quiche/quic/core/quic_packets.h"
#include "net/url_request/url_request_context.h"
#include "url/gurl.h"

// Must come after all headers that specialize FromJniType() / ToJniType().
#include "components/cronet/android/cronet_jni_headers/CronetBidirectionalStream_jni.h"
#include "components/cronet/android/cronet_jni_headers/CronetMetrics_jni.h"

using base::android::ConvertUTF8ToJavaString;
using base::android::ConvertJavaStringToUTF8;
using base::android::JavaRef;
using base::android::ScopedJavaLocalRef;

namespace cronet {

namespace {

// As |GetArrayLength| makes no guarantees about the returned value (e.g., it
// may be -1 if |array| is not a valid Java array), provide a safe wrapper
// that always returns a valid, non-negative size.
template <typename JavaArrayType>
size_t SafeGetArrayLength(JNIEnv* env, JavaArrayType jarray) {
  DCHECK(jarray);
  jsize length = env->GetArrayLength(jarray);
  DCHECK_GE(length, 0) << "Invalid array length: " << length;
  return static_cast<size_t>(std::max(0, length));
}

}  // namespace

CronetMetrics::CronetMetrics() = default;
CronetMetrics::CronetMetrics(const CronetMetrics& metrics) = default;
CronetMetrics::~CronetMetrics() = default;

PendingWriteData::PendingWriteData(
    JNIEnv* env,
    const JavaRef<jobjectArray>& jwrite_buffer_list,
    const JavaRef<jintArray>& jwrite_buffer_pos_list,
    const JavaRef<jintArray>& jwrite_buffer_limit_list,
    jboolean jwrite_end_of_stream) {
  this->jwrite_buffer_list.Reset(jwrite_buffer_list);
  this->jwrite_buffer_pos_list.Reset(jwrite_buffer_pos_list);
  this->jwrite_buffer_limit_list.Reset(jwrite_buffer_limit_list);
  this->jwrite_end_of_stream = jwrite_end_of_stream;
}

PendingWriteData::~PendingWriteData() {
  // Reset global references.
  jwrite_buffer_list.Reset();
  jwrite_buffer_pos_list.Reset();
  jwrite_buffer_limit_list.Reset();
}

static jlong JNI_CronetBidirectionalStream_CreateBidirectionalStream(
    JNIEnv* env,
    const base::android::JavaParamRef<jobject>& jbidi_stream,
    jlong jurl_request_context_adapter,
    jboolean jsend_request_headers_automatically,
    jboolean jtraffic_stats_tag_set,
    jint jtraffic_stats_tag,
    jboolean jtraffic_stats_uid_set,
    jint jtraffic_stats_uid,
    jlong jnetwork_handle) {
  CronetContextAdapter* context_adapter =
      reinterpret_cast<CronetContextAdapter*>(jurl_request_context_adapter);
  DCHECK(context_adapter);

  CronetBidirectionalStreamAdapter* adapter =
      new CronetBidirectionalStreamAdapter(
          context_adapter, env, jbidi_stream,
          jsend_request_headers_automatically, jtraffic_stats_tag_set,
          jtraffic_stats_tag, jtraffic_stats_uid_set, jtraffic_stats_uid,
          jnetwork_handle);

  return reinterpret_cast<jlong>(adapter);
}

CronetBidirectionalStreamAdapter::CronetBidirectionalStreamAdapter(
    CronetContextAdapter* context,
    JNIEnv* env,
    const base::android::JavaParamRef<jobject>& jbidi_stream,
    bool send_request_headers_automatically,
    bool traffic_stats_tag_set,
    int32_t traffic_stats_tag,
    bool traffic_stats_uid_set,
    int32_t traffic_stats_uid,
    net::handles::NetworkHandle network)
    : context_(context),
      owner_(env, jbidi_stream),
      send_request_headers_automatically_(send_request_headers_automatically),
      traffic_stats_tag_set_(traffic_stats_tag_set),
      traffic_stats_tag_(traffic_stats_tag),
      traffic_stats_uid_set_(traffic_stats_uid_set),
      traffic_stats_uid_(traffic_stats_uid),
      network_(network),
      stream_failed_(false) {}

CronetBidirectionalStreamAdapter::~CronetBidirectionalStreamAdapter() {
  DCHECK(context_->IsOnNetworkThread());
}

void CronetBidirectionalStreamAdapter::SendRequestHeaders(JNIEnv* env) {
  context_->PostTaskToNetworkThread(
      FROM_HERE,
      base::BindOnce(
          &CronetBidirectionalStreamAdapter::SendRequestHeadersOnNetworkThread,
          base::Unretained(this)));
}

jint CronetBidirectionalStreamAdapter::Start(
    JNIEnv* env,
    const base::android::JavaParamRef<jstring>& jurl,
    jint jpriority,
    const base::android::JavaParamRef<jstring>& jmethod,
    const base::android::JavaParamRef<jobjectArray>& jheaders,
    jboolean jend_of_stream) {
  // Prepare request info here to be able to return the error.
  std::unique_ptr<net::BidirectionalStreamRequestInfo> request_info(
      new net::BidirectionalStreamRequestInfo());
  request_info->url = GURL(ConvertJavaStringToUTF8(env, jurl));
  request_info->priority = static_cast<net::RequestPriority>(jpriority);
  // Http method is a token, just as header name.
  request_info->method = ConvertJavaStringToUTF8(env, jmethod);
  if (!net::HttpUtil::IsValidHeaderName(request_info->method))
    return -1;

  std::vector<std::string> headers;
  base::android::AppendJavaStringArrayToStringVector(env, jheaders, &headers);
  for (size_t i = 0; i < headers.size(); i += 2) {
    std::string name(headers[i]);
    std::string value(headers[i + 1]);
    if (!net::HttpUtil::IsValidHeaderName(name) ||
        !net::HttpUtil::IsValidHeaderValue(value)) {
      return i + 1;
    }
    request_info->extra_headers.SetHeader(name, value);
  }
  request_info->end_stream_on_headers = jend_of_stream;
  if (traffic_stats_tag_set_ || traffic_stats_uid_set_) {
    request_info->socket_tag = net::SocketTag(
        traffic_stats_uid_set_ ? traffic_stats_uid_ : net::SocketTag::UNSET_UID,
        traffic_stats_tag_set_ ? traffic_stats_tag_
                               : net::SocketTag::UNSET_TAG);
  }

  context_->PostTaskToNetworkThread(
      FROM_HERE,
      base::BindOnce(&CronetBidirectionalStreamAdapter::StartOnNetworkThread,
                     base::Unretained(this), std::move(request_info)));
  return 0;
}

jboolean CronetBidirectionalStreamAdapter::ReadData(
    JNIEnv* env,
    const base::android::JavaParamRef<jobject>& jbyte_buffer,
    jint jposition,
    jint jlimit) {
  DCHECK_LT(jposition, jlimit);

  scoped_refptr<IOBufferWithByteBuffer> read_buffer(
      new IOBufferWithByteBuffer(env, jbyte_buffer, jposition, jlimit));

  int remaining_capacity = jlimit - jposition;

  context_->PostTaskToNetworkThread(
      FROM_HERE,
      base::BindOnce(&CronetBidirectionalStreamAdapter::ReadDataOnNetworkThread,
                     base::Unretained(this), read_buffer, remaining_capacity));
  return JNI_TRUE;
}

jboolean CronetBidirectionalStreamAdapter::WritevData(
    JNIEnv* env,
    const base::android::JavaParamRef<jobjectArray>& jbyte_buffers,
    const base::android::JavaParamRef<jintArray>& jbyte_buffers_pos,
    const base::android::JavaParamRef<jintArray>& jbyte_buffers_limit,
    jboolean jend_of_stream) {
  size_t buffers_array_size = SafeGetArrayLength(env, jbyte_buffers.obj());
  size_t pos_array_size = SafeGetArrayLength(env, jbyte_buffers.obj());
  size_t limit_array_size = SafeGetArrayLength(env, jbyte_buffers.obj());
  if (buffers_array_size != pos_array_size ||
      buffers_array_size != limit_array_size) {
    DLOG(ERROR) << "Illegal arguments.";
    return JNI_FALSE;
  }

  std::unique_ptr<PendingWriteData> pending_write_data;
  pending_write_data.reset(
      new PendingWriteData(env, jbyte_buffers, jbyte_buffers_pos,
                           jbyte_buffers_limit, jend_of_stream));
  for (size_t i = 0; i < buffers_array_size; ++i) {
    auto jbuffer = ScopedJavaLocalRef<jobject>::Adopt(
        env, env->GetObjectArrayElement(
                 pending_write_data->jwrite_buffer_list.obj(), i));
    void* data = env->GetDirectBufferAddress(jbuffer.obj());
    if (!data)
      return JNI_FALSE;
    jint pos;
    env->GetIntArrayRegion(pending_write_data->jwrite_buffer_pos_list.obj(), i,
                           1, &pos);
    jint limit;
    env->GetIntArrayRegion(pending_write_data->jwrite_buffer_limit_list.obj(),
                           i, 1, &limit);
    auto write_buffer = base::MakeRefCounted<net::WrappedIOBuffer>(
        base::span(static_cast<char*>(data), base::checked_cast<size_t>(limit))
            .subspan(base::checked_cast<size_t>(pos)));
    pending_write_data->write_buffer_list.push_back(write_buffer);
    pending_write_data->write_buffer_len_list.push_back(write_buffer->size());
  }

  context_->PostTaskToNetworkThread(
      FROM_HERE,
      base::BindOnce(
          &CronetBidirectionalStreamAdapter::WritevDataOnNetworkThread,
          base::Unretained(this), std::move(pending_write_data)));
  return JNI_TRUE;
}

void CronetBidirectionalStreamAdapter::Destroy(JNIEnv* env) {
  // Destroy could be called from any thread, including network thread (if
  // posting task to executor throws an exception), but is posted, so |this|
  // is valid until calling task is complete. Destroy() is always called from
  // within a synchronized java block that guarantees no future posts to the
  // network thread with the adapter pointer.
  context_->PostTaskToNetworkThread(
      FROM_HERE,
      base::BindOnce(
          &CronetBidirectionalStreamAdapter::DestroyOnNetworkThreadThenCallback,
          base::Unretained(this)));
}

void CronetBidirectionalStreamAdapter::OnStreamReady(
    bool request_headers_sent) {
  DCHECK(context_->IsOnNetworkThread());
  JNIEnv* env = base::android::AttachCurrentThread();
  cronet::Java_CronetBidirectionalStream_onStreamReady(
      env, owner_, request_headers_sent ? JNI_TRUE : JNI_FALSE);
}

void CronetBidirectionalStreamAdapter::OnHeadersReceived(
    const quiche::HttpHeaderBlock& response_headers) {
  DCHECK(context_->IsOnNetworkThread());
  JNIEnv* env = base::android::AttachCurrentThread();
  // Get http status code from response headers.
  jint http_status_code = 0;
  const auto http_status_header = response_headers.find(":status");
  if (http_status_header != response_headers.end()) {
    base::StringToInt(http_status_header->second, &http_status_code);
  }

  std::string protocol;
  switch (bidi_stream_->GetProtocol()) {
    case net::NextProto::kProtoHTTP2:
      protocol = "h2";
      break;
    case net::NextProto::kProtoQUIC:
      protocol = "quic/1+spdy/3";
      break;
    default:
      break;
  }

  cronet::Java_CronetBidirectionalStream_onResponseHeadersReceived(
      env, owner_, http_status_code, ConvertUTF8ToJavaString(env, protocol),
      GetHeadersArray(env, response_headers),
      bidi_stream_->GetTotalReceivedBytes());
}

void CronetBidirectionalStreamAdapter::OnDataRead(int bytes_read) {
  DCHECK(context_->IsOnNetworkThread());
  JNIEnv* env = base::android::AttachCurrentThread();
  cronet::Java_CronetBidirectionalStream_onReadCompleted(
      env, owner_, read_buffer_->byte_buffer(), bytes_read,
      read_buffer_->initial_position(), read_buffer_->initial_limit(),
      bidi_stream_->GetTotalReceivedBytes());
  // Free the read buffer. This lets the Java ByteBuffer be freed, if the
  // embedder releases it, too.
  read_buffer_ = nullptr;
}

void CronetBidirectionalStreamAdapter::OnDataSent() {
  DCHECK(context_->IsOnNetworkThread());
  DCHECK(pending_write_data_);

  JNIEnv* env = base::android::AttachCurrentThread();
  // Call into Java.
  cronet::Java_CronetBidirectionalStream_onWritevCompleted(
      env, owner_, pending_write_data_->jwrite_buffer_list,
      pending_write_data_->jwrite_buffer_pos_list,
      pending_write_data_->jwrite_buffer_limit_list,
      pending_write_data_->jwrite_end_of_stream);
  // Free the java objects. This lets the Java ByteBuffers be freed, if the
  // embedder releases it, too.
  pending_write_data_.reset();
}

void CronetBidirectionalStreamAdapter::OnTrailersReceived(
    const quiche::HttpHeaderBlock& response_trailers) {
  DCHECK(context_->IsOnNetworkThread());
  JNIEnv* env = base::android::AttachCurrentThread();
  cronet::Java_CronetBidirectionalStream_onResponseTrailersReceived(
      env, owner_, GetHeadersArray(env, response_trailers));
}

void CronetBidirectionalStreamAdapter::OnFailed(int error) {
  DCHECK(context_->IsOnNetworkThread());
  stream_failed_ = true;
  JNIEnv* env = base::android::AttachCurrentThread();
  net::NetErrorDetails net_error_details;
  bidi_stream_->PopulateNetErrorDetails(&net_error_details);
  cronet::Java_CronetBidirectionalStream_onError(
      env, owner_, NetErrorToUrlRequestError(error), error,
      net_error_details.quic_connection_error,
      (int)NetSourceToJavaSource(net_error_details.source),
      ConvertUTF8ToJavaString(env, net::ErrorToString(error)),
      bidi_stream_->GetTotalReceivedBytes());
}

void CronetBidirectionalStreamAdapter::StartOnNetworkThread(
    std::unique_ptr<net::BidirectionalStreamRequestInfo> request_info) {
  DCHECK(context_->IsOnNetworkThread());
  DCHECK(!bidi_stream_);

  request_info->detect_broken_connection =
      context_->cronet_url_request_context()
          ->bidi_stream_detect_broken_connection();
  request_info->heartbeat_interval =
      context_->cronet_url_request_context()->heartbeat_interval();
  request_info->extra_headers.SetHeaderIfMissing(
      net::HttpRequestHeaders::kUserAgent,
      context_->GetURLRequestContext(network_)
          ->http_user_agent_settings()
          ->GetUserAgent());
  bidi_stream_.reset(
      new net::BidirectionalStream(std::move(request_info),
                                   context_->GetURLRequestContext(network_)
                                       ->http_transaction_factory()
                                       ->GetSession(),
                                   send_request_headers_automatically_, this));
}

void CronetBidirectionalStreamAdapter::SendRequestHeadersOnNetworkThread() {
  DCHECK(context_->IsOnNetworkThread());
  DCHECK(!send_request_headers_automatically_);

  if (stream_failed_) {
    // If stream failed between the time when SendRequestHeaders is invoked and
    // SendRequestHeadersOnNetworkThread is executed, do not call into
    // |bidi_stream_| since the underlying stream might have been destroyed.
    // Do not invoke Java callback either, since onError is posted when
    // |stream_failed_| is set to true.
    return;
  }
  bidi_stream_->SendRequestHeaders();
}

void CronetBidirectionalStreamAdapter::ReadDataOnNetworkThread(
    scoped_refptr<IOBufferWithByteBuffer> read_buffer,
    int buffer_size) {
  DCHECK(context_->IsOnNetworkThread());
  DCHECK(read_buffer);
  DCHECK(!read_buffer_);

  read_buffer_ = read_buffer;

  int bytes_read = bidi_stream_->ReadData(read_buffer_.get(), buffer_size);
  // If IO is pending, wait for the BidirectionalStream to call OnDataRead.
  if (bytes_read == net::ERR_IO_PENDING)
    return;

  if (bytes_read < 0) {
    OnFailed(bytes_read);
    return;
  }
  OnDataRead(bytes_read);
}

void CronetBidirectionalStreamAdapter::WritevDataOnNetworkThread(
    std::unique_ptr<PendingWriteData> pending_write_data) {
  DCHECK(context_->IsOnNetworkThread());
  DCHECK(pending_write_data);
  DCHECK(!pending_write_data_);

  if (stream_failed_) {
    // If stream failed between the time when WritevData is invoked and
    // WritevDataOnNetworkThread is executed, do not call into |bidi_stream_|
    // since the underlying stream might have been destroyed. Do not invoke
    // Java callback either, since onError is posted when |stream_failed_| is
    // set to true.
    return;
  }

  pending_write_data_ = std::move(pending_write_data);
  bool end_of_stream = pending_write_data_->jwrite_end_of_stream == JNI_TRUE;
  bidi_stream_->SendvData(pending_write_data_->write_buffer_list,
                          pending_write_data_->write_buffer_len_list,
                          end_of_stream);
}

void CronetBidirectionalStreamAdapter::DestroyOnNetworkThreadThenCallback() {
  DCHECK(context_->IsOnNetworkThread());
  JNIEnv* env = base::android::AttachCurrentThread();
  std::optional<CronetMetrics> metrics = GetMetrics();
  cronet::Java_CronetBidirectionalStream_onNativeStreamAdapterDestroyed(
      env, owner_, metrics.has_value() ? metrics->cronet_metrics : nullptr,
      /*quicConnectionMigrationAttempted= */
      metrics.has_value() ? metrics->quic_connection_migration_attempted
                          : false,
      /*quicConnectionMigrationSuccessful */
      metrics.has_value() ? metrics->quic_connection_migration_successful
                          : false);
  delete this;
}

base::android::ScopedJavaLocalRef<jobjectArray>
CronetBidirectionalStreamAdapter::GetHeadersArray(
    JNIEnv* env,
    const quiche::HttpHeaderBlock& header_block) {
  DCHECK(context_->IsOnNetworkThread());

  std::vector<std::string> headers;
  for (const auto& header : header_block) {
    auto value = std::string(header.second);
    size_t start = 0;
    size_t end = 0;
    // The do loop will split headers by '\0' so that applications can skip it.
    do {
      end = value.find('\0', start);
      std::string split_value;
      if (end != value.npos) {
        split_value = value.substr(start, end - start);
      } else {
        split_value = value.substr(start);
      }
      headers.push_back(std::string(header.first));
      headers.push_back(split_value);
      start = end + 1;
    } while (end != value.npos);
  }
  return base::android::ToJavaArrayOfStrings(env, headers);
}

std::optional<CronetMetrics> CronetBidirectionalStreamAdapter::GetMetrics() {
  JNIEnv* env = base::android::AttachCurrentThread();
  // There can exist a scenario where we don't have an underlying bidi stream
  // but the adapter has been created (e.g. user passes bad headers). In this
  // scenario, there will be no metrics.
  if (!bidi_stream_) {
    return std::nullopt;
  }
  CronetMetrics metrics;
  net::LoadTimingInfo load_timing_info;
  net::NetErrorDetails net_error_details;
  bidi_stream_->GetLoadTimingInfo(&load_timing_info);
  bidi_stream_->PopulateNetErrorDetails(&net_error_details);
  base::Time start_time = load_timing_info.request_start_time;
  base::TimeTicks start_ticks = load_timing_info.request_start;
  metrics.cronet_metrics = cronet::Java_CronetMetrics_Constructor(
      env, metrics_util::ConvertTime(start_ticks, start_ticks, start_time),
      metrics_util::ConvertTime(
          load_timing_info.connect_timing.domain_lookup_start, start_ticks,
          start_time),
      metrics_util::ConvertTime(
          load_timing_info.connect_timing.domain_lookup_end, start_ticks,
          start_time),
      metrics_util::ConvertTime(load_timing_info.connect_timing.connect_start,
                                start_ticks, start_time),
      metrics_util::ConvertTime(load_timing_info.connect_timing.connect_end,
                                start_ticks, start_time),
      metrics_util::ConvertTime(load_timing_info.connect_timing.ssl_start,
                                start_ticks, start_time),
      metrics_util::ConvertTime(load_timing_info.connect_timing.ssl_end,
                                start_ticks, start_time),
      metrics_util::ConvertTime(load_timing_info.send_start, start_ticks,
                                start_time),
      metrics_util::ConvertTime(load_timing_info.send_end, start_ticks,
                                start_time),
      metrics_util::ConvertTime(load_timing_info.push_start, start_ticks,
                                start_time),
      metrics_util::ConvertTime(load_timing_info.push_end, start_ticks,
                                start_time),
      metrics_util::ConvertTime(load_timing_info.receive_headers_end,
                                start_ticks, start_time),
      metrics_util::ConvertTime(base::TimeTicks::Now(), start_ticks,
                                start_time),
      load_timing_info.socket_reused, bidi_stream_->GetTotalSentBytes(),
      bidi_stream_->GetTotalReceivedBytes()),

  metrics.quic_connection_migration_attempted =
      net_error_details.quic_connection_migration_attempted;
  metrics.quic_connection_migration_successful =
      net_error_details.quic_connection_migration_successful;
  return metrics;
}

}  // namespace cronet