#include "server.h"

// Ignore the following warnings in this header when doing Yocto builds.
#include <sys/stat.h>

#include <algorithm>
#include <chrono>  // NOLINT
#include <cstddef>
#include <cstdint>
#include <cstdio>
#include <cstring>
#include <filesystem>  // NOLINT
#include <fstream>
#include <functional>
#include <ios>
#include <iterator>
#include <map>
#include <memory>
#include <optional>
#include <queue>
#include <string>
#include <string_view>
#include <system_error>  // NOLINT
#include <thread>        // NOLINT
#include <unordered_set>
#include <utility>
#include <vector>

#include "tlbmc/credentials/credential_manager.h"
#include "tlbmc/hal/sysfs/led.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 "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 "app.hpp"
#include "grpc/grpc.h"
#include "grpcpp/support/status.h"
#include <nlohmann/json_fwd.hpp>
#include "peer_identity.h"
#include "usb_fru_config.pb.h"
#include "tlbmc/hal/fru_scanner_gpio.h"
#include "tlbmc/hal/fru_scanner_usb.h"
#include "tlbmc/hal/shared_mem/static_client_impl.h"
#include "tlbmc/hal/shared_mem/static_server_impl.h"
#include "tlbmc/hft/core/hft_service.h"
#include "tlbmc/store/octeon10_evk_store.h"

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

// `B200` is defined in a macro while also in an enum. Need to undefine it to
// avoid build errors.
// More context at https://yaqs.corp.google.com/eng/q/1495189578728865792
#undef B200

#include "gbmc-hal/api/app/inventory/smbios_inventory.h"
#include "gbmc-hal/api/system/codecs/smbios_codec.h"
#include "gbmc-hal/api/system/registries.h"
#include "gbmc-hal/api/system/registry_configs.h"
#include "bus_topology.pb.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 "zatar/struct_proto_conversion.h"
#include "http_request.hpp"
#include "http_response.hpp"
#include "async_resp.hpp"
#include <nlohmann/json.hpp>
#include "oauth_utils.h"
#include "tlbmc/service/fru_service.h"
#include "tlbmc/adapter/data_source.h"
#include "payload.pb.h"
#include "sensor_identifier.pb.h"
#include "sensor_payload.pb.h"
#include "tlbmc/central_config/config.h"
#include "tlbmc/collector/fake_fru_collector.h"
#include "tlbmc/collector/fake_gpio_collector.h"
#include "tlbmc/collector/gpio_collector.h"
#include "tlbmc/configs/proto_config_parser.h"
#include "tlbmc/configs/rate_limiter_config.h"
#include "tlbmc/deterministic_bmc/offline_config_parser/offline_config_parser.h"
#include "tlbmc/deterministic_bmc/offline_config_parser/proto_reader.h"
#include "tlbmc/hal/fake_fru_scanner_fram.h"
#include "tlbmc/hal/fru_scanner.h"
#include "tlbmc/hal/fru_scanner_adc.h"
#include "tlbmc/hal/fru_scanner_cable.h"
#include "tlbmc/hal/fru_scanner_fram.h"
#include "tlbmc/hal/fru_scanner_i2c.h"
#include "tlbmc/hal/nic_veeprom/interface.h"
#include "tlbmc/hal/shared_mem/server_interface.h"
#include "tlbmc/hal/sysfs/i2c.h"
#include "tlbmc/hal/sysfs/i3c.h"
#include "tlbmc/hal/sysfs/iio.h"
#include "tlbmc/hal/sysfs/peci.h"
#include "tlbmc/hft/core/manager.h"
#include "tlbmc/hft/core/manager_fake.h"
#include "tlbmc/hft/core/manager_impl.h"
#include "tlbmc/hft/task_scheduler_adapter.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.h"
#include "tlbmc/store/store_hft_adapter.h"
#include "tlbmc/store/store_impl.h"
#include "tlbmc/trace/tracer.h"
#include "g3/grpc_headers.h"
#include "bmcweb_authorizer_singleton.h"
#include "rate_limiter.h"
#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::StaticSharedMemoryClient;
using ::milotic_tlbmc::StaticSharedMemoryServer;

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,
      crow::RateLimiter* rate_limiter)
      : 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),
        rate_limiter_(rate_limiter) {}

  ~RedfishV1Impl() override = default;

  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);
  }
  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);
  }
  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);
  }
  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);
  }
  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;
      milotic_tlbmc::StaticSharedMemoryServer::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;
      boost::asio::post(*strand, [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 bool IsEmmcMounted() {
    struct stat st_mount;
    struct stat st_parent;
    if (stat("/mnt/luks-mmcblk0_fs", &st_mount) != 0) {
      return false;
    }
    if (stat("/mnt", &st_parent) != 0) {
      return false;
    }
    return st_mount.st_dev != st_parent.st_dev;
  }

  static absl::StatusOr<
      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) {
      return absl::InvalidArgumentError(
          absl::StrFormat("URL %s is invalid", request->url()));
    }

    // 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";

        // If eMMC is not mounted, we should not write to the file system.
        // Instead, we will write to the tmpfs.
        bool is_emmc_mounted = IsEmmcMounted();
        if (!is_emmc_mounted) {
          return absl::InternalError(
              "eMMC is not mounted. Can't process multipart update.");
        }
#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;
      }
    }

    absl::StatusOr<boost::beast::http::request<boost::beast::http::string_body>>
        boost_request = CreateBoostRequest(request, method);
    if (!boost_request.ok()) {
      reactor->Finish(
          grpc::Status(grpc::StatusCode::INTERNAL,
                       std::string(boost_request.status().message())));
      return reactor;
    }

    if (rate_limiter_ != nullptr) {
      if (std::optional<crow::RateLimiterDropInfo> drop_info =
              rate_limiter_->AdmitRequest(request->url());
          drop_info.has_value()) {
        reactor->Finish(grpc::Status(
            grpc::StatusCode::RESOURCE_EXHAUSTED,
            absl::StrFormat(
                "Request rate limited: bucket level %.2f, capacity %.2f, leak "
                "rate %.2f reqs/sec.",
                drop_info->bucket_level, drop_info->bucket_capacity,
                drop_info->leak_rate_per_sec)));
        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;
  crow::RateLimiter* rate_limiter_ = 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!";
}

void GrpcRedfishService::RegisterTlbmcRoutesAndValidate(
    milotic_tlbmc::RedfishApp* app, const std::string& system_root_path) {
  if (app != nullptr) {
    milotic_tlbmc::RegisterAllRoutes(*app, system_root_path);
    app->Validate();
    if (app->GetSmartRouter() != nullptr) {
      app->GetSmartRouter()->UpdateTlbmcOwnedUrls();
    }
  }
}

std::unique_ptr<milotic_tlbmc::RedfishApp> GrpcRedfishService::CreateTlbmcApp(
    const RedfishServiceConfig& config) {
  if (!config.enable_tlbmc) {
    return nullptr;
  }
  if (config.enable_tlbmc_octeon10_evk) {
    return std::make_unique<milotic_tlbmc::RedfishApp>(
        std::make_unique<milotic_tlbmc::Octeon10EvkStore>());
  }
  using ::milotic_tlbmc::AdHocScannerType;
  using ::milotic_tlbmc::FakeGpioCollector;
  using ::milotic_tlbmc::FruScanner;
  using ::milotic_tlbmc::FruScannerAdc;
  using ::milotic_tlbmc::FruScannerCable;
  using ::milotic_tlbmc::FruScannerFram;
  using ::milotic_tlbmc::FruScannerGpio;
  using ::milotic_tlbmc::FruScannerI2c;
  using ::milotic_tlbmc::FruScannerUsb;
  using ::milotic_tlbmc::GpioCollector;
  using ::milotic_tlbmc::I2cSysfs;
  using ::milotic_tlbmc::I3cSysfs;
  using ::milotic_tlbmc::IioSysfs;
  using ::milotic_tlbmc::InitializeTlbmcConfig;
  using ::milotic_tlbmc::LedSysfs;
  using ::milotic_tlbmc::PeciSysfs;
  using ::milotic_tlbmc::RateLimiterConfig;
  using ::milotic_tlbmc::RedfishApp;
  using ::milotic_tlbmc::StoreImpl;
  using ::milotic_tlbmc::Tracer;
  InitializeTlbmcConfig(config.tlbmc_config_bundle_path,
                        config.tlbmc_release_file_path);
  RateLimiterConfig::Init(std::filesystem::path(config.tlbmc_config_bundle_path)
                              .parent_path()
                              .string());
  // Initialize tlbmc
  if (config.enable_tlbmc_trace) {
    Tracer::Initialize(true);
  }

  // Initialize deterministic bmc.
  bool enable_deterministic_bmc = milotic_tlbmc::GetTlbmcConfig()
                                      .fru_collector_module()
                                      .enable_deterministic_bmc();
  uhmm::I2cRoot i2c_root;
  uhmm::BmcMonitoredComponentsResponse bmc_monitored_components_response;

  milotic_tlbmc::StoreImpl::Options options;

  if (enable_deterministic_bmc) {
    deterministic_bmc::OfflineConfigParser::ConfigParserOptions
        config_parser_options;

    config_parser_options.uhmm_offline_data_proto_format =
        config.uhmm_offline_data_format;

    if (config.uhmm_offline_data_format ==
        deterministic_bmc::ProtoFormat::kTextProto) {
      config_parser_options.uhmm_offline_data_path_txtpb =
          config.uhmm_offline_data_path;
    } else if (config.uhmm_offline_data_format ==
               deterministic_bmc::ProtoFormat::kBinaryProto) {
      config_parser_options.uhmm_offline_data_path_pb =
          config.uhmm_offline_data_path;
    } else {
      LOG(ERROR) << "Invalid uhmm_offline_data_format: "
                 << config.uhmm_offline_data_format;
      enable_deterministic_bmc = false;
    }

    absl::StatusOr<std::unique_ptr<deterministic_bmc::OfflineConfigParser>>
        offline_config_parser = deterministic_bmc::OfflineConfigParser::Create(
            config_parser_options);
    std::unique_ptr<deterministic_bmc::OfflineConfigParser>
        offline_config_parser_ptr;
    if (!offline_config_parser.ok()) {
      LOG(ERROR) << "Uhmm is not ready for deterministic bmc, "
                    "OfflineConfigParser::Create failed with status: "
                 << offline_config_parser.status();
      enable_deterministic_bmc = false;
    } else {
      offline_config_parser_ptr = std::move(*offline_config_parser);
      if (!offline_config_parser_ptr->isUhmmConfigValid()) {
        LOG(ERROR) << "Uhmm config is not valid.";
        enable_deterministic_bmc = false;
      } else {
        // Deterministic bmc is enabled, get configs from the parser.
        absl::StatusOr<uhmm::BmcMonitoredComponentsResponse>
            bmc_monitored_components_response_status_or =
                offline_config_parser_ptr->GetBmcMonitoredComponentsResponse();
        if (!bmc_monitored_components_response_status_or.ok()) {
          // For absl::StatusOr access practice, this shall never happen as
          // we already checked isUhmmConfigValid() above.
          LOG(ERROR) << "Uhmm is not valid, getBmcMonitoredComponentsResponse "
                        "failed with status: "
                     << bmc_monitored_components_response_status_or.status();
          enable_deterministic_bmc = false;
        } else {
          bmc_monitored_components_response =
              *bmc_monitored_components_response_status_or;
        }
        // Get the I2cRoot config from the parser.
        absl::StatusOr<uhmm::I2cRoot> i2c_root_status_or =
            offline_config_parser_ptr->GetI2cRoot();
        if (!i2c_root_status_or.ok()) {
          // For absl::StatusOr access practice, this shall never happen as
          // we already checked isUhmmConfigValid() above.
          LOG(ERROR) << "Uhmm is not valid, getI2cRoot failed with status: "
                     << i2c_root_status_or.status();
          enable_deterministic_bmc = false;
        } else {
          i2c_root = *i2c_root_status_or;
        }
      }
    }
  }

  absl::flat_hash_map<
      uhmm::BmcMonitoredComponentsResponse::MonitoredComponent::ProbePath,
      std::unique_ptr<FruScanner>>
      deterministic_fru_scanners;
  if (enable_deterministic_bmc) {
    absl::flat_hash_map<
        uhmm::BmcMonitoredComponentsResponse::MonitoredComponent::ProbePath,
        std::vector<uhmm::BmcMonitoredComponentsResponse::MonitoredComponent>>
        probe_path_to_components;

    DLOG(INFO) << "Deterministic BMC: bmc_monitored_components_response: "
               << bmc_monitored_components_response;
    // Group components by probe path.
    for (const auto& component :
         bmc_monitored_components_response.monitored_components()) {
      if (component.has_probe_path()) {
        probe_path_to_components[component.probe_path()].push_back(component);
      }
    }

    // Create deterministic fru scanners for each probe path.
    for (const auto& [probe_path, components] : probe_path_to_components) {
      switch (probe_path) {
        case uhmm::BmcMonitoredComponentsResponse::MonitoredComponent::
            PROBE_PATH_I2C: {
          // TODO(haoooamazing): create the deterministic FruScannerI2c
          // here. We now still use the existing FruScannerI2c for all I2C
          // devices.
          break;
        }
        case uhmm::BmcMonitoredComponentsResponse::MonitoredComponent::
            PROBE_PATH_NCSI: {
          // Create FruScannerCable for NCSI.
          DLOG(INFO)
              << "Deterministic BMC: Creating FruScannerCable for NCSI with:"
              << components.size()
              << " components and root_dir: " << config.tlbmc_root_dir;
          milotic_tlbmc::FruScannerCable::Options fru_scanner_cable_options = {
              .root_dir = config.tlbmc_root_dir,
              .components = components,
          };
          std::unique_ptr<FruScannerCable> fru_scanner_cable =
              FruScannerCable::Create(fru_scanner_cable_options);
          deterministic_fru_scanners[probe_path] = std::move(fru_scanner_cable);
          break;
        }
        case uhmm::BmcMonitoredComponentsResponse::MonitoredComponent::
            PROBE_PATH_ADC: {
          // Create FruScannerAdc for ADC.
          DLOG(INFO)
              << "Deterministic BMC: Creating FruScannerAdc for ADC with:"
              << components.size() << " components.";
          milotic_tlbmc::FruScannerAdc::Options fru_scanner_adc_options = {
              .components = components,
          };
          std::unique_ptr<FruScannerAdc> fru_scanner_adc =
              FruScannerAdc::Create(fru_scanner_adc_options);
          deterministic_fru_scanners[probe_path] = std::move(fru_scanner_adc);
          break;
        }
        case uhmm::BmcMonitoredComponentsResponse::MonitoredComponent::
            PROBE_PATH_SMBIOS: {
          options.expected_smbios_components = components;
          break;
        }
        default:
          LOG(ERROR) << "Unsupported probe path: "
                     << uhmm::BmcMonitoredComponentsResponse::
                            MonitoredComponent::ProbePath_Name(probe_path);
      }
    }
  }

  LOG(WARNING) << "enable_deterministic_bmc: " << enable_deterministic_bmc;

  // Enable deterministic periodic fru scanning only if both deterministic bmc
  // and deterministic periodic fru scanning are enabled in the config.
  bool enable_deterministic_periodic_fru_scanning =
      enable_deterministic_bmc &&
      config.enable_deterministic_periodic_fru_scanning;

  LOG(WARNING) << "enable_deterministic_periodic_fru_scanning: "
               << enable_deterministic_periodic_fru_scanning;

  // Create FruScannerI2c instance.
  FruScannerI2c::Options fru_scanner_options;
  fru_scanner_options.root_dir = config.tlbmc_root_dir;
  fru_scanner_options.deterministic_scan = enable_deterministic_bmc;
  fru_scanner_options.i2c_root = i2c_root;
  fru_scanner_options.i2c_devices_sysfs_dir = config.tlbmc_i2c_sysfs_path;
  DLOG(INFO) << "i2c_devices_sysfs_dir: "
             << fru_scanner_options.i2c_devices_sysfs_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) {
    DLOG(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);
  }

  options.proto_parser = milotic_tlbmc::ProtoConfigParser::Create(
      config.tlbmc_proto_config_location);
  options.proto_parser->LoadProtoConfigs();

  std::unique_ptr<GpioCollector> gpio_collector;
  if (milotic_tlbmc::GetTlbmcConfig().gpio_collector_module().enabled()) {
    absl::StatusOr<std::unique_ptr<GpioCollector>> gpio_collector_status_or;
    if (config.tlbmc_in_unit_test &&
        config.tlbmc_unit_test_enable_fake_gpio_collector) {
      gpio_collector_status_or = FakeGpioCollector::Create();
    } else {
      gpio_collector_status_or = options.collector_factory->CreateGpioCollector(
          {.gpio_configs = options.proto_parser->GetGpioConfigs()});
    }
    if (!gpio_collector_status_or.ok()) {
      LOG(ERROR) << "Failed to create GPIO collector: "
                 << gpio_collector_status_or.status();
      return nullptr;
    }
    gpio_collector = std::move(*gpio_collector_status_or);
  }

  // TODO: b/509637698 - Migrate after the new model is ready.
  // Create FruScannerGpio instance.
  std::unique_ptr<FruScannerGpio> fru_scanner_gpio = nullptr;
  if (gpio_collector != nullptr) {
    FruScannerGpio::Options fru_scanner_gpio_options;
    auto gpio_fru_component_configs =
        options.proto_parser->GetGpioFruComponentConfigs();
    fru_scanner_gpio_options.components =
        std::vector<milotic_tlbmc::GpioFruComponent>(
            gpio_fru_component_configs.gpio_fru_components().begin(),
            gpio_fru_component_configs.gpio_fru_components().end());
    fru_scanner_gpio_options.gpio_collector = gpio_collector.get();
    absl::StatusOr<std::unique_ptr<FruScannerGpio>> fru_scanner_gpio_status_or =
        FruScannerGpio::Create(fru_scanner_gpio_options);
    if (!fru_scanner_gpio_status_or.ok()) {
      LOG(ERROR) << "Failed to create FruScannerGpio: "
                 << fru_scanner_gpio_status_or.status();
      return nullptr;
    }
    fru_scanner_gpio = std::move(*fru_scanner_gpio_status_or);
  }

  // TODO: b/526494991 - Migrate after the new model is ready.
  // Create FruScannerUsb instance if config is not empty.
  const milotic_tlbmc::UsbFruComponentConfigs usb_fru_component_configs =
      options.proto_parser->GetUsbFruComponentConfigs();
  std::unique_ptr<FruScannerUsb> fru_scanner_usb = nullptr;
  if (!usb_fru_component_configs.usb_fru_components().empty()) {
    FruScannerUsb::Options fru_scanner_usb_options = {
        .soc_path = usb_fru_component_configs.soc_path(),
        .components = std::vector<milotic_tlbmc::UsbFruComponent>(
            usb_fru_component_configs.usb_fru_components().begin(),
            usb_fru_component_configs.usb_fru_components().end()),
    };

    fru_scanner_usb = FruScannerUsb::Create(fru_scanner_usb_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);
  // Don't pass null fru scanner to the fru collector.
  if (fru_scanner_gpio != nullptr) {
    fru_scanners[milotic_tlbmc::AdHocScannerType::SCANNER_TYPE_GPIO] =
        std::move(fru_scanner_gpio);
  }
  if (fru_scanner_usb != nullptr) {
    fru_scanners[milotic_tlbmc::AdHocScannerType::SCANNER_TYPE_USB] =
        std::move(fru_scanner_usb);
  }

  std::string disable_tlbmc_file_path = config.tlbmc_disable_file;
  if (std::filesystem::exists(disable_tlbmc_file_path)) {
    LOG(WARNING) << "Forcefully disable tlbmc";
    return nullptr;
  }

  options.gpio_collector = std::move(gpio_collector);
  options.fru_scanners = std::move(fru_scanners);
  options.deterministic_fru_scanners = std::move(deterministic_fru_scanners);
  options.iio_sysfs = std::make_unique<IioSysfs>(milotic_tlbmc::IioSysfsConfig{
      .iio_sysfs_path = config.tlbmc_iio_sysfs_path});
  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.led_sysfs = std::make_unique<LedSysfs>(milotic_tlbmc::LedSysfsConfig{
      .led_sysfs_path = config.tlbmc_led_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.root_path = config.tlbmc_root_dir;
  options.tlbmc_unit_test_enable_fake_gpio_collector =
      config.tlbmc_unit_test_enable_fake_gpio_collector;
  options.smbios_single_host_file_path = config.smbios_single_host_file_path;
  options.smbios_multihost_host_file_path_prefix =
      config.smbios_multihost_host_file_path_prefix;
  options.smbios_single_host_status_file_path =
      config.smbios_single_host_status_file_path;
  options.smbios_multihost_host_status_file_path_prefix =
      config.smbios_multihost_host_status_file_path_prefix;

  // Create SmbiosCodec
  platforms::gbmc::hal::SmbiosCodecConfig codec_config;
  absl::StatusOr<std::shared_ptr<platforms::gbmc::hal::SmbiosCodec>>
      smbios_codec = platforms::gbmc::hal::SmbiosCodecRegistry::Create(
          "SmbiosCodec", codec_config);
  if (!smbios_codec.ok()) {
    LOG(ERROR) << "Failed to create SmbiosCodec: " << smbios_codec.status();
  } else {
    // Create SmbiosInventory
    platforms::gbmc::hal::SmbiosInventoryConfig inventory_config;
    inventory_config.codec = smbios_codec->get();
    inventory_config.search_paths = config.smbios_search_paths;

    absl::StatusOr<std::shared_ptr<platforms::gbmc::hal::SmbiosInventory>>
        smbios_inventory =
            platforms::gbmc::hal::SmbiosInventoryRegistry::Create(
                "SmbiosInventory", inventory_config);
    if (!smbios_inventory.ok()) {
      LOG(ERROR) << "Failed to create SmbiosInventory: "
                 << smbios_inventory.status();
    } else {
      options.smbios_inventory = *std::move(smbios_inventory);
      options.smbios_codec = *std::move(smbios_codec);
    }
  }

  options.private_key_path = config.private_key_path;
  options.cert_path = config.own_cert_path;
  options.trust_bundle_path = config.trust_bundle_path;
  options.trust_bundle_signature_path = config.trust_bundle_signature_path;
  options.owner_verification_cert_path = config.owner_verification_cert_path;
  options.owner_verification_config_path =
      config.owner_verification_config_path;
  options.bmc_ssh_trusted_user_ca_keys_path =
      config.bmc_ssh_trusted_user_ca_keys_path;
  options.os_verification_cert_path = config.os_verification_cert_path;
  options.os_verification_key_path = config.os_verification_key_path;
  options.serial_console_trusted_user_ca_keys_path =
      config.serial_console_trusted_user_ca_keys_path;
  options.serial_console_detatched_signature_path =
      config.serial_console_detatched_signature_path;
  options.syslog_client_conf_override_path =
      config.syslog_client_conf_override_path;
  options.syslog_root_cert_path = config.syslog_root_cert_path;
  options.offline_node_entities_path = config.offline_node_entity_path;
  options.enable_deterministic_bmc = enable_deterministic_bmc;
  options.enable_deterministic_periodic_fru_scanning =
      enable_deterministic_periodic_fru_scanning;
  options.periodic_deterministic_fru_scan_interval_ms =
      config.periodic_deterministic_fru_scan_interval_ms;
  options.system_registry = config.system_registry;
  options.enable_hft_test_sensor = config.enable_hft_test_sensor;
  options.credential_manager = config.credential_manager;

  if (config.nic_accessor_factory != nullptr) {
    options.nic_accessor_factory =
        [factory = config.nic_accessor_factory](
            ::milotic_tlbmc::nic_veeprom::NicTelemetryVersion version, int bus,
            int address)
        -> std::unique_ptr<::milotic_tlbmc::nic_veeprom::Accessor> {
      return (*factory)(version, bus, address);
    };
  }

  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()) {
    std::vector<std::string> all_shm_sensor_keys =
        (*tlbmc_store)->GetAllSharedMemorySensorKeys();
    absl::Status shm_init_status =
        !config.tlbmc_in_unit_test
            ? StaticSharedMemoryServer::InitializeSharedMemory(
                  all_shm_sensor_keys)
            : StaticSharedMemoryServer::InitializeSharedMemoryForUnitTest(
                  all_shm_sensor_keys);

    if (!shm_init_status.ok()) {
      // Shared memory initialization can fail due to various reasons, such
      // as insufficient permissions, resource limits (e.g., running out of
      // memory or exceeding the maximum number of segments), or conflicts
      // with existing segments. This is not treated as a fatal error for
      // bmcweb, but sensor collection via shared memory will not work.
      LOG(WARNING) << "Failed to initialize shared memory segment: "
                   << shm_init_status;
    }

    if (config.tlbmc_in_unit_test) {
      StaticSharedMemoryServer::SetupInstanceForUnitTest();
      StaticSharedMemoryClient::SetupInstanceForUnitTest();
    }
    // Start the sensor & thermal collection after the shared memory server is
    // initialized.
    (*tlbmc_store)->StartSensorCollection();
    (*tlbmc_store)->StartThermalCollection();

    // Schedule periodic deterministic FRU scanning.
    (*tlbmc_store)->GetFruCollector()->SetUpPeriodicDeterministicFruScanning();
    LOG(WARNING) << "tlbmc enabled with real store";
    auto tlbmc_app =
        std::make_unique<milotic_tlbmc::RedfishApp>(std::move(*tlbmc_store));
    if (!config.tlbmc_defer_router_registration_and_validation) {
      RegisterTlbmcRoutesAndValidate(tlbmc_app.get(), config.tlbmc_root_dir);
    }
    return tlbmc_app;
  }
  LOG(ERROR) << "Cannot create tlBMC store!! Error: " << tlbmc_store.status()
             << " - Enabling tlBMC with dummy store.";
  auto store = std::make_unique<milotic_tlbmc::EmptyStore>();
  auto tlbmc_app =
      std::make_unique<milotic_tlbmc::RedfishApp>(std::move(store));
  if (!config.tlbmc_defer_router_registration_and_validation) {
    RegisterTlbmcRoutesAndValidate(tlbmc_app.get(), config.tlbmc_root_dir);
  }
  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;

  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);
  }

  if (milotic_tlbmc::GetTlbmcConfig().redfish_rate_limiter_module().enabled() &&
      tlbmc_app != nullptr) {
    rate_limiter_ = std::make_unique<crow::RateLimiter>(
        milotic_tlbmc::RateLimiterConfig::GetConfig(), config.clock);
    const_cast<milotic_tlbmc::RedfishApp*>(tlbmc_app)
        ->SetRateLimiterUpdateCallback(
            [this](const milotic_tlbmc::RedfishRateLimiterModule& config) {
              if (rate_limiter_) {
                rate_limiter_->UpdateConfig(config);
              }
            });
  }

  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_, rate_limiter_.get());

  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()) {
      subscription_manager = milotic_hft::SubscriptionManagerImpl::Create(
          std::move(*data_source),
          std::make_unique<milotic_tlbmc::HftTaskSchedulerAdapter>());
    } else {
      LOG(ERROR) << "Failed to create hft source adapter: "
                 << data_source.status();
    }
  }

  if (subscription_manager != nullptr && config_.enable_hft) {
    milotic_hft::HftServiceOptions hft_options;
    hft_options.is_serving_allowed = [cert_provider = cert_provider_.get()]() {
      return cert_provider != nullptr &&
             cert_provider->GetServerStatus() ==
                 ::milotic::redfish::BmcWebCertProvider::ServerStatus::
                     kWithRootCertsAndProdSignedCert;
    };
    hft_service_ = std::make_unique<milotic_hft::HftServiceImpl>(
        std::move(hft_options), std::move(subscription_manager),
        &::milotic::authz::BmcWebAuthorizerSingleton::GetInstance()
             .GetRedfishAuthorizer());
    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(),
    };
    if (tlbmc_app != nullptr &&
        tlbmc_app->GetStore()->IsDeterministicBmcEnabled()) {
      fru_service_options.enable_deterministic_bmc = true;
      fru_service_options.fru_collector =
          tlbmc_app->GetStore()->GetFruCollector();
    } else if (config_
                   .tlbmc_unit_test_enable_fake_fru_collector_for_fast_sanity) {
      fru_service_options.enable_deterministic_bmc = true;
      static auto* fake_fru_collector =
          new milotic_tlbmc::FakeFruCollectorForTest();
      fru_service_options.fru_collector = fake_fru_collector;
    }

#ifdef UNIT_TEST_BUILD
    fru_service_options_ = fru_service_options;
#endif

    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);

  // Set keepalive to kill zombie connections as buildup of connections can lead
  // to fd exhaustion. Keepalive will be sent every 2 hours by default.
  builder.AddChannelArgument(GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS, 1);
  server_ = builder.BuildAndStart();
}

}  // namespace milotic

#pragma GCC diagnostic pop