#include "server.h"

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

#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"

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

#include "tlbmc/feed_client_grpc.h"
#include "tlbmc/feed_client_interface.h"
#include "absl/log/log.h"
#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 "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 "tlbmc/grpc_tls_options.h"
#include "zatar/struct_proto_conversion.h"
#include "http_request.hpp"
#include "http_response.hpp"
#include "async_resp.hpp"
#include "grpc/grpc.h"
#include "grpcpp/ext/proto_server_reflection_plugin.h"
#include "grpcpp/security/auth_context.h"
#include "grpcpp/security/credentials.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/service/fru_service.h"
#include "tlbmc/adapter/data_source.h"
#include "tlbmc/subscription/manager.h"
#include "tlbmc/subscription/manager_fake.h"
#include "tlbmc/subscription/manager_impl.h"
#include "payload.pb.h"
#include "sensor_identifier.pb.h"
#include "sensor_payload.pb.h"
#include "tlbmc/central_config/config.h"
#include "tlbmc/configs/proto_config_parser.h"
#include "tlbmc/hal/fake_fru_scanner_fram.h"
#include "tlbmc/hal/fru_scanner.h"
#include "tlbmc/hal/fru_scanner_fram.h"
#include "tlbmc/hal/fru_scanner_i2c.h"
#include "tlbmc/hal/shared_mem/client.h"
#include "tlbmc/hal/shared_mem/server.h"
#include "tlbmc/hal/sysfs/i2c.h"
#include "tlbmc/hal/sysfs/i3c.h"
#include "tlbmc/hal/sysfs/peci.h"
#include "tlbmc/redfish/app.h"
#include "tlbmc/redfish/request.h"
#include "tlbmc/redfish/response.h"
#include "tlbmc/redfish/routes/all_routes.h"
#include "tlbmc/store/store_hft_adapter.h"
#include "tlbmc/store/store_impl.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"
#include "openssl/sha.h"  // NOLINT: just for sanity check. No crypto is done.

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 && tlbmc_app_ != nullptr},
         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()) {
      std::string stripped_url = url->path();
      if (stripped_url.starts_with(milotic_tlbmc::kTlbmcPrefix)) {
        stripped_url =
            stripped_url.substr(milotic_tlbmc::kTlbmcPrefix.size() - 1);
      }
      if (!stripped_url.empty() && stripped_url.back() == '/') {
        stripped_url.pop_back();
      }
      if (stripped_url != "/redfish/v1/UpdateService/MultipartUpdate" ||
          method != boost::beast::http::verb::post) {
        boost_request.body() = request->octet_stream();
      } else {
        // MultipartUpdate is a special case that we want to avoid copying the
        // entire content to the body. We will write the content to a
        // temporary file and then pass the file path to the handler.
        // Breaking the API boundary is not perfect but I didn't find other good
        // way to avoid copying the request body.
        std::string_view file_content = request->octet_stream();
#ifdef UNIT_TEST_BUILD
        std::string bundle_file_tmp_path = std::tmpnam(nullptr);
#else
        // TODO(nanzhou): handle concurrent multipart update requests better
        std::string bundle_file_tmp_path =
            "/mnt/luks-mmcblk0_fs/firmware_bundle.tar.gz.tmp";
#endif

        std::ofstream file(bundle_file_tmp_path, std::ios::binary);
        bool hash_check_good = true;
        SHA256_CTX sha256_ctx;
        if (SHA256_Init(&sha256_ctx) != 1) {
          hash_check_good = false;
        }
        constexpr std::size_t kBufferSize1MB = 1024 * 1024;
        std::size_t offset = 0;
        while (offset < file_content.size()) {
          std::size_t remaining_size = file_content.size() - offset;
          std::size_t current_chunk_size =
              std::min(kBufferSize1MB, remaining_size);
          std::string_view chunk =
              file_content.substr(offset, current_chunk_size);
          file.write(chunk.data(), static_cast<std::streamsize>(chunk.size()));
          offset += current_chunk_size;
          if (hash_check_good &&
              SHA256_Update(&sha256_ctx, chunk.data(), chunk.size()) != 1) {
            hash_check_good = false;
          }
        }
        file.flush();
        if (file.good()) {
          boost_request.body() = bundle_file_tmp_path;
        }
        unsigned char hash[SHA256_DIGEST_LENGTH];
        if (SHA256_Final(hash, &sha256_ctx) != 1) {
          hash_check_good = false;
        }
        if (hash_check_good) {
          std::string server_hash(reinterpret_cast<char*>(hash),
                                  SHA256_DIGEST_LENGTH);
          boost_request.base().insert("Multipart-SHA256-ServerSide",
                                      server_hash);
        }
      }
    }

    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 (!milotic_tlbmc::GetTlbmcConfig()
             .trust_bundle_install_module()
             .enabled() &&
        !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!";
}

std::unique_ptr<milotic_tlbmc::RedfishApp> GrpcRedfishService::CreateTlbmcApp(
    const RedfishServiceConfig& config) {
  using ::milotic_tlbmc::AdHocScannerType;
  using ::milotic_tlbmc::FruScanner;
  using ::milotic_tlbmc::FruScannerFram;
  using ::milotic_tlbmc::FruScannerI2c;
  using ::milotic_tlbmc::I2cSysfs;
  using ::milotic_tlbmc::I3cSysfs;
  using ::milotic_tlbmc::InitializeTlbmcConfig;
  using ::milotic_tlbmc::PeciSysfs;
  using ::milotic_tlbmc::RedfishApp;
  using ::milotic_tlbmc::StoreImpl;
  using ::milotic_tlbmc::Tracer;
  InitializeTlbmcConfig(config.tlbmc_config_bundle_path,
                        config.tlbmc_release_file_path);
  std::unique_ptr<RedfishApp> tlbmc_app = nullptr;
  // Initialize tlbmc
  if (config.enable_tlbmc_trace) {
    Tracer::Initialize(true);
  }

  // Create FruScannerI2c instance.
  FruScannerI2c::Options fru_scanner_options;
  fru_scanner_options.root_dir = config.tlbmc_root_dir;
  std::unique_ptr<FruScannerI2c> fru_scanner_i2c =
      FruScannerI2c::Create(fru_scanner_options);

  // Create FruScannerFram instance.
  FruScannerFram::Options fru_scanner_fram_options;
  std::unique_ptr<FruScannerFram> fru_scanner_fram;
  if (config.tlbmc_in_unit_test &&
      config.tlbmc_unit_test_enable_fake_fram_scanner) {
    LOG(INFO) << "Using FakeFruScannerFram for unit test.";
    fru_scanner_fram = std::make_unique<milotic_tlbmc::FakeFruScannerFram>();
  } else {
    fru_scanner_fram = FruScannerFram::Create(fru_scanner_fram_options);
  }

  absl::flat_hash_map<milotic_tlbmc::AdHocScannerType,
                      std::unique_ptr<FruScanner>>
      fru_scanners;
  fru_scanners[milotic_tlbmc::AdHocScannerType::SCANNER_TYPE_DEFAULT_IPMI_I2C] =
      std::move(fru_scanner_i2c);
  fru_scanners[milotic_tlbmc::AdHocScannerType::SCANNER_TYPE_CAVIUM_HSM_FRAM] =
      std::move(fru_scanner_fram);

  bool enable_tlbmc = config.enable_tlbmc;
  std::string disable_tlbmc_file_path = config.tlbmc_disable_file;
  if (std::filesystem::exists(disable_tlbmc_file_path)) {
    LOG(WARNING) << "Forcefully disable tlbmc";
    enable_tlbmc = false;
  }
  if (enable_tlbmc) {
    milotic_tlbmc::StoreImpl::Options options;
    options.fru_scanners = std::move(fru_scanners);
    options.i2c_sysfs =
        std ::make_unique<I2cSysfs>(milotic_tlbmc::I2cSysfsConfig{
            .i2c_sysfs_path = config.tlbmc_i2c_sysfs_path});
    options.i3c_sysfs =
        std ::make_unique<I3cSysfs>(milotic_tlbmc::I3cSysfsConfig{
            .i3c_sysfs_path = config.tlbmc_i3c_sysfs_path});
    options.peci_sysfs =
        std ::make_unique<PeciSysfs>(milotic_tlbmc::PeciSysfsConfig{
            .peci_sysfs_path = config.tlbmc_peci_sysfs_path});
    options.config_location = config.tlbmc_entity_config_location;
    options.proto_parser = milotic_tlbmc::ProtoConfigParser::Create(
        config.tlbmc_proto_config_location);

    if (config.tlbmc_in_unit_test && config.executor_command_map.has_value()) {
      options.executor_command_map = *config.executor_command_map;
    }

    absl::StatusOr<std::unique_ptr<milotic_tlbmc::StoreImpl>> tlbmc_store =
        milotic_tlbmc::StoreImpl::Create(std::move(options));
    if (tlbmc_store.ok()) {
      if (!config.tlbmc_in_unit_test) {
        milotic_tlbmc::SharedMemoryServer::Initialize();
      } else {
        milotic_tlbmc::SharedMemoryServer::SetUpInstanceForUnitTest();
        milotic_tlbmc::SharedMemoryClient::SetUpInstanceForUnitTest();
      }
      // Start the sensor collection after the shared memory server is
      // initialized.
      (*tlbmc_store)->StartSensorCollection();
      tlbmc_app =
          std::make_unique<milotic_tlbmc::RedfishApp>(std::move(*tlbmc_store));
      milotic_tlbmc::RegisterAllRoutes(*tlbmc_app);
      tlbmc_app->Validate();
    } else {
      LOG(ERROR) << "Cannot create tlBMC store!! Error: "
                 << tlbmc_store.status() << " - Disabling tlBMC.";
      tlbmc_app = nullptr;
      enable_tlbmc = false;
    }
  }
  LOG(WARNING) << "tlbmc enabled: " << (enable_tlbmc ? "true" : "false");
  return tlbmc_app;
}

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_fast_sanity::FruServiceImpl;
  using ::milotic_hft::HftServiceImpl;
  using ::milotic_hft::HighFrequencySensorReading;
  using ::milotic_hft::HighFrequencySensorsReadings;
  using ::milotic_hft::Payload;
  using ::milotic_hft::SubscriptionManager;
  using ::milotic_hft::SubscriptionManagerFake;
  using ::milotic_hft::SubscriptionManagerImpl;
  using ::milotic_tlbmc::StoreHftAdapter;
  using ::platforms_syshealth::collection::feed::FeedClientAnycastUser;

  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,
          .tlbmc_trust_bundle_install_module_is_enabled =
              milotic_tlbmc::GetTlbmcConfig()
                  .trust_bundle_install_module()
                  .enabled(),
      },
      /*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);
  } else if (tlbmc_app != nullptr) {
    absl::StatusOr<std::unique_ptr<milotic_hft::DataSource>> data_source =
        milotic_tlbmc::StoreHftAdapter::Create(tlbmc_app->GetStore());
    if (data_source.ok()) {
      // Hardcode the peer role to FeedService for now as we only have one
      // anycast user and is not expected to be changed.
      std::string anycast_address =
          BmcWebAuthorizerSingleton::GetInstance().GetAnycastAddress(
              "FeedService");
      if (config_.enable_feed_client) {
        if (anycast_address.empty()) {
          LOG(WARNING) << "FeedService anycast address is empty, initializing "
                          "feed client in dry run mode.";
        }
        platforms_syshealth::collection::feed::FeedClientConfig
            feed_client_config;
        std::shared_ptr<grpc::ChannelCredentials> credentials;
        feed_client_config.set_target_address(anycast_address);
        if (absl::StartsWith(anycast_address, "[::1]") ||
            absl::StartsWith(anycast_address, "localhost")) {
          credentials = grpc::InsecureChannelCredentials();
        } else {
          ecclesia::StaticBufferBasedTlsOptions grpc_options;
          grpc_options.SetToTls(config_.private_key_path, config_.own_cert_path,
                                config_.trust_bundle_path);
          credentials = grpc_options.GetChannelCredentials();
        }
        feed_client_ = std::make_shared<
            platforms_syshealth::collection::feed::FeedClientGrpc>(
            feed_client_config, credentials);
        feed_client_anycast_user_ =
            std::make_unique<FeedClientAnycastUser>(feed_client_);
        BmcWebAuthorizerSingleton::GetInstance().RegisterAnycastUser(
            "FeedService", feed_client_anycast_user_.get());
        subscription_manager = milotic_hft::SubscriptionManagerImpl::Create(
            std::move(*data_source), feed_client_.get(),
            {.get_policies_interval = config_.feed_client_get_policies_interval,
             .endpoint_type = platforms_syshealth::collection::feed::
                 EndpointIdentifier::ENDPOINT_TYPE_BMCWEB,
             .offline_node_entities_path = config_.offline_node_entity_path});
      }

      // If feed client is not enabled, create a subscription manager without
      // feed client.
      if (subscription_manager == nullptr) {
        LOG(WARNING) << "Creating subscription manager without feed client.";
        subscription_manager = milotic_hft::SubscriptionManagerImpl::Create(
            std::move(*data_source));
      }
    } else {
      LOG(ERROR) << "Failed to create hft source adapter: "
                 << data_source.status();
    }
  }

  if (subscription_manager != nullptr && config_.enable_hft) {
    hft_service_ = std::make_unique<milotic_hft::HftServiceImpl>(
        milotic_hft::HftServiceOptions{
            .cert_provider = cert_provider_.get(),
        },
        std::move(subscription_manager));
    builder.RegisterService(hft_service_.get());
    // TODO(nanzhou): better abstraction so tlbmc_app becomes a real frontend
    // and owns the hft service.
    // The const_cast is safe because we access the pointer before any URLs
    // start to be handled.
    if (tlbmc_app != nullptr) {
      const_cast<milotic_tlbmc::RedfishApp*>(tlbmc_app)->SetHftService(
          hft_service_.get());
    }
  }

  if (config_.enable_fast_sanity) {
    milotic_fast_sanity::FruServiceOptions fru_service_options{
        .cert_provider = cert_provider_.get(),
    };
    fru_service_ = std::make_unique<milotic_fast_sanity::FruServiceImpl>(
        app, fru_service_options);
    builder.RegisterService(fru_service_.get());
    LOG(WARNING) << "Fast-Sanity is enabled.";
  }

  builder.RegisterService(service_.get());
  // Set max message size to 500MB, this is to allow large request like Firmware
  // Update.
  constexpr int k500MB = 500 * 1024 * 1024;
  builder.AddChannelArgument(GRPC_ARG_MAX_SEND_MESSAGE_LENGTH, k500MB);
  builder.AddChannelArgument(GRPC_ARG_MAX_RECEIVE_MESSAGE_LENGTH, k500MB);
  server_ = builder.BuildAndStart();
}

}  // namespace milotic

#pragma GCC diagnostic pop