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

#include "components/cast/message_port/fuchsia/message_port_fuchsia.h"

#include <lib/fpromise/result.h>

#include <string_view>

#include "base/fuchsia/fuchsia_logging.h"
#include "base/fuchsia/mem_buffer_util.h"
#include "base/memory/weak_ptr.h"
#include "base/notreached.h"

namespace cast_api_bindings {
namespace {

// A MessagePortFuchsia instantiated with an
// InterfaceHandle<fuchsia::web::MessagePort>. Acts as a client of a
// MessagePortFuchsiaServer.
class MessagePortFuchsiaClient : public MessagePortFuchsia {
 public:
  explicit MessagePortFuchsiaClient(
      fidl::InterfaceHandle<::fuchsia::web::MessagePort> port)
      : MessagePortFuchsia(PortType::HANDLE), port_(port.Bind()) {}

  ~MessagePortFuchsiaClient() override {}

  // MessagePortFuchsia implementation.
  fidl::InterfaceHandle<::fuchsia::web::MessagePort> TakeClientHandle() final {
    CHECK(!receiver_);
    CHECK(port_.is_bound());
    return port_.Unbind();
  }

  fidl::InterfaceRequest<::fuchsia::web::MessagePort> TakeServiceRequest()
      final {
    NOTREACHED();
  }

  // MessagePort implementation.
  void SetReceiver(cast_api_bindings::MessagePort::Receiver* receiver) final {
    CHECK(receiver);
    CHECK(!receiver_);
    receiver_ = receiver;
    port_.set_error_handler(
        [this](zx_status_t status) { MessagePortFuchsia::OnZxError(status); });
    ReadNextMessage();
  }

  void Close() final {
    if (port_.is_bound()) {
      port_.Unbind();
    }
  }

  bool CanPostMessage() const final { return port_.is_bound(); }

 private:
  void OnDeliverMessageToFidlComplete(
      fuchsia::web::MessagePort_PostMessage_Result result) {
    if (result.is_err()) {
      LOG(ERROR) << "PostMessage failed, reason: "
                 << static_cast<int32_t>(result.err());
      ReportPipeError();
      return;
    }

    message_queue_.pop_front();
    DeliverMessageToFidl();
  }

  void DeliverMessageToFidl() final {
    if (message_queue_.empty()) {
      return;
    }

    port_->PostMessage(
        std::move(message_queue_.front()),
        fit::bind_member(
            this, &MessagePortFuchsiaClient::OnDeliverMessageToFidlComplete));
  }

  // Helpers for reading and writing messages on the |port_|
  void OnMessageReady(fuchsia::web::WebMessage message) {
    base::WeakPtr<MessagePortFuchsia> weak_this =
        weak_factory_.GetMutableWeakPtr();
    auto status = ExtractAndHandleMessageFromFidl(std::move(message));
    if (!weak_this)
      return;

    if (status) {
      LOG(WARNING) << "Received bad message, error: "
                   << static_cast<int32_t>(*status);
      ReportPipeError();
      return;
    }

    ReadNextMessage();
  }

  void ReadNextMessage() {
    CHECK(receiver_);
    CHECK(port_);

    port_->ReceiveMessage(
        fit::bind_member(this, &MessagePortFuchsiaClient::OnMessageReady));
  }

  fuchsia::web::MessagePortPtr port_;

  const base::WeakPtrFactory<MessagePortFuchsia> weak_factory_{this};
};

// A MessagePortFuchsia instantiated with an
// InterfaceRequest<fuchsia::web::MessagePort>. Acts as the server for a
// MessagePortFuchsiaClient.
class MessagePortFuchsiaServer : public MessagePortFuchsia,
                                 public fuchsia::web::MessagePort {
 public:
  explicit MessagePortFuchsiaServer(
      fidl::InterfaceRequest<::fuchsia::web::MessagePort> port)
      : MessagePortFuchsia(PortType::REQUEST), binding_(this) {
    binding_.Bind(std::move(port));
  }

  ~MessagePortFuchsiaServer() override {}

  // MessagePortFuchsia implementation.
  fidl::InterfaceHandle<::fuchsia::web::MessagePort> TakeClientHandle() final {
    NOTREACHED();
  }

  fidl::InterfaceRequest<::fuchsia::web::MessagePort> TakeServiceRequest()
      final {
    return binding_.Unbind();
  }

  // MessagePort implementation.
  void SetReceiver(cast_api_bindings::MessagePort::Receiver* receiver) final {
    CHECK(receiver);
    CHECK(!receiver_);
    receiver_ = receiver;
    binding_.set_error_handler(
        [this](zx_status_t status) { MessagePortFuchsia::OnZxError(status); });
  }

  void Close() final {
    if (binding_.is_bound()) {
      binding_.Unbind();
    }
  }

  bool CanPostMessage() const final { return binding_.is_bound(); }

 private:
  void DeliverMessageToFidl() final {
    // Do nothing if the client hasn't requested a read, or if there's nothing
    // to read.
    if (!pending_receive_message_callback_)
      return;

    if (message_queue_.empty())
      return;

    pending_receive_message_callback_(std::move(message_queue_.front()));
    pending_receive_message_callback_ = {};
    message_queue_.pop_front();
  }

  // Avoid hiding warning by other PostMessage
  using MessagePortFuchsia::PostMessage;

  // fuchsia::web::MessagePort implementation.
  void PostMessage(fuchsia::web::WebMessage message,
                   PostMessageCallback callback) final {
    base::WeakPtr<MessagePortFuchsia> weak_this =
        weak_factory_.GetMutableWeakPtr();
    auto status = ExtractAndHandleMessageFromFidl(std::move(message));
    if (!weak_this)
      return;

    if (status) {
      LOG(WARNING) << "Received bad message, error: "
                   << static_cast<int32_t>(*status);
      ReportPipeError();
      return;
    }

    callback(fpromise::ok());
  }

  void ReceiveMessage(ReceiveMessageCallback callback) final {
    if (pending_receive_message_callback_) {
      LOG(WARNING)
          << "ReceiveMessage called multiple times without acknowledgement.";
      ReportPipeError();
      return;
    }

    pending_receive_message_callback_ = std::move(callback);
    DeliverMessageToFidl();
  }

  fidl::Binding<fuchsia::web::MessagePort> binding_;
  ReceiveMessageCallback pending_receive_message_callback_;

  const base::WeakPtrFactory<MessagePortFuchsia> weak_factory_{this};
};
}  // namespace

// static
void MessagePortFuchsia::CreatePair(std::unique_ptr<MessagePort>* client,
                                    std::unique_ptr<MessagePort>* server) {
  fidl::InterfaceHandle<fuchsia::web::MessagePort> port0;
  fidl::InterfaceRequest<fuchsia::web::MessagePort> port1 = port0.NewRequest();
  *client = MessagePortFuchsia::Create(std::move(port0));
  *server = MessagePortFuchsia::Create(std::move(port1));
}

// static
std::unique_ptr<MessagePort> MessagePortFuchsia::Create(
    fidl::InterfaceHandle<::fuchsia::web::MessagePort> handle) {
  return std::make_unique<MessagePortFuchsiaClient>(std::move(handle));
}

// static
std::unique_ptr<MessagePort> MessagePortFuchsia::Create(
    fidl::InterfaceRequest<::fuchsia::web::MessagePort> request) {
  return std::make_unique<MessagePortFuchsiaServer>(std::move(request));
}

MessagePortFuchsia* MessagePortFuchsia::FromMessagePort(MessagePort* port) {
  CHECK(port);
  // This is safe because there is one MessagePort implementation per platform
  // and this is called internally to the implementation.
  return static_cast<MessagePortFuchsia*>(port);
}

// static
fuchsia::web::WebMessage MessagePortFuchsia::CreateWebMessage(
    std::string_view message,
    std::vector<std::unique_ptr<MessagePort>> ports) {
  fuchsia::web::WebMessage message_fidl;
  message_fidl.set_data(base::MemBufferFromString(message, message));
  if (!ports.empty()) {
    PortType expected_port_type = FromMessagePort(ports[0].get())->port_type_;
    std::vector<fuchsia::web::IncomingTransferable> incoming_transferables;
    std::vector<fuchsia::web::OutgoingTransferable> receiver_transferables;
    for (auto& port : ports) {
      MessagePortFuchsia* port_fuchsia = FromMessagePort(port.get());
      PortType port_type = port_fuchsia->port_type_;

      CHECK_EQ(expected_port_type, port_type)
          << "Only one implementation of MessagePortFuchsia can be transmitted "
             "in the same message.";
      if (expected_port_type != port_type) {
        continue;
      }

      switch (port_type) {
        case PortType::HANDLE: {
          fuchsia::web::IncomingTransferable in;
          in.set_message_port(
              reinterpret_cast<MessagePortFuchsiaClient*>(port_fuchsia)
                  ->TakeClientHandle());
          incoming_transferables.emplace_back(std::move(in));
          break;
        }
        case PortType::REQUEST: {
          fuchsia::web::OutgoingTransferable out;
          out.set_message_port(
              reinterpret_cast<MessagePortFuchsiaServer*>(port_fuchsia)
                  ->TakeServiceRequest());
          receiver_transferables.emplace_back(std::move(out));
          break;
        }
      }
    }

    message_fidl.set_incoming_transfer(std::move(incoming_transferables));
    message_fidl.set_outgoing_transfer(std::move(receiver_transferables));
  }

  return message_fidl;
}

MessagePortFuchsia::MessagePortFuchsia(PortType port_type)
    : receiver_(nullptr), port_type_(port_type) {}
MessagePortFuchsia::~MessagePortFuchsia() = default;

std::optional<fuchsia::web::FrameError>
MessagePortFuchsia::ExtractAndHandleMessageFromFidl(
    fuchsia::web::WebMessage message) {
  CHECK(receiver_);
  if (!message.has_data()) {
    return fuchsia::web::FrameError::NO_DATA_IN_MESSAGE;
  }

  std::optional<std::string> data = base::StringFromMemBuffer(message.data());
  if (!data) {
    return fuchsia::web::FrameError::BUFFER_NOT_UTF8;
  }

  std::vector<std::unique_ptr<MessagePort>> ports;
  if (message.has_incoming_transfer()) {
    for (fuchsia::web::IncomingTransferable& transferable :
         *message.mutable_incoming_transfer()) {
      ports.emplace_back(Create(std::move(transferable.message_port())));
    }
  }

  if (message.mutable_outgoing_transfer()) {
    for (fuchsia::web::OutgoingTransferable& transferable :
         *message.mutable_outgoing_transfer()) {
      ports.emplace_back(Create(std::move(transferable.message_port())));
    }
  }

  if (!receiver_->OnMessage(std::move(*data), std::move(ports))) {
    return fuchsia::web::FrameError::INTERNAL_ERROR;
  }

  return std::nullopt;
}

void MessagePortFuchsia::OnZxError(zx_status_t status) {
  ZX_LOG_IF(ERROR, status != ZX_ERR_PEER_CLOSED && status != ZX_ERR_CANCELED,
            status)
      << " MessagePort disconnected.";
  ReportPipeError();
}

void MessagePortFuchsia::ReportPipeError() {
  CHECK(receiver_);
  receiver_->OnPipeError();
}

// cast_api_bindings::MessagePortFuchsia implementation
bool MessagePortFuchsia::PostMessage(std::string_view message) {
  return PostMessageWithTransferables(message, {});
}

bool MessagePortFuchsia::PostMessageWithTransferables(
    std::string_view message,
    std::vector<std::unique_ptr<MessagePort>> ports) {
  CHECK(receiver_);
  message_queue_.emplace_back(CreateWebMessage(message, std::move(ports)));

  // Start draining the queue if it was empty beforehand.
  if (message_queue_.size() == 1u) {
    DeliverMessageToFidl();
  }

  return true;
}

}  // namespace cast_api_bindings