// Copyright 2017 The Fuchsia 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 "peridot/bin/sessionmgr/message_queue/message_queue_manager.h"

#include <fuchsia/modular/cpp/fidl.h>
#include <lib/fidl/cpp/binding_set.h>
#include <lib/fsl/vmo/strings.h>

#include <algorithm>
#include <deque>
#include <utility>

#include "peridot/bin/sessionmgr/message_queue/persistent_queue.h"
#include "peridot/bin/sessionmgr/storage/constants_and_utils.h"
#include "peridot/lib/fidl/array_to_string.h"
#include "peridot/lib/fidl/json_xdr.h"
#include "peridot/lib/ledger_client/operations.h"
#include "peridot/lib/ledger_client/page_client.h"
#include "src/lib/fxl/strings/string_printf.h"

namespace modular {

struct MessageQueueInfo {
  std::string component_namespace;
  std::string component_instance_id;
  std::string queue_name;
  std::string queue_token;

  bool is_complete() const {
    return !component_instance_id.empty() && !queue_name.empty();
  }

  bool operator==(const MessageQueueInfo& a) const {
    return component_namespace == a.component_namespace &&
           component_instance_id == a.component_instance_id &&
           queue_name == a.queue_name && queue_token == a.queue_token;
  }
};

namespace {

void XdrMessageQueueInfo_v1(XdrContext* const xdr,
                            MessageQueueInfo* const data) {
  xdr->Field("component_namespace", &data->component_namespace);
  xdr->Field("component_instance_id", &data->component_instance_id);
  xdr->Field("queue_name", &data->queue_name);
  xdr->Field("queue_token", &data->queue_token);
}

void XdrMessageQueueInfo_v2(XdrContext* const xdr,
                            MessageQueueInfo* const data) {
  if (!xdr->Version(2)) {
    return;
  }
  xdr->Field("component_namespace", &data->component_namespace);
  xdr->Field("component_instance_id", &data->component_instance_id);
  xdr->Field("queue_name", &data->queue_name);
  xdr->Field("queue_token", &data->queue_token);
}

constexpr XdrFilterType<MessageQueueInfo> XdrMessageQueueInfo[] = {
    XdrMessageQueueInfo_v2,
    XdrMessageQueueInfo_v1,
    nullptr,
};

}  // namespace

class MessageQueueStorage;

// This class implements the |fuchsia::modular::MessageQueue| fidl interface,
// and is owned by |MessageQueueStorage|. It forwards all calls to its owner,
// and expects its owner to manage outstanding
// |fuchsia::modular::MessageQueue.Receive| calls. It also notifies its owner on
// object destruction.
//
// Interface is public, because bindings are outside of the class.
class MessageQueueConnection : public fuchsia::modular::MessageQueue {
 public:
  explicit MessageQueueConnection(MessageQueueStorage* queue_storage);
  ~MessageQueueConnection() override;

 private:
  // |fuchsia::modular::MessageQueue|
  void RegisterReceiver(
      fidl::InterfaceHandle<fuchsia::modular::MessageReader> receiver) override;

  // |fuchsia::modular::MessageQueue|
  void GetToken(GetTokenCallback callback) override;

  MessageQueueStorage* const queue_storage_;
};

// Class for managing a particular message queue, its tokens and its storage.
// Implementations of |fuchsia::modular::MessageQueue| and
// |fuchsia::modular::MessageSender| call into this class to manipulate the
// message queue. Owned by |MessageQueueManager|.
class MessageQueueStorage : fuchsia::modular::MessageSender {
 public:
  MessageQueueStorage(std::string queue_name, std::string queue_token,
                      const std::string& file_name_)
      : queue_name_(std::move(queue_name)),
        queue_token_(std::move(queue_token)),
        queue_data_(file_name_) {}

  ~MessageQueueStorage() override = default;

  void RegisterReceiver(
      fidl::InterfaceHandle<fuchsia::modular::MessageReader> receiver) {
    if (message_receiver_) {
      FXL_DLOG(WARNING)
          << "Existing fuchsia::modular::MessageReader is being replaced for "
             "message queue. queue name="
          << queue_name_;
    }

    message_receiver_.Bind(std::move(receiver));
    message_receiver_.set_error_handler(
        [this](zx_status_t status) {
          if (receive_ack_pending_) {
            FXL_DLOG(WARNING)
                << "MessageReceiver closed, but OnReceive acknowledgement still"
                   " pending.";
          }
          message_receiver_.Unbind();
          receive_ack_pending_ = false;
        });

    MaybeSendNextMessage();
  }

  const std::string& queue_token() const { return queue_token_; }

  void AddMessageSenderBinding(
      fidl::InterfaceRequest<fuchsia::modular::MessageSender> request) {
    message_sender_bindings_.AddBinding(this, std::move(request));
  }

  void AddMessageQueueBinding(
      fidl::InterfaceRequest<fuchsia::modular::MessageQueue> request) {
    message_queue_bindings_.AddBinding(
        std::make_unique<MessageQueueConnection>(this), std::move(request));
  }

  void RegisterWatcher(fit::function<void()> watcher) {
    watcher_ = std::move(watcher);
    if (watcher_ && !queue_data_.IsEmpty()) {
      watcher_();
    }
  }

  void DropWatcher() { watcher_ = nullptr; }

 private:
  void MaybeSendNextMessage() {
    if (!message_receiver_ || receive_ack_pending_ || queue_data_.IsEmpty()) {
      return;
    }

    receive_ack_pending_ = true;

    fsl::SizedVmo queue_data_vmo;
    FXL_CHECK(VmoFromString(queue_data_.Peek(), &queue_data_vmo));
    message_receiver_->OnReceive(std::move(queue_data_vmo).ToTransport(),
                                 [this] {
                                   receive_ack_pending_ = false;
                                   queue_data_.Dequeue();
                                   MaybeSendNextMessage();
                                 });
  }

  // |fuchsia::modular::MessageSender|
  void Send(fuchsia::mem::Buffer message) override {
    std::string msg_str;
    FXL_CHECK(fsl::StringFromVmo(message, &msg_str));
    queue_data_.Enqueue(msg_str);
    MaybeSendNextMessage();
    if (watcher_) {
      watcher_();
    }
  }

  const std::string queue_name_;
  const std::string queue_token_;

  fit::function<void()> watcher_;

  PersistentQueue queue_data_;

  bool receive_ack_pending_ = false;
  fuchsia::modular::MessageReaderPtr message_receiver_;

  // When a |fuchsia::modular::MessageQueue| connection closes, the
  // corresponding MessageQueueConnection instance gets removed.
  fidl::BindingSet<fuchsia::modular::MessageQueue,
                   std::unique_ptr<MessageQueueConnection>>
      message_queue_bindings_;

  fidl::BindingSet<fuchsia::modular::MessageSender> message_sender_bindings_;
};

// MessageQueueConnection -----------------------------------------------------

MessageQueueConnection::MessageQueueConnection(
    MessageQueueStorage* const queue_storage)
    : queue_storage_(queue_storage) {}

MessageQueueConnection::~MessageQueueConnection() = default;

void MessageQueueConnection::RegisterReceiver(
    fidl::InterfaceHandle<fuchsia::modular::MessageReader> receiver) {
  queue_storage_->RegisterReceiver(std::move(receiver));
}

void MessageQueueConnection::GetToken(GetTokenCallback callback) {
  callback(queue_storage_->queue_token());
}

// MessageQueueManager --------------------------------------------------------

namespace {

std::string GenerateQueueToken() {
  // Get 256 bits of pseudo-randomness.
  constexpr size_t kBitCount = 256;
  constexpr size_t kBitsPerByte = 8;
  constexpr size_t kCharsPerByte = 2;
  constexpr size_t kByteCount = kBitCount / kBitsPerByte;
  constexpr char kHex[] = "0123456789ABCDEF";
  uint8_t bytes[kByteCount] = {};
  zx_cprng_draw(bytes, kByteCount);
  std::string token(kByteCount * kCharsPerByte, '\0');
  for (size_t i = 0; i < kByteCount; ++i) {
    uint8_t byte = bytes[i];
    token[2 * i] = kHex[byte & 0x0F];
    token[2 * i + 1] = kHex[byte / 0x10];
  }
  return token;
}

}  // namespace

class MessageQueueManager::GetQueueTokenCall
    : public PageOperation<fidl::StringPtr> {
 public:
  GetQueueTokenCall(fuchsia::ledger::Page* const page,
                    std::string component_namespace,
                    std::string component_instance_id,
                    const std::string& queue_name, ResultCall result_call)
      : PageOperation("MessageQueueManager::GetQueueTokenCall", page,
                      std::move(result_call), queue_name),
        component_namespace_(std::move(component_namespace)),
        component_instance_id_(std::move(component_instance_id)),
        queue_name_(queue_name) {}

 private:
  void Run() override {
    FlowToken flow{this, &result_};

    snapshot_.set_error_handler([](zx_status_t status) {
      FXL_LOG(WARNING) << "Error on snapshot connection";
    });
    page()->GetSnapshot(snapshot_.NewRequest(),
                        fidl::VectorPtr<uint8_t>::New(0), nullptr);

    key_ = MakeMessageQueueTokenKey(component_namespace_,
                                    component_instance_id_, queue_name_);
    snapshot_->Get(
        to_array(key_),
        [this, flow](fuchsia::ledger::PageSnapshot_Get_Result result) {
          if (result.is_err() &&
              result.err() == fuchsia::ledger::Error::KEY_NOT_FOUND) {
            // Key wasn't found, that's not an error.
            return;
          }

          if (result.is_err()) {
            FXL_LOG(ERROR) << trace_name() << " " << key_ << " "
                           << "PageSnapshot.Get() "
                           << fidl::ToUnderlying(result.err());
            return;
          }

          std::string queue_token;
          if (!fsl::StringFromVmo(result.response().buffer, &queue_token)) {
            FXL_LOG(ERROR) << trace_name() << " " << key_ << " "
                           << "VMO could not be copied.";
            return;
          }
          result_ = queue_token;
        });
  }

  const std::string component_namespace_;
  const std::string component_instance_id_;
  const std::string queue_name_;
  fuchsia::ledger::PageSnapshotPtr snapshot_;
  std::string key_;

  fidl::StringPtr result_;
};

class MessageQueueManager::GetMessageSenderCall : public PageOperation<> {
 public:
  GetMessageSenderCall(
      MessageQueueManager* const message_queue_manager,
      fuchsia::ledger::Page* const page, std::string token,
      fidl::InterfaceRequest<fuchsia::modular::MessageSender> request)
      : PageOperation("MessageQueueManager::GetMessageSenderCall", page, [] {}),
        message_queue_manager_(message_queue_manager),
        token_(std::move(token)),
        request_(std::move(request)) {}

 private:
  void Run() override {
    FlowToken flow{this};

    page()->GetSnapshot(snapshot_.NewRequest(), std::vector<uint8_t>(),
                        nullptr);
    std::string key = MakeMessageQueueKey(token_);
    snapshot_->Get(
        to_array(key),
        [this, flow](fuchsia::ledger::PageSnapshot_Get_Result result) {
          if (result.is_err()) {
            if (result.err() != fuchsia::ledger::Error::KEY_NOT_FOUND) {
              // It's expected that the key is not found when the link
              // is accessed for the first time. Don't log an error
              // then.
              FXL_LOG(ERROR)
                  << trace_name() << " " << token_ << " "
                  << "PageSnapshot.Get() " << fidl::ToUnderlying(result.err());
            }
            return;
          }

          std::string value_as_string;
          if (result.is_response()) {
            if (!fsl::StringFromVmo(result.response().buffer,
                                    &value_as_string)) {
              FXL_LOG(ERROR) << trace_name() << " " << token_ << " "
                             << "VMO could not be copied.";
              return;
            }
          }

          if (!XdrRead(value_as_string, &result_, XdrMessageQueueInfo)) {
            return;
          }

          if (!result_.is_complete()) {
            FXL_LOG(WARNING) << trace_name() << " " << token_ << " "
                             << "Queue token not found in the ledger.";
            return;
          }

          message_queue_manager_->GetMessageQueueStorage(result_)
              ->AddMessageSenderBinding(std::move(request_));
        });
  }

  MessageQueueManager* const message_queue_manager_;  // not owned
  const std::string token_;
  fidl::InterfaceRequest<fuchsia::modular::MessageSender> request_;

  fuchsia::ledger::PageSnapshotPtr snapshot_;
  std::string key_;

  MessageQueueInfo result_;
};

class MessageQueueManager::ObtainMessageQueueCall : public PageOperation<> {
 public:
  ObtainMessageQueueCall(
      MessageQueueManager* const message_queue_manager,
      fuchsia::ledger::Page* const page, const std::string& component_namespace,
      const std::string& component_instance_id, const std::string& queue_name,
      fidl::InterfaceRequest<fuchsia::modular::MessageQueue> request)
      : PageOperation(
            "MessageQueueManager::ObtainMessageQueueCall", page, [] {},
            queue_name),
        message_queue_manager_(message_queue_manager),
        request_(std::move(request)) {
    message_queue_info_.component_namespace = component_namespace;
    message_queue_info_.component_instance_id = component_instance_id;
    message_queue_info_.queue_name = queue_name;
  }

 private:
  void Run() override {
    FlowToken flow{this};

    operation_collection_.Add(std::make_unique<GetQueueTokenCall>(
        page(), message_queue_info_.component_namespace,
        message_queue_info_.component_instance_id,
        message_queue_info_.queue_name, [this, flow](fidl::StringPtr token) {
          if (token) {
            // Queue token was found in the ledger.
            message_queue_info_.queue_token = token.get();
            Finish(flow);
            return;
          }

          Cont(flow);
        }));
  }

  void Cont(FlowToken flow) {
    // Not found in the ledger, time to create a new message
    // queue.
    message_queue_info_.queue_token = GenerateQueueToken();

    page()->StartTransaction();
    const std::string message_queue_token_key =
        MakeMessageQueueTokenKey(message_queue_info_.component_namespace,
                                 message_queue_info_.component_instance_id,
                                 message_queue_info_.queue_name);

    page()->Put(to_array(message_queue_token_key),
                to_array(message_queue_info_.queue_token));

    const std::string message_queue_key =
        MakeMessageQueueKey(message_queue_info_.queue_token);

    std::string json;
    XdrWrite(&json, &message_queue_info_, XdrMessageQueueInfo);

    page()->Put(to_array(message_queue_key), to_array(json));

    page()->Commit();
    FXL_LOG(INFO) << trace_name() << " "
                  << "Created message queue: "
                  << message_queue_info_.queue_token;

    Finish(flow);
  }

  void Finish(FlowToken /*flow*/) {
    message_queue_manager_->GetMessageQueueStorage(message_queue_info_)
        ->AddMessageQueueBinding(std::move(request_));
  }

  MessageQueueManager* const message_queue_manager_;  // not owned
  fidl::InterfaceRequest<fuchsia::modular::MessageQueue> request_;

  MessageQueueInfo message_queue_info_;
  fuchsia::ledger::PageSnapshotPtr snapshot_;

  OperationCollection operation_collection_;
};

class MessageQueueManager::DeleteMessageQueueCall : public PageOperation<> {
 public:
  DeleteMessageQueueCall(MessageQueueManager* const message_queue_manager,
                         fuchsia::ledger::Page* const page,
                         const std::string& component_namespace,
                         const std::string& component_instance_id,
                         const std::string& queue_name)
      : PageOperation(
            "MessageQueueManager::DeleteMessageQueueCall", page, [] {},
            queue_name),
        message_queue_manager_(message_queue_manager) {
    message_queue_info_.component_namespace = component_namespace;
    message_queue_info_.component_instance_id = component_instance_id;
    message_queue_info_.queue_name = queue_name;
  }

 private:
  void Run() override {
    FlowToken flow{this};

    operation_collection_.Add(std::make_unique<GetQueueTokenCall>(
        page(), message_queue_info_.component_namespace,
        message_queue_info_.component_instance_id,
        message_queue_info_.queue_name, [this, flow](fidl::StringPtr token) {
          if (!token) {
            FXL_LOG(WARNING)
                << trace_name() << " " << message_queue_info_.queue_name << " "
                << "Request to delete queue not found in ledger"
                << " for component instance "
                << message_queue_info_.component_instance_id << ".";
            return;
          }

          message_queue_info_.queue_token = token.get();

          std::string message_queue_key =
              MakeMessageQueueKey(message_queue_info_.queue_token);

          std::string message_queue_token_key = MakeMessageQueueTokenKey(
              message_queue_info_.component_namespace,
              message_queue_info_.component_instance_id,
              message_queue_info_.queue_name);

          // Delete the ledger entries.
          page()->StartTransaction();

          page()->Delete(to_array(message_queue_key));

          page()->Delete(to_array(message_queue_token_key));

          message_queue_manager_->ClearMessageQueueStorage(message_queue_info_);

          page()->Commit();

          FXL_LOG(INFO) << trace_name() << " "
                        << "Deleted message queue: "
                        << message_queue_info_.component_instance_id << "/"
                        << message_queue_info_.queue_name;
        }));
  }

  MessageQueueManager* const message_queue_manager_;  // not owned
  MessageQueueInfo message_queue_info_;
  fuchsia::ledger::PageSnapshotPtr snapshot_;

  OperationCollection operation_collection_;
};

class MessageQueueManager::DeleteNamespaceCall : public PageOperation<> {
 public:
  DeleteNamespaceCall(MessageQueueManager* const message_queue_manager,
                      fuchsia::ledger::Page* const page,
                      const std::string& component_namespace,
                      fit::function<void()> done)
      : PageOperation("MessageQueueManager::DeleteNamespaceCall", page,
                      std::move(done), component_namespace),
        message_queue_manager_(message_queue_manager),
        component_namespace_(component_namespace),
        message_queues_key_prefix_(
            MakeMessageQueuesPrefix(component_namespace)) {}

 private:
  void Run() override {
    FlowToken flow{this};
    page()->GetSnapshot(snapshot_.NewRequest(),
                        to_array(message_queues_key_prefix_), nullptr);
    GetEntries(snapshot_.get(), &component_entries_,
               [this, flow] { ProcessKeysToDelete(flow); });
  }

  void ProcessKeysToDelete(FlowToken flow) {
    std::vector<std::string> keys_to_delete;
    for (const auto& entry : component_entries_) {
      FXL_DCHECK(entry.value) << "Value vmo handle is null";

      keys_to_delete.push_back(to_string(entry.key));

      std::string queue_token;
      if (!fsl::StringFromVmo(*entry.value, &queue_token)) {
        FXL_LOG(ERROR) << trace_name() << " " << to_string(entry.key)
                       << "VMO could not be copied.";
        continue;
      }

      keys_to_delete.push_back(MakeMessageQueueKey(queue_token));
    }

    for (auto& i : keys_to_delete) {
      page()->Delete(to_array(i));
    }

    message_queue_manager_->ClearMessageQueueStorage(component_namespace_);
  }

  MessageQueueManager* const message_queue_manager_;  // not owned
  fuchsia::ledger::PageSnapshotPtr snapshot_;
  const std::string component_namespace_;
  const std::string message_queues_key_prefix_;
  std::vector<fuchsia::ledger::Entry> component_entries_;
};

MessageQueueManager::MessageQueueManager(LedgerClient* const ledger_client,
                                         fuchsia::ledger::PageId page_id,
                                         std::string local_path)
    : PageClient("MessageQueueManager", ledger_client, std::move(page_id)),
      local_path_(std::move(local_path)) {}

MessageQueueManager::~MessageQueueManager() = default;

void MessageQueueManager::ObtainMessageQueue(
    const std::string& component_namespace,
    const std::string& component_instance_id, const std::string& queue_name,
    fidl::InterfaceRequest<fuchsia::modular::MessageQueue> request) {
  operation_collection_.Add(std::make_unique<ObtainMessageQueueCall>(
      this, page(), component_namespace, component_instance_id, queue_name,
      std::move(request)));
}

template <typename ValueType>
typename std::map<std::string, ValueType>*
MessageQueueManager::FindComponentQueues(
    ComponentQueueNameMap<ValueType>& queue_map, const MessageQueueInfo& info) {
  auto it1 = queue_map.find(info.component_namespace);
  if (it1 != queue_map.end()) {
    auto it2 = it1->second.find(info.component_instance_id);
    if (it2 != it1->second.end()) {
      return &(it2->second);
    }
  }

  return nullptr;
}

template <typename ValueType>
const ValueType* MessageQueueManager::FindQueueName(
    ComponentQueueNameMap<ValueType>& queue_map, const MessageQueueInfo& info) {
  auto map = FindComponentQueues(queue_map, info);
  if (map) {
    auto it3 = map->find(info.queue_name);
    if (it3 != map->end()) {
      return &(it3->second);
    }
  }
  return nullptr;
}

template <typename Callable>
fit::function<Callable> MessageQueueManager::ReMoveByQueueName(
    ComponentQueueNameMap<fit::function<Callable>>& queue_map,
    const MessageQueueInfo& info) {
  auto map = FindComponentQueues(queue_map, info);
  if (map) {
    auto it3 = map->find(info.queue_name);
    if (it3 != map->end()) {
      auto found = std::move(it3->second);
      map->erase(info.queue_name);
      return found;
    }
  }
  return nullptr;
}

template <typename ValueType>
void MessageQueueManager::EraseQueueName(
    ComponentQueueNameMap<ValueType>& queue_map, const MessageQueueInfo& info) {
  auto it1 = queue_map.find(info.component_namespace);
  if (it1 != queue_map.end()) {
    auto it2 = it1->second.find(info.component_instance_id);
    if (it2 != it1->second.end()) {
      it2->second.erase(info.queue_name);
    }
  }
}

template <typename ValueType>
void MessageQueueManager::EraseNamespace(
    ComponentQueueNameMap<ValueType>& queue_map,
    const std::string& component_namespace) {
  auto it1 = queue_map.find(component_namespace);
  if (it1 != queue_map.end()) {
    it1->second.erase(component_namespace);
  }
}

MessageQueueStorage* MessageQueueManager::GetMessageQueueStorage(
    const MessageQueueInfo& info) {
  auto it = message_queues_.find(info.queue_token);
  if (it == message_queues_.end()) {
    // Not found, create one.
    bool inserted;
    std::string path = local_path_;
    path.push_back('/');
    path.append(info.queue_token);
    path.append(".json");
    auto new_queue = std::make_unique<MessageQueueStorage>(
        info.queue_name, info.queue_token, std::move(path));
    std::tie(it, inserted) = message_queues_.insert(
        std::make_pair(info.queue_token, std::move(new_queue)));
    FXL_DCHECK(inserted);

    message_queue_infos_[info.queue_token] = info;

    message_queue_tokens_[info.component_namespace][info.component_instance_id]
                         [info.queue_name] = info.queue_token;

    fit::function<void()> watcher =
        ReMoveByQueueName(pending_watcher_callbacks_, info);
    if (watcher) {
      EraseQueueName(pending_watcher_callbacks_, info);
      it->second->RegisterWatcher(std::move(watcher));
    }
  }
  return it->second.get();
}

void MessageQueueManager::ClearMessageQueueStorage(
    const MessageQueueInfo& info) {
  // Remove the |MessageQueueStorage| and delete it which in turn will
  // close all outstanding fuchsia::modular::MessageSender and
  // fuchsia::modular::MessageQueue interface connections, and delete all
  // messages on the queue permanently.
  message_queues_.erase(info.queue_token);

  // Clear entries in message_queue_tokens_ and
  // pending_watcher_callbacks_.
  EraseQueueName(pending_watcher_callbacks_, info);
  EraseQueueName(message_queue_tokens_, info);

  auto deletion_watchers = FindQueueName(deletion_watchers_, info);
  for (const auto& component_namespace_iter : *deletion_watchers) {
    for (const auto& component_instance_iter :
         component_namespace_iter.second) {
      component_instance_iter.second();
    }
  }

  EraseQueueName(deletion_watchers_, info);
}

void MessageQueueManager::ClearMessageQueueStorage(
    const std::string& component_namespace) {
  const auto& namespace_to_delete = deletion_watchers_[component_namespace];
  for (const auto& instances_in_namespace : namespace_to_delete) {
    for (const auto& queue_names : instances_in_namespace.second) {
      for (const auto& watcher_namespace : queue_names.second) {
        for (const auto& watcher_instance : watcher_namespace.second) {
          watcher_instance.second();
        }
      }
    }
  }

  EraseNamespace(pending_watcher_callbacks_, component_namespace);
  EraseNamespace(message_queue_tokens_, component_namespace);
  EraseNamespace(deletion_watchers_, component_namespace);
}

void MessageQueueManager::DeleteMessageQueue(
    const std::string& component_namespace,
    const std::string& component_instance_id, const std::string& queue_name) {
  operation_collection_.Add(std::make_unique<DeleteMessageQueueCall>(
      this, page(), component_namespace, component_instance_id, queue_name));
}

void MessageQueueManager::DeleteNamespace(
    const std::string& component_namespace, fit::function<void()> done) {
  operation_collection_.Add(std::make_unique<DeleteNamespaceCall>(
      this, page(), component_namespace, std::move(done)));
}

void MessageQueueManager::GetMessageSender(
    const std::string& queue_token,
    fidl::InterfaceRequest<fuchsia::modular::MessageSender> request) {
  const auto& it = message_queues_.find(queue_token);
  if (it != message_queues_.cend()) {
    // Found the message queue already.
    it->second->AddMessageSenderBinding(std::move(request));
    return;
  }

  operation_collection_.Add(std::make_unique<GetMessageSenderCall>(
      this, page(), queue_token, std::move(request)));
}

void MessageQueueManager::RegisterMessageWatcher(
    const std::string& component_namespace,
    const std::string& component_instance_id, const std::string& queue_name,
    fit::function<void()> watcher) {
  const std::string* const token =
      FindQueueName(message_queue_tokens_,
                    MessageQueueInfo{component_namespace, component_instance_id,
                                     queue_name, ""});
  if (!token) {
    pending_watcher_callbacks_[component_namespace][component_instance_id]
        .emplace(queue_name, std::move(watcher));
    return;
  }

  auto msq_it = message_queues_.find(*token);
  FXL_DCHECK(msq_it != message_queues_.end());
  msq_it->second->RegisterWatcher(std::move(watcher));
}

void MessageQueueManager::RegisterDeletionWatcher(
    const std::string& component_namespace,
    const std::string& component_instance_id, const std::string& queue_token,
    fit::function<void()> watcher) {
  auto it = message_queue_infos_.find(queue_token);
  if (it == message_queue_infos_.end()) {
    return;
  }

  const MessageQueueInfo queue_info = it->second;

  deletion_watchers_[queue_info.component_namespace]
                    [queue_info.component_instance_id][queue_info.queue_name]
                    [component_namespace]
                        .emplace(component_instance_id, std::move(watcher));
}

void MessageQueueManager::DropMessageWatcher(
    const std::string& component_namespace,
    const std::string& component_instance_id, const std::string& queue_name) {
  MessageQueueInfo queue_info{component_namespace, component_instance_id,
                              queue_name, ""};
  const std::string* const token =
      FindQueueName(message_queue_tokens_, queue_info);
  if (token) {
    // The |MessageQueueStorage| doesn't exist yet.
    EraseQueueName(message_queue_tokens_, queue_info);
    return;
  }

  auto msq_it = message_queues_.find(*token);
  if (msq_it == message_queues_.end()) {
    FXL_LOG(WARNING) << "Asked to DropWatcher for a queue that doesn't exist";
    return;
  }
  msq_it->second->DropWatcher();
}

void MessageQueueManager::DropDeletionWatcher(
    const std::string& watcher_namespace,
    const std::string& watcher_instance_id, const std::string& queue_token) {
  auto it = message_queue_infos_.find(queue_token);
  if (it == message_queue_infos_.end()) {
    return;
  }

  const MessageQueueInfo queue_info = it->second;
  deletion_watchers_[queue_info.component_namespace]
                    [queue_info.component_instance_id][queue_info.queue_name]
                    [watcher_namespace]
                        .erase(watcher_instance_id);
}

}  // namespace modular