// Copyright 2018 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 "components/cast_channel/cast_message_handler.h"

#include <tuple>

#include "base/rand_util.h"
#include "base/strings/stringprintf.h"
#include "base/time/default_tick_clock.h"
#include "components/cast_channel/cast_socket_service.h"
#include "net/traffic_annotation/network_traffic_annotation.h"

namespace cast_channel {

namespace {

constexpr base::TimeDelta kRequestTimeout = base::TimeDelta::FromSeconds(5);

constexpr net::NetworkTrafficAnnotationTag kMessageTrafficAnnotation =
    net::DefineNetworkTrafficAnnotation("cast_message_handler", R"(
        semantics {
          sender: "Cast Message Handler"
          description:
            "A Cast protocol or application-level message sent to a Cast "
            "device."
          trigger:
            "Triggered by user gesture from using Cast functionality, or "
            "a webpage using the Presentation API, or "
            "Cast device discovery internal logic."
          data:
            "A serialized Cast protocol or application-level protobuf message. "
            "A non-exhaustive list of Cast protocol messages:\n"
            "- Virtual connection requests,\n"
            "- App availability / media status / receiver status requests,\n"
            "- Launch / stop Cast session requests,\n"
            "- Media commands, such as play/pause.\n"
            "Application-level messages may contain data specific to the Cast "
            "application."
          destination: OTHER
          destination_other:
            "Data will be sent to a Cast device in local network."
        }
        policy {
          cookies_allowed: NO
          setting:
            "This request cannot be disabled, but it would not be sent if user "
            "does not connect a Cast device to the local network."
          policy_exception_justification: "Not implemented."
        })");

}  // namespace

GetAppAvailabilityRequest::GetAppAvailabilityRequest(
    int request_id,
    GetAppAvailabilityCallback callback,
    const base::TickClock* clock,
    const std::string& app_id)
    : PendingRequest(request_id, std::move(callback), clock), app_id(app_id) {}

GetAppAvailabilityRequest::~GetAppAvailabilityRequest() = default;

VirtualConnection::VirtualConnection(int channel_id,
                                     const std::string& source_id,
                                     const std::string& destination_id)
    : channel_id(channel_id),
      source_id(source_id),
      destination_id(destination_id) {}
VirtualConnection::~VirtualConnection() = default;

bool VirtualConnection::operator<(const VirtualConnection& other) const {
  return std::tie(channel_id, source_id, destination_id) <
         std::tie(other.channel_id, other.source_id, other.destination_id);
}

InternalMessage::InternalMessage(CastMessageType type, base::Value message)
    : type(type), message(std::move(message)) {}
InternalMessage::~InternalMessage() = default;

CastMessageHandler::CastMessageHandler(CastSocketService* socket_service,
                                       const std::string& user_agent,
                                       const std::string& browser_version,
                                       const std::string& locale)
    : sender_id_(base::StringPrintf("sender-%d", base::RandInt(0, 1000000))),
      user_agent_(user_agent),
      browser_version_(browser_version),
      locale_(locale),
      socket_service_(socket_service),
      clock_(base::DefaultTickClock::GetInstance()),
      weak_ptr_factory_(this) {
  DETACH_FROM_SEQUENCE(sequence_checker_);
  socket_service_->task_runner()->PostTask(
      FROM_HERE, base::BindOnce(&CastSocketService::AddObserver,
                                base::Unretained(socket_service_),
                                base::Unretained(this)));
}

CastMessageHandler::~CastMessageHandler() {
  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  socket_service_->RemoveObserver(this);
}

void CastMessageHandler::EnsureConnection(int channel_id,
                                          const std::string& source_id,
                                          const std::string& destination_id) {
  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  CastSocket* socket = socket_service_->GetSocket(channel_id);
  if (!socket) {
    DVLOG(2) << __func__ << ": socket not found: " << channel_id;
    return;
  }

  DoEnsureConnection(socket, source_id, destination_id);
}

CastMessageHandler::PendingRequests*
CastMessageHandler::GetOrCreatePendingRequests(int channel_id) {
  CastMessageHandler::PendingRequests* requests = nullptr;
  auto pending_it = pending_requests_.find(channel_id);
  if (pending_it != pending_requests_.end()) {
    return pending_it->second.get();
  }

  auto new_requests = std::make_unique<CastMessageHandler::PendingRequests>();
  requests = new_requests.get();
  pending_requests_.emplace(channel_id, std::move(new_requests));
  return requests;
}

void CastMessageHandler::RequestAppAvailability(
    CastSocket* socket,
    const std::string& app_id,
    GetAppAvailabilityCallback callback) {
  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);

  int channel_id = socket->id();
  auto* requests = GetOrCreatePendingRequests(channel_id);
  int request_id = NextRequestId();

  DVLOG(2) << __func__ << ", channel_id: " << channel_id
           << ", app_id: " << app_id << ", request_id: " << request_id;
  if (requests->AddAppAvailabilityRequest(
          std::make_unique<GetAppAvailabilityRequest>(
              request_id, std::move(callback), clock_, app_id))) {
    SendCastMessage(socket, CreateGetAppAvailabilityRequest(
                                sender_id_, request_id, app_id));
  }
}

void CastMessageHandler::SendBroadcastMessage(
    int channel_id,
    const std::vector<std::string>& app_ids,
    const BroadcastRequest& request) {
  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);

  CastSocket* socket = socket_service_->GetSocket(channel_id);
  if (!socket) {
    DVLOG(2) << __func__ << ": socket not found: " << channel_id;
    return;
  }

  int request_id = NextRequestId();
  DVLOG(2) << __func__ << ", channel_id: " << channel_id
           << ", request_id: " << request_id;

  // Note: Even though the message is formatted like a request, we don't care
  // about the response, as broadcasts are fire-and-forget.
  CastMessage message =
      CreateBroadcastRequest(sender_id_, request_id, app_ids, request);
  SendCastMessage(socket, message);
}

void CastMessageHandler::LaunchSession(int channel_id,
                                       const std::string& app_id,
                                       base::TimeDelta launch_timeout,
                                       LaunchSessionCallback callback) {
  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  CastSocket* socket = socket_service_->GetSocket(channel_id);
  if (!socket) {
    DVLOG(2) << __func__ << ": socket not found: " << channel_id;
    std::move(callback).Run(LaunchSessionResponse());
    return;
  }

  auto* requests = GetOrCreatePendingRequests(channel_id);
  int request_id = NextRequestId();
  DVLOG(2) << __func__ << ", channel_id: " << channel_id
           << ", request_id: " << request_id;

  if (requests->AddLaunchRequest(std::make_unique<LaunchSessionRequest>(
                                     request_id, std::move(callback), clock_),
                                 launch_timeout)) {
    SendCastMessage(
        socket, CreateLaunchRequest(sender_id_, request_id, app_id, locale_));
  }
}

void CastMessageHandler::SendAppMessage(int channel_id,
                                        const CastMessage& message) {
  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  DCHECK(!IsCastInternalNamespace(message.namespace_()))
      << ": unexpected app message namespace: " << message.namespace_();

  CastSocket* socket = socket_service_->GetSocket(channel_id);
  if (!socket) {
    DVLOG(2) << __func__ << ": socket not found: " << channel_id;
    return;
  }

  SendCastMessage(socket, message);
}

void CastMessageHandler::AddObserver(Observer* observer) {
  observers_.AddObserver(observer);
}

void CastMessageHandler::RemoveObserver(Observer* observer) {
  observers_.RemoveObserver(observer);
}

void CastMessageHandler::OnError(const CastSocket& socket,
                                 ChannelError error_state) {
  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  int channel_id = socket.id();

  base::EraseIf(virtual_connections_,
                [&channel_id](const VirtualConnection& connection) {
                  return connection.channel_id == channel_id;
                });

  pending_requests_.erase(channel_id);
}

void CastMessageHandler::OnMessage(const CastSocket& socket,
                                   const CastMessage& message) {
  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  DVLOG(2) << __func__ << ", channel_id: " << socket.id()
           << ", message: " << CastMessageToString(message);

  // TODO(crbug.com/698940): Support Observers for both kinds of messages.
  if (IsCastInternalNamespace(message.namespace_())) {
    HandleCastInternalMessage(socket, message);
  } else {
    DVLOG(2) << "Got app message from cast channel with namespace: "
             << message.namespace_();

    for (auto& observer : observers_)
      observer.OnAppMessage(socket.id(), message);
  }
}

void CastMessageHandler::HandleCastInternalMessage(const CastSocket& socket,
                                                   const CastMessage& message) {
  // TODO(crbug.com/698940): Handle other messages (VIRTUAL_CONNECT_CLOSE).
  std::unique_ptr<base::DictionaryValue> payload =
      GetDictionaryFromCastMessage(message);
  if (!payload)
    return;

  int request_id = 0;
  if (GetRequestIdFromResponse(*payload, &request_id) && request_id > 0) {
    auto requests_it = pending_requests_.find(socket.id());
    if (requests_it != pending_requests_.end())
      requests_it->second->HandlePendingRequest(request_id, *payload);
  }

  CastMessageType type = ParseMessageTypeFromPayload(*payload);
  if (type == CastMessageType::kOther) {
    DVLOG(2) << "Unknown message type: " << CastMessageToString(message);
    return;
  }

  InternalMessage internal_message(type, std::move(*payload));
  for (auto& observer : observers_)
    observer.OnInternalMessage(socket.id(), internal_message);
}

void CastMessageHandler::SendCastMessage(CastSocket* socket,
                                         const CastMessage& message) {
  // A virtual connection must be opened to the receiver before other messages
  // can be sent.
  DoEnsureConnection(socket, message.source_id(), message.destination_id());
  socket->transport()->SendMessage(
      message,
      base::Bind(&CastMessageHandler::OnMessageSent,
                 weak_ptr_factory_.GetWeakPtr()),
      kMessageTrafficAnnotation);
}

void CastMessageHandler::DoEnsureConnection(CastSocket* socket,
                                            const std::string& source_id,
                                            const std::string& destination_id) {
  VirtualConnection connection(socket->id(), source_id, destination_id);
  if (virtual_connections_.find(connection) != virtual_connections_.end())
    return;

  DVLOG(1) << "Creating VC for channel: " << connection.channel_id
           << ", source: " << connection.source_id
           << ", dest: " << connection.destination_id;
  CastMessage virtual_connection_request = CreateVirtualConnectionRequest(
      connection.source_id, connection.destination_id,
      connection.destination_id == kPlatformReceiverId
          ? VirtualConnectionType::kStrong
          : VirtualConnectionType::kInvisible,
      user_agent_, browser_version_);
  socket->transport()->SendMessage(
      virtual_connection_request,
      base::Bind(&CastMessageHandler::OnMessageSent,
                 weak_ptr_factory_.GetWeakPtr()),
      kMessageTrafficAnnotation);

  // We assume the virtual connection request will succeed; otherwise this
  // will eventually self-correct.
  virtual_connections_.insert(connection);
}

void CastMessageHandler::OnMessageSent(int result) {
  DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  DVLOG_IF(2, result < 0) << "SendMessage failed with code: " << result;
}

CastMessageHandler::PendingRequests::PendingRequests() {}
CastMessageHandler::PendingRequests::~PendingRequests() {
  for (auto& request : pending_app_availability_requests) {
    std::move(request.second->callback)
        .Run(request.second->app_id, GetAppAvailabilityResult::kUnknown);
  }

  if (pending_launch_session_request) {
    LaunchSessionResponse response;
    response.result = LaunchSessionResponse::kError;
    std::move(pending_launch_session_request->callback)
        .Run(std::move(response));
  }
}

bool CastMessageHandler::PendingRequests::AddAppAvailabilityRequest(
    std::unique_ptr<GetAppAvailabilityRequest> request) {
  std::string app_id = request->app_id;
  if (base::ContainsKey(pending_app_availability_requests, request->app_id))
    return false;

  int request_id = request->request_id;
  request->timeout_timer.Start(
      FROM_HERE, kRequestTimeout,
      base::Bind(&CastMessageHandler::PendingRequests::AppAvailabilityTimedOut,
                 base::Unretained(this), request_id));
  pending_app_availability_requests.emplace(app_id, std::move(request));
  return true;
}

bool CastMessageHandler::PendingRequests::AddLaunchRequest(
    std::unique_ptr<LaunchSessionRequest> request,
    base::TimeDelta timeout) {
  if (pending_launch_session_request)
    return false;

  int request_id = request->request_id;
  request->timeout_timer.Start(
      FROM_HERE, timeout,
      base::Bind(&CastMessageHandler::PendingRequests::LaunchSessionTimedOut,
                 base::Unretained(this), request_id));
  pending_launch_session_request = std::move(request);
  return true;
}

void CastMessageHandler::PendingRequests::HandlePendingRequest(
    int request_id,
    const base::DictionaryValue& response) {
  auto app_availability_it =
      std::find_if(pending_app_availability_requests.begin(),
                   pending_app_availability_requests.end(),
                   [&request_id](const auto& entry) {
                     return entry.second->request_id == request_id;
                   });
  if (app_availability_it != pending_app_availability_requests.end()) {
    GetAppAvailabilityResult result = GetAppAvailabilityResultFromResponse(
        response, app_availability_it->second->app_id);
    std::move(app_availability_it->second->callback)
        .Run(app_availability_it->second->app_id, result);
    pending_app_availability_requests.erase(app_availability_it);
    return;
  }

  if (pending_launch_session_request &&
      pending_launch_session_request->request_id == request_id) {
    std::move(pending_launch_session_request->callback)
        .Run(GetLaunchSessionResponse(response));
    pending_launch_session_request.reset();
    return;
  }
}

void CastMessageHandler::PendingRequests::AppAvailabilityTimedOut(
    int request_id) {
  DVLOG(1) << __func__ << ", request_id: " << request_id;

  auto it = std::find_if(pending_app_availability_requests.begin(),
                         pending_app_availability_requests.end(),
                         [&request_id](const auto& entry) {
                           return entry.second->request_id == request_id;
                         });

  CHECK(it != pending_app_availability_requests.end());
  std::move(it->second->callback)
      .Run(it->second->app_id, GetAppAvailabilityResult::kUnknown);
  pending_app_availability_requests.erase(it);
}

void CastMessageHandler::PendingRequests::LaunchSessionTimedOut(
    int request_id) {
  DVLOG(1) << __func__ << ", request_id: " << request_id;
  CHECK(pending_launch_session_request);
  CHECK(pending_launch_session_request->request_id == request_id);

  LaunchSessionResponse response;
  response.result = LaunchSessionResponse::kTimedOut;
  std::move(pending_launch_session_request->callback).Run(std::move(response));
  pending_launch_session_request.reset();
}

}  // namespace cast_channel