// Ignore the following warnings in this header when doing Yocto builds.
#include <cstdint>
#include <cstring>
#include <string_view>

#include "absl/status/status.h"
#include "absl/strings/numbers.h"
#include "absl/strings/str_format.h"
#include "absl/strings/string_view.h"
#include "absl/time/time.h"
#include "tlbmc/subscription/manager.h"
#include "tlbmc/subscription/manager_fake.h"
#include "payload.pb.h"
#include "sensor_identifier.pb.h"
#include "sensor_payload.pb.h"

#pragma GCC diagnostic push
#pragma GCC diagnostic warning \
    "-Wdeprecated-declarations"  // There is not a good way to avoid this as we
                                 // need backward compatibility

#include <chrono>  // NOLINT
#include <cstddef>
#include <filesystem>  // NOLINT
#include <fstream>
#include <functional>
#include <iterator>
#include <map>
#include <memory>
#include <optional>
#include <queue>
#include <string>
#include <system_error>  // NOLINT
#include <thread>        // NOLINT
#include <unordered_set>
#include <utility>

#include "absl/base/thread_annotations.h"
#include "absl/container/flat_hash_map.h"
#include "absl/strings/match.h"
#include "absl/synchronization/mutex.h"
#include "absl/time/clock.h"
#include "subscription.h"
#include "nlohmann/json_fwd.hpp"
#include "tlbmc/service/hft_service.h"
#include "server.h"

#ifdef ENABLE_LOAS3_VALIDATION
#include "security/zatar/loas3_validation/validation/validation.h"
#endif

#include "absl/log/log.h"
#include "boost/asio/error.hpp"  // NOLINT
#include "boost/asio/io_context.hpp"  // NOLINT
#include "boost/asio/post.hpp"  // NOLINT
#include "boost/beast/http/verb.hpp"  // NOLINT
#include "boost/uuid/uuid.hpp"  // NOLINT
#include "boost/uuid/uuid_generators.hpp"  // NOLINT
#include "boost/uuid/uuid_io.hpp"  // NOLINT
#include "redfish_v1.grpc.pb.h"
#include "redfish_v1.pb.h"
#include "redfish_v1_grpc_include.h"
#include "zatar/struct_proto_conversion.h"
#include "grpcpp/ext/proto_server_reflection_plugin.h"
#include "grpcpp/security/auth_context.h"
#include "grpcpp/security/server_credentials.h"
#include "grpcpp/server_builder.h"
#include "grpcpp/server_context.h"
#include "grpcpp/support/server_callback.h"
#include "grpcpp/support/status.h"
#include "grpcpp/support/string_ref.h"
#include "nlohmann/json.hpp"
#include "config_parser.h"
#include "oauth_utils.h"
#include "tlbmc/hal/shared_mem/server.h"
#include "tlbmc/redfish/app.h"
#include "tlbmc/redfish/request.h"
#include "tlbmc/redfish/response.h"
#include "tlbmc/trace/tracer.h"
#include "bmcweb_authorizer_singleton.h"
#include "app.hpp"
#include "zatar/bmcweb_cert_provider.h"
#include "zatar/bmcweb_cert_provider_impl.h"
#include "zatar/bmcweb_credentials.h"

namespace milotic {

using ::milotic_tlbmc::RedfishRequest;
using ::milotic_tlbmc::RedfishResponse;
using ::milotic_tlbmc::SharedMemoryServer;

grpc::Status SubscriptionPreCheck(
    const grpc::AuthContext& auth_context,
    const std::multimap<grpc::string_ref, grpc::string_ref>& client_metadata,
    const ::redfish::v1::Request* request,
    ::milotic::redfish::BmcWebCertProvider::ServerStatus server_status) {
  ::milotic::authz::BmcWebAuthorizerSingleton& authorizer =
      ::milotic::authz::BmcWebAuthorizerSingleton::GetInstance();
  // Abort if it is not mutual authenticated
  if (server_status != ::milotic::redfish::BmcWebCertProvider::ServerStatus::
                           kWithRootCertsAndProdSignedCert) {
    return grpc::Status(grpc::StatusCode::PERMISSION_DENIED,
                        "The server is not in mutual authenticated mode thus "
                        "no subscription is "
                        "allowed. Please check Server's credential status.");
  }

  if (!auth_context.IsPeerAuthenticated()) {
    return grpc::Status(
        grpc::StatusCode::PERMISSION_DENIED,
        "The request is not authenticated thus no subscription is "
        "allowed. Please check Server's credential status.");
  }

  // Abort if target doesn't match
  if (grpc::Status status =
          authorizer.CheckTarget(auth_context, client_metadata, *request);
      !status.ok()) {
    return status;
  }

  return grpc::Status::OK;
}

namespace internal {

// The implementation of the ServiceWriteReactor.
// This implementation maintains a queue of events, which is supposed to be
// filled by the SubscriptionService.
class Reactor final : public grpc::ServerWriteReactor<::redfish::v1::Response> {
 public:
  // The status of the reactor (also the stream).
  enum class Status : char {
    kIdle,           // The stream is idle, no write is in the flight.
    kWriteInFlight,  // There is a write in flight.
    kFinishCalled,   // The stream will be terminated soon at any given time as
                     // `Finish` has been called. A stream can fall into this
                     // state because of "previous write failed" or "client
                     // cancelled the stream". The stream is waiting for OnDone
                     // to be called to declare the stream fully finished.
    kFinished,  // The stream has been terminated. The `reactor` pointer might
                // be still valid but will be released at any given time.
  };

  Reactor(ecclesia::SubscriptionService* subscription_service,
          std::function<void(Reactor*)>&& on_done,
          size_t maximum_event_queue_size)
      : subscription_service_(subscription_service),
        on_done_(std::move(on_done)),
        maximum_event_queue_size_(maximum_event_queue_size) {}

  // NOTE: set this once before any StartWrite is called.
  void SetPeerSpiffeIdentity(const ::milotic::authz::PeerSpiffeIdentity& peer) {
    peer_ = peer;
  }

  void SafeFinish(const grpc::Status& status) {
    absl::MutexLock lock(&status_mutex_);
    if (status_ == Status::kFinishCalled || status_ == Status::kFinished) {
      return;
    }
    status_ = Status::kFinishCalled;
    Finish(status);
  }

  void OnWriteDone(bool ok) override {
    {
      absl::MutexLock status_lock(&status_mutex_);
      if (status_ != Status::kWriteInFlight) {
        return;
      }
      if (!ok) {
        LOG(ERROR) << "Failed to write response";
        status_ = Status::kFinishCalled;
        Finish(grpc::Status(grpc::StatusCode::INTERNAL,
                            "Failed to write response"));
        return;
      }
      // Now pop the last event as it won't be referenced again
      {
        absl::MutexLock events_lock(&events_mutex_);
        events_.pop();
      }
      status_ = Status::kIdle;
    }
    if (is_testing_events_) {
      GenerateSingleTestingEvent();
    }
    // The above code block will ensure no lock is held when the block goes out
    // of scope.
    WriteAnEventIfAny();
  }

  void SetToTestMode() { is_testing_events_ = true; }

  void OnCancel() override {
    absl::MutexLock lock(&status_mutex_);
    if (status_ == Status::kFinishCalled || status_ == Status::kFinished) {
      return;
    }
    status_ = Status::kFinishCalled;
    Finish(grpc::Status(grpc::StatusCode::CANCELLED,
                        "Client cancelled requests."));
  }

  void OnDone() override {
    {
      absl::MutexLock lock(&status_mutex_);
      status_ = Status::kFinished;
    }
    if (peer_) {
      grpc::Status status =
          ::milotic::authz::BmcWebAuthorizerSingleton::GetInstance()
              .RecordNewUnsubscription(*peer_);
      if (!status.ok()) {
        // Conversion from grpc::Status to std::ostream is not supported in grpc
        // v1.56.2 which is what was used causing builds to break. See
        // b/323972900
        LOG(ERROR) << "RecordNewUnsubscription failed with code "
                   << status.error_code() << " and message "
                   << status.error_message();
      }
    }
    {
      absl::MutexLock lock(&subscription_id_mutex_);
      if (subscription_id_ && subscription_service_ != nullptr) {
        subscription_service_->DeleteSubscription(*subscription_id_);
      }
    }
    if (on_done_) {
      on_done_(this);
    }
  }

  void SetSubscriptionId(const ecclesia::SubscriptionId& id) {
    absl::MutexLock lock(&subscription_id_mutex_);
    subscription_id_ = id;
  }

  void AddAnEvent(const ::redfish::v1::Response& event) {
    // We only push an event to the queue if the stream is idle or has a write
    // in flight.
    absl::MutexLock status_lock(&status_mutex_);
    if (status_ != Status::kIdle && status_ != Status::kWriteInFlight) {
      return;
    }
    // TODO: check subscription IDs match before enqueue.
    absl::MutexLock events_lock(&events_mutex_);
    if (events_.size() >= maximum_event_queue_size_) {
      LOG(WARNING) << "Too many events to queue; new events have been ignored";
      return;
    }
    events_.push(event);
  }

  Status GetStatus() {
    absl::MutexLock lock(&status_mutex_);
    return status_;
  }

  void WriteAnEventIfAny() {
    // We only start a new StartWrite if the stream is idle.
    absl::MutexLock status_lock(&status_mutex_);
    if (status_ != Status::kIdle) {
      return;
    }
    absl::MutexLock lock(&events_mutex_);
    if (events_.empty()) {
      return;
    }
    status_ = Status::kWriteInFlight;
    StartWrite(&events_.front());
  }

  void StartTestingEvents() {
    GenerateSingleTestingEvent();
    WriteAnEventIfAny();
  }

 protected:
  void GenerateSingleTestingEvent() {
    ::redfish::v1::Response event;
    nlohmann::json json_response;
    event.set_code(200);
    json_response["number"] = test_event_count_;
    event.set_json_str(json_response.dump());
    test_event_count_++;
    AddAnEvent(event);
  }

 private:
  // The data members below are guaranteed to be created before OnWrite starts,
  // afterwards they are read-only, so no need to be protected by a lock
  std::optional<::milotic::authz::PeerSpiffeIdentity> peer_;
  bool is_testing_events_ = false;
  ecclesia::SubscriptionService* subscription_service_ = nullptr;
  std::function<void(Reactor*)> on_done_;
  size_t maximum_event_queue_size_ = 0;

  // The data members below is guaranteed to be accessed only by one thread at a
  // time.
  int test_event_count_ = 0;

  // `events_` might be modified by different thread: gRPC Event Engine threads,
  // subscription service threads.
  absl::Mutex events_mutex_;
  std::queue<::redfish::v1::Response> events_ ABSL_GUARDED_BY(events_mutex_);

  // `subscription_id_` might be modified by different thread. For example, the
  // stream gets closed very early before any SetSubscriptionId is called.
  absl::Mutex subscription_id_mutex_;
  std::optional<ecclesia::SubscriptionId> subscription_id_
      ABSL_GUARDED_BY(subscription_id_mutex_);

  // `status_` might be modified by different thread.
  // `status_` is to avoid extra Write to be fired when the stream is
  // supposed to be finished.
  absl::Mutex status_mutex_;
  Status status_ ABSL_GUARDED_BY(status_mutex_) = Status::kIdle;
};

class RedfishV1Impl final : public ecclesia::GrpcRedfishV1::CallbackService {
 public:
  RedfishV1Impl(
      App* app, const milotic_tlbmc::RedfishApp* tlbmc_app,
      const std::shared_ptr<boost::asio::io_context>& io_context_main_thread,
      const std::shared_ptr<boost::asio::io_context>& io_context_worker_threads,
      ::milotic::redfish::BmcWebCertProvider& cert_provider,
      RedfishServiceConfig& config,
      ecclesia::SubscriptionService* subscription_service)
      : hardware_concurrency_(std::thread::hardware_concurrency()),
        app_(app),
        tlbmc_app_(tlbmc_app),
        io_context_main_thread_(io_context_main_thread),
        io_context_worker_threads_(io_context_worker_threads),
        cert_provider_(cert_provider),
        config_(config),
        subscription_service_(subscription_service) {}

  ~RedfishV1Impl() override = default;

  inline grpc::ServerUnaryReactor* Get(
      grpc::CallbackServerContext* context,
      const ::redfish::v1::Request* request,
      ::redfish::v1::Response* response) override {
    return RequestUri(context, boost::beast::http::verb::get, request,
                      response);
  }
  inline grpc::ServerUnaryReactor* Post(
      grpc::CallbackServerContext* context,
      const ::redfish::v1::Request* request,
      ::redfish::v1::Response* response) override {
    return RequestUri(context, boost::beast::http::verb::post, request,
                      response);
  }
  inline grpc::ServerUnaryReactor* Put(
      grpc::CallbackServerContext* context,
      const ::redfish::v1::Request* request,
      ::redfish::v1::Response* response) override {
    return RequestUri(context, boost::beast::http::verb::put, request,
                      response);
  }
  inline grpc::ServerUnaryReactor* Patch(
      grpc::CallbackServerContext* context,
      const ::redfish::v1::Request* request,
      ::redfish::v1::Response* response) override {
    return RequestUri(context, boost::beast::http::verb::patch, request,
                      response);
  }
  inline grpc::ServerUnaryReactor* Delete(
      grpc::CallbackServerContext* context,
      const ::redfish::v1::Request* request,
      ::redfish::v1::Response* response) override {
    return RequestUri(context, boost::beast::http::verb::delete_, request,
                      response);
  }

  grpc::ServerUnaryReactor* GetOverridePolicy(
      grpc::CallbackServerContext* context,
      const ::redfish::v1::GetOverridePolicyRequest* /*unused*/,
      ::redfish::v1::GetOverridePolicyResponse* response) override {
    grpc::ServerUnaryReactor* reactor = context->DefaultReactor();

    namespace fs = std::filesystem;
    fs::file_status status =
        fs::status(fs::path(config_.redfish_override_policy_path));
    if (!fs::is_regular_file(status)) {
      reactor->Finish(grpc::Status(
          grpc::StatusCode::NOT_FOUND,
          "Policy file not found: " + config_.redfish_override_policy_path));
      return reactor;
    }

    std::ifstream policy_file(config_.redfish_override_policy_path);

    if (!policy_file) {
      reactor->Finish(grpc::Status(
          grpc::StatusCode::INTERNAL,
          "Policy file read error: " + config_.redfish_override_policy_path));
      return reactor;
    }

    response->set_policy(
        std::string(std::istreambuf_iterator<char>{policy_file}, {}));
    reactor->Finish(grpc::Status::OK);
    return reactor;
  }

  grpc::ServerWriteReactor<::redfish::v1::Response>* Subscribe(
      grpc::CallbackServerContext* context,
      const ::redfish::v1::Request* request) override {
    // A reactor will only released when all references are cleared. References
    // include gRPC core layer, and the subscription service.
    auto reactor = std::make_shared<Reactor>(
        subscription_service_,
        [this](Reactor* reactor) { RemoveReactor(reactor); },
        config_.maximum_event_queue_size);
    StoreReactor(reactor);

    if (!config_.generate_testing_events &&
        (config_.disable_eventing || subscription_service_ == nullptr)) {
      reactor->SafeFinish(
          grpc::Status(grpc::StatusCode::UNIMPLEMENTED,
                       "Eventing is disabled or not implemented."));
      return reactor.get();
    }

    if (config_.generate_testing_events) {
      reactor->SetToTestMode();
    }

    // Sanity check: the server has to be in mutual TLS mode and the requester
    // has been authenticated
    if (grpc::Status status = SubscriptionPreCheck(
            *context->auth_context(), context->client_metadata(), request,
            cert_provider_.GetServerStatus());
        !status.ok()) {
      reactor->SafeFinish(status);
      LOG(WARNING) << "AuthorizeSubscription failed: "
                   << status.error_message();
      return reactor.get();
    }

    // Rate limit
    ::milotic::authz::PeerSpiffeIdentity peer;
    if (grpc::Status status = ::milotic::authz::BmcWebAuthorizerSingleton::
            GetPeerIdentityFromAuthContext(*context->auth_context(), peer);
        !status.ok()) {
      reactor->SafeFinish(status);
      LOG(WARNING) << "GetPeerIdentityFromAuthContext failed"
                   << status.error_message();
      return reactor.get();
    }
    if (grpc::Status status =
            ::milotic::authz::BmcWebAuthorizerSingleton::GetInstance()
                .RecordNewSubscription(peer);
        !status.ok()) {
      reactor->SafeFinish(status);
      LOG(WARNING) << "RecordNewSubscription failed" << status.error_message();
      return reactor.get();
    }
    reactor->SetPeerSpiffeIdentity(peer);

    nlohmann::json event_config =
        nlohmann::json::parse(request->json_str(), nullptr, false);

    // Create subscription asynchronously.
    if (subscription_service_ != nullptr) {
      std::unordered_set<std::string> peer_privileges;
      if (grpc::Status status =
              ::milotic::authz::BmcWebAuthorizerSingleton::GetInstance()
                  .GetPrivilegesViaMTls(*context->auth_context(),
                                        peer_privileges);
          !status.ok()) {
        reactor->SafeFinish(status);
        LOG(WARNING) << "GetPrivilegesViaMTls failed" << status.error_message();
        return reactor.get();
      }
      subscription_service_->CreateSubscription(
          event_config, peer_privileges,
          [this, reactor](
              const absl::StatusOr<ecclesia::SubscriptionId>& subscription_id) {
            if (!subscription_id.ok()) {
              reactor->SafeFinish(
                  grpc::Status(grpc::StatusCode::INTERNAL,
                               subscription_id.status().ToString()));
              return;
            }
            reactor->SetSubscriptionId(*subscription_id);

            // Note: any StartWrite must be after any Finish calls above
            if (config_.generate_testing_events) {
              reactor->StartTestingEvents();
            }
          },
          [reactor](const nlohmann::json& event_json) mutable {
            if (reactor == nullptr) return;
            // If reactor ever reports that it is finished, we should never
            // access it again.
            if (reactor->GetStatus() == Reactor::Status::kFinished) {
              reactor = nullptr;
              return;
            }
            // If reactor is about to finish, we should return and wait until
            // it is actually finished. We must not start a new StartWrite().
            if (reactor->GetStatus() == Reactor::Status::kFinishCalled) {
              return;
            }
            ::redfish::v1::Response event;
            event.set_json_str(event_json.dump());
            event.set_code(200);
            reactor->AddAnEvent(event);
            reactor->WriteAnEventIfAny();
          });
      return reactor.get();
    }

    if (config_.generate_testing_events) {
      reactor->StartTestingEvents();
    }
    return reactor.get();
  }

 private:
  struct ProfilingTimeStamps {
    std::chrono::time_point<std::chrono::high_resolution_clock>
        request_uri_start;
    std::chrono::time_point<std::chrono::high_resolution_clock>
        entering_io_context;
    std::chrono::time_point<std::chrono::high_resolution_clock> handler_start;
    std::chrono::time_point<std::chrono::high_resolution_clock> handler_end;
    std::chrono::time_point<std::chrono::high_resolution_clock> response_end;
    absl::Time request_start;
  };

  std::optional<crow::Request> GetInternalRequestFromProto(
      boost::beast::http::request<boost::beast::http::string_body>&&
          boost_request,
      grpc::ServerUnaryReactor* reactor,
      grpc::CallbackServerContext* context) const {
    std::error_code error;
    crow::Request crow_request(std::move(boost_request), error);

    if (error) {
      LOG(ERROR) << "Request failed to construct " << error;
      reactor->Finish(grpc::Status(grpc::StatusCode::INTERNAL,
                                   "Request failed to construct"));
      return std::nullopt;
    }

    // If enable_insecure_server is true, set fromGrpc to false; otherwise
    // fromGrpc should be false.
    crow_request.fromGrpc = !config_.enable_insecure_server;

    crow_request.with_trust_bundle =
        cert_provider_.GetServerStatus() ==
            ::milotic::redfish::BmcWebCertProvider::ServerStatus::
                kWithRootCertsAndProdSignedCert ||
        cert_provider_.GetServerStatus() ==
            ::milotic::redfish::BmcWebCertProvider::ServerStatus::
                kWithRootCertsAndSelfSignedCert;

    crow_request.peer_authenticated =
        context->auth_context()->IsPeerAuthenticated();
    return crow_request;
  }

  grpc::Status SetPeerPrivileges(
      const grpc::AuthContext& auth_context,
      const ::redfish::v1::Request* request,
      std::unordered_set<std::string>& peer_privileges,
      bool with_trust_bundle) const {
    std::optional<std::string> oauth_token =
        ::milotic::authz::GetOAuthTokenFromRequest(*request);

    grpc::Status privileges_status;
    if (config_.enable_insecure_server) {
      peer_privileges.clear();
      privileges_status = grpc::Status::OK;
    } else if (oauth_token != std::nullopt) {
      privileges_status =
          ::milotic::authz::BmcWebAuthorizerSingleton::GetInstance()
              .GetPrivilegesViaOAuth(auth_context, *oauth_token,
                                     peer_privileges);
    } else {
      privileges_status =
          ::milotic::authz::BmcWebAuthorizerSingleton::GetInstance()
              .GetPrivilegesViaMTls(auth_context, peer_privileges);
      // Allow recovery RPCs when there is no trust bundle.
      if (!privileges_status.ok() && !with_trust_bundle) {
        LOG(WARNING) << "Server doesn't have trust bundle; only recovery RPCs "
                        "are allowed";
        peer_privileges.clear();
        privileges_status = grpc::Status::OK;
      }
    }

    return privileges_status;
  }

  static std::chrono::time_point<std::chrono::high_resolution_clock>
  GetCurrentTime(bool do_profiling) {
    if (!do_profiling)
      return std::chrono::high_resolution_clock::time_point::min();
    return std::chrono::high_resolution_clock::now();
  }

  static void LogProfileTimeInResp(
      std::chrono::time_point<std::chrono::high_resolution_clock> start,
      std::chrono::time_point<std::chrono::high_resolution_clock> end,
      const std::string& profile_label, bool do_profiling,
      crow::Response& res) {
    if (!do_profiling) return;
    auto time_delta =
        std::chrono::duration_cast<std::chrono::nanoseconds>(end - start);
    res.jsonValue[profile_label] = time_delta.count();
  }

  static void UpdateResponseMetrics(int code, std::string_view resource_url,
                                    bool tlbmc_enabled,
                                    absl::Time request_start) {
    if (tlbmc_enabled) {
      absl::Duration response_time = absl::Now() - request_start;
      SharedMemoryServer::GetInstance().UpdateMetricsResponse(
          response_time, code, resource_url);
    }
  }

  static void AsyncResponseHandler(::redfish::v1::Response* response,
                                   grpc::ServerUnaryReactor* reactor,
                                   crow::Response& res,
                                   ProfilingTimeStamps profiling_time_stamps,
                                   bool do_profiling, bool tlbmc_enabled,
                                   std::string_view resource_url) {
    if (!res.stringResponse) {
      LOG(ERROR) << "HTTP response is empty!";
      reactor->Finish(grpc::Status(grpc::StatusCode::INTERNAL,
                                   "HTTP response from BMCWeb is empty."));
      UpdateResponseMetrics(static_cast<int>(res.result()), resource_url,
                            tlbmc_enabled, profiling_time_stamps.request_start);
      return;
    }

    auto* headers = response->mutable_headers();
    for (const auto& element : res.stringResponse->base()) {
      // Repeated headers are combined into a ',' separated list. See
      // https://www.rfc-editor.org/rfc/rfc9110.html#section-5.3-3.
      auto found = headers->find(element.name_string());
      if (found == headers->end()) {
        (*headers)[element.name_string()] = element.value();
      } else {
        found->second += ",";
        found->second += element.value();
      }
    }

    profiling_time_stamps.response_end = GetCurrentTime(do_profiling);

    LogProfileTimeInResp(profiling_time_stamps.request_uri_start,
                         profiling_time_stamps.response_end,
                         "time-total-response", do_profiling, res);

    if ((*res.stringResponse)["Content-Type"] == "application/octet-stream" ||
        (*res.stringResponse)["OData-Version"] != "4.0") {
      *response->mutable_octet_stream() = res.body();
    } else {
      *response->mutable_json_str() = res.jsonValue.dump(
          2, ' ', true, nlohmann::json::error_handler_t::replace);
    }
    response->set_code(static_cast<unsigned int>(res.result()));
    reactor->Finish(grpc::Status::OK);
    UpdateResponseMetrics(static_cast<int>(res.result()), resource_url,
                          tlbmc_enabled, profiling_time_stamps.request_start);
  }

  void HandleTlbmcRequest(
      grpc::CallbackServerContext* context,
      const ::redfish::v1::Request* grpc_request,
      boost::beast::http::request<boost::beast::http::string_body>&&
          boost_request,
      ::redfish::v1::Response* grpc_response, grpc::ServerUnaryReactor* reactor,
      bool with_trust_bundle, bool enable_insecure_server) const {
    milotic_tlbmc::Tracer::GetInstance().AddRepeatedDatapoint(
        "Tlbmc-Parse-Request-Begin", absl::Now());
    absl::StatusOr<milotic_tlbmc::RedfishRequest> request =
        milotic_tlbmc::RedfishRequest::Create(std::move(boost_request));
    if (!request.ok()) {
      reactor->Finish(grpc::Status(grpc::StatusCode::INTERNAL,
                                   request.status().ToString()));
      return;
    }
    request->SetWithTrustBundle(with_trust_bundle);
    // If enable_insecure_server is true, set `FromGrpc` to false; otherwise
    // `FromGrpc` should be true.
    request->SetFromGrpc(!enable_insecure_server);
    request->SetPeerAuthenticated(
        context->auth_context()->IsPeerAuthenticated());

    LOG(INFO) << "Received TLBMC request; |url|=" << request->Target()
              << "; |method|=" << static_cast<int>(request->Method());

    std::unordered_set<std::string> peer_privileges;
    if (grpc::Status privileges_status =
            SetPeerPrivileges(*context->auth_context(), grpc_request,
                              peer_privileges, with_trust_bundle);
        !privileges_status.ok() && !enable_insecure_server) {
      reactor->Finish(privileges_status);
      return;
    }

    request->SetPeerPrivileges(peer_privileges);
    milotic_tlbmc::Tracer::GetInstance().AddRepeatedDatapoint(
        "Tlbmc-Parse-Request-End", absl::Now());

    RedfishResponse response(reactor, *grpc_response);
    milotic_tlbmc::Tracer::GetInstance().AddRepeatedDatapoint(
        "Tlbmc-Redfish-Routing-Begin", absl::Now());
    tlbmc_app_->Handle(std::move(*request), std::move(response));
  }

  void HandleNonTlbmcRequest(
      grpc::CallbackServerContext* context,
      const ::redfish::v1::Request* request,
      boost::beast::http::request<boost::beast::http::string_body>&&
          boost_request,
      ::redfish::v1::Response* response, grpc::ServerUnaryReactor* reactor,
      ProfilingTimeStamps profiling_time_stamps) const {
    std::optional<crow::Request> crow_request =
        GetInternalRequestFromProto(std::move(boost_request), reactor, context);
    if (!crow_request) {
      return;
    }

    LOG(INFO) << "Received non-TLBMC request; |url|=" << crow_request->target()
              << "; |method|=" << crow_request->methodString();

    if (grpc::Status privileges_status = SetPeerPrivileges(
            *context->auth_context(), request, crow_request->peer_privileges,
            crow_request->with_trust_bundle);
        !privileges_status.ok()) {
      reactor->Finish(privileges_status);
      return;
    }

    auto async_resp = std::make_shared<bmcweb::AsyncResp>(nullptr);
    if (auto it = request->headers().find(bmcweb::BMCWEB_HINT_MAX_AGE_SEC);
        it != request->headers().end()) {
      int64_t max_age_sec{};
      bool success = absl::SimpleAtoi(it->second, &max_age_sec);
      if (!success || max_age_sec <= 0) {
        reactor->Finish(grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
                                     "Invalid BMCWEB_HINT_MAX_AGE_SEC,"
                                     "must be a positive integer"));
        return;
      }
      async_resp->hintMaxAgeSec = max_age_sec;
      async_resp->hintCacheMaxAgeSec = max_age_sec;
    }

    profiling_time_stamps.request_start = absl::Now();
    std::string resource_url = std::string(crow_request->target());
    async_resp->res.setCompleteRequestHandler(
        [response, reactor, profiling_time_stamps,
         do_profiling{config_.do_profiling},
         tlbmc_enabled{config_.enable_tlbmc},
         resource_url(std::move(resource_url))](crow::Response& res) {
          AsyncResponseHandler(response, reactor, res, profiling_time_stamps,
                               do_profiling, tlbmc_enabled, resource_url);
        });

    profiling_time_stamps.entering_io_context =
        GetCurrentTime(config_.do_profiling);

    if (crow_request->method() == boost::beast::http::verb::get &&
        config_.multi_thread_get) {
      auto strand = std::make_shared<boost::asio::io_context::strand>(
          *io_context_worker_threads_);
      async_resp->strand_ = strand;
      strand->post([this, crow_request{std::move(crow_request)},
                    async_resp{std::move(async_resp)}]() mutable {
        app_->handle(*crow_request, async_resp);
      });
    } else {
      boost::asio::post(
          *io_context_main_thread_, [crow_request{std::move(crow_request)},
                                     /*async_resp has to be moved to prevent
                                       the handler being called in the grpc
                                       thread due to expand queries.*/
                                     async_resp{std::move(async_resp)},
                                     profiling_time_stamps, this]() mutable {
            profiling_time_stamps.handler_start =
                GetCurrentTime(config_.do_profiling);

            app_->handle(*crow_request, async_resp);

            profiling_time_stamps.handler_end =
                GetCurrentTime(config_.do_profiling);

            LogProfileTimeInResp(profiling_time_stamps.request_uri_start,
                                 profiling_time_stamps.entering_io_context,
                                 "time-before-io-context", config_.do_profiling,
                                 async_resp->res);
            LogProfileTimeInResp(profiling_time_stamps.entering_io_context,
                                 profiling_time_stamps.handler_start,
                                 "time-io-context-queue-wait",
                                 config_.do_profiling, async_resp->res);
            LogProfileTimeInResp(profiling_time_stamps.handler_start,
                                 profiling_time_stamps.handler_end,
                                 "time-handler", config_.do_profiling,
                                 async_resp->res);
          });
    }
  }

  static bool IsTlbmcRequest(const ::redfish::v1::Request* request) {
    return absl::StartsWith(request->url(), milotic_tlbmc::kTlbmcPrefix);
  }

  static std::optional<
      boost::beast::http::request<boost::beast::http::string_body>>
  CreateBoostRequest(const ::redfish::v1::Request* request,
                     boost::beast::http::verb method) {
    boost::beast::http::request<boost::beast::http::string_body> boost_request;

    boost::system::result<boost::urls::url> url =
        boost::urls::parse_relative_ref(request->url());
    if (!url) {
      LOG(INFO) << absl::StrFormat("URL %s is invalid", request->url());
      return std::nullopt;
    }

    // https://www.dmtf.org/sites/default/files/standards/documents/DSP0266_1.8.0.pdf
    // Redfish spec states that # is URL must be a fragment. (6.1)
    // Fragments will be ignored by the server. (Also 6.1)
    boost_request.target(url->remove_fragment());
    boost_request.method(method);

    // Check if json_str here first since that's preferred.
    if (!request->json_str().empty()) {
      boost_request.body() = request->json_str();
    } else if (request->has_json()) {
      nlohmann::json json_req = ecclesia::StructToJson(request->json());
      if (!json_req.is_null()) {
        boost_request.body() = json_req.dump();
      }
    } else if (request->has_octet_stream()) {
      boost_request.body() = request->octet_stream();
    }

    for (const auto& [key, value] : request->headers()) {
      boost_request.base().insert(key, value);
    };

    return boost_request;
  }
  grpc::ServerUnaryReactor* RequestUri(grpc::CallbackServerContext* context,
                                       boost::beast::http::verb method,
                                       const ::redfish::v1::Request* request,
                                       ::redfish::v1::Response* response) {
    ProfilingTimeStamps profiling_time_stamps;
    profiling_time_stamps.request_uri_start =
        GetCurrentTime(config_.do_profiling);

    using ::milotic::authz::BmcWebAuthorizerSingleton;

    grpc::ServerUnaryReactor* reactor = context->DefaultReactor();

    milotic_tlbmc::Tracer::GetInstance().AddRepeatedDatapoint(
        "Authz-CheckTarget-Begin", absl::Now());
    // Check target only when there is trust bundle and server is in secure
    // mode.
    bool with_trust_bundle =
        (cert_provider_.GetServerStatus() ==
             ::milotic::redfish::BmcWebCertProvider::ServerStatus::
                 kWithRootCertsAndProdSignedCert ||
         cert_provider_.GetServerStatus() ==
             ::milotic::redfish::BmcWebCertProvider::ServerStatus::
                 kWithRootCertsAndSelfSignedCert);
    if (!config_.enable_insecure_server && with_trust_bundle) {
      grpc::Status status =
          BmcWebAuthorizerSingleton::GetInstance().CheckTarget(
              *context->auth_context(), context->client_metadata(), *request);
      if (!status.ok()) {
        LOG(WARNING) << "Check target failed: " << status.error_message();
        reactor->Finish(status);
        return reactor;
      }
    }

    std::optional<boost::beast::http::request<boost::beast::http::string_body>>
        boost_request = CreateBoostRequest(request, method);
    if (!boost_request) {
      reactor->Finish(
          grpc::Status(grpc::StatusCode::INVALID_ARGUMENT, "Invalid URL"));
      return reactor;
    }

    milotic_tlbmc::Tracer::GetInstance().AddRepeatedDatapoint(
        "Authz-CheckTarget-End", absl::Now());

    if (config_.enable_tlbmc && IsTlbmcRequest(request)) {
      HandleTlbmcRequest(context, request, std::move(boost_request.value()),
                         response, reactor, with_trust_bundle,
                         config_.enable_insecure_server);
    } else {
      HandleNonTlbmcRequest(context, request, std::move(boost_request.value()),
                            response, reactor, profiling_time_stamps);
    }

    return reactor;
  }

  void StoreReactor(const std::shared_ptr<Reactor>& reactor) {
    absl::MutexLock lock(&mutex_);
    reactors_store_.insert({reactor.get(), reactor});
  }

  void RemoveReactor(Reactor* reactor) {
    absl::MutexLock lock(&mutex_);
    reactors_store_.erase(reactor);
  }

  const unsigned int hardware_concurrency_;
  App* app_;
  const milotic_tlbmc::RedfishApp* tlbmc_app_;
  std::shared_ptr<boost::asio::io_context> io_context_main_thread_;
  std::shared_ptr<boost::asio::io_context> io_context_worker_threads_;
  ::milotic::redfish::BmcWebCertProvider& cert_provider_;
  RedfishServiceConfig& config_;
  ::redfish::v1::Response first_test_response_;
  ecclesia::SubscriptionService* subscription_service_ = nullptr;

  absl::Mutex mutex_;
  absl::flat_hash_map<Reactor*, std::shared_ptr<Reactor>> reactors_store_
      ABSL_GUARDED_BY(mutex_);
};

}  // namespace internal

void GrpcRedfishService::Shutdown() {
  LOG(INFO) << "Dumping trace: \n"
            << milotic_tlbmc::Tracer::GetInstance().Dump();
  LOG(WARNING) << "Shutdown gRPC server...";
  server_->Shutdown(std::chrono::system_clock::now() + std::chrono::seconds(1));
  server_->Wait();
  LOG(WARNING) << "gRPC server stopped!";
}

GrpcRedfishService::GrpcRedfishService(
    App* app, const milotic_tlbmc::RedfishApp* tlbmc_app,
    const std::shared_ptr<boost::asio::io_context>& io_context_main_thread,
    const std::shared_ptr<boost::asio::io_context>& io_context_worker_threads,
    const RedfishServiceConfig& config,
    ecclesia::SubscriptionService* subscription_service)
    : cert_provider_(
          std::make_unique<::milotic::redfish::BmcWebCertProviderImpl>(
              config.trust_bundle_path, config.private_key_path,
              config.own_cert_path, config.self_signed_key_cert_path,
              boost::uuids::to_string(boost::uuids::random_generator()()),
              config.authority_policy_file_binary,
              /*enable_insecure_server=*/config.enable_insecure_server,
              /*check_loas3_policy=*/config.check_loas3_policy)),
      config_(config),
      subscription_service_(subscription_service) {
  using ::milotic::authz::BmcWebAuthorizerSingleton;
  using ::milotic::redfish::BmcWebCertProvider;
  using ::milotic::redfish::BmcWebCertProviderImpl;
  using ::milotic::redfish::GetCredentialsWithoutAuthz;
  using ::milotic_hft::HftServiceImpl;
  using ::milotic_hft::HighFrequencySensorReading;
  using ::milotic_hft::HighFrequencySensorsReadings;
  using ::milotic_hft::Payload;
  using ::milotic_hft::SubscriptionManager;
  using ::milotic_hft::SubscriptionManagerFake;

  LOG(WARNING) << "Start gRPC server...";
  LOG(WARNING) << "Initialize authorization layer...";
#ifdef ENABLE_LOAS3_VALIDATION
  security::SetCertificateAuthorityPolicyFilePath(
      config_.authority_policy_file_binary);
#endif
  BmcWebAuthorizerSingleton::Initialize(
      {
          .configuration_path = config_.authz_config_path,
          .platform_configuration_path = config_.authz_platform_config_path,
          .base_privileges_folder = config_.persistent_base_privileges_folder,
          .pattern_entity_overrides_path =
              config.pattern_to_entity_overrides_path,
          .offline_node_entities_path = config.offline_node_entity_path,
          .google_machine_identity_path = config_.gmi_file_path,
      },
      /*oauth_key_path=*/config.oauth_key_path);
  if (!BmcWebAuthorizerSingleton::GetInstance()
           .IsBasePrivilegeRegistryFound()) {
    LOG(WARNING)
        << "Could not find Privilege Registry at /var/google/authz_policies";
    BmcWebAuthorizerSingleton::GetInstance().SetBasePrivilegesFolder(
        config_.rofs_base_privileges_folder);
  }

  LOG(WARNING) << "Authorization layer initialized!";
  grpc::reflection::InitProtoReflectionServerBuilderPlugin();
  grpc::ServerBuilder builder;

  if (config_.enable_insecure_server) {
    builder.AddListeningPort("[::]:" + std::to_string(config_.port),
                             grpc::InsecureServerCredentials());
  } else {
    builder.AddListeningPort(
        "[::]:" + std::to_string(config_.port),
        GetCredentialsWithoutAuthz(*cert_provider_, config_.crl_directory));
  }

  service_ = std::make_unique<internal::RedfishV1Impl>(
      app, tlbmc_app, io_context_main_thread, io_context_worker_threads,
      *cert_provider_, config_, subscription_service_);

  std::unique_ptr<SubscriptionManager> subscription_manager;
  if (config_.enable_hft_fake_manager) {
    auto fake_manager =
        std::make_unique<SubscriptionManagerFake>();
    Payload payload;
    HighFrequencySensorsReadings* readings =
        payload.mutable_high_frequency_sensors_readings_batch()
            ->add_high_frequency_sensors();
    readings->mutable_sensor_identifier()->set_name("test_sensor");
    HighFrequencySensorReading* timestamped_reading =
        readings->add_timestamped_readings();
    timestamped_reading->set_timestamp_ns(1234567890);
    timestamped_reading->set_float_reading(1.23456789f);
    fake_manager->SetFakeData(std::move(payload));
    subscription_manager = std::move(fake_manager);
  }

  hft_service_ = std::make_unique<milotic_hft::HftServiceImpl>(
      milotic_hft::HftServiceOptions{
          .cert_provider = cert_provider_.get(),
      },
      std::move(subscription_manager));
  builder.RegisterService(service_.get());

  if (config_.enable_hft) {
    builder.RegisterService(hft_service_.get());
  }

  server_ = builder.BuildAndStart();
}

}  // namespace milotic

#pragma GCC diagnostic pop