SERVER-70053 Serialize and deserialize DatabaseName correctly in multitenancy mode

35 files changed, 488 insertions, 63 deletions

diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index a16121cf5a2..a848fd32367 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -214,7 +214,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/util/concurrency/admission_context',
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         'auth/auth',
         'auth/user_acquisition_stats',
         'prepare_conflict_tracker',
@@ -525,6 +525,7 @@ env.Library(
         'change_stream_state.idl',
     ],
     LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         'server_base',
     ],
 )
@@ -776,6 +777,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/auth/authprivilege',
         '$BUILD_DIR/mongo/db/timeseries/timeseries_options',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         'server_base',
     ],
 )
@@ -787,7 +789,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/pipeline/document_sources_idl',
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         'server_base',
     ],
 )
@@ -822,7 +824,7 @@ env.Library(
         '$BUILD_DIR/mongo/rpc/command_status',
         '$BUILD_DIR/mongo/rpc/rewrite_state_change_errors',
         '$BUILD_DIR/mongo/rpc/rpc',
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         'audit',
         'coll_mod_command_idl',
         'index_commands_idl',
@@ -1727,6 +1729,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/auth/authprivilege',
         '$BUILD_DIR/mongo/db/catalog/commit_quorum_options',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         'server_base',
     ],
 )
@@ -2102,7 +2105,7 @@ env.Library(
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/repl/optime',
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         'server_base',
         'service_context',
     ],
diff --git a/src/mongo/db/basic_types.idl b/src/mongo/db/basic_types.idl
index 4a234f6bab4..c10917f26a7 100644
--- a/src/mongo/db/basic_types.idl
+++ b/src/mongo/db/basic_types.idl
@@ -34,6 +34,7 @@ global:
         - "mongo/db/logical_time.h"
         - "mongo/db/namespace_string.h"
         - "mongo/db/tenant_id.h"
+        - "mongo/util/database_name_util.h"
         - "mongo/util/namespace_string_util.h"
         - "mongo/util/uuid.h"
 
@@ -285,8 +286,8 @@ types:
         bson_serialization_type: string
         description: "A MongoDB DatabaseName"
         cpp_type: "mongo::DatabaseName"
-        serializer: "mongo::DatabaseName::toString"
-        deserializer: "mongo::DatabaseName"
+        serializer: "::mongo::DatabaseNameUtil::serialize"
+        deserializer: "::mongo::DatabaseNameUtil::deserialize"
         deserialize_with_tenant: true
 
 enums:
diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript
index db5d7a4f352..7d495903c0f 100644
--- a/src/mongo/db/catalog/SConscript
+++ b/src/mongo/db/catalog/SConscript
@@ -101,7 +101,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/concurrency/deferred_writer',
         '$BUILD_DIR/mongo/db/server_base',
         '$BUILD_DIR/mongo/db/service_context',
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         'collection_options',
     ],
 )
@@ -309,7 +309,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/storage/storage_options',
         '$BUILD_DIR/mongo/db/views/util',
         '$BUILD_DIR/mongo/db/views/views',
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         'collection',
     ],
 )
@@ -543,7 +543,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/ttl_collection_cache',
         '$BUILD_DIR/mongo/db/views/view_catalog_helpers',
         '$BUILD_DIR/mongo/db/views/views',
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         'cannot_convert_index_to_unique_info',
         'clustered_collection_options',
         'collection_options',
@@ -589,7 +589,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/server_base',
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
diff --git a/src/mongo/db/catalog/drop_indexes.cpp b/src/mongo/db/catalog/drop_indexes.cpp
index 22b55c90e71..88005e6d51e 100644
--- a/src/mongo/db/catalog/drop_indexes.cpp
+++ b/src/mongo/db/catalog/drop_indexes.cpp
@@ -572,9 +572,10 @@ Status dropIndexesForApplyOps(OperationContext* opCtx,
                               const NamespaceString& nss,
                               const BSONObj& cmdObj) try {
     BSONObjBuilder bob(cmdObj);
-    bob.append("$db", nss.db());
+    bob.append("$db", nss.dbName().db());
     auto cmdObjWithDb = bob.obj();
-    auto parsed = DropIndexes::parse(IDLParserContext{"dropIndexes"}, cmdObjWithDb);
+    auto parsed = DropIndexes::parse(
+        IDLParserContext{"dropIndexes", false /* apiStrict */, nss.tenantId()}, cmdObjWithDb);
 
     return writeConflictRetry(opCtx, "dropIndexes", nss.db(), [opCtx, &nss, &cmdObj, &parsed] {
         AutoGetCollection collection(opCtx, nss, MODE_X);
diff --git a/src/mongo/db/commands.cpp b/src/mongo/db/commands.cpp
index e3371fde4e4..b1c1e586d4d 100644
--- a/src/mongo/db/commands.cpp
+++ b/src/mongo/db/commands.cpp
@@ -59,6 +59,7 @@
 #include "mongo/rpc/write_concern_error_detail.h"
 #include "mongo/s/stale_exception.h"
 #include "mongo/util/assert_util.h"
+#include "mongo/util/database_name_util.h"
 #include "mongo/util/fail_point.h"
 #include "mongo/util/str.h"
 
@@ -873,7 +874,8 @@ public:
         : CommandInvocation(command),
           _command(command),
           _request(request),
-          _dbName(_request.getValidatedTenantId(), _request.getDatabase()) {}
+          _dbName(DatabaseNameUtil::deserialize(_request.getValidatedTenantId(),
+                                                _request.getDatabase())) {}
 
 private:
     void run(OperationContext* opCtx, rpc::ReplyBuilderInterface* result) override {
diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript
index 27af0649b6d..0074dc859c6 100644
--- a/src/mongo/db/commands/SConscript
+++ b/src/mongo/db/commands/SConscript
@@ -161,6 +161,7 @@ env.Library(
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/read_write_concern_defaults',
         '$BUILD_DIR/mongo/db/server_base',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
@@ -179,6 +180,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/commands',
         '$BUILD_DIR/mongo/db/server_base',
         '$BUILD_DIR/mongo/rpc/client_metadata',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         '$BUILD_DIR/mongo/util/net/ssl_manager',
         'test_commands_enabled',
     ],
@@ -233,6 +235,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/auth/authprivilege',
         '$BUILD_DIR/mongo/db/server_base',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
@@ -244,6 +247,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/auth/authprivilege',
         '$BUILD_DIR/mongo/db/server_base',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
@@ -262,6 +266,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/server_base',
         '$BUILD_DIR/mongo/db/server_options',
         '$BUILD_DIR/mongo/db/timeseries/timeseries_options',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
@@ -272,6 +277,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/server_base',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
@@ -283,6 +289,7 @@ env.Library(
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/server_base',
         '$BUILD_DIR/mongo/idl/cluster_server_parameter',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
@@ -435,6 +442,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/server_base',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
@@ -445,7 +453,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/server_base',
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
@@ -456,6 +464,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/server_base',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
@@ -469,6 +478,7 @@ env.Library(
         '$BUILD_DIR/mongo/bson/bson_validate',
         '$BUILD_DIR/mongo/db/server_base',
         '$BUILD_DIR/mongo/util/fail_point',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
@@ -680,6 +690,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/server_options_core',
         '$BUILD_DIR/mongo/db/serverless/serverless_types_idl',
         '$BUILD_DIR/mongo/idl/idl_parser',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 env.Library(
@@ -692,6 +703,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/auth/authprivilege',
         '$BUILD_DIR/mongo/db/server_base',
         '$BUILD_DIR/mongo/idl/idl_parser',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
@@ -705,6 +717,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/server_base',
         '$BUILD_DIR/mongo/db/write_concern_options',
         '$BUILD_DIR/mongo/idl/idl_parser',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
diff --git a/src/mongo/db/commands/count_cmd.cpp b/src/mongo/db/commands/count_cmd.cpp
index c1cd599a6f7..21ec11df16f 100644
--- a/src/mongo/db/commands/count_cmd.cpp
+++ b/src/mongo/db/commands/count_cmd.cpp
@@ -45,6 +45,7 @@
 #include "mongo/db/query/view_response_formatter.h"
 #include "mongo/db/s/collection_sharding_state.h"
 #include "mongo/logv2/log.h"
+#include "mongo/util/database_name_util.h"
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kCommand
 
@@ -138,7 +139,8 @@ public:
                    const OpMsgRequest& opMsgRequest,
                    ExplainOptions::Verbosity verbosity,
                    rpc::ReplyBuilderInterface* result) const override {
-        DatabaseName dbName(opMsgRequest.getValidatedTenantId(), opMsgRequest.getDatabase());
+        DatabaseName dbName = DatabaseNameUtil::deserialize(opMsgRequest.getValidatedTenantId(),
+                                                            opMsgRequest.getDatabase());
         const BSONObj& cmdObj = opMsgRequest.body;
         // Acquire locks. The RAII object is optional, because in the case
         // of a view, the locks need to be released.
@@ -151,7 +153,10 @@ public:
 
         CountCommandRequest request(NamespaceStringOrUUID(NamespaceString{}));
         try {
-            request = CountCommandRequest::parse(IDLParserContext("count"), opMsgRequest);
+            request = CountCommandRequest::parse(
+                IDLParserContext(
+                    "count", false /* apiStrict */, opMsgRequest.getValidatedTenantId()),
+                opMsgRequest);
         } catch (...) {
             return exceptionToStatus();
         }
@@ -234,7 +239,8 @@ public:
         CurOpFailpointHelpers::waitWhileFailPointEnabled(
             &hangBeforeCollectionCount, opCtx, "hangBeforeCollectionCount", []() {}, nss);
 
-        auto request = CountCommandRequest::parse(IDLParserContext("count"), cmdObj);
+        auto request = CountCommandRequest::parse(
+            IDLParserContext("count", false /* apiStrict */, dbName.tenantId()), cmdObj);
         if (shouldDoFLERewrite(request)) {
             processFLECountD(opCtx, nss, &request);
         }
diff --git a/src/mongo/db/commands/distinct.cpp b/src/mongo/db/commands/distinct.cpp
index c4c89cbf66c..070da8d285c 100644
--- a/src/mongo/db/commands/distinct.cpp
+++ b/src/mongo/db/commands/distinct.cpp
@@ -59,6 +59,7 @@
 #include "mongo/db/s/collection_sharding_state.h"
 #include "mongo/db/views/resolved_view.h"
 #include "mongo/logv2/log.h"
+#include "mongo/util/database_name_util.h"
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kQuery
 
@@ -146,7 +147,8 @@ public:
                    const OpMsgRequest& request,
                    ExplainOptions::Verbosity verbosity,
                    rpc::ReplyBuilderInterface* result) const override {
-        const DatabaseName dbName(request.getValidatedTenantId(), request.getDatabase());
+        const DatabaseName dbName =
+            DatabaseNameUtil::deserialize(request.getValidatedTenantId(), request.getDatabase());
         const BSONObj& cmdObj = request.body;
         // Acquire locks. The RAII object is optional, because in the case of a view, the locks
         // need to be released.
diff --git a/src/mongo/db/commands/explain_cmd.cpp b/src/mongo/db/commands/explain_cmd.cpp
index 85f70ed20e9..137d418b389 100644
--- a/src/mongo/db/commands/explain_cmd.cpp
+++ b/src/mongo/db/commands/explain_cmd.cpp
@@ -33,6 +33,7 @@
 #include "mongo/db/commands.h"
 #include "mongo/db/explain_gen.h"
 #include "mongo/db/query/explain.h"
+#include "mongo/util/database_name_util.h"
 #include "mongo/util/str.h"
 
 namespace mongo {
@@ -101,7 +102,8 @@ public:
                std::unique_ptr<CommandInvocation> innerInvocation)
         : CommandInvocation(explainCommand),
           _outerRequest{&request},
-          _dbName(_outerRequest->getValidatedTenantId(), _outerRequest->getDatabase()),
+          _dbName(DatabaseNameUtil::deserialize(_outerRequest->getValidatedTenantId(),
+                                                _outerRequest->getDatabase())),
           _verbosity{std::move(verbosity)},
           _innerRequest{std::move(innerRequest)},
           _innerInvocation{std::move(innerInvocation)} {}
diff --git a/src/mongo/db/commands/find_cmd.cpp b/src/mongo/db/commands/find_cmd.cpp
index ecbdd20ab9a..d048a2ec616 100644
--- a/src/mongo/db/commands/find_cmd.cpp
+++ b/src/mongo/db/commands/find_cmd.cpp
@@ -65,6 +65,7 @@
 #include "mongo/db/transaction/transaction_participant.h"
 #include "mongo/logv2/log.h"
 #include "mongo/rpc/get_status_from_command_result.h"
+#include "mongo/util/database_name_util.h"
 #include "mongo/util/fail_point.h"
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kQuery
@@ -231,7 +232,8 @@ public:
         Invocation(const FindCmd* definition, const OpMsgRequest& request)
             : CommandInvocation(definition),
               _request(request),
-              _dbName(_request.getValidatedTenantId(), _request.getDatabase()) {
+              _dbName(DatabaseNameUtil::deserialize(_request.getValidatedTenantId(),
+                                                    _request.getDatabase())) {
             invariant(_request.body.isOwned());
         }
 
diff --git a/src/mongo/db/commands/list_databases_common.h b/src/mongo/db/commands/list_databases_common.h
index 14dc0884689..2143131f882 100644
--- a/src/mongo/db/commands/list_databases_common.h
+++ b/src/mongo/db/commands/list_databases_common.h
@@ -41,6 +41,7 @@
 #include "mongo/db/operation_context.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/storage/storage_engine.h"
+#include "mongo/util/database_name_util.h"
 
 
 namespace mongo {
@@ -97,7 +98,8 @@ int64_t setReplyItems(OperationContext* opCtx,
             continue;
         }
 
-        ReplyItemType item(dbName.db());
+        // If setTenantId is true, always return the dbName without the tenantId
+        ReplyItemType item(setTenantId ? dbName.db() : DatabaseNameUtil::serialize(dbName));
         if (setTenantId) {
             initializeItemWithTenantId(item, dbName);
         }
diff --git a/src/mongo/db/commands/map_reduce_command_base.h b/src/mongo/db/commands/map_reduce_command_base.h
index a235b9345da..ce0a4ea0661 100644
--- a/src/mongo/db/commands/map_reduce_command_base.h
+++ b/src/mongo/db/commands/map_reduce_command_base.h
@@ -32,6 +32,7 @@
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/mr_common.h"
 #include "mongo/db/repl/replication_coordinator.h"
+#include "mongo/util/database_name_util.h"
 
 namespace mongo {
 
@@ -102,7 +103,8 @@ public:
         auto explain = boost::make_optional(verbosity);
         try {
             _explainImpl(opCtx,
-                         DatabaseName(request.getValidatedTenantId(), request.getDatabase()),
+                         DatabaseNameUtil::deserialize(request.getValidatedTenantId(),
+                                                       request.getDatabase()),
                          request.body,
                          builder,
                          explain);
diff --git a/src/mongo/db/commands/pipeline_command.cpp b/src/mongo/db/commands/pipeline_command.cpp
index 8b6129b9b6f..b2fb0341568 100644
--- a/src/mongo/db/commands/pipeline_command.cpp
+++ b/src/mongo/db/commands/pipeline_command.cpp
@@ -47,6 +47,7 @@
 #include "mongo/db/query/query_knobs_gen.h"
 #include "mongo/idl/idl_parser.h"
 #include "mongo/stdx/unordered_set.h"
+#include "mongo/util/database_name_util.h"
 
 namespace mongo {
 namespace {
@@ -84,7 +85,8 @@ public:
         boost::optional<ExplainOptions::Verbosity> explainVerbosity) override {
         const auto aggregationRequest = aggregation_request_helper::parseFromBSON(
             opCtx,
-            DatabaseName(opMsgRequest.getValidatedTenantId(), opMsgRequest.getDatabase()),
+            DatabaseNameUtil::deserialize(opMsgRequest.getValidatedTenantId(),
+                                          opMsgRequest.getDatabase()),
             opMsgRequest.body,
             explainVerbosity,
             APIParameters::get(opCtx).getAPIStrict().value_or(false));
@@ -119,7 +121,8 @@ public:
                    PrivilegeVector privileges)
             : CommandInvocation(cmd),
               _request(request),
-              _dbName(request.getValidatedTenantId(), request.getDatabase()),
+              _dbName(DatabaseNameUtil::deserialize(request.getValidatedTenantId(),
+                                                    request.getDatabase())),
               _aggregationRequest(std::move(aggregationRequest)),
               _liteParsedPipeline(_aggregationRequest),
               _privileges(std::move(privileges)) {
diff --git a/src/mongo/db/commands/user_management_commands.cpp b/src/mongo/db/commands/user_management_commands.cpp
index e955d44d8c4..328b214ae5e 100644
--- a/src/mongo/db/commands/user_management_commands.cpp
+++ b/src/mongo/db/commands/user_management_commands.cpp
@@ -80,6 +80,7 @@
 #include "mongo/s/write_ops/batched_command_response.h"
 #include "mongo/stdx/unordered_set.h"
 #include "mongo/transport/service_entry_point.h"
+#include "mongo/util/database_name_util.h"
 #include "mongo/util/icu.h"
 #include "mongo/util/net/ssl_manager.h"
 #include "mongo/util/password_digest.h"
@@ -735,7 +736,7 @@ public:
             as->grantInternalAuthorization(_client.get());
         }
 
-        _dbName = DatabaseName(tenant, kAdminDB);
+        _dbName = DatabaseNameUtil::deserialize(tenant, kAdminDB);
 
         AlternativeClientRegion clientRegion(_client);
         _sessionInfo.setStartTransaction(true);
diff --git a/src/mongo/db/commands/validate_db_metadata_cmd.cpp b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
index ff51cd3b2d2..6cfc2fc939a 100644
--- a/src/mongo/db/commands/validate_db_metadata_cmd.cpp
+++ b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
@@ -43,6 +43,7 @@
 #include "mongo/db/multitenancy.h"
 #include "mongo/db/views/view_catalog_helpers.h"
 #include "mongo/logv2/log.h"
+#include "mongo/util/database_name_util.h"
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kCommand
 
@@ -125,8 +126,9 @@ public:
             // If there is no database name present in the input, run validation against all the
             // databases.
             auto dbNames = validateCmdRequest.getDb()
-                ? std::vector<DatabaseName>{DatabaseName(getActiveTenant(opCtx),
-                                                         validateCmdRequest.getDb()->toString())}
+                ? std::vector<DatabaseName>{DatabaseNameUtil::deserialize(
+                      validateCmdRequest.getDbName().tenantId(),
+                      validateCmdRequest.getDb()->toString())}
                 : collectionCatalog->getAllDbNames();
 
             for (const auto& dbName : dbNames) {
diff --git a/src/mongo/db/curop.cpp b/src/mongo/db/curop.cpp
index eac0c5cbdfd..73ddcb6827e 100644
--- a/src/mongo/db/curop.cpp
+++ b/src/mongo/db/curop.cpp
@@ -66,6 +66,18 @@ namespace {
 
 auto& oplogGetMoreStats = makeServerStatusMetric<TimerStats>("repl.network.oplogGetMoresProcessed");
 
+BSONObj serializeDollarDbInOpDescription(boost::optional<TenantId> tenantId,
+                                         const BSONObj& cmdObj) {
+    auto db = cmdObj["$db"];
+    if (!db) {
+        return cmdObj;
+    }
+
+    auto dbName = DatabaseNameUtil::deserialize(tenantId, db.String());
+    auto newCmdObj =
+        cmdObj.addField(BSON("$db" << DatabaseNameUtil::serialize(dbName)).firstElement());
+    return newCmdObj;
+}
 }  // namespace
 
 /**
@@ -256,6 +268,10 @@ void CurOp::reportCurrentOpForClient(OperationContext* opCtx,
 #endif
 }
 
+void CurOp::setOpDescription_inlock(const BSONObj& opDescription) {
+    _opDescription = serializeDollarDbInOpDescription(_nss.tenantId(), opDescription);
+}
+
 void CurOp::setGenericCursor_inlock(GenericCursor gc) {
     _genericCursor = std::move(gc);
 }
@@ -306,7 +322,7 @@ void CurOp::setGenericOpRequestDetails(OperationContext* opCtx,
     _isCommand = _debug.iscommand = isCommand;
     _logicalOp = _debug.logicalOp = logicalOp;
     _networkOp = _debug.networkOp = op;
-    _opDescription = cmdObj;
+    _opDescription = serializeDollarDbInOpDescription(nss.tenantId(), cmdObj);
     _command = command;
     _nss = std::move(nss);
 }
@@ -657,7 +673,6 @@ void CurOp::reportState(OperationContext* opCtx, BSONObjBuilder* builder, bool t
     // contrast, the $currentOp aggregation stage does not have this restriction. If 'truncateOps'
     // is true, limit the size of each op to 1000 bytes. Otherwise, do not truncate.
     const boost::optional<size_t> maxQuerySize{truncateOps, 1000};
-
     appendAsObjOrString(
         "command", appendCommentField(opCtx, _opDescription), maxQuerySize, builder);
 
diff --git a/src/mongo/db/curop.h b/src/mongo/db/curop.h
index 9cccd05d9ff..9f83bd92cf1 100644
--- a/src/mongo/db/curop.h
+++ b/src/mongo/db/curop.h
@@ -705,9 +705,7 @@ public:
      * 'opDescription' must be either an owned BSONObj or guaranteed to outlive the OperationContext
      * it is associated with.
      */
-    void setOpDescription_inlock(const BSONObj& opDescription) {
-        _opDescription = opDescription;
-    }
+    void setOpDescription_inlock(const BSONObj& opDescription);
 
     /**
      * Sets the original command object.
diff --git a/src/mongo/db/namespace_string.cpp b/src/mongo/db/namespace_string.cpp
index 5455dd717be..a45f99552f5 100644
--- a/src/mongo/db/namespace_string.cpp
+++ b/src/mongo/db/namespace_string.cpp
@@ -197,7 +197,7 @@ const NamespaceString NamespaceString::kConfigQueryAnalyzersNamespace(NamespaceS
 
 NamespaceString NamespaceString::parseFromStringExpectTenantIdInMultitenancyMode(StringData ns) {
     if (!gMultitenancySupport) {
-        return NamespaceString(ns, boost::none);
+        return NamespaceString(boost::none, ns);
     }
 
     auto tenantDelim = ns.find('_');
@@ -205,11 +205,19 @@ NamespaceString NamespaceString::parseFromStringExpectTenantIdInMultitenancyMode
     // If the first '_' is after the '.' that separates the db and coll names, the '_' is part
     // of the coll name and is not a db prefix.
     if (tenantDelim == std::string::npos || collDelim < tenantDelim) {
-        return NamespaceString(ns, boost::none);
+        return NamespaceString(boost::none, ns);
     }
 
-    const TenantId tenantId(OID(ns.substr(0, tenantDelim)));
-    return NamespaceString(ns.substr(tenantDelim + 1, ns.size() - 1 - tenantDelim), tenantId);
+    auto swOID = OID::parse(ns.substr(0, tenantDelim));
+    if (swOID.getStatus() == ErrorCodes::BadValue) {
+        // If we fail to parse an OID, either the size of the substring is incorrect, or there is an
+        // invalid character. This indicates that the db has the "_" character, but it does not act
+        // as a delimeter for a tenantId prefix.
+        return NamespaceString(boost::none, ns);
+    }
+
+    const TenantId tenantId(swOID.getValue());
+    return NamespaceString(tenantId, ns.substr(tenantDelim + 1, ns.size() - 1 - tenantDelim));
 }
 
 bool NamespaceString::isListCollectionsCursorNS() const {
diff --git a/src/mongo/db/namespace_string.h b/src/mongo/db/namespace_string.h
index 4bf5cc78967..904bac9f99d 100644
--- a/src/mongo/db/namespace_string.h
+++ b/src/mongo/db/namespace_string.h
@@ -347,6 +347,8 @@ public:
      * Constructs a NamespaceString from the string 'ns'. Should only be used when reading a
      * namespace from disk. 'ns' is expected to contain a tenantId when running in Serverless mode.
      */
+    // TODO SERVER-70013 Move this function into NamespaceStringUtil, and delegate overlapping
+    // functionality to DatabaseNameUtil::parseDbNameFromStringExpectTenantIdInMultitenancyMode.
     static NamespaceString parseFromStringExpectTenantIdInMultitenancyMode(StringData ns);
 
     /**
diff --git a/src/mongo/db/ops/SConscript b/src/mongo/db/ops/SConscript
index 695633c9342..1c0819eec70 100644
--- a/src/mongo/db/ops/SConscript
+++ b/src/mongo/db/ops/SConscript
@@ -79,6 +79,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/crypto/encrypted_field_config',
         '$BUILD_DIR/mongo/crypto/fle_fields',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
diff --git a/src/mongo/db/pipeline/process_interface/SConscript b/src/mongo/db/pipeline/process_interface/SConscript
index b332ea2b352..c590971845a 100644
--- a/src/mongo/db/pipeline/process_interface/SConscript
+++ b/src/mongo/db/pipeline/process_interface/SConscript
@@ -25,7 +25,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/operation_time_tracker',
         '$BUILD_DIR/mongo/db/pipeline/field_path',
         '$BUILD_DIR/mongo/s/sharding_router_api',
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
diff --git a/src/mongo/db/query/SConscript b/src/mongo/db/query/SConscript
index 69b982bd9ba..4db9f6b6920 100644
--- a/src/mongo/db/query/SConscript
+++ b/src/mongo/db/query/SConscript
@@ -192,7 +192,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/server_base',
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
@@ -216,7 +216,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/repl/optime',
         '$BUILD_DIR/mongo/rpc/command_status',
         '$BUILD_DIR/mongo/rpc/rpc',
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         'query_request',
     ],
     LIBDEPS_PRIVATE=[
@@ -243,6 +243,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/commands/test_commands_enabled',
         '$BUILD_DIR/mongo/db/pipeline/runtime_constants_idl',
         '$BUILD_DIR/mongo/db/repl/read_concern_args',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         'hint_parser',
     ],
     LIBDEPS_PRIVATE=[
diff --git a/src/mongo/db/query/parsed_distinct.cpp b/src/mongo/db/query/parsed_distinct.cpp
index 61562191b6c..21781211771 100644
--- a/src/mongo/db/query/parsed_distinct.cpp
+++ b/src/mongo/db/query/parsed_distinct.cpp
@@ -246,7 +246,7 @@ StatusWith<ParsedDistinct> ParsedDistinct::parse(OperationContext* opCtx,
                                                  const ExtensionsCallback& extensionsCallback,
                                                  bool isExplain,
                                                  const CollatorInterface* defaultCollator) {
-    IDLParserContext ctx("distinct");
+    IDLParserContext ctx("distinct", false /* apiStrict */, nss.tenantId());
 
     DistinctCommandRequest parsedDistinct(nss);
     try {
diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index f0107256592..9d87b25ba6d 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -96,7 +96,7 @@ env.Library(
         '$BUILD_DIR/mongo/idl/idl_parser',
         '$BUILD_DIR/mongo/rpc/command_status',
         '$BUILD_DIR/mongo/s/common_s',
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         'dbcheck',
         'image_collection_entry',
         'repl_coordinator_interface',
@@ -553,7 +553,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/session/kill_sessions_local',
         '$BUILD_DIR/mongo/db/session/session_catalog_mongod',
         '$BUILD_DIR/mongo/db/storage/historical_ident_tracker',
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         'drop_pending_collection_reaper',
     ],
 )
@@ -575,7 +575,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/server_base',
         '$BUILD_DIR/mongo/db/server_feature_flags',
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
@@ -1295,6 +1295,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/server_base',
         '$BUILD_DIR/mongo/idl/idl_parser',
         '$BUILD_DIR/mongo/rpc/metadata',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         'optime',
     ],
 )
diff --git a/src/mongo/db/service_entry_point_common.cpp b/src/mongo/db/service_entry_point_common.cpp
index f7f2ad0bdbd..0c4766739e4 100644
--- a/src/mongo/db/service_entry_point_common.cpp
+++ b/src/mongo/db/service_entry_point_common.cpp
@@ -1967,12 +1967,14 @@ void curOpCommandSetup(OperationContext* opCtx, const OpMsgRequest& request) {
 
     // We construct a legacy $cmd namespace so we can fill in curOp using
     // the existing logic that existed for OP_QUERY commands
-    NamespaceString nss(request.getDatabase(), "$cmd");
+    NamespaceString nss(
+        DatabaseNameUtil::deserialize(request.getValidatedTenantId(), request.getDatabase()),
+        "$cmd");
 
     stdx::lock_guard<Client> lk(*opCtx->getClient());
+    curop->setNS_inlock(nss);
     curop->setOpDescription_inlock(request.body);
     curop->markCommand_inlock();
-    curop->setNS_inlock(nss);
 }
 
 Future<void> parseCommand(std::shared_ptr<HandleRequest::ExecutionContext> execContext) try {
diff --git a/src/mongo/db/views/SConscript b/src/mongo/db/views/SConscript
index bee9cc79c65..99023cd03eb 100644
--- a/src/mongo/db/views/SConscript
+++ b/src/mongo/db/views/SConscript
@@ -50,7 +50,7 @@ env.Library(
         'util.cpp',
     ],
     LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         'views',
     ],
 )
diff --git a/src/mongo/idl/SConscript b/src/mongo/idl/SConscript
index 9ac6fe00850..ffc13c8d153 100644
--- a/src/mongo/idl/SConscript
+++ b/src/mongo/idl/SConscript
@@ -135,7 +135,7 @@ env.CppUnitTest(
         '$BUILD_DIR/mongo/db/service_context',
         '$BUILD_DIR/mongo/rpc/message',
         '$BUILD_DIR/mongo/util/cmdline_utils/cmdline_utils',
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         '$BUILD_DIR/mongo/util/options_parser/options_parser',
         'cluster_server_parameter',
     ],
diff --git a/src/mongo/idl/idl_test.cpp b/src/mongo/idl/idl_test.cpp
index 69ddf6214e6..32da3905236 100644
--- a/src/mongo/idl/idl_test.cpp
+++ b/src/mongo/idl/idl_test.cpp
@@ -2382,15 +2382,17 @@ TEST(IDLCommand, TestConcatentateWithDb) {
 }
 
 TEST(IDLCommand, TestConcatentateWithDb_WithTenant) {
+    RAIIServerParameterControllerForTest multitenanyController("multitenancySupport", true);
     IDLParserContext ctxt("root");
 
     const auto tenantId = TenantId(OID::gen());
+    const auto prefixedDb = std::string(str::stream() << tenantId.toString() << "_db");
 
     auto testDoc = BSONObjBuilder{}
                        .append(BasicConcatenateWithDbCommand::kCommandName, "coll1")
                        .append("field1", 3)
                        .append("field2", "five")
-                       .append("$db", "db")
+                       .append("$db", prefixedDb)
                        .obj();
 
     auto testStruct =
@@ -2525,16 +2527,19 @@ TEST(IDLCommand, TestConcatentateWithDbOrUUID_TestNSS) {
 }
 
 TEST(IDLCommand, TestConcatentateWithDbOrUUID_TestNSS_WithTenant) {
+    RAIIServerParameterControllerForTest multitenanyController("multitenancySupport", true);
     IDLParserContext ctxt("root");
 
+    const auto tenantId = TenantId(OID::gen());
+    const auto prefixedDb = std::string(str::stream() << tenantId.toString() << "_db");
+
     auto testDoc = BSONObjBuilder{}
                        .append(BasicConcatenateWithDbOrUUIDCommand::kCommandName, "coll1")
                        .append("field1", 3)
                        .append("field2", "five")
-                       .append("$db", "db")
+                       .append("$db", prefixedDb)
                        .obj();
 
-    const auto tenantId = TenantId(OID::gen());
     auto testStruct =
         BasicConcatenateWithDbOrUUIDCommand::parse(ctxt, makeOMRWithTenant(testDoc, tenantId));
     ASSERT_EQUALS(testStruct.getDbName(), DatabaseName(tenantId, "db"));
@@ -2595,19 +2600,21 @@ TEST(IDLCommand, TestConcatentateWithDbOrUUID_TestUUID) {
 }
 
 TEST(IDLCommand, TestConcatentateWithDbOrUUID_TestUUID_WithTenant) {
+    RAIIServerParameterControllerForTest multitenanyController("multitenancySupport", true);
     IDLParserContext ctxt("root");
 
     UUID uuid = UUID::gen();
+    const auto tenantId = TenantId(OID::gen());
+    const auto prefixedDb = std::string(str::stream() << tenantId.toString() << "_db");
 
     auto testDoc =
         BSONObjBuilder{}
             .appendElements(BSON(BasicConcatenateWithDbOrUUIDCommand::kCommandName << uuid))
             .append("field1", 3)
             .append("field2", "five")
-            .append("$db", "db")
+            .append("$db", prefixedDb)
             .obj();
 
-    const auto tenantId = TenantId(OID::gen());
     auto testStruct =
         BasicConcatenateWithDbOrUUIDCommand::parse(ctxt, makeOMRWithTenant(testDoc, tenantId));
     ASSERT_EQUALS(testStruct.getDbName(), DatabaseName(tenantId, "db"));
@@ -3985,14 +3992,14 @@ TEST(IDLCommand, TestCommandTypeNamespaceCommand_WithMultitenancySupportOn) {
     const auto tenantId = TenantId(OID::gen());
     const auto nssWithPrefixedTenantId =
         std::string(str::stream() << tenantId.toString() << "_db.coll1");
-    auto testDoc = BSON(CommandTypeNamespaceCommand::kCommandName << nssWithPrefixedTenantId
-                                                                  << "field1" << 3 << "$db"
-                                                                  << "admin");
+    const auto prefixedAdminDb = std::string(str::stream() << tenantId.toString() << "_admin");
+
+    auto testDoc = BSON(CommandTypeNamespaceCommand::kCommandName
+                        << nssWithPrefixedTenantId << "field1" << 3 << "$db" << prefixedAdminDb);
 
     auto testStruct = CommandTypeNamespaceCommand::parse(ctxt, makeOMR(testDoc));
-    // TODO SERVER-70053: Expect tenantId to be the extracted prefix once DatabaseName uses the
-    // correct serializer/deserializer.
-    ASSERT_EQUALS(testStruct.getDbName(), DatabaseName(boost::none, "admin"));
+
+    ASSERT_EQUALS(testStruct.getDbName(), DatabaseName(tenantId, "admin"));
     // Deserialize called from parse correctly sets the tenantId field.
     ASSERT_EQUALS(testStruct.getCommandParameter(), NamespaceString(tenantId, "db.coll1"));
     assert_same_types<decltype(testStruct.getCommandParameter()), const NamespaceString&>();
diff --git a/src/mongo/rpc/SConscript b/src/mongo/rpc/SConscript
index 4ec2445e872..9d1dafcc1b6 100644
--- a/src/mongo/rpc/SConscript
+++ b/src/mongo/rpc/SConscript
@@ -44,6 +44,7 @@ protoEnv.Library(
         '$BUILD_DIR/mongo/db/bson/dotted_path_support',
         '$BUILD_DIR/mongo/db/server_base',
         '$BUILD_DIR/mongo/db/server_options_core',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         '$BUILD_DIR/third_party/wiredtiger/wiredtiger_checksum' if wiredtiger else [],
     ],
 )
diff --git a/src/mongo/rpc/op_msg.cpp b/src/mongo/rpc/op_msg.cpp
index e6213828ed4..5f8a87438a0 100644
--- a/src/mongo/rpc/op_msg.cpp
+++ b/src/mongo/rpc/op_msg.cpp
@@ -43,6 +43,7 @@
 #include "mongo/logv2/log.h"
 #include "mongo/rpc/object_check.h"
 #include "mongo/util/bufreader.h"
+#include "mongo/util/database_name_util.h"
 #include "mongo/util/hex.h"
 
 #ifdef MONGO_CONFIG_WIREDTIGER_ENABLED
@@ -300,9 +301,12 @@ OpMsgRequest OpMsgRequestBuilder::create(const DatabaseName& dbName,
     auto dollarTenant = parseDollarTenant(body);
     BSONObjBuilder bodyBuilder(std::move(body));
     bodyBuilder.appendElements(extraFields);
-    bodyBuilder.append("$db", dbName.db());
     if (dbName.tenantId()) {
+        // If we append $tenant, never include the prefix in $db as well.
+        bodyBuilder.append("$db", dbName.db());
         appendDollarTenant(bodyBuilder, dbName.tenantId().value(), dollarTenant);
+    } else {
+        bodyBuilder.append("$db", DatabaseNameUtil::serialize(dbName));
     }
 
     OpMsgRequest request;
diff --git a/src/mongo/s/SConscript b/src/mongo/s/SConscript
index 82eb4d98ee6..91c67eba373 100644
--- a/src/mongo/s/SConscript
+++ b/src/mongo/s/SConscript
@@ -143,7 +143,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/server_base',
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
@@ -235,7 +235,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/server_base',
         '$BUILD_DIR/mongo/db/timeseries/timeseries_options',
-        '$BUILD_DIR/mongo/util/namespace_string_util',
+        '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
     ],
 )
 
diff --git a/src/mongo/util/SConscript b/src/mongo/util/SConscript
index 7666304bddd..e90c9ffb348 100644
--- a/src/mongo/util/SConscript
+++ b/src/mongo/util/SConscript
@@ -228,8 +228,9 @@ env.Library(
 )
 
 env.Library(
-    target='namespace_string_util',
+    target='namespace_string_database_name_util',
     source=[
+        'database_name_util.cpp',
         'namespace_string_util.cpp',
     ],
     LIBDEPS=[
@@ -246,7 +247,17 @@ env.CppUnitTest(
         'namespace_string_util_test.cpp',
     ],
     LIBDEPS=[
-        'namespace_string_util',
+        'namespace_string_database_name_util',
+    ],
+)
+
+env.CppUnitTest(
+    target='database_name_util_test',
+    source=[
+        'database_name_util_test.cpp',
+    ],
+    LIBDEPS=[
+        'namespace_string_database_name_util',
     ],
 )
 
diff --git a/src/mongo/util/database_name_util.cpp b/src/mongo/util/database_name_util.cpp
new file mode 100644
index 00000000000..ce8712207e4
--- /dev/null
+++ b/src/mongo/util/database_name_util.cpp
@@ -0,0 +1,106 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/util/database_name_util.h"
+#include "mongo/db/database_name.h"
+#include "mongo/db/multitenancy_gen.h"
+#include "mongo/db/server_feature_flags_gen.h"
+#include "mongo/util/str.h"
+#include <ostream>
+
+namespace mongo {
+
+std::string DatabaseNameUtil::serialize(const DatabaseName& dbName) {
+    if (!gMultitenancySupport) {
+        return dbName.toString();
+    }
+
+    if (serverGlobalParams.featureCompatibility.isVersionInitialized() &&
+        gFeatureFlagRequireTenantID.isEnabled(serverGlobalParams.featureCompatibility)) {
+        return dbName.toString();
+    }
+    return dbName.toStringWithTenantId();
+}
+
+DatabaseName parseDbNameFromStringExpectTenantIdInMultitenancyMode(StringData dbName) {
+    if (!gMultitenancySupport) {
+        return DatabaseName(boost::none, dbName);
+    }
+
+    auto tenantDelim = dbName.find('_');
+    if (tenantDelim == std::string::npos) {
+        return DatabaseName(boost::none, dbName);
+    }
+
+    auto swOID = OID::parse(dbName.substr(0, tenantDelim));
+    if (swOID.getStatus() == ErrorCodes::BadValue) {
+        // If we fail to parse an OID, either the size of the substring is incorrect, or there is an
+        // invalid character. This indicates that the db has the "_" character, but it does not act
+        // as a delimeter for a tenantId prefix.
+        return DatabaseName(boost::none, dbName);
+    }
+
+    const TenantId tenantId(swOID.getValue());
+    return DatabaseName(tenantId, dbName.substr(tenantDelim + 1));
+}
+
+DatabaseName DatabaseNameUtil::deserialize(boost::optional<TenantId> tenantId, StringData db) {
+    if (db.empty()) {
+        return DatabaseName();
+    }
+
+    if (!gMultitenancySupport) {
+        massert(7005302, "TenantId must not be set", tenantId == boost::none);
+        return DatabaseName(boost::none, db);
+    }
+
+    if (serverGlobalParams.featureCompatibility.isVersionInitialized() &&
+        gFeatureFlagRequireTenantID.isEnabled(serverGlobalParams.featureCompatibility)) {
+        // TODO SERVER-62491 Remove this conditional, the tenantId should be kSystemTenantId.
+        // TODO SERVER-70876 Uncomment out this conditional to check that we always have a tenantId.
+        /* if (db != "admin" && db != "config" && db != "local")
+            massert(7005300, "TenantId must be set", tenantId != boost::none);
+        */
+
+        return DatabaseName(std::move(tenantId), db);
+    }
+
+    auto dbName = parseDbNameFromStringExpectTenantIdInMultitenancyMode(db);
+    // TenantId could be prefixed, or passed in separately (or both) and namespace is always
+    // constructed with the tenantId separately.
+    if (tenantId != boost::none) {
+        if (!dbName.tenantId()) {
+            return DatabaseName(std::move(tenantId), db);
+        }
+        massert(7005301, "TenantId must match that in db prefix", tenantId == dbName.tenantId());
+    }
+    return dbName;
+}
+
+}  // namespace mongo
diff --git a/src/mongo/util/database_name_util.h b/src/mongo/util/database_name_util.h
new file mode 100644
index 00000000000..2aa47d0be62
--- /dev/null
+++ b/src/mongo/util/database_name_util.h
@@ -0,0 +1,77 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/database_name.h"
+#include "mongo/db/tenant_id.h"
+
+namespace mongo {
+
+class DatabaseNameUtil {
+public:
+    /**
+     * Serializes a DatabaseName object.
+     *
+     * If multitenancySupport is enabled and featureFlagRequireTenantID is enabled, then tenantId is
+     * not included in the serialization.
+     * eg. serialize(DatabaseName(tenantID, "foo")) -> "foo"
+     *
+     * If multitenancySupport is enabled and featureFlagRequireTenantID is disabled, then tenantId
+     * is included in the serialization.
+     * eg. serialize(DatabaseName(tenantID, "foo")) -> "tenantID_foo"
+     *
+     * If multitenancySupport is disabled, the tenantID is not set in the DatabaseName Object.
+     * eg. serialize(DatabaseName(boost::none, "foo")) -> "foo"
+     */
+    static std::string serialize(const DatabaseName& dbName);
+
+    /**
+     * Deserializes StringData dbName to a DatabaseName object.
+     *
+     * If multitenancySupport is enabled and featureFlagRequireTenantID is enabled, then a
+     * DatabaseName object is constructed using the tenantId passed in to the constructor. The
+     * invariant requires tenantID to be initialized and passed to the constructor.
+     * eg. deserialize(tenantID, "foo") -> DatabaseName(tenantID, "foo")
+     *
+     * If multitenancySupport is enabled and featureFlagRequireTenantID is disabled, then dbName is
+     * required to be prefixed with a tenantID. The tenantID parameter is ignored and
+     * DatabaseName is constructed using only ns. The invariant requires that if a tenantID
+     * is a parameter, then the tenantID is equal to the prefixed tenantID.
+     * eg. deserialize(boost::none, "preTenantID_foo") -> DatabaseName(preTenantId,
+     * "foo")
+     *
+     * If multitenancySupport is disabled then the invariant requires tenantID to not be initialized
+     * and DatabaseName is constructed without the tenantID.
+     * eg. deserialize(boost::none, "foo") -> DatabaseName(boost::none, "foo")
+     */
+    static DatabaseName deserialize(boost::optional<TenantId> tenantId, StringData db);
+};
+
+}  // namespace mongo
diff --git a/src/mongo/util/database_name_util_test.cpp b/src/mongo/util/database_name_util_test.cpp
new file mode 100644
index 00000000000..2e541c6c451
--- /dev/null
+++ b/src/mongo/util/database_name_util_test.cpp
@@ -0,0 +1,149 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/database_name.h"
+#include "mongo/db/multitenancy_gen.h"
+#include "mongo/idl/server_parameter_test_util.h"
+#include "mongo/unittest/death_test.h"
+#include "mongo/unittest/unittest.h"
+#include "mongo/util/database_name_util.h"
+
+namespace mongo {
+
+// TenantID is not included in serialization when multitenancySupport and
+// featureFlagRequireTenantID are enabled.
+TEST(DatabaseNameUtilTest, SerializeMultitenancySupportOnFeatureFlagRequireTenantIDOn) {
+    RAIIServerParameterControllerForTest multitenanyController("multitenancySupport", true);
+    RAIIServerParameterControllerForTest featureFlagController("featureFlagRequireTenantID", true);
+    TenantId tenantId(OID::gen());
+    DatabaseName dbName(tenantId, "foo");
+    ASSERT_EQ(DatabaseNameUtil::serialize(dbName), "foo");
+}
+
+// TenantID is included in serialization when multitenancySupport is enabled and
+// featureFlagRequireTenantID is disabled.
+TEST(DatabaseNameUtilTest, SerializeMultitenancySupportOnFeatureFlagRequireTenantIDOff) {
+    RAIIServerParameterControllerForTest multitenanyController("multitenancySupport", true);
+    RAIIServerParameterControllerForTest featureFlagController("featureFlagRequireTenantID", false);
+    TenantId tenantId(OID::gen());
+    std::string tenantDbStr = str::stream() << tenantId.toString() << "_foo";
+    DatabaseName dbName(tenantId, "foo");
+    ASSERT_EQ(DatabaseNameUtil::serialize(dbName), tenantDbStr);
+}
+
+// Serialize correctly when multitenancySupport is disabled.
+TEST(DatabaseNameUtilTest, SerializeMultitenancySupportOff) {
+    RAIIServerParameterControllerForTest multitenanyController("multitenancySupport", false);
+    DatabaseName dbName(boost::none, "foo");
+    ASSERT_EQ(DatabaseNameUtil::serialize(dbName), "foo");
+}
+
+// Assert that if multitenancySupport and featureFlagRequireTenantID are on, then tenantId is set.
+// TODO SERVER-70876 Uncomment out the test case below.
+/* TEST(DatabaseNameUtilTest,
+     DeserializeAssertTenantIdSetMultitenancySupportOnFeatureFlagRequireTenantIDOn) {
+    RAIIServerParameterControllerForTest multitenanyController("multitenancySupport", true);
+    RAIIServerParameterControllerForTest featureFlagController("featureFlagRequireTenantID", true);
+    ASSERT_THROWS_CODE(
+        DatabaseNameUtil::deserialize(boost::none, "foo"), AssertionException, 7005300);
+}
+*/
+
+// If the database is an inernal db, it's acceptable not to have a tenantId even if
+// multitenancySupport and featureFlagRequireTenantID are on.
+// TODO SERVER-62491 This test should throw once we use kSystemTenantId.
+TEST(DatabaseNameUtilTest,
+     DeserializeInternalDbTenantIdSetMultitenancySupportOnFeatureFlagRequireTenantIDOn) {
+    RAIIServerParameterControllerForTest multitenanyController("multitenancySupport", true);
+    RAIIServerParameterControllerForTest featureFlagController("featureFlagRequireTenantID", true);
+    DatabaseName dbName = DatabaseNameUtil::deserialize(boost::none, "local");
+    ASSERT_EQ(dbName, DatabaseName(boost::none, "local"));
+}
+
+// Deserialize DatabaseName using the tenantID as a parameter to the DatabaseName constructor
+// when multitenancySupport and featureFlagRequireTenantID are enabled and ns does not have prefixed
+// tenantID.
+TEST(DatabaseNameUtilTest,
+     DeserializeNSSWithoutPrefixedTenantIDMultitenancySupportOnFeatureFlagRequireTenantIDOn) {
+    RAIIServerParameterControllerForTest multitenanyController("multitenancySupport", true);
+    RAIIServerParameterControllerForTest featureFlagController("featureFlagRequireTenantID", true);
+    TenantId tenantId(OID::gen());
+    DatabaseName dbName = DatabaseNameUtil::deserialize(tenantId, "foo");
+    ASSERT_EQ(dbName.db(), "foo");
+    ASSERT(dbName.tenantId());
+    ASSERT_EQ(dbName, DatabaseName(tenantId, "foo"));
+}
+
+// Deserialize DatabaseName when multitenancySupport is enabled and featureFlagRequireTenantID is
+// disabled.
+TEST(DatabaseNameUtilTest, DeserializeMultitenancySupportOnFeatureFlagRequireTenantIDOff) {
+    RAIIServerParameterControllerForTest multitenanyController("multitenancySupport", true);
+    RAIIServerParameterControllerForTest featureFlagController("featureFlagRequireTenantID", false);
+    TenantId tenantId(OID::gen());
+    std::string tenantDbStr = str::stream() << tenantId.toString() << "_foo";
+    DatabaseName dbName = DatabaseNameUtil::deserialize(boost::none, tenantDbStr);
+    DatabaseName dbName1 = DatabaseNameUtil::deserialize(tenantId, tenantDbStr);
+    ASSERT_EQ(dbName.db(), "foo");
+    ASSERT(dbName.tenantId());
+    ASSERT_EQ(dbName, DatabaseName(tenantId, "foo"));
+    ASSERT_EQ(dbName, dbName1);
+}
+
+// Assert tenantID is not initialized when multitenancySupport is disabled.
+TEST(DatabaseNameUtilTest, DeserializeMultitenancySupportOff) {
+    RAIIServerParameterControllerForTest multitenanyController("multitenancySupport", false);
+    TenantId tenantId(OID::gen());
+    ASSERT_THROWS_CODE(DatabaseNameUtil::deserialize(tenantId, "foo"), AssertionException, 7005302);
+}
+
+// Deserialize DatabaseName with prefixed tenantId when multitenancySupport and
+// featureFlagRequireTenantId are disabled.
+TEST(DatabaseNameUtilTest,
+     DeserializeWithTenantIdInStringMultitenancySupportOffFeatureFlagRequireTenantIDOff) {
+    RAIIServerParameterControllerForTest multitenanyController("multitenancySupport", false);
+    TenantId tenantId(OID::gen());
+    std::string dbNameStr = str::stream() << tenantId.toString() << "_foo";
+    DatabaseName dbName = DatabaseNameUtil::deserialize(boost::none, dbNameStr);
+    ASSERT_EQ(dbName.tenantId(), boost::none);
+    ASSERT_EQ(dbName.db(), dbNameStr);
+}
+
+// Deserialize DatabaseName when multitenancySupport and featureFlagRequireTenantID are disabled.
+TEST(DatabaseNameUtilTest, DeserializeMultitenancySupportOffFeatureFlagRequireTenantIDOff) {
+    RAIIServerParameterControllerForTest multitenanyController("multitenancySupport", false);
+    RAIIServerParameterControllerForTest featureFlagController("featureFlagRequireTenantID", false);
+    DatabaseName dbName = DatabaseNameUtil::deserialize(boost::none, "foo");
+    ASSERT_EQ(dbName.db(), "foo");
+    ASSERT(!dbName.tenantId());
+    ASSERT_EQ(dbName, DatabaseName(boost::none, "foo"));
+}
+
+}  // namespace mongo