SERVER-63712 Add support for FLE 2 insert into MongoS

16 files changed, 1547 insertions, 46 deletions

diff --git a/src/mongo/base/error_codes.yml b/src/mongo/base/error_codes.yml
index 6615502ad4d..4d963e94e42 100644
--- a/src/mongo/base/error_codes.yml
+++ b/src/mongo/base/error_codes.yml
@@ -473,6 +473,9 @@ error_codes:
 
     - {code: 366, name: WouldChangeOwningShardDeletedNoDocument}
 
+    - {code: 367, name: FLECompactionPlaceholder}
+    - {code: 368, name: FLEStateCollectionContention}
+
     # Error codes 4000-8999 are reserved.
 
     # Non-sequential error codes for compatibility only)
diff --git a/src/mongo/crypto/SConscript b/src/mongo/crypto/SConscript
index dfb0809531d..78dc83d6596 100644
--- a/src/mongo/crypto/SConscript
+++ b/src/mongo/crypto/SConscript
@@ -100,6 +100,7 @@ env.Library(
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/bson/util/bson_extract',
         'aead_encryption',
+        'encrypted_field_config',
         'fle_fields',
         'sha_block_${MONGO_CRYPTO}',
     ],
@@ -130,6 +131,7 @@ env.CppUnitTest(
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/base/secure_allocator',
         'aead_encryption',
+        'encrypted_field_config',
         'fle_crypto',
         'sha_block_${MONGO_CRYPTO}',
     ],
diff --git a/src/mongo/crypto/fle_crypto.cpp b/src/mongo/crypto/fle_crypto.cpp
index 24557b26a13..17ddd50e8bf 100644
--- a/src/mongo/crypto/fle_crypto.cpp
+++ b/src/mongo/crypto/fle_crypto.cpp
@@ -54,9 +54,11 @@
 #include "mongo/bson/bsonobj.h"
 #include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/bson/bsontypes.h"
+#include "mongo/bson/oid.h"
 #include "mongo/bson/util/bson_extract.h"
 #include "mongo/bson/util/builder.h"
 #include "mongo/crypto/aead_encryption.h"
+#include "mongo/crypto/encryption_fields_gen.h"
 #include "mongo/crypto/encryption_fields_util.h"
 #include "mongo/crypto/fle_data_frames.h"
 #include "mongo/crypto/fle_field_schema_gen.h"
@@ -66,10 +68,22 @@
 #include "mongo/idl/basic_types.h"
 #include "mongo/idl/idl_parser.h"
 #include "mongo/rpc/object_check.h"
+#include "mongo/stdx/unordered_map.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/scopeguard.h"
 #include "mongo/util/str.h"
 
+// Optional defines to help with debugging
+//
+// Appends unencrypted fields to the state collections to aid in debugging
+//#define FLE2_DEBUG_STATE_COLLECTIONS
+
+// Verbose std::cout to troubleshoot the EmuBinary algorithm
+//#define DEBUG_ENUM_BINARY 1
+
+#ifdef FLE2_DEBUG_STATE_COLLECTIONS
+static_assert(kDebugBuild == 1, "Only use in debug builds");
+#endif
 
 namespace mongo {
 
@@ -92,6 +106,7 @@ constexpr uint64_t kTwiceDerivedTokenFromESCValue = 2;
 constexpr uint64_t kTwiceDerivedTokenFromECCTag = 1;
 constexpr uint64_t kTwiceDerivedTokenFromECCValue = 2;
 
+constexpr int32_t kEncryptionInformationSchemaVersion = 1;
 
 constexpr auto kECCNullId = 0;
 constexpr auto kECCNonNullId = 1;
@@ -108,6 +123,17 @@ constexpr auto kId = "_id";
 constexpr auto kValue = "value";
 constexpr auto kFieldName = "fieldName";
 
+constexpr auto kEncryptedFields = "encryptedFields";
+
+#ifdef FLE2_DEBUG_STATE_COLLECTIONS
+constexpr auto kDebugId = "_debug_id";
+constexpr auto kDebugValuePosition = "_debug_value_position";
+constexpr auto kDebugValueCount = "_debug_value_count";
+
+constexpr auto kDebugValueStart = "_debug_value_start";
+constexpr auto kDebugValueEnd = "_debug_value_end";
+#endif
+
 using UUIDBuf = std::array<uint8_t, UUID::kNumBytes>;
 
 static_assert(sizeof(PrfBlock) == SHA256Block::kHashLength);
@@ -419,16 +445,14 @@ StatusWith<std::tuple<T1, T2>> decryptAndUnpack(ConstDataRange cdr, FLEToken<Tok
 }
 
 
-//#define DEBUG_ENUM_BINARY 1
-
 template <typename collectionT, typename tagTokenT, typename valueTokenT>
-uint64_t emuBinaryCommon(FLEStateCollectionReader* reader,
-                         tagTokenT tagToken,
-                         valueTokenT valueToken) {
+boost::optional<uint64_t> emuBinaryCommon(FLEStateCollectionReader* reader,
+                                          tagTokenT tagToken,
+                                          valueTokenT valueToken) {
 
     // Default search parameters
     uint64_t lambda = 0;
-    uint64_t i = 0;
+    boost::optional<uint64_t> i = 0;
 
     // Step 2:
     // Search for null record
@@ -439,7 +463,8 @@ uint64_t emuBinaryCommon(FLEStateCollectionReader* reader,
     if (!nullDoc.isEmpty()) {
         auto swNullEscDoc = collectionT::decryptNullDocument(valueToken, nullDoc);
         uassertStatusOK(swNullEscDoc.getStatus());
-        lambda = swNullEscDoc.getValue().pos + 1;
+        lambda = swNullEscDoc.getValue().position + 1;
+        i = boost::none;
 #ifdef DEBUG_ENUM_BINARY
         std::cout << fmt::format("start: null_document: lambda {}, i: {}", lambda, i) << std::endl;
 #endif
@@ -478,7 +503,7 @@ uint64_t emuBinaryCommon(FLEStateCollectionReader* reader,
     uint64_t median = 0, min = 1, max = rho;
 
     // Step 9
-    uint64_t maxIterations = ceil(log2(rho));
+    uint64_t maxIterations = rho > 0 ? ceil(log2(rho)) : 0;
 
 #ifdef DEBUG_ENUM_BINARY
     std::cout << fmt::format("start2: maxIterations {}", maxIterations) << std::endl;
@@ -676,6 +701,78 @@ FLE2InsertUpdatePayload EDCClientPayload::serialize(FLEIndexKeyAndId indexKey,
     return iupayload;
 }
 
+
+/**
+ * Lightweight class to build a singly linked list of field names to represent the current field
+ * name
+ *
+ * Avoids heap allocations until getFieldPath() is called
+ */
+class SinglyLinkedFieldPath {
+public:
+    SinglyLinkedFieldPath() : _predecessor(nullptr) {}
+
+    SinglyLinkedFieldPath(StringData fieldName, const SinglyLinkedFieldPath* predecessor)
+        : _currentField(fieldName), _predecessor(predecessor) {}
+
+
+    std::string getFieldPath(StringData fieldName) const;
+
+private:
+    // Name of the current field that is being parsed.
+    const StringData _currentField;
+
+    // Pointer to a parent parser context.
+    // This provides a singly linked list of parent pointers, and use to produce a full path to a
+    // field with an error.
+    const SinglyLinkedFieldPath* _predecessor;
+};
+
+
+std::string SinglyLinkedFieldPath::getFieldPath(StringData fieldName) const {
+    dassert(!fieldName.empty());
+    if (_predecessor == nullptr) {
+        str::stream builder;
+
+        if (!_currentField.empty()) {
+            builder << _currentField << ".";
+        }
+
+        builder << fieldName;
+
+        return builder;
+    } else {
+        std::stack<StringData> pieces;
+
+        pieces.push(fieldName);
+
+        if (!_currentField.empty()) {
+            pieces.push(_currentField);
+        }
+
+        const SinglyLinkedFieldPath* head = _predecessor;
+        while (head) {
+            if (!head->_currentField.empty()) {
+                pieces.push(head->_currentField);
+            }
+            head = head->_predecessor;
+        }
+
+        str::stream builder;
+
+        while (!pieces.empty()) {
+            builder << pieces.top();
+            pieces.pop();
+
+            if (!pieces.empty()) {
+                builder << ".";
+            }
+        }
+
+        return builder;
+    }
+}
+
 /**
  * Copies an input document to the output but provides callers a way to customize how encrypted
  * fields are handled.
@@ -740,8 +837,8 @@ BSONObj transformBSON(
  * Callers can pass a function doVisit(Original bindata content, Field Name).
  */
 void visitEncryptedBSON(const BSONObj& object,
-                        const std::function<void(ConstDataRange, StringData)>& doVisit) {
-    std::stack<BSONObjIterator> frameStack;
+                        const std::function<void(ConstDataRange, std::string)>& doVisit) {
+    std::stack<std::pair<SinglyLinkedFieldPath, BSONObjIterator>> frameStack;
 
     const ScopeGuard frameStackGuard([&] {
         while (!frameStack.empty()) {
@@ -749,24 +846,28 @@ void visitEncryptedBSON(const BSONObj& object,
         }
     });
 
-    frameStack.emplace(BSONObjIterator(object));
+    frameStack.emplace(SinglyLinkedFieldPath(), BSONObjIterator(object));
 
-    while (frameStack.size() > 1 || frameStack.top().more()) {
+    while (frameStack.size() > 1 || frameStack.top().second.more()) {
         uassert(6373511,
                 "Object too deep to be encrypted. Exceeded stack depth.",
                 frameStack.size() < BSONDepth::kDefaultMaxAllowableDepth);
         auto& iterator = frameStack.top();
-        if (iterator.more()) {
-            BSONElement elem = iterator.next();
+        if (iterator.second.more()) {
+            BSONElement elem = iterator.second.next();
             if (elem.type() == BSONType::Object) {
-                frameStack.emplace(BSONObjIterator(elem.Obj()));
+                frameStack.emplace(
+                    SinglyLinkedFieldPath(elem.fieldNameStringData(), &iterator.first),
+                    BSONObjIterator(elem.Obj()));
             } else if (elem.type() == BSONType::Array) {
-                frameStack.emplace(BSONObjIterator(elem.Obj()));
+                frameStack.emplace(
+                    SinglyLinkedFieldPath(elem.fieldNameStringData(), &iterator.first),
+                    BSONObjIterator(elem.Obj()));
             } else if (elem.isBinData(BinDataType::Encrypt)) {
                 int len;
                 const char* data(elem.binData(len));
                 ConstDataRange cdr(data, len);
-                doVisit(cdr, elem.fieldNameStringData());
+                doVisit(cdr, iterator.first.getFieldPath(elem.fieldNameStringData()));
             }
         } else {
             frameStack.pop();
@@ -933,6 +1034,17 @@ void decryptField(FLEKeyVault* keyVault,
     builder->appendAs(obj.firstElement(), fieldPath);
 }
 
+void collectIndexedFields(std::vector<EDCIndexedFields>* pFields,
+                          ConstDataRange cdr,
+                          StringData fieldPath) {
+    auto [encryptedTypeBinding, subCdr] = fromEncryptedConstDataRange(cdr);
+    if (encryptedTypeBinding != EncryptedBinDataType::kFLE2EqualityIndexedValue) {
+        return;
+    }
+
+    pFields->push_back({cdr, fieldPath.toString()});
+}
+
 }  // namespace
 
 
@@ -1148,6 +1260,11 @@ BSONObj ESCCollection::generateNullDocument(ESCTwiceDerivedTagToken tagToken,
     BSONObjBuilder builder;
     toBinData(kId, block, &builder);
     toBinData(kValue, swCipherText.getValue(), &builder);
+#ifdef FLE2_DEBUG_STATE_COLLECTIONS
+    builder.append(kDebugId, "NULL DOC");
+    builder.append(kDebugValuePosition, static_cast<int64_t>(pos));
+    builder.append(kDebugValueCount, static_cast<int64_t>(count));
+#endif
 
     return builder.obj();
 }
@@ -1165,6 +1282,10 @@ BSONObj ESCCollection::generateInsertDocument(ESCTwiceDerivedTagToken tagToken,
     BSONObjBuilder builder;
     toBinData(kId, block, &builder);
     toBinData(kValue, swCipherText.getValue(), &builder);
+#ifdef FLE2_DEBUG_STATE_COLLECTIONS
+    builder.append(kDebugId, static_cast<int64_t>(index));
+    builder.append(kDebugValueCount, static_cast<int64_t>(count));
+#endif
 
     return builder.obj();
 }
@@ -1183,6 +1304,11 @@ BSONObj ESCCollection::generatePositionalDocument(ESCTwiceDerivedTagToken tagTok
     BSONObjBuilder builder;
     toBinData(kId, block, &builder);
     toBinData(kValue, swCipherText.getValue(), &builder);
+#ifdef FLE2_DEBUG_STATE_COLLECTIONS
+    builder.append(kDebugId, static_cast<int64_t>(index));
+    builder.append(kDebugValuePosition, static_cast<int64_t>(pos));
+    builder.append(kDebugValueCount, static_cast<int64_t>(count));
+#endif
 
     return builder.obj();
 }
@@ -1245,9 +1371,9 @@ StatusWith<ESCDocument> ESCCollection::decryptDocument(ESCTwiceDerivedValueToken
 }
 
 
-uint64_t ESCCollection::emuBinary(FLEStateCollectionReader* reader,
-                                  ESCTwiceDerivedTagToken tagToken,
-                                  ESCTwiceDerivedValueToken valueToken) {
+boost::optional<uint64_t> ESCCollection::emuBinary(FLEStateCollectionReader* reader,
+                                                   ESCTwiceDerivedTagToken tagToken,
+                                                   ESCTwiceDerivedValueToken valueToken) {
     return emuBinaryCommon<ESCCollection, ESCTwiceDerivedTagToken, ESCTwiceDerivedValueToken>(
         reader, tagToken, valueToken);
 }
@@ -1273,6 +1399,10 @@ BSONObj ECCCollection::generateNullDocument(ECCTwiceDerivedTagToken tagToken,
     BSONObjBuilder builder;
     toBinData(kId, block, &builder);
     toBinData(kValue, swCipherText.getValue(), &builder);
+#ifdef FLE2_DEBUG_STATE_COLLECTIONS
+    builder.append(kDebugId, "NULL DOC");
+    builder.append(kDebugValueCount, static_cast<int64_t>(count));
+#endif
 
     return builder.obj();
 }
@@ -1290,6 +1420,11 @@ BSONObj ECCCollection::generateDocument(ECCTwiceDerivedTagToken tagToken,
     BSONObjBuilder builder;
     toBinData(kId, block, &builder);
     toBinData(kValue, swCipherText.getValue(), &builder);
+#ifdef FLE2_DEBUG_STATE_COLLECTIONS
+    builder.append(kDebugId, static_cast<int64_t>(index));
+    builder.append(kDebugValueStart, static_cast<int64_t>(start));
+    builder.append(kDebugValueEnd, static_cast<int64_t>(end));
+#endif
 
     return builder.obj();
 }
@@ -1313,6 +1448,11 @@ BSONObj ECCCollection::generateCompactionDocument(ECCTwiceDerivedTagToken tagTok
     BSONObjBuilder builder;
     toBinData(kId, block, &builder);
     toBinData(kValue, swCipherText.getValue(), &builder);
+#ifdef FLE2_DEBUG_STATE_COLLECTIONS
+    builder.append(kDebugId, static_cast<int64_t>(index));
+    builder.append(kDebugValueStart, static_cast<int64_t>(kECCompactionRecordValue));
+    builder.append(kDebugValueEnd, static_cast<int64_t>(kECCompactionRecordValue));
+#endif
 
     return builder.obj();
 }
@@ -1361,15 +1501,16 @@ StatusWith<ECCDocument> ECCCollection::decryptDocument(ECCTwiceDerivedValueToken
                        std::get<1>(value)};
 }
 
-uint64_t ECCCollection::emuBinary(FLEStateCollectionReader* reader,
-                                  ECCTwiceDerivedTagToken tagToken,
-                                  ECCTwiceDerivedValueToken valueToken) {
+boost::optional<uint64_t> ECCCollection::emuBinary(FLEStateCollectionReader* reader,
+                                                   ECCTwiceDerivedTagToken tagToken,
+                                                   ECCTwiceDerivedValueToken valueToken) {
     return emuBinaryCommon<ECCCollection, ECCTwiceDerivedTagToken, ECCTwiceDerivedValueToken>(
         reader, tagToken, valueToken);
 }
 
 BSONObj ECOCollection::generateDocument(StringData fieldName, ConstDataRange payload) {
     BSONObjBuilder builder;
+    builder.append(kId, OID::gen());
     builder.append(kFieldName, fieldName);
     toBinData(kValue, payload, &builder);
     return builder.obj();
@@ -1558,6 +1699,10 @@ StatusWith<std::vector<uint8_t>> FLE2IndexedEqualityEncryptedValue::serialize(
     return serializedServerValue;
 }
 
+ESCDerivedFromDataTokenAndContentionFactorToken EDCServerPayloadInfo::getESCToken() const {
+    return FLETokenFromCDR<FLETokenType::ESCDerivedFromDataTokenAndContentionFactorToken>(
+        payload.getEscDerivedToken());
+}
 
 std::vector<EDCServerPayloadInfo> EDCServerCollection::getEncryptedFieldInfo(BSONObj& obj) {
     std::vector<EDCServerPayloadInfo> fields;
@@ -1582,7 +1727,7 @@ PrfBlock EDCServerCollection::generateTag(const EDCServerPayloadInfo& payload) {
 }
 
 BSONObj EDCServerCollection::finalizeForInsert(
-    BSONObj& doc, const std::vector<EDCServerPayloadInfo>& serverPayload) {
+    const BSONObj& doc, const std::vector<EDCServerPayloadInfo>& serverPayload) {
 
     std::vector<TagInfo> tags;
     // TODO - improve size estimate after range is supported since it no longer be 1 to 1
@@ -1635,5 +1780,45 @@ BSONObj EDCServerCollection::finalizeForInsert(
     return builder.obj();
 }
 
+std::vector<EDCIndexedFields> EDCServerCollection::getEncryptedIndexedFields(BSONObj& obj) {
+    std::vector<EDCIndexedFields> fields;
+
+    visitEncryptedBSON(obj, [&fields](ConstDataRange cdr, StringData fieldPath) {
+        collectIndexedFields(&fields, cdr, fieldPath);
+    });
+
+    return fields;
+}
+
+
+BSONObj EncryptionInformationHelpers::encryptionInformationSerialize(NamespaceString& nss,
+                                                                     EncryptedFieldConfig& ef) {
+    EncryptionInformation ei;
+    ei.setType(kEncryptionInformationSchemaVersion);
+
+    ei.setSchema(BSON(nss.toString() << ef.toBSON()));
+
+    return ei.toBSON();
+}
+
+EncryptedFieldConfig EncryptionInformationHelpers::getAndValidateSchema(
+    const NamespaceString& nss, const EncryptionInformation& ei) {
+    BSONObj schema = ei.getSchema();
+
+    auto element = schema.getField(nss.toString());
+
+    uassert(6371205,
+            "Expected an object for schema in EncryptionInformation",
+            !element.eoo() && element.type() == Object);
+
+    auto efc = EncryptedFieldConfig::parse(IDLParserErrorContext("schema"), element.Obj());
+
+    uassert(6371206, "Expected a value for eccCollection", efc.getEccCollection().has_value());
+    uassert(6371207, "Expected a value for escCollection", efc.getEscCollection().has_value());
+    uassert(6371208, "Expected a value for ecocCollection", efc.getEcocCollection().has_value());
+
+    return efc;
+}
+
 
 }  // namespace mongo
diff --git a/src/mongo/crypto/fle_crypto.h b/src/mongo/crypto/fle_crypto.h
index 3be7c7ea0b2..22f47983cc6 100644
--- a/src/mongo/crypto/fle_crypto.h
+++ b/src/mongo/crypto/fle_crypto.h
@@ -410,7 +410,7 @@ public:
 
 struct ESCNullDocument {
     // Id is not included as it is HMAC generated and cannot be reversed
-    uint64_t pos;
+    uint64_t position;
     uint64_t count;
 };
 
@@ -497,9 +497,9 @@ public:
     /**
      * Search for the highest document id for a given field/value pair based on the token.
      */
-    static uint64_t emuBinary(FLEStateCollectionReader* reader,
-                              ESCTwiceDerivedTagToken tagToken,
-                              ESCTwiceDerivedValueToken valueToken);
+    static boost::optional<uint64_t> emuBinary(FLEStateCollectionReader* reader,
+                                               ESCTwiceDerivedTagToken tagToken,
+                                               ESCTwiceDerivedValueToken valueToken);
 };
 
 
@@ -579,7 +579,7 @@ enum class ECCValueType : uint64_t {
 
 struct ECCNullDocument {
     // Id is not included as it HMAC generated and cannot be reversed
-    uint64_t pos;
+    uint64_t position;
 };
 
 
@@ -645,9 +645,9 @@ public:
     /**
      * Search for the highest document id for a given field/value pair based on the token.
      */
-    static uint64_t emuBinary(FLEStateCollectionReader* reader,
-                              ECCTwiceDerivedTagToken tagToken,
-                              ECCTwiceDerivedValueToken valueToken);
+    static boost::optional<uint64_t> emuBinary(FLEStateCollectionReader* reader,
+                                               ECCTwiceDerivedTagToken tagToken,
+                                               ECCTwiceDerivedValueToken valueToken);
 };
 
 /**
@@ -841,12 +841,19 @@ struct FLE2IndexedEqualityEncryptedValue {
 
 
 struct EDCServerPayloadInfo {
-    FLE2InsertUpdatePayload payload;
+    ESCDerivedFromDataTokenAndContentionFactorToken getESCToken() const;
 
+    FLE2InsertUpdatePayload payload;
     std::string fieldPathName;
     uint64_t count;
 };
 
+struct EDCIndexedFields {
+    ConstDataRange value;
+
+    std::string fieldPathName;
+};
+
 /**
  * Manipulates the EDC collection.
  *
@@ -877,8 +884,21 @@ public:
      * Converts FLE2InsertUpdatePayload to a final insert payload and updates __safeContent__ with
      * new tags.
      */
-    static BSONObj finalizeForInsert(BSONObj& doc,
+    static BSONObj finalizeForInsert(const BSONObj& doc,
                                      const std::vector<EDCServerPayloadInfo>& serverPayload);
+
+    /**
+     * Get a list of encrypted, indexed fields.
+     */
+    static std::vector<EDCIndexedFields> getEncryptedIndexedFields(BSONObj& obj);
+};
+
+class EncryptionInformationHelpers {
+public:
+    static BSONObj encryptionInformationSerialize(NamespaceString& nss, EncryptedFieldConfig& ef);
+
+    static EncryptedFieldConfig getAndValidateSchema(const NamespaceString& nss,
+                                                     const EncryptionInformation& ei);
 };
 
 }  // namespace mongo
diff --git a/src/mongo/crypto/fle_crypto_test.cpp b/src/mongo/crypto/fle_crypto_test.cpp
index 6cf4513f186..24bf3089a0b 100644
--- a/src/mongo/crypto/fle_crypto_test.cpp
+++ b/src/mongo/crypto/fle_crypto_test.cpp
@@ -283,7 +283,7 @@ TEST(FLE_ESC, RoundTrip) {
             ESCCollection::generateNullDocument(escTwiceTag, escTwiceValue, 123, 123456789);
         auto swDoc = ESCCollection::decryptNullDocument(escTwiceValue, doc);
         ASSERT_OK(swDoc.getStatus());
-        ASSERT_EQ(swDoc.getValue().pos, 123);
+        ASSERT_EQ(swDoc.getValue().position, 123);
         ASSERT_EQ(swDoc.getValue().count, 123456789);
     }
 
@@ -342,7 +342,7 @@ TEST(FLE_ECC, RoundTrip) {
         BSONObj doc = ECCCollection::generateNullDocument(twiceTag, twiceValue, 123456789);
         auto swDoc = ECCCollection::decryptNullDocument(twiceValue, doc);
         ASSERT_OK(swDoc.getStatus());
-        ASSERT_EQ(swDoc.getValue().pos, 123456789);
+        ASSERT_EQ(swDoc.getValue().position, 123456789);
     }
 
 
@@ -405,6 +405,33 @@ private:
     std::vector<BSONObj> _docs;
 };
 
+// Test Empty Collection
+TEST(FLE_ESC, EmuBinary_Empty) {
+
+    TestDocumentCollection coll;
+    ConstDataRange value(testValue);
+
+    auto c1 = FLELevel1TokenGenerator::generateCollectionsLevel1Token(indexKey);
+    auto escToken = FLECollectionTokenGenerator::generateESCToken(c1);
+
+    ESCDerivedFromDataToken escDatakey =
+        FLEDerivedFromDataTokenGenerator::generateESCDerivedFromDataToken(escToken, value);
+
+    auto escDerivedToken = FLEDerivedFromDataTokenAndContentionFactorTokenGenerator::
+        generateESCDerivedFromDataTokenAndContentionFactorToken(escDatakey, 0);
+
+    auto escTwiceTag =
+        FLETwiceDerivedTokenGenerator::generateESCTwiceDerivedTagToken(escDerivedToken);
+    auto escTwiceValue =
+        FLETwiceDerivedTokenGenerator::generateESCTwiceDerivedValueToken(escDerivedToken);
+
+
+    auto i = ESCCollection::emuBinary(&coll, escTwiceTag, escTwiceValue);
+
+    ASSERT_TRUE(i.has_value());
+    ASSERT_EQ(i.value(), 0);
+}
+
 // Test one new field in esc
 TEST(FLE_ESC, EmuBinary) {
 
@@ -430,10 +457,10 @@ TEST(FLE_ESC, EmuBinary) {
         coll.insert(doc);
     }
 
-    uint64_t i = ESCCollection::emuBinary(&coll, escTwiceTag, escTwiceValue);
+    auto i = ESCCollection::emuBinary(&coll, escTwiceTag, escTwiceValue);
 
-    std::cout << "i: " << i << std::endl;
-    ASSERT_EQ(i, 5);
+    ASSERT_TRUE(i.has_value());
+    ASSERT_EQ(i.value(), 5);
 }
 
 
@@ -480,15 +507,46 @@ TEST(FLE_ESC, EmuBinary2) {
         coll.insert(doc);
     }
 
-    uint64_t i = ESCCollection::emuBinary(&coll, escTwiceTag, escTwiceValue);
+    auto i = ESCCollection::emuBinary(&coll, escTwiceTag, escTwiceValue);
 
-    ASSERT_EQ(i, 13);
+    ASSERT_TRUE(i.has_value());
+    ASSERT_EQ(i.value(), 13);
 
     i = ESCCollection::emuBinary(&coll, escTwiceTag2, escTwiceValue2);
 
-    ASSERT_EQ(i, 5);
+    ASSERT_TRUE(i.has_value());
+    ASSERT_EQ(i.value(), 5);
 }
 
+// Test Emulated Binary with null record
+TEST(FLE_ESC, EmuBinary_NullRecord) {
+
+    TestDocumentCollection coll;
+    ConstDataRange value(testValue);
+
+    auto c1 = FLELevel1TokenGenerator::generateCollectionsLevel1Token(indexKey);
+    auto escToken = FLECollectionTokenGenerator::generateESCToken(c1);
+
+    ESCDerivedFromDataToken escDatakey =
+        FLEDerivedFromDataTokenGenerator::generateESCDerivedFromDataToken(escToken, value);
+
+    auto escDerivedToken = FLEDerivedFromDataTokenAndContentionFactorTokenGenerator::
+        generateESCDerivedFromDataTokenAndContentionFactorToken(escDatakey, 0);
+
+    auto escTwiceTag =
+        FLETwiceDerivedTokenGenerator::generateESCTwiceDerivedTagToken(escDerivedToken);
+    auto escTwiceValue =
+        FLETwiceDerivedTokenGenerator::generateESCTwiceDerivedValueToken(escDerivedToken);
+
+    BSONObj doc = ESCCollection::generateNullDocument(escTwiceTag, escTwiceValue, 7, 7);
+    coll.insert(doc);
+
+    auto i = ESCCollection::emuBinary(&coll, escTwiceTag, escTwiceValue);
+
+    ASSERT_FALSE(i.has_value());
+}
+
+
 std::vector<char> generatePlaceholder(BSONElement value) {
     FLE2EncryptionPlaceholder ep;
 
@@ -514,8 +572,6 @@ BSONObj encryptDocument(BSONObj obj, FLEKeyVault* keyVault) {
     // Start Server Side
     auto serverPayload = EDCServerCollection::getEncryptedFieldInfo(result);
 
-    ASSERT_EQ(serverPayload.size(), 1);
-
     // TODO set count based on EmuBinary
     for (auto& payload : serverPayload) {
         payload.count = 1;
@@ -895,6 +951,160 @@ TEST(FLE_EDC, DuplicateSafeContent_IncompatibleType) {
     ASSERT_THROWS_CODE(encryptDocument(builder.obj(), &keyVault), DBException, 6373510);
 }
 
+template <typename T, typename Func>
+bool vectorContains(const std::vector<T>& vec, Func func) {
+    return std::find_if(vec.begin(), vec.end(), func) != vec.end();
+}
+
+EncryptedFieldConfig getTestEncryptedFieldConfig() {
+
+    constexpr auto schema = R"({
+    "escCollection": "esc",
+    "eccCollection": "ecc",
+    "ecocCollection": "ecoc",
+    "fields": [
+        {
+            "keyId":
+                            {
+                                "$uuid": "12345678-1234-9876-1234-123456789012"
+                            }
+                        ,
+            "path": "encrypted",
+            "bsonType": "string",
+            "queries": {"queryType": "equality"}
+
+        },
+        {
+            "keyId":
+                            {
+                                "$uuid": "12345678-1234-9876-1234-123456789012"
+                            }
+                        ,
+            "path": "nested.encrypted",
+            "bsonType": "string",
+            "queries": {"queryType": "equality"}
+
+        },
+        {
+            "keyId":
+                            {
+                                "$uuid": "12345678-1234-9876-1234-123456789012"
+                            }
+                        ,
+            "path": "nested.notindexed",
+            "bsonType": "string"
+        }
+    ]
+})";
+
+    return EncryptedFieldConfig::parse(IDLParserErrorContext("root"), fromjson(schema));
+}
+
+TEST(EncryptionInformation, RoundTrip) {
+    NamespaceString ns("test.test");
+
+    EncryptedFieldConfig efc = getTestEncryptedFieldConfig();
+    auto obj = EncryptionInformationHelpers::encryptionInformationSerialize(ns, efc);
+
+
+    EncryptedFieldConfig efc2 = EncryptionInformationHelpers::getAndValidateSchema(
+        ns, EncryptionInformation::parse(IDLParserErrorContext("foo"), obj));
+
+    ASSERT_BSONOBJ_EQ(efc.toBSON(), efc2.toBSON());
+}
+
+TEST(EncryptionInformation, BadSchema) {
+    EncryptionInformation ei;
+    ei.setType(1);
+
+    ei.setSchema(BSON("a"
+                      << "b"));
+
+    auto obj = ei.toBSON();
+
+    NamespaceString ns("test.test");
+    ASSERT_THROWS_CODE(EncryptionInformationHelpers::getAndValidateSchema(
+                           ns, EncryptionInformation::parse(IDLParserErrorContext("foo"), obj)),
+                       DBException,
+                       6371205);
+}
+
+TEST(EncryptionInformation, MissingStateCollection) {
+    NamespaceString ns("test.test");
+
+    {
+        EncryptedFieldConfig efc = getTestEncryptedFieldConfig();
+        efc.setEscCollection(boost::none);
+        auto obj = EncryptionInformationHelpers::encryptionInformationSerialize(ns, efc);
+        ASSERT_THROWS_CODE(EncryptionInformationHelpers::getAndValidateSchema(
+                               ns, EncryptionInformation::parse(IDLParserErrorContext("foo"), obj)),
+                           DBException,
+                           6371207);
+    }
+    {
+        EncryptedFieldConfig efc = getTestEncryptedFieldConfig();
+        efc.setEccCollection(boost::none);
+        auto obj = EncryptionInformationHelpers::encryptionInformationSerialize(ns, efc);
+        ASSERT_THROWS_CODE(EncryptionInformationHelpers::getAndValidateSchema(
+                               ns, EncryptionInformation::parse(IDLParserErrorContext("foo"), obj)),
+                           DBException,
+                           6371206);
+    }
+    {
+        EncryptedFieldConfig efc = getTestEncryptedFieldConfig();
+        efc.setEcocCollection(boost::none);
+        auto obj = EncryptionInformationHelpers::encryptionInformationSerialize(ns, efc);
+        ASSERT_THROWS_CODE(EncryptionInformationHelpers::getAndValidateSchema(
+                               ns, EncryptionInformation::parse(IDLParserErrorContext("foo"), obj)),
+                           DBException,
+                           6371208);
+    }
+}
+
+TEST(IndexedFields, FetchTwoLevels) {
+    TestKeyVault keyVault;
+
+    auto doc = BSON("value"
+                    << "123456");
+    auto element = doc.firstElement();
+    auto inputDoc = BSON("__safeContent__" << BSON_ARRAY(1 << 2 << 4) << "encrypted" << element);
+
+    auto buf = generatePlaceholder(element);
+    BSONObjBuilder builder;
+    builder.append("__safeContent__", BSON_ARRAY(1 << 2 << 4));
+    builder.appendBinData("encrypted", buf.size(), BinDataType::Encrypt, buf.data());
+    {
+        BSONObjBuilder sub(builder.subobjStart("nested"));
+        sub.appendBinData("encrypted", buf.size(), BinDataType::Encrypt, buf.data());
+        {
+            BSONObjBuilder sub2(sub.subobjStart("nested2"));
+            sub2.appendBinData("encrypted", buf.size(), BinDataType::Encrypt, buf.data());
+        }
+    }
+
+    auto obj = builder.obj();
+
+    auto noIndexedFields = EDCServerCollection::getEncryptedIndexedFields(obj);
+
+    ASSERT_EQ(noIndexedFields.size(), 0);
+
+    auto finalDoc = encryptDocument(obj, &keyVault);
+
+    auto indexedFields = EDCServerCollection::getEncryptedIndexedFields(finalDoc);
+
+    ASSERT_EQ(indexedFields.size(), 3);
+
+
+    ASSERT(vectorContains(indexedFields,
+                          [](EDCIndexedFields i) { return i.fieldPathName == "encrypted"; }));
+    ASSERT(vectorContains(
+        indexedFields, [](EDCIndexedFields i) { return i.fieldPathName == "nested.encrypted"; }));
+    ASSERT(vectorContains(indexedFields, [](EDCIndexedFields i) {
+        return i.fieldPathName == "nested.nested2.encrypted";
+    }));
+}
+
+
 }  // namespace mongo
 
 #endif
diff --git a/src/mongo/crypto/fle_field_schema.idl b/src/mongo/crypto/fle_field_schema.idl
index 68bbd2b5e9f..16d15bf704a 100644
--- a/src/mongo/crypto/fle_field_schema.idl
+++ b/src/mongo/crypto/fle_field_schema.idl
@@ -211,13 +211,16 @@ structs:
             description: "The version number"
             type: safeInt
             default: 1
+            unstable: true
           deleteTokens:
             description: "A map of field paths to FLEDeletePayload"
             type: object
             optional: true
+            unstable: true
           schema:
             description: "A map of NamespaceString to EncryptedFieldConfig"
             type: object
+            unstable: true
 
     ecocDocument:
         description: "foo"
@@ -226,7 +229,6 @@ structs:
           _id:
             description: "Random id"
             type: objectid
-            optional: true
           fieldName:
             description: "Name of field"
             type: string
diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index a690a3fc9de..b7f13a73605 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -854,6 +854,24 @@ env.Library(
 )
 
 env.Library(
+    target='fle_crud',
+    source=[
+        'fle_crud.cpp',
+    ],
+    LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/crypto/encrypted_field_config',
+        '$BUILD_DIR/mongo/crypto/fle_crypto',
+        '$BUILD_DIR/mongo/db/ops/write_ops_parsers',
+        '$BUILD_DIR/mongo/db/query/query_request',
+        '$BUILD_DIR/mongo/executor/task_executor_pool',
+        '$BUILD_DIR/mongo/s/grid',
+        '$BUILD_DIR/mongo/s/sharding_router_api',
+        'logical_session_id',
+        'transaction_api',
+    ],
+)
+
+env.Library(
     target='dbdirectclient',
     source=[
         'dbdirectclient.cpp',
@@ -2559,6 +2577,7 @@ if wiredtiger:
             'field_parser_test.cpp',
             'field_ref_set_test.cpp',
             'field_ref_test.cpp',
+            'fle_crud_test.cpp',
             'hasher_test.cpp',
             'index_build_entry_helpers_test.cpp',
             'index_builds_coordinator_mongod_test.cpp',
@@ -2607,6 +2626,8 @@ if wiredtiger:
             '$BUILD_DIR/mongo/base',
             '$BUILD_DIR/mongo/bson/util/bson_extract',
             '$BUILD_DIR/mongo/client/read_preference',
+            '$BUILD_DIR/mongo/crypto/encrypted_field_config',
+            '$BUILD_DIR/mongo/crypto/fle_crypto',
             '$BUILD_DIR/mongo/db/auth/auth',
             '$BUILD_DIR/mongo/db/auth/security_token',
             '$BUILD_DIR/mongo/db/catalog/catalog_test_fixture',
@@ -2636,6 +2657,7 @@ if wiredtiger:
             'dbdirectclient',
             'dbmessage',
             'fcv_op_observer',
+            'fle_crud',
             'index_build_entry_helpers',
             'index_builds_coordinator_mongod',
             'keys_collection_client_direct',
diff --git a/src/mongo/db/fle_crud.cpp b/src/mongo/db/fle_crud.cpp
new file mode 100644
index 00000000000..e30b46d8dba
--- /dev/null
+++ b/src/mongo/db/fle_crud.cpp
@@ -0,0 +1,396 @@
+/**
+ *    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 <cstdint>
+#include <memory>
+#include <string>
+#include <utility>
+
+#include "mongo/db/fle_crud.h"
+
+#include "mongo/bson/bsonmisc.h"
+#include "mongo/bson/bsonobj.h"
+#include "mongo/bson/bsonobjbuilder.h"
+#include "mongo/bson/bsontypes.h"
+#include "mongo/crypto/encryption_fields_gen.h"
+#include "mongo/crypto/fle_crypto.h"
+#include "mongo/db/namespace_string.h"
+#include "mongo/db/ops/write_ops_gen.h"
+#include "mongo/db/query/find_command_gen.h"
+#include "mongo/db/transaction_api.h"
+#include "mongo/idl/idl_parser.h"
+#include "mongo/s/grid.h"
+#include "mongo/s/transaction_router_resource_yielder.h"
+#include "mongo/s/write_ops/batch_write_exec.h"
+#include "mongo/util/assert_util.h"
+#include "mongo/util/concurrency/thread_pool.h"
+
+namespace mongo {
+namespace {
+
+class FLEQueryInterfaceImpl : public FLEQueryInterface {
+public:
+    FLEQueryInterfaceImpl(const txn_api::TransactionClient& txnClient) : _txnClient(txnClient) {}
+
+    BSONObj getById(const NamespaceString& nss, PrfBlock block) final;
+
+    uint64_t countDocuments(const NamespaceString& nss) final;
+
+    void insertDocument(const NamespaceString& nss, BSONObj obj, bool translateDuplicateKey) final;
+
+private:
+    const txn_api::TransactionClient& _txnClient;
+};
+
+BSONObj FLEQueryInterfaceImpl::getById(const NamespaceString& nss, PrfBlock block) {
+    FindCommandRequest find(nss);
+    find.setFilter(BSON("_id" << BSONBinData(block.data(), block.size(), BinDataGeneral)));
+    find.setSingleBatch(true);
+
+    // Throws on error
+    auto docs = _txnClient.exhaustiveFind(find).get();
+
+    if (docs.size() == 0) {
+        return BSONObj();
+    } else {
+        // We only expect one document in the state collection considering that _id is a unique
+        // index
+        uassert(6371201,
+                "Unexpected to find more then one FLE state collection document",
+                docs.size() == 1);
+        return docs[0];
+    }
+}
+
+uint64_t FLEQueryInterfaceImpl::countDocuments(const NamespaceString& nss) {
+    // TODO - what about
+    // count cmd
+    // $collStats
+    // approxCount
+
+    // Build the following pipeline:
+    //
+    //{ aggregate : "testColl", pipeline: [{$match:{}}, {$group : {_id: null, n : {$sum:1}
+    //}} ], cursor: {}}
+
+    BSONObjBuilder builder;
+    // $db - TXN API DOES THIS FOR US by building OP_MSG
+    builder.append("aggregate", nss.coll());
+
+    AggregateCommandRequest request(nss);
+
+    std::vector<BSONObj> pipeline;
+    pipeline.push_back(BSON("$match" << BSONObj()));
+
+    {
+        BSONObjBuilder sub;
+        {
+            BSONObjBuilder sub2(sub.subobjStart("$group"));
+            sub2.appendNull("_id");
+            {
+                BSONObjBuilder sub3(sub.subobjStart("n"));
+                sub3.append("$sum", 1);
+            }
+        }
+
+        pipeline.push_back(sub.obj());
+    }
+
+    request.setPipeline(pipeline);
+
+    auto commandResponse = _txnClient.runCommand(nss.db(), request.toBSON({})).get();
+
+    uint64_t docCount = 0;
+    auto cursorResponse = uassertStatusOK(CursorResponse::parseFromBSON(commandResponse));
+
+    auto firstBatch = cursorResponse.getBatch();
+    if (!firstBatch.empty()) {
+        auto countObj = firstBatch.front();
+        docCount = countObj.getIntField("n"_sd);
+    }
+
+    return docCount;
+}
+
+void FLEQueryInterfaceImpl::insertDocument(const NamespaceString& nss,
+                                           BSONObj obj,
+                                           bool translateDuplicateKey) {
+    write_ops::InsertCommandRequest insertRequest(nss);
+    insertRequest.setDocuments({obj});
+
+    // TODO SERVER-64143 - propagate the retryable statement ids to runCRUDOp
+    auto response = _txnClient.runCRUDOp(BatchedCommandRequest(insertRequest), {}).get();
+
+    auto status = response.toStatus();
+    if (translateDuplicateKey && status.code() == ErrorCodes::DuplicateKey) {
+        uassertStatusOK(Status(ErrorCodes::FLEStateCollectionContention, status.reason()));
+    }
+
+    uassertStatusOK(status);
+}
+
+/**
+ * Implementation of FLEStateCollectionReader for txn_api::TransactionClient
+ *
+ * Document count is cached since we only need it once per esc or ecc collection.
+ */
+class TxnCollectionReader : public FLEStateCollectionReader {
+public:
+    TxnCollectionReader(uint64_t count, FLEQueryInterface* queryImpl, const NamespaceString& nss)
+        : _count(count), _queryImpl(queryImpl), _nss(nss) {}
+
+    uint64_t getDocumentCount() override {
+        return _count;
+    }
+
+    BSONObj getById(PrfBlock block) override {
+        return _queryImpl->getById(_nss, block);
+    }
+
+private:
+    uint64_t _count;
+    FLEQueryInterface* _queryImpl;
+    const NamespaceString& _nss;
+};
+
+StatusWith<txn_api::CommitResult> runInTxnWithRetry(
+    OperationContext* opCtx,
+    std::shared_ptr<txn_api::TransactionWithRetries> trun,
+    std::function<SemiFuture<void>(const txn_api::TransactionClient& txnClient,
+                                   ExecutorPtr txnExec)> callback) {
+
+    bool inClientTransaction = opCtx->inMultiDocumentTransaction();
+
+    // TODO SERVER-59566 - how much do we retry before we give up?
+    while (true) {
+
+        // Result will get the status of the TXN
+        // Non-client initiated txns get retried automatically.
+        // Client txns are the user responsibility to retry and so if we hit a contention
+        // placeholder, we need to abort and defer to the client
+        auto swResult = trun->runSyncNoThrow(opCtx, callback);
+        if (swResult.isOK()) {
+            return swResult;
+        }
+
+        auto commitResult = swResult.getValue();
+        if (commitResult.getEffectiveStatus().isOK()) {
+            return commitResult;
+        }
+
+        // We cannot retry the transaction if initiated by a user
+        if (inClientTransaction) {
+            return swResult;
+        }
+
+        // - DuplicateKeyException - suggestions contention on ESC
+        // - FLEContention
+        if (swResult.getStatus().code() != ErrorCodes::FLECompactionPlaceholder &&
+            swResult.getStatus().code() != ErrorCodes::FLEStateCollectionContention) {
+            return swResult;
+        }
+    }
+}
+
+
+StatusWith<FLEBatchResult> processInsert(
+    OperationContext* opCtx,
+    const write_ops::InsertCommandRequest& insertRequest,
+    std::function<std::shared_ptr<txn_api::TransactionWithRetries>(OperationContext*)> getTxns) {
+
+    auto documents = insertRequest.getDocuments();
+    // TODO - support larger batches?
+    // TODO - how to check if a document will be too large???
+    uassert(6371202, "Only single insert batches are supported in FLE2", documents.size() == 1);
+
+    auto document = documents[0];
+    auto serverPayload = std::make_shared<std::vector<EDCServerPayloadInfo>>(
+        EDCServerCollection::getEncryptedFieldInfo(document));
+
+    if (serverPayload->size() == 0) {
+        // No actual FLE2 indexed fields
+        return FLEBatchResult::kNotProcessed;
+    }
+
+    auto ei = insertRequest.getEncryptionInformation().get();
+
+    auto edcNss = insertRequest.getNamespace();
+    auto efc = EncryptionInformationHelpers::getAndValidateSchema(insertRequest.getNamespace(), ei);
+
+    std::shared_ptr<txn_api::TransactionWithRetries> trun = getTxns(opCtx);
+
+    // The function that handles the transaction may outlive this function so we need to use
+    // shared_ptrs since it runs on another thread
+    auto ownedDocument = document.getOwned();
+    auto insertBlock = std::tie(edcNss, efc, serverPayload);
+    auto sharedInsertBlock = std::make_shared<decltype(insertBlock)>(insertBlock);
+
+    auto swResult = runInTxnWithRetry(
+        opCtx,
+        trun,
+        [sharedInsertBlock, ownedDocument](const txn_api::TransactionClient& txnClient,
+                                           ExecutorPtr txnExec) {
+            FLEQueryInterfaceImpl queryImpl(txnClient);
+
+            auto [edcNss2, efc2, serverPayload2] = *sharedInsertBlock.get();
+
+            processInsert(&queryImpl, edcNss2, *serverPayload2.get(), efc2, ownedDocument);
+
+            return SemiFuture<void>::makeReady();
+        });
+    if (!swResult.isOK()) {
+        return swResult.getStatus();
+    }
+
+    return FLEBatchResult::kProcessed;
+}
+
+}  // namespace
+
+FLEQueryInterface::~FLEQueryInterface() {}
+
+void processInsert(FLEQueryInterface* queryImpl,
+                   const NamespaceString& edcNss,
+                   std::vector<EDCServerPayloadInfo>& serverPayload,
+                   const EncryptedFieldConfig& efc,
+                   BSONObj document) {
+    NamespaceString nssEsc(edcNss.db(), efc.getEscCollection().get());
+
+    auto docCount = queryImpl->countDocuments(nssEsc);
+
+    TxnCollectionReader reader(docCount, queryImpl, nssEsc);
+
+    for (auto& payload : serverPayload) {
+
+        auto escToken = payload.getESCToken();
+        auto tagToken = FLETwiceDerivedTokenGenerator::generateESCTwiceDerivedTagToken(escToken);
+        auto valueToken =
+            FLETwiceDerivedTokenGenerator::generateESCTwiceDerivedValueToken(escToken);
+
+        int position = 1;
+        int count = 1;
+        auto alpha = ESCCollection::emuBinary(&reader, tagToken, valueToken);
+
+        if (alpha.has_value() && alpha.value() == 0) {
+            position = 1;
+            count = 1;
+        } else if (!alpha.has_value()) {
+            auto block = ESCCollection::generateId(tagToken, boost::none);
+
+            auto r_esc = reader.getById(block);
+            uassert(6371203, "ESC document not found", !r_esc.isEmpty());
+
+            auto escNullDoc =
+                uassertStatusOK(ESCCollection::decryptNullDocument(valueToken, r_esc));
+
+            position = escNullDoc.position + 2;
+            count = escNullDoc.count + 1;
+        } else {
+            auto block = ESCCollection::generateId(tagToken, alpha);
+
+            auto r_esc = reader.getById(block);
+            uassert(6371204, "ESC document not found", !r_esc.isEmpty());
+
+            auto escDoc = uassertStatusOK(ESCCollection::decryptDocument(valueToken, r_esc));
+
+            position = alpha.value() + 1;
+            count = escDoc.count + 1;
+
+            if (escDoc.compactionPlaceholder) {
+                uassertStatusOK(Status(ErrorCodes::FLECompactionPlaceholder,
+                                       "Found ESC contention placeholder"));
+            }
+        }
+
+        payload.count = count;
+
+        queryImpl->insertDocument(
+            nssEsc,
+            ESCCollection::generateInsertDocument(tagToken, valueToken, position, count),
+            true);
+
+        NamespaceString nssEcoc(edcNss.db(), efc.getEcocCollection().get());
+
+        // TODO - should we make this a batch of ECOC updates?
+        queryImpl->insertDocument(nssEcoc,
+                                  ECOCollection::generateDocument(
+                                      payload.fieldPathName, payload.payload.getEncryptedTokens()),
+                                  false);
+    }
+
+    auto finalDoc = EDCServerCollection::finalizeForInsert(document, serverPayload);
+
+    queryImpl->insertDocument(edcNss, finalDoc, false);
+}
+
+FLEBatchResult processFLEBatch(OperationContext* opCtx,
+                               const BatchedCommandRequest& request,
+                               BatchWriteExecStats* stats,
+                               BatchedCommandResponse* response,
+                               boost::optional<OID> targetEpoch) {
+
+    if (!gFeatureFlagFLE2.isEnabledAndIgnoreFCV()) {
+        uasserted(6371209, "Feature flag FLE2 is not enabled");
+    }
+
+    if (request.getBatchType() != BatchedCommandRequest::BatchType_Insert) {
+        return FLEBatchResult::kNotProcessed;
+    }
+
+    auto getTxn = [](OperationContext* opCtx) {
+        return std::make_shared<txn_api::TransactionWithRetries>(
+            opCtx,
+            Grid::get(opCtx)->getExecutorPool()->getFixedExecutor(),
+            TransactionRouterResourceYielder::make());
+    };
+
+    if (request.getBatchType() == BatchedCommandRequest::BatchType_Insert) {
+        auto insertRequest = request.getInsertRequest();
+
+        auto swResult = processInsert(opCtx, insertRequest, getTxn);
+
+        if (!swResult.isOK()) {
+            response->setStatus(swResult.getStatus());
+            response->setN(0);
+
+            return FLEBatchResult::kProcessed;
+        } else if (swResult.getValue() == FLEBatchResult::kProcessed) {
+            response->setStatus(Status::OK());
+            response->setN(1);
+        }
+
+        return swResult.getValue();
+    }
+
+    MONGO_UNREACHABLE;
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/fle_crud.h b/src/mongo/db/fle_crud.h
new file mode 100644
index 00000000000..8fa6da6a892
--- /dev/null
+++ b/src/mongo/db/fle_crud.h
@@ -0,0 +1,111 @@
+/**
+ *    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/bson/bsonobj.h"
+#include "mongo/bson/oid.h"
+#include "mongo/crypto/fle_crypto.h"
+#include "mongo/db/namespace_string.h"
+#include "mongo/db/operation_context.h"
+#include "mongo/s/write_ops/batch_write_exec.h"
+#include "mongo/s/write_ops/batched_command_response.h"
+
+namespace mongo {
+
+/**
+ * FLE Result enum
+ */
+enum class FLEBatchResult {
+    /**
+     * FLE CRUD code decided input document requires FLE processing. Caller should not do any CRUD.
+     */
+    kProcessed,
+
+    /**
+     * FLE CRUD code decided it did not have to do any CRUD. For instance, it has no encrypted
+     * fields that require further processing. Caller should process the request normally.
+     */
+    kNotProcessed
+};
+
+/**
+ * Process a batch from mongos.
+ */
+FLEBatchResult processFLEBatch(OperationContext* opCtx,
+                               const BatchedCommandRequest& request,
+                               BatchWriteExecStats* stats,
+                               BatchedCommandResponse* response,
+                               boost::optional<OID> targetEpoch);
+
+/**
+ * Abstraction layer for FLE
+ */
+class FLEQueryInterface {
+public:
+    virtual ~FLEQueryInterface();
+
+    /**
+     * Retrieve a single document by _id == PrfBlock from nss.
+     *
+     * Returns an empty BSONObj if no document is found.
+     * Expected to throw an error if it detects more then one documents.
+     */
+    virtual BSONObj getById(const NamespaceString& nss, PrfBlock block) = 0;
+
+    /**
+     * Count the documents in the collection.
+     *
+     * Throws if the collection is not found.
+     */
+    virtual uint64_t countDocuments(const NamespaceString& nss) = 0;
+
+    /**
+     * Insert a document into the given collection.
+     *
+     * If translateDuplicateKey == true and the insert returns DuplicateKey, returns
+     * FLEStateCollectionContention instead
+     */
+    virtual void insertDocument(const NamespaceString& nss,
+                                BSONObj obj,
+                                bool translateDuplicateKey) = 0;
+};
+
+/**
+ * Process a FLE insert with the query interface
+ *
+ * Used by unit tests.
+ */
+void processInsert(FLEQueryInterface* queryImpl,
+                   const NamespaceString& edcNss,
+                   std::vector<EDCServerPayloadInfo>& serverPayload,
+                   const EncryptedFieldConfig& efc,
+                   BSONObj document);
+
+}  // namespace mongo
diff --git a/src/mongo/db/fle_crud_test.cpp b/src/mongo/db/fle_crud_test.cpp
new file mode 100644
index 00000000000..0c97f8844ea
--- /dev/null
+++ b/src/mongo/db/fle_crud_test.cpp
@@ -0,0 +1,525 @@
+/**
+ *    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 <algorithm>
+#include <array>
+#include <string>
+#include <third_party/murmurhash3/MurmurHash3.h>
+#include <unordered_map>
+#include <vector>
+
+#include "mongo/base/data_range.h"
+#include "mongo/base/string_data.h"
+#include "mongo/bson/bsonelement.h"
+#include "mongo/bson/bsonmisc.h"
+#include "mongo/bson/bsonobj.h"
+#include "mongo/bson/bsonobjbuilder.h"
+#include "mongo/bson/bsontypes.h"
+#include "mongo/bson/json.h"
+#include "mongo/crypto/encryption_fields_gen.h"
+#include "mongo/crypto/fle_crypto.h"
+#include "mongo/crypto/fle_field_schema_gen.h"
+#include "mongo/db/catalog/collection_options.h"
+#include "mongo/db/fle_crud.h"
+#include "mongo/db/matcher/schema/encrypt_schema_gen.h"
+#include "mongo/db/namespace_string.h"
+#include "mongo/db/repl/replication_coordinator_mock.h"
+#include "mongo/db/repl/storage_interface.h"
+#include "mongo/db/repl/storage_interface_impl.h"
+#include "mongo/db/service_context.h"
+#include "mongo/db/service_context_d_test_fixture.h"
+#include "mongo/executor/network_interface_mock.h"
+#include "mongo/idl/idl_parser.h"
+#include "mongo/platform/random.h"
+#include "mongo/unittest/unittest.h"
+#include "mongo/util/assert_util.h"
+#include "mongo/util/uuid.h"
+
+namespace mongo {
+namespace {
+class FLEQueryTestImpl : public FLEQueryInterface {
+public:
+    FLEQueryTestImpl(OperationContext* opCtx, repl::StorageInterface* storage)
+        : _opCtx(opCtx), _storage(storage) {}
+    ~FLEQueryTestImpl() = default;
+
+    BSONObj getById(const NamespaceString& nss, PrfBlock block);
+
+    uint64_t countDocuments(const NamespaceString& nss);
+
+    void insertDocument(const NamespaceString& nss, BSONObj obj, bool translateDuplicateKey);
+
+    BSONObj deleteWithPreimage(const NamespaceString& nss, BSONObj query);
+
+private:
+    OperationContext* _opCtx;
+    repl::StorageInterface* _storage;
+};
+
+BSONObj FLEQueryTestImpl::getById(const NamespaceString& nss, PrfBlock block) {
+    auto obj = BSON("_id" << BSONBinData(block.data(), block.size(), BinDataGeneral));
+    auto swDoc = _storage->findById(_opCtx, nss, obj.firstElement());
+    if (swDoc.getStatus() == ErrorCodes::NoSuchKey) {
+        return BSONObj();
+    }
+
+    return uassertStatusOK(swDoc);
+}
+
+uint64_t FLEQueryTestImpl::countDocuments(const NamespaceString& nss) {
+    return uassertStatusOK(_storage->getCollectionCount(_opCtx, nss));
+}
+
+void FLEQueryTestImpl::insertDocument(const NamespaceString& nss,
+                                      BSONObj obj,
+                                      bool translateDuplicateKey) {
+    repl::TimestampedBSONObj tb;
+    tb.obj = obj;
+
+    auto status = _storage->insertDocument(_opCtx, nss, tb, 0);
+
+    uassertStatusOK(status);
+}
+
+FLEIndexKey indexKey(KeyMaterial{0x6e, 0xda, 0x88, 0xc8, 0x49, 0x6e, 0xc9, 0x90, 0xf5, 0xd5, 0x51,
+                                 0x8d, 0xd2, 0xad, 0x6f, 0x3d, 0x9c, 0x33, 0xb6, 0x05, 0x59, 0x04,
+                                 0xb1, 0x20, 0xf1, 0x2d, 0xe8, 0x29, 0x11, 0xfb, 0xd9, 0x33});
+
+FLEUserKey userKey(KeyMaterial{0x1b, 0xd4, 0x32, 0xd4, 0xce, 0x54, 0x7d, 0xd7, 0xeb, 0xfb, 0x30,
+                               0x9a, 0xea, 0xd6, 0xc6, 0x95, 0xfe, 0x53, 0xff, 0xe9, 0xc4, 0xb1,
+                               0xc4, 0xf0, 0x6f, 0x36, 0x3c, 0xf0, 0x7b, 0x00, 0x28, 0xaf});
+
+constexpr auto kIndexKeyId = "12345678-1234-9876-1234-123456789012"_sd;
+constexpr auto kUserKeyId = "ABCDEFAB-1234-9876-1234-123456789012"_sd;
+static UUID indexKeyId = uassertStatusOK(UUID::parse(kIndexKeyId.toString()));
+static UUID userKeyId = uassertStatusOK(UUID::parse(kUserKeyId.toString()));
+
+std::vector<char> testValue = {0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, 0x19};
+std::vector<char> testValue2 = {0x20, 0x21, 0x22, 0x23, 0x24, 0x25, 0x26, 0x27, 0x28, 0x29};
+
+class TestKeyVault : public FLEKeyVault {
+public:
+    TestKeyVault() : _random(123456) {}
+
+    KeyMaterial getKey(UUID uuid) override;
+
+    uint64_t getCount() const {
+        return _dynamicKeys.size();
+    }
+
+private:
+    PseudoRandom _random;
+    stdx::unordered_map<UUID, KeyMaterial, UUID::Hash> _dynamicKeys;
+};
+
+KeyMaterial TestKeyVault::getKey(UUID uuid) {
+    if (uuid == indexKeyId) {
+        return indexKey.data;
+    } else if (uuid == userKeyId) {
+        return userKey.data;
+    } else {
+        if (_dynamicKeys.find(uuid) != _dynamicKeys.end()) {
+            return _dynamicKeys[uuid];
+        }
+
+        KeyMaterial material;
+        _random.fill(&material, sizeof(material));
+        _dynamicKeys.insert({uuid, material});
+        return material;
+    }
+}
+
+UUID fieldNameToUUID(StringData field) {
+    std::array<uint8_t, UUID::kNumBytes> buf;
+
+    MurmurHash3_x86_128(field.rawData(), field.size(), 123456, buf.data());
+
+    return UUID::fromCDR(buf);
+}
+
+std::string fieldNameFromInt(uint64_t i) {
+    return "field" + std::to_string(i);
+}
+
+class FleCrudTest : public ServiceContextMongoDTest {
+private:
+    void setUp() final;
+    void tearDown() final;
+
+protected:
+    void createCollection(const NamespaceString& ns);
+
+    void assertDocumentCounts(uint64_t edc, uint64_t esc, uint64_t ecc, uint64_t ecoc);
+
+    void doSingleInsert(int id, BSONElement element);
+    void doSingleInsert(int id, BSONObj obj);
+
+    using ValueGenerator = std::function<std::string(StringData fieldName, uint64_t row)>;
+
+    void doSingleWideInsert(int id, uint64_t fieldCount, ValueGenerator func);
+
+    ESCTwiceDerivedTagToken getTestESCToken(BSONElement value);
+    ESCTwiceDerivedTagToken getTestESCToken(BSONObj obj);
+    ESCTwiceDerivedTagToken getTestESCToken(StringData name, StringData value);
+
+    std::vector<char> generatePlaceholder(UUID keyId, BSONElement value);
+
+protected:
+    /**
+     * Looks up the current ReplicationCoordinator.
+     * The result is cast to a ReplicationCoordinatorMock to provide access to test features.
+     */
+    repl::ReplicationCoordinatorMock* _getReplCoord() const;
+
+    ServiceContext::UniqueOperationContext _opCtx;
+
+    repl::StorageInterface* _storage{nullptr};
+
+    std::unique_ptr<FLEQueryTestImpl> _queryImpl;
+
+    TestKeyVault _keyVault;
+
+    NamespaceString _edcNs{"test.edc"};
+    NamespaceString _escNs{"test.esc"};
+    NamespaceString _eccNs{"test.ecc"};
+    NamespaceString _ecocNs{"test.ecoc"};
+};
+
+void FleCrudTest::setUp() {
+    ServiceContextMongoDTest::setUp();
+    auto service = getServiceContext();
+
+    repl::ReplicationCoordinator::set(service,
+                                      std::make_unique<repl::ReplicationCoordinatorMock>(service));
+
+    _opCtx = cc().makeOperationContext();
+
+    repl::StorageInterface::set(service, std::make_unique<repl::StorageInterfaceImpl>());
+    _storage = repl::StorageInterface::get(service);
+
+    _queryImpl = std::make_unique<FLEQueryTestImpl>(_opCtx.get(), _storage);
+
+    createCollection(_edcNs);
+    createCollection(_escNs);
+    createCollection(_eccNs);
+    createCollection(_ecocNs);
+}
+
+void FleCrudTest::tearDown() {
+    _opCtx = {};
+    ServiceContextMongoDTest::tearDown();
+}
+
+void FleCrudTest::createCollection(const NamespaceString& ns) {
+    CollectionOptions collectionOptions;
+    collectionOptions.uuid = UUID::gen();
+    auto statusCC = _storage->createCollection(
+        _opCtx.get(), NamespaceString(ns.db(), ns.coll()), collectionOptions);
+    ASSERT_OK(statusCC);
+}
+
+ConstDataRange toCDR(BSONElement element) {
+    return ConstDataRange(element.value(), element.value() + element.valuesize());
+}
+
+ESCTwiceDerivedTagToken FleCrudTest::getTestESCToken(BSONElement element) {
+    auto c1token = FLELevel1TokenGenerator::generateCollectionsLevel1Token(
+        _keyVault.getIndexKeyById(indexKeyId).key);
+    auto escToken = FLECollectionTokenGenerator::generateESCToken(c1token);
+
+    auto escDataToken =
+        FLEDerivedFromDataTokenGenerator::generateESCDerivedFromDataToken(escToken, toCDR(element));
+    auto escContentionToken = FLEDerivedFromDataTokenAndContentionFactorTokenGenerator::
+        generateESCDerivedFromDataTokenAndContentionFactorToken(escDataToken, 0);
+
+    return FLETwiceDerivedTokenGenerator::generateESCTwiceDerivedTagToken(escContentionToken);
+}
+
+ESCTwiceDerivedTagToken FleCrudTest::getTestESCToken(BSONObj obj) {
+    return getTestESCToken(obj.firstElement());
+}
+
+ESCTwiceDerivedTagToken FleCrudTest::getTestESCToken(StringData name, StringData value) {
+
+    auto doc = BSON("i" << value);
+    auto element = doc.firstElement();
+
+    UUID keyId = fieldNameToUUID(name);
+
+    auto c1token = FLELevel1TokenGenerator::generateCollectionsLevel1Token(
+        _keyVault.getIndexKeyById(keyId).key);
+    auto escToken = FLECollectionTokenGenerator::generateESCToken(c1token);
+
+    auto escDataToken =
+        FLEDerivedFromDataTokenGenerator::generateESCDerivedFromDataToken(escToken, toCDR(element));
+    auto escContentionToken = FLEDerivedFromDataTokenAndContentionFactorTokenGenerator::
+        generateESCDerivedFromDataTokenAndContentionFactorToken(escDataToken, 0);
+
+    return FLETwiceDerivedTokenGenerator::generateESCTwiceDerivedTagToken(escContentionToken);
+}
+
+std::vector<char> FleCrudTest::generatePlaceholder(UUID keyId, BSONElement value) {
+    FLE2EncryptionPlaceholder ep;
+
+    ep.setAlgorithm(mongo::Fle2AlgorithmInt::kEquality);
+    ep.setUserKeyId(keyId);
+    ep.setIndexKeyId(keyId);
+    ep.setValue(value);
+    ep.setType(mongo::Fle2PlaceholderType::kInsert);
+    ep.setMaxContentionCounter(0);
+
+    BSONObj obj = ep.toBSON();
+
+    std::vector<char> v;
+    v.resize(obj.objsize() + 1);
+    v[0] = static_cast<uint8_t>(EncryptedBinDataType::kFLE2Placeholder);
+    std::copy(obj.objdata(), obj.objdata() + obj.objsize(), v.begin() + 1);
+    return v;
+}
+
+EncryptedFieldConfig getTestEncryptedFieldConfig() {
+
+    constexpr auto schema = R"({
+    "escCollection": "esc",
+    "eccCollection": "ecc",
+    "ecocCollection": "ecoc",
+    "fields": [
+        {
+            "keyId":
+                            {
+                                "$uuid": "12345678-1234-9876-1234-123456789012"
+                            }
+                        ,
+            "path": "encrypted",
+            "bsonType": "string",
+            "queries": {"queryType": "equality"}
+
+        }
+    ]
+})";
+
+    return EncryptedFieldConfig::parse(IDLParserErrorContext("root"), fromjson(schema));
+}
+
+void FleCrudTest::assertDocumentCounts(uint64_t edc, uint64_t esc, uint64_t ecc, uint64_t ecoc) {
+    ASSERT_EQ(_queryImpl->countDocuments(_edcNs), edc);
+    ASSERT_EQ(_queryImpl->countDocuments(_escNs), esc);
+    ASSERT_EQ(_queryImpl->countDocuments(_eccNs), ecc);
+    ASSERT_EQ(_queryImpl->countDocuments(_ecocNs), ecoc);
+}
+
+
+// Auto generate key ids from field id
+void FleCrudTest::doSingleWideInsert(int id, uint64_t fieldCount, ValueGenerator func) {
+    BSONObjBuilder builder;
+    builder.append("_id", id);
+    builder.append("plainText", "sample");
+
+    for (uint64_t i = 0; i < fieldCount; i++) {
+        auto name = fieldNameFromInt(i);
+        auto value = func(name, id);
+        auto doc = BSON("I" << value);
+        UUID uuid = fieldNameToUUID(name);
+        auto buf = generatePlaceholder(uuid, doc.firstElement());
+        builder.appendBinData(name, buf.size(), BinDataType::Encrypt, buf.data());
+    }
+
+    auto clientDoc = builder.obj();
+
+    auto result = FLEClientCrypto::generateInsertOrUpdateFromPlaceholders(clientDoc, &_keyVault);
+
+    auto serverPayload = EDCServerCollection::getEncryptedFieldInfo(result);
+
+    auto efc = getTestEncryptedFieldConfig();
+
+    processInsert(_queryImpl.get(), _edcNs, serverPayload, efc, result);
+}
+
+// Use different keys for index and user
+std::vector<char> generateSinglePlaceholder(BSONElement value) {
+    FLE2EncryptionPlaceholder ep;
+
+    ep.setAlgorithm(mongo::Fle2AlgorithmInt::kEquality);
+    ep.setUserKeyId(userKeyId);
+    ep.setIndexKeyId(indexKeyId);
+    ep.setValue(value);
+    ep.setType(mongo::Fle2PlaceholderType::kInsert);
+    ep.setMaxContentionCounter(0);
+
+    BSONObj obj = ep.toBSON();
+
+    std::vector<char> v;
+    v.resize(obj.objsize() + 1);
+    v[0] = static_cast<uint8_t>(EncryptedBinDataType::kFLE2Placeholder);
+    std::copy(obj.objdata(), obj.objdata() + obj.objsize(), v.begin() + 1);
+    return v;
+}
+
+void FleCrudTest::doSingleInsert(int id, BSONElement element) {
+    auto buf = generateSinglePlaceholder(element);
+    BSONObjBuilder builder;
+    builder.append("_id", id);
+    builder.append("plainText", "sample");
+    builder.appendBinData("encrypted", buf.size(), BinDataType::Encrypt, buf.data());
+
+    auto clientDoc = builder.obj();
+
+    auto result = FLEClientCrypto::generateInsertOrUpdateFromPlaceholders(clientDoc, &_keyVault);
+
+    auto serverPayload = EDCServerCollection::getEncryptedFieldInfo(result);
+
+    auto efc = getTestEncryptedFieldConfig();
+
+    processInsert(_queryImpl.get(), _edcNs, serverPayload, efc, result);
+}
+
+void FleCrudTest::doSingleInsert(int id, BSONObj obj) {
+    doSingleInsert(id, obj.firstElement());
+}
+
+// Insert one document
+TEST_F(FleCrudTest, InsertOne) {
+    auto doc = BSON("encrypted"
+                    << "secret");
+    auto element = doc.firstElement();
+
+    doSingleInsert(1, element);
+
+    assertDocumentCounts(1, 1, 0, 1);
+
+    ASSERT_FALSE(_queryImpl->getById(_escNs, ESCCollection::generateId(getTestESCToken(element), 1))
+                     .isEmpty());
+}
+
+// Insert two documents with same values
+TEST_F(FleCrudTest, InsertTwoSame) {
+
+    auto doc = BSON("encrypted"
+                    << "secret");
+    auto element = doc.firstElement();
+    doSingleInsert(1, element);
+    doSingleInsert(2, element);
+
+    assertDocumentCounts(2, 2, 0, 2);
+
+    ASSERT_FALSE(_queryImpl->getById(_escNs, ESCCollection::generateId(getTestESCToken(element), 1))
+                     .isEmpty());
+    ASSERT_FALSE(_queryImpl->getById(_escNs, ESCCollection::generateId(getTestESCToken(element), 2))
+                     .isEmpty());
+}
+
+// Insert two documents with different values
+TEST_F(FleCrudTest, InsertTwoDifferent) {
+
+    doSingleInsert(1,
+                   BSON("encrypted"
+                        << "secret"));
+    doSingleInsert(2,
+                   BSON("encrypted"
+                        << "topsecret"));
+
+    assertDocumentCounts(2, 2, 0, 2);
+
+    ASSERT_FALSE(_queryImpl
+                     ->getById(_escNs,
+                               ESCCollection::generateId(getTestESCToken(BSON("encrypted"
+                                                                              << "secret")),
+                                                         1))
+                     .isEmpty());
+    ASSERT_FALSE(_queryImpl
+                     ->getById(_escNs,
+                               ESCCollection::generateId(getTestESCToken(BSON("encrypted"
+                                                                              << "topsecret")),
+                                                         1))
+                     .isEmpty());
+}
+
+// Insert 1 document with 100 fields
+TEST_F(FleCrudTest, Insert100Fields) {
+
+    uint64_t fieldCount = 100;
+    ValueGenerator valueGenerator = [](StringData fieldName, uint64_t row) {
+        return fieldName.toString();
+    };
+    doSingleWideInsert(1, fieldCount, valueGenerator);
+
+    assertDocumentCounts(1, fieldCount, 0, fieldCount);
+
+    for (uint64_t field = 0; field < fieldCount; field++) {
+        auto fieldName = fieldNameFromInt(field);
+
+        ASSERT_FALSE(
+            _queryImpl
+                ->getById(
+                    _escNs,
+                    ESCCollection::generateId(
+                        getTestESCToken(fieldName, valueGenerator(fieldNameFromInt(field), 0)), 1))
+                .isEmpty());
+    }
+}
+
+// Insert 100 documents each with 20 fields with 7 distinct values per field
+TEST_F(FleCrudTest, Insert20Fields50Rows) {
+
+    uint64_t fieldCount = 20;
+    uint64_t rowCount = 50;
+
+    ValueGenerator valueGenerator = [](StringData fieldName, uint64_t row) {
+        return fieldName.toString() + std::to_string(row % 7);
+    };
+
+
+    for (uint64_t row = 0; row < rowCount; row++) {
+        doSingleWideInsert(row, fieldCount, valueGenerator);
+    }
+
+    assertDocumentCounts(rowCount, rowCount * fieldCount, 0, rowCount * fieldCount);
+
+    for (uint64_t row = 0; row < rowCount; row++) {
+        for (uint64_t field = 0; field < fieldCount; field++) {
+            auto fieldName = fieldNameFromInt(field);
+
+            int count = (row / 7) + 1;
+
+            ASSERT_FALSE(
+                _queryImpl
+                    ->getById(_escNs,
+                              ESCCollection::generateId(
+                                  getTestESCToken(fieldName,
+                                                  valueGenerator(fieldNameFromInt(field), row)),
+                                  count))
+                    .isEmpty());
+        }
+    }
+}
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/db/ops/SConscript b/src/mongo/db/ops/SConscript
index e2f6ce1a44d..69a93922ab9 100644
--- a/src/mongo/db/ops/SConscript
+++ b/src/mongo/db/ops/SConscript
@@ -48,6 +48,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/crypto/encrypted_field_config',
+        '$BUILD_DIR/mongo/crypto/fle_fields',
     ],
 )
 
diff --git a/src/mongo/db/ops/write_ops.idl b/src/mongo/db/ops/write_ops.idl
index 41ee21b5346..cc2ab757c81 100644
--- a/src/mongo/db/ops/write_ops.idl
+++ b/src/mongo/db/ops/write_ops.idl
@@ -33,6 +33,7 @@ global:
         - "mongo/db/repl/optime.h"
 
 imports:
+    - "mongo/crypto/fle_field_schema.idl"
     - "mongo/db/auth/action_type.idl"
     - "mongo/db/logical_session_id.idl"
     - "mongo/db/pipeline/legacy_runtime_constants.idl"
@@ -95,7 +96,7 @@ structs:
                 optional: true
                 unstable: false
             writeErrors:
-                description: "Contains all the errors encoutered."
+                description: "Contains all the errors encountered."
                 type: array<write_error>
                 optional: true
                 unstable: false
@@ -196,6 +197,11 @@ structs:
                 type: uuid
                 optional: true
                 unstable: true
+            encryptionInformation:
+                description: "Encryption Information schema and other tokens for CRUD commands"
+                type: EncryptionInformation
+                optional: true
+                unstable: true
 
     UpdateOpEntry:
         description: "Parser for the entries in the 'updates' array of an update command."
diff --git a/src/mongo/s/SConscript b/src/mongo/s/SConscript
index 7be21eef763..75522bca2fb 100644
--- a/src/mongo/s/SConscript
+++ b/src/mongo/s/SConscript
@@ -34,6 +34,7 @@ env.Library(
         'cluster_write.cpp',
     ],
     LIBDEPS=[
+        '$BUILD_DIR/mongo/db/fle_crud',
         '$BUILD_DIR/mongo/db/not_primary_error_tracker',
         '$BUILD_DIR/mongo/db/timeseries/timeseries_conversion_util',
         '$BUILD_DIR/mongo/db/timeseries/timeseries_options',
diff --git a/src/mongo/s/cluster_write.cpp b/src/mongo/s/cluster_write.cpp
index 5ac8f4e0511..441e8a19f6a 100644
--- a/src/mongo/s/cluster_write.cpp
+++ b/src/mongo/s/cluster_write.cpp
@@ -35,6 +35,7 @@
 
 #include "mongo/s/cluster_write.h"
 
+#include "mongo/db/fle_crud.h"
 #include "mongo/db/not_primary_error_tracker.h"
 #include "mongo/s/chunk_manager_targeter.h"
 #include "mongo/s/grid.h"
@@ -47,6 +48,15 @@ void write(OperationContext* opCtx,
            BatchWriteExecStats* stats,
            BatchedCommandResponse* response,
            boost::optional<OID> targetEpoch) {
+    if (request.hasEncryptionInformation()) {
+        FLEBatchResult result = processFLEBatch(opCtx, request, stats, response, targetEpoch);
+        if (result == FLEBatchResult::kProcessed) {
+            return;
+        }
+
+        // fall through
+    }
+
     NotPrimaryErrorTracker::Disabled scopeDisabledTracker(
         &NotPrimaryErrorTracker::get(opCtx->getClient()));
 
diff --git a/src/mongo/s/write_ops/batched_command_request.cpp b/src/mongo/s/write_ops/batched_command_request.cpp
index c916b5b58ba..107f1a49204 100644
--- a/src/mongo/s/write_ops/batched_command_request.cpp
+++ b/src/mongo/s/write_ops/batched_command_request.cpp
@@ -95,6 +95,11 @@ const NamespaceString& BatchedCommandRequest::getNS() const {
     return _visit([](auto&& op) -> decltype(auto) { return op.getNamespace(); });
 }
 
+bool BatchedCommandRequest::hasEncryptionInformation() const {
+    return _visit(
+        [](auto&& op) -> decltype(auto) { return op.getEncryptionInformation().has_value(); });
+}
+
 std::size_t BatchedCommandRequest::sizeWriteOps() const {
     struct Visitor {
         auto operator()(const write_ops::InsertCommandRequest& op) const {
diff --git a/src/mongo/s/write_ops/batched_command_request.h b/src/mongo/s/write_ops/batched_command_request.h
index f900d52140a..eea7f7bbe11 100644
--- a/src/mongo/s/write_ops/batched_command_request.h
+++ b/src/mongo/s/write_ops/batched_command_request.h
@@ -74,6 +74,8 @@ public:
 
     bool getBypassDocumentValidation() const;
 
+    bool hasEncryptionInformation() const;
+
     const auto& getInsertRequest() const {
         invariant(_insertReq);
         return *_insertReq;