<commit_before>/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
/*
 *     Copyright 2020 Couchbase, Inc.
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 */

#pragma once

#include <utility>

#include <tao/json.hpp>

#include <asio.hpp>

#include <platform/uuid.h>

#include <io/binary_message.hxx>
#include <io/binary_parser.hxx>

#include <protocol/hello_feature.hxx>
#include <protocol/client_request.hxx>
#include <protocol/client_response.hxx>
#include <protocol/server_request.hxx>
#include <protocol/cmd_hello.hxx>
#include <protocol/cmd_sasl_list_mechs.hxx>
#include <protocol/cmd_sasl_auth.hxx>
#include <protocol/cmd_sasl_step.hxx>
#include <protocol/cmd_select_bucket.hxx>
#include <protocol/cmd_get_cluster_config.hxx>
#include <protocol/cmd_get_error_map.hxx>
#include <protocol/cmd_get_collections_manifest.hxx>
#include <protocol/cmd_get.hxx>
#include <protocol/cmd_cluster_map_change_notification.hxx>

#include <cbsasl/client.h>

#include <spdlog/fmt/bin_to_hex.h>

#include <errors.hxx>
#include <version.hxx>

namespace couchbase::io
{

class key_value_session : public std::enable_shared_from_this<key_value_session>
{
    class message_handler
    {
      public:
        virtual void handle(binary_message&& msg) = 0;

        virtual ~message_handler() = default;

        virtual void stop()
        {
        }
    };

    class bootstrap_handler : public message_handler
    {
      private:
        std::shared_ptr<key_value_session> session_;
        sasl::ClientContext sasl_;

      public:
        bootstrap_handler(bootstrap_handler&&) = default;
        ~bootstrap_handler() override = default;

        explicit bootstrap_handler(std::shared_ptr<key_value_session> session)
          : session_(std::move(session))
          , sasl_([this]() -> std::string { return session_->username_; },
                  [this]() -> std::string { return session_->password_; },
                  { "SCRAM-SHA512", "SCRAM-SHA256", "SCRAM-SHA1", "PLAIN" })
        {
            tao::json::value user_agent{
                { "a", fmt::format("ruby_sdk/{}.{}.{}", BACKEND_VERSION_MAJOR, BACKEND_VERSION_MINOR, BACKEND_VERSION_PATCH) },
                { "i", fmt::format("{}/{}", uuid::to_string(session_->client_id_), uuid::to_string(session_->id_)) }
            };
            protocol::client_request<protocol::hello_request_body> hello_req;
            hello_req.opaque(session_->next_opaque());
            hello_req.body().user_agent(tao::json::to_string(user_agent));
            session_->write(hello_req.data());

            protocol::client_request<protocol::sasl_list_mechs_request_body> list_req;
            list_req.opaque(session_->next_opaque());
            session_->write(list_req.data());

            protocol::client_request<protocol::sasl_auth_request_body> auth_req;
            sasl::error sasl_code;
            std::string_view sasl_payload;
            std::tie(sasl_code, sasl_payload) = sasl_.start();
            auth_req.opaque(session_->next_opaque());
            auth_req.body().mechanism(sasl_.get_name());
            auth_req.body().sasl_data(sasl_payload);
            session_->write(auth_req.data());

            session_->flush();
        }

        void complete(std::error_code ec)
        {
            session_->invoke_bootstrap_handler(ec);
            if (!ec) {
                session_->handler_ = std::make_unique<normal_handler>(session_);
            }
        }

        void auth_success()
        {
            spdlog::debug("{} authentication successful", sasl_.get_name());
            session_->authenticated_ = true;
            if (session_->supports_feature(protocol::hello_feature::xerror)) {
                protocol::client_request<protocol::get_error_map_request_body> errmap_req;
                errmap_req.opaque(session_->next_opaque());
                session_->write(errmap_req.data());
            }
            if (session_->bucket_name_) {
                protocol::client_request<protocol::select_bucket_request_body> sb_req;
                sb_req.opaque(session_->next_opaque());
                sb_req.body().bucket_name(session_->bucket_name_.value());
                session_->write(sb_req.data());

                protocol::client_request<protocol::get_collections_manifest_request_body> gcm_req;
                gcm_req.opaque(session_->next_opaque());
                session_->write(gcm_req.data());
            }
            protocol::client_request<protocol::get_cluster_config_request_body> cfg_req;
            cfg_req.opaque(session_->next_opaque());
            session_->write(cfg_req.data());
            session_->flush();
        }

        void handle(binary_message&& msg) override
        {
            Expects(protocol::is_valid_client_opcode(msg.header.opcode));
            auto opcode = static_cast<protocol::client_opcode>(msg.header.opcode);
            switch (opcode) {
                case protocol::client_opcode::hello: {
                    protocol::client_response<protocol::hello_response_body> resp(msg);
                    if (resp.status() == protocol::status::success) {
                        session_->supported_features_ = resp.body().supported_features();
                    } else {
                        spdlog::warn("unexpected message status during bootstrap: {}", resp.error_message());
                        return complete(std::make_error_code(error::network_errc::handshake_failure));
                    }
                } break;
                case protocol::client_opcode::sasl_list_mechs: {
                    protocol::client_response<protocol::sasl_list_mechs_response_body> resp(msg);
                    if (resp.status() != protocol::status::success) {
                        spdlog::warn("unexpected message status during bootstrap: {}", resp.error_message());
                        return complete(std::make_error_code(error::common_errc::authentication_failure));
                    }
                } break;
                case protocol::client_opcode::sasl_auth: {
                    protocol::client_response<protocol::sasl_auth_response_body> resp(msg);
                    if (resp.status() == protocol::status::success) {
                        return auth_success();
                    }
                    if (resp.status() == protocol::status::auth_continue) {
                        sasl::error sasl_code;
                        std::string_view sasl_payload;
                        std::tie(sasl_code, sasl_payload) = sasl_.step(resp.body().value());
                        if (sasl_code == sasl::error::OK) {
                            return auth_success();
                        }
                        if (sasl_code == sasl::error::CONTINUE) {
                            protocol::client_request<protocol::sasl_step_request_body> req;
                            req.opaque(session_->next_opaque());
                            req.body().mechanism(sasl_.get_name());
                            req.body().sasl_data(sasl_payload);
                            session_->write_and_flush(req.data());
                        } else {
                            spdlog::error("unable to authenticate: sasl_code={}", sasl_code);
                            return complete(std::make_error_code(error::common_errc::authentication_failure));
                        }
                    } else {
                        spdlog::warn("unexpected message status during bootstrap: {}", resp.error_message());
                        return complete(std::make_error_code(error::common_errc::authentication_failure));
                    }
                } break;
                case protocol::client_opcode::sasl_step: {
                    protocol::client_response<protocol::sasl_step_response_body> resp(msg);
                    if (resp.status() == protocol::status::success) {
                        return auth_success();
                    }
                    return complete(std::make_error_code(error::common_errc::authentication_failure));
                }
                case protocol::client_opcode::get_error_map: {
                    protocol::client_response<protocol::get_error_map_response_body> resp(msg);
                    if (resp.status() == protocol::status::success) {
                        session_->errmap_.emplace(resp.body().errmap());
                    } else {
                        spdlog::warn("unexpected message status during bootstrap: {}", resp.error_message());
                        return complete(std::make_error_code(error::network_errc::protocol_error));
                    }
                } break;
                case protocol::client_opcode::get_collections_manifest: {
                    protocol::client_response<protocol::get_collections_manifest_response_body> resp(msg);
                    if (resp.status() == protocol::status::success) {
                        session_->manifest_.emplace(resp.body().manifest());
                        spdlog::trace(
                          "collections manifest for bucket \"{}\": {}", session_->bucket_name_.value_or(""), *session_->manifest_);
                    } else if (resp.status() == protocol::status::no_collections_manifest) {
                        spdlog::trace("collection manifest is not available for bucket \"{}\": {}",
                                      session_->bucket_name_.value_or(""),
                                      resp.error_message());
                    } else {
                        spdlog::warn("unexpected message status during bootstrap: {}", resp.error_message());
                        return complete(std::make_error_code(error::network_errc::protocol_error));
                    }
                } break;
                case protocol::client_opcode::select_bucket: {
                    protocol::client_response<protocol::select_bucket_response_body> resp(msg);
                    if (resp.status() == protocol::status::success) {
                        spdlog::trace("selected bucket: {}", session_->bucket_name_.value_or(""));
                        session_->bucket_selected_ = true;
                    } else if (resp.status() == protocol::status::no_access) {
                        spdlog::trace("unable to select bucket: {}, probably the bucket does not exist",
                                      session_->bucket_name_.value_or(""));
                        session_->bucket_selected_ = false;
                        return complete(std::make_error_code(error::common_errc::bucket_not_found));
                    } else {
                        spdlog::warn("unexpected message status during bootstrap: {}", resp.error_message());
                        return complete(std::make_error_code(error::common_errc::bucket_not_found));
                    }
                } break;
                case protocol::client_opcode::get_cluster_config: {
                    protocol::client_response<protocol::get_cluster_config_response_body> resp(msg);
                    if (resp.status() == protocol::status::success) {
                        session_->update_configuration(resp.body().config());
                        complete({});
                    } else {
                        spdlog::warn("unexpected message status during bootstrap: {}", resp.error_message());
                        return complete(std::make_error_code(error::network_errc::protocol_error));
                    }
                } break;
                default:
                    spdlog::trace("unexpected message during bootstrap: {}", opcode);
                    return complete(std::make_error_code(error::network_errc::protocol_error));
            }
        }
    };

    class normal_handler : public message_handler
    {
      private:
        std::shared_ptr<key_value_session> session_;
        asio::steady_timer heartbeat_timer_;
        bool stopped_{ false };

      public:
        normal_handler(normal_handler&&) = default;
        ~normal_handler() override = default;

        explicit normal_handler(std::shared_ptr<key_value_session> session)
          : session_(session)
          , heartbeat_timer_(session_->ctx_)
        {
            fetch_config({});
        }

        void stop() override
        {
            if (!stopped_) {
                heartbeat_timer_.cancel();
                session_.reset();
            }
        }

        void handle(binary_message&& msg) override
        {
            if (stopped_ || !session_) {
                return;
            }
            Expects(protocol::is_valid_magic(msg.header.magic));
            switch (auto magic = static_cast<protocol::magic>(msg.header.magic)) {
                case protocol::magic::client_response:
                    Expects(protocol::is_valid_client_opcode(msg.header.opcode));
                    switch (auto opcode = static_cast<protocol::client_opcode>(msg.header.opcode)) {
                        case protocol::client_opcode::get_cluster_config: {
                            protocol::client_response<protocol::get_cluster_config_response_body> resp(msg);
                            if (resp.status() == protocol::status::success) {
                                if (session_) {
                                    session_->update_configuration(resp.body().config());
                                }
                            } else {
                                spdlog::warn("unexpected message status: {}", resp.error_message());
                            }
                        } break;
                        case protocol::client_opcode::get:
                        case protocol::client_opcode::upsert:
                        case protocol::client_opcode::remove:
                        case protocol::client_opcode::subdoc_multi_lookup:
                        case protocol::client_opcode::subdoc_multi_mutation: {
                            std::uint32_t opaque = msg.header.opaque;
                            std::uint16_t status = ntohs(msg.header.specific);
                            auto handler = session_->command_handlers_.find(opaque);
                            if (handler != session_->command_handlers_.end()) {
                                handler->second(session_->map_status_code(opcode, status), std::move(msg));
                                session_->command_handlers_.erase(handler);
                            } else {
                                spdlog::trace("unexpected orphan response opcode={}, opaque={}", msg.header.opcode, msg.header.opaque);
                            }
                        } break;
                        default:
                            spdlog::trace("unexpected client response: {}", opcode);
                    }
                    break;
                case protocol::magic::server_request:
                    Expects(protocol::is_valid_server_request_opcode(msg.header.opcode));
                    switch (auto opcode = static_cast<protocol::server_opcode>(msg.header.opcode)) {
                        case protocol::server_opcode::cluster_map_change_notification: {
                            protocol::server_request<protocol::cluster_map_change_notification_request_body> req(msg);
                            if (session_) {
                                if (session_->bucket_name_->empty()) {
                                    if (req.body().config().bucket->empty()) {
                                        session_->update_configuration(req.body().config());
                                    }
                                } else if (session_->bucket_name_.value() == req.body().config().bucket) {
                                    session_->update_configuration(req.body().config());
                                }
                            }
                        } break;
                        default:
                            spdlog::trace("unexpected server request: {}", opcode);
                    }
                    break;
                case protocol::magic::client_request:
                case protocol::magic::alt_client_request:
                case protocol::magic::alt_client_response:
                case protocol::magic::server_response:
                    spdlog::trace("unexpected magic: {}, opcode={}, opaque={}{}", magic, msg.header.opcode, msg.header.opaque);
                    break;
            }
        }

        void fetch_config(std::error_code ec)
        {
            if (ec == asio::error::operation_aborted) {
                return;
            }
            if (stopped_ || !session_) {<commit_msg>🚑 fix check for bucket name when using config<commit_after>/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
/*
 *     Copyright 2020 Couchbase, Inc.
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 */

#pragma once

#include <utility>

#include <tao/json.hpp>

#include <asio.hpp>

#include <platform/uuid.h>

#include <io/binary_message.hxx>
#include <io/binary_parser.hxx>

#include <protocol/hello_feature.hxx>
#include <protocol/client_request.hxx>
#include <protocol/client_response.hxx>
#include <protocol/server_request.hxx>
#include <protocol/cmd_hello.hxx>
#include <protocol/cmd_sasl_list_mechs.hxx>
#include <protocol/cmd_sasl_auth.hxx>
#include <protocol/cmd_sasl_step.hxx>
#include <protocol/cmd_select_bucket.hxx>
#include <protocol/cmd_get_cluster_config.hxx>
#include <protocol/cmd_get_error_map.hxx>
#include <protocol/cmd_get_collections_manifest.hxx>
#include <protocol/cmd_get.hxx>
#include <protocol/cmd_cluster_map_change_notification.hxx>

#include <cbsasl/client.h>

#include <spdlog/fmt/bin_to_hex.h>

#include <errors.hxx>
#include <version.hxx>

namespace couchbase::io
{

class key_value_session : public std::enable_shared_from_this<key_value_session>
{
    class message_handler
    {
      public:
        virtual void handle(binary_message&& msg) = 0;

        virtual ~message_handler() = default;

        virtual void stop()
        {
        }
    };

    class bootstrap_handler : public message_handler
    {
      private:
        std::shared_ptr<key_value_session> session_;
        sasl::ClientContext sasl_;

      public:
        bootstrap_handler(bootstrap_handler&&) = default;
        ~bootstrap_handler() override = default;

        explicit bootstrap_handler(std::shared_ptr<key_value_session> session)
          : session_(std::move(session))
          , sasl_([this]() -> std::string { return session_->username_; },
                  [this]() -> std::string { return session_->password_; },
                  { "SCRAM-SHA512", "SCRAM-SHA256", "SCRAM-SHA1", "PLAIN" })
        {
            tao::json::value user_agent{
                { "a", fmt::format("ruby_sdk/{}.{}.{}", BACKEND_VERSION_MAJOR, BACKEND_VERSION_MINOR, BACKEND_VERSION_PATCH) },
                { "i", fmt::format("{}/{}", uuid::to_string(session_->client_id_), uuid::to_string(session_->id_)) }
            };
            protocol::client_request<protocol::hello_request_body> hello_req;
            hello_req.opaque(session_->next_opaque());
            hello_req.body().user_agent(tao::json::to_string(user_agent));
            session_->write(hello_req.data());

            protocol::client_request<protocol::sasl_list_mechs_request_body> list_req;
            list_req.opaque(session_->next_opaque());
            session_->write(list_req.data());

            protocol::client_request<protocol::sasl_auth_request_body> auth_req;
            sasl::error sasl_code;
            std::string_view sasl_payload;
            std::tie(sasl_code, sasl_payload) = sasl_.start();
            auth_req.opaque(session_->next_opaque());
            auth_req.body().mechanism(sasl_.get_name());
            auth_req.body().sasl_data(sasl_payload);
            session_->write(auth_req.data());

            session_->flush();
        }

        void complete(std::error_code ec)
        {
            session_->invoke_bootstrap_handler(ec);
            if (!ec) {
                session_->handler_ = std::make_unique<normal_handler>(session_);
            }
        }

        void auth_success()
        {
            spdlog::debug("{} authentication successful", sasl_.get_name());
            session_->authenticated_ = true;
            if (session_->supports_feature(protocol::hello_feature::xerror)) {
                protocol::client_request<protocol::get_error_map_request_body> errmap_req;
                errmap_req.opaque(session_->next_opaque());
                session_->write(errmap_req.data());
            }
            if (session_->bucket_name_) {
                protocol::client_request<protocol::select_bucket_request_body> sb_req;
                sb_req.opaque(session_->next_opaque());
                sb_req.body().bucket_name(session_->bucket_name_.value());
                session_->write(sb_req.data());

                protocol::client_request<protocol::get_collections_manifest_request_body> gcm_req;
                gcm_req.opaque(session_->next_opaque());
                session_->write(gcm_req.data());
            }
            protocol::client_request<protocol::get_cluster_config_request_body> cfg_req;
            cfg_req.opaque(session_->next_opaque());
            session_->write(cfg_req.data());
            session_->flush();
        }

        void handle(binary_message&& msg) override
        {
            Expects(protocol::is_valid_client_opcode(msg.header.opcode));
            auto opcode = static_cast<protocol::client_opcode>(msg.header.opcode);
            switch (opcode) {
                case protocol::client_opcode::hello: {
                    protocol::client_response<protocol::hello_response_body> resp(msg);
                    if (resp.status() == protocol::status::success) {
                        session_->supported_features_ = resp.body().supported_features();
                    } else {
                        spdlog::warn("unexpected message status during bootstrap: {}", resp.error_message());
                        return complete(std::make_error_code(error::network_errc::handshake_failure));
                    }
                } break;
                case protocol::client_opcode::sasl_list_mechs: {
                    protocol::client_response<protocol::sasl_list_mechs_response_body> resp(msg);
                    if (resp.status() != protocol::status::success) {
                        spdlog::warn("unexpected message status during bootstrap: {}", resp.error_message());
                        return complete(std::make_error_code(error::common_errc::authentication_failure));
                    }
                } break;
                case protocol::client_opcode::sasl_auth: {
                    protocol::client_response<protocol::sasl_auth_response_body> resp(msg);
                    if (resp.status() == protocol::status::success) {
                        return auth_success();
                    }
                    if (resp.status() == protocol::status::auth_continue) {
                        sasl::error sasl_code;
                        std::string_view sasl_payload;
                        std::tie(sasl_code, sasl_payload) = sasl_.step(resp.body().value());
                        if (sasl_code == sasl::error::OK) {
                            return auth_success();
                        }
                        if (sasl_code == sasl::error::CONTINUE) {
                            protocol::client_request<protocol::sasl_step_request_body> req;
                            req.opaque(session_->next_opaque());
                            req.body().mechanism(sasl_.get_name());
                            req.body().sasl_data(sasl_payload);
                            session_->write_and_flush(req.data());
                        } else {
                            spdlog::error("unable to authenticate: sasl_code={}", sasl_code);
                            return complete(std::make_error_code(error::common_errc::authentication_failure));
                        }
                    } else {
                        spdlog::warn("unexpected message status during bootstrap: {}", resp.error_message());
                        return complete(std::make_error_code(error::common_errc::authentication_failure));
                    }
                } break;
                case protocol::client_opcode::sasl_step: {
                    protocol::client_response<protocol::sasl_step_response_body> resp(msg);
                    if (resp.status() == protocol::status::success) {
                        return auth_success();
                    }
                    return complete(std::make_error_code(error::common_errc::authentication_failure));
                }
                case protocol::client_opcode::get_error_map: {
                    protocol::client_response<protocol::get_error_map_response_body> resp(msg);
                    if (resp.status() == protocol::status::success) {
                        session_->errmap_.emplace(resp.body().errmap());
                    } else {
                        spdlog::warn("unexpected message status during bootstrap: {}", resp.error_message());
                        return complete(std::make_error_code(error::network_errc::protocol_error));
                    }
                } break;
                case protocol::client_opcode::get_collections_manifest: {
                    protocol::client_response<protocol::get_collections_manifest_response_body> resp(msg);
                    if (resp.status() == protocol::status::success) {
                        session_->manifest_.emplace(resp.body().manifest());
                        spdlog::trace(
                          "collections manifest for bucket \"{}\": {}", session_->bucket_name_.value_or(""), *session_->manifest_);
                    } else if (resp.status() == protocol::status::no_collections_manifest) {
                        spdlog::trace("collection manifest is not available for bucket \"{}\": {}",
                                      session_->bucket_name_.value_or(""),
                                      resp.error_message());
                    } else {
                        spdlog::warn("unexpected message status during bootstrap: {}", resp.error_message());
                        return complete(std::make_error_code(error::network_errc::protocol_error));
                    }
                } break;
                case protocol::client_opcode::select_bucket: {
                    protocol::client_response<protocol::select_bucket_response_body> resp(msg);
                    if (resp.status() == protocol::status::success) {
                        spdlog::trace("selected bucket: {}", session_->bucket_name_.value_or(""));
                        session_->bucket_selected_ = true;
                    } else if (resp.status() == protocol::status::no_access) {
                        spdlog::trace("unable to select bucket: {}, probably the bucket does not exist",
                                      session_->bucket_name_.value_or(""));
                        session_->bucket_selected_ = false;
                        return complete(std::make_error_code(error::common_errc::bucket_not_found));
                    } else {
                        spdlog::warn("unexpected message status during bootstrap: {}", resp.error_message());
                        return complete(std::make_error_code(error::common_errc::bucket_not_found));
                    }
                } break;
                case protocol::client_opcode::get_cluster_config: {
                    protocol::client_response<protocol::get_cluster_config_response_body> resp(msg);
                    if (resp.status() == protocol::status::success) {
                        session_->update_configuration(resp.body().config());
                        complete({});
                    } else {
                        spdlog::warn("unexpected message status during bootstrap: {}", resp.error_message());
                        return complete(std::make_error_code(error::network_errc::protocol_error));
                    }
                } break;
                default:
                    spdlog::trace("unexpected message during bootstrap: {}", opcode);
                    return complete(std::make_error_code(error::network_errc::protocol_error));
            }
        }
    };

    class normal_handler : public message_handler
    {
      private:
        std::shared_ptr<key_value_session> session_;
        asio::steady_timer heartbeat_timer_;
        bool stopped_{ false };

      public:
        normal_handler(normal_handler&&) = default;
        ~normal_handler() override = default;

        explicit normal_handler(std::shared_ptr<key_value_session> session)
          : session_(session)
          , heartbeat_timer_(session_->ctx_)
        {
            fetch_config({});
        }

        void stop() override
        {
            if (!stopped_) {
                heartbeat_timer_.cancel();
                session_.reset();
            }
        }

        void handle(binary_message&& msg) override
        {
            if (stopped_ || !session_) {
                return;
            }
            Expects(protocol::is_valid_magic(msg.header.magic));
            switch (auto magic = static_cast<protocol::magic>(msg.header.magic)) {
                case protocol::magic::client_response:
                    Expects(protocol::is_valid_client_opcode(msg.header.opcode));
                    switch (auto opcode = static_cast<protocol::client_opcode>(msg.header.opcode)) {
                        case protocol::client_opcode::get_cluster_config: {
                            protocol::client_response<protocol::get_cluster_config_response_body> resp(msg);
                            if (resp.status() == protocol::status::success) {
                                if (session_) {
                                    session_->update_configuration(resp.body().config());
                                }
                            } else {
                                spdlog::warn("unexpected message status: {}", resp.error_message());
                            }
                        } break;
                        case protocol::client_opcode::get:
                        case protocol::client_opcode::upsert:
                        case protocol::client_opcode::remove:
                        case protocol::client_opcode::subdoc_multi_lookup:
                        case protocol::client_opcode::subdoc_multi_mutation: {
                            std::uint32_t opaque = msg.header.opaque;
                            std::uint16_t status = ntohs(msg.header.specific);
                            auto handler = session_->command_handlers_.find(opaque);
                            if (handler != session_->command_handlers_.end()) {
                                handler->second(session_->map_status_code(opcode, status), std::move(msg));
                                session_->command_handlers_.erase(handler);
                            } else {
                                spdlog::trace("unexpected orphan response opcode={}, opaque={}", msg.header.opcode, msg.header.opaque);
                            }
                        } break;
                        default:
                            spdlog::trace("unexpected client response: {}", opcode);
                    }
                    break;
                case protocol::magic::server_request:
                    Expects(protocol::is_valid_server_request_opcode(msg.header.opcode));
                    switch (auto opcode = static_cast<protocol::server_opcode>(msg.header.opcode)) {
                        case protocol::server_opcode::cluster_map_change_notification: {
                            protocol::server_request<protocol::cluster_map_change_notification_request_body> req(msg);
                            if (session_) {
                                if (session_->bucket_name_) {
                                    if (req.body().config().bucket == session_->bucket_name_.value()) {
                                        session_->update_configuration(req.body().config());
                                    }
                                } else {
                                    if (req.body().config().bucket->empty()) {
                                        session_->update_configuration(req.body().config());
                                    }
                                }
                            }
                        } break;
                        default:
                            spdlog::trace("unexpected server request: {}", opcode);
                    }
                    break;
                case protocol::magic::client_request:
                case protocol::magic::alt_client_request:
                case protocol::magic::alt_client_response:
                case protocol::magic::server_response:
                    spdlog::trace("unexpected magic: {}, opcode={}, opaque={}{}", magic, msg.header.opcode, msg.header.opaque);
                    break;
            }
        }

        void fetch_config(std::error_code ec)
        {
            if (ec == asio::error::operation_aborted) {
                return;
            }
            if (stopped_ || !session_) {
{'fraction_non_alphanumeric': 0.11346226115109993, 'fraction_numerical': 0.0012719336281906744, 'mean_word_length': 1.7817626246381473, 'pattern_counts': {'":': 4, '<': 107, '<?xml version=': 0, '>': 214, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 24, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3938133', 'n_tokens_mistral': 8958, 'n_tokens_neox': 8591, 'n_words': 1326}