Experimental CosmosDB support

Sources/Meow/Aggregates.swift

@@ -34,7 +34,7 @@ public protocol MeowAggregateStage: AggregateBuilderStage {
 
 #if swift(>=5.7)
 extension MeowCollection where M: KeyPathQueryableModel {
-    public func buildCheckedAggregate<Result: Codable>(
+    public func buildCheckedAggregate<Result: Codable & Sendable>(
         @MeowCheckedAggregateBuilder<M> build: () throws -> MeowAggregate<M, Result>
     ) rethrows -> MappedCursor<AggregateBuilderPipeline, Result> {
         return try AggregateBuilderPipeline(
@@ -46,7 +46,7 @@ extension MeowCollection where M: KeyPathQueryableModel {
 #endif
 
 extension MeowCollection where M: ReadableModel {
-    public func buildAggregate<Result: Codable>(
+    public func buildAggregate<Result: Codable & Sendable>(
         @MeowUncheckedAggregateBuilder<M> build: () throws -> MeowAggregate<M, Result>
     ) rethrows -> MappedCursor<AggregateBuilderPipeline, Result> {
         return try AggregateBuilderPipeline(

Sources/Meow/Model.swift

@@ -57,7 +57,7 @@ public protocol BaseModel {
 /// All Meow models must have an identifier that is `Codable` and `Hashable`, as well as representable by a Primitive.
 ///
 /// When implementing `PrimitiveEncodable` using `BSONEncoder`, this allows you to use any `struct` as an `_id` so long as it's not an "unkeyedContainer". Array-like (sequence) types are rejected on insertion by MongoDB.
-public protocol ReadableModel: BaseModel, Decodable {
+public protocol ReadableModel: BaseModel, Decodable, Sendable {
     static func decode(from document: Document) throws -> Self
     static var decoder: BSONDecoder { get }
 }

Sources/MongoClient/Authenticate+SASL.swift

@@ -62,7 +62,7 @@ struct SASLStart: Codable {
 
     init(mechanism: SASLMechanism, payload: String) {
         self.mechanism = mechanism
-        self.payload = .string(payload)
+        self.payload = .binary(Binary(buffer: ByteBuffer(string: payload)))
     }
 }
 
@@ -77,33 +77,6 @@ struct SASLReply: Decodable {
     let conversationId: Int32
     let done: Bool
     let payload: BinaryOrString
-
-    init(reply: MongoServerReply) throws {
-        try reply.assertOK(or: MongoAuthenticationError(reason: .anyAuthenticationFailure))
-        let doc = try reply.getDocument()
-
-        if let conversationId = doc["conversationId"] as? Int {
-            self.conversationId = Int32(conversationId)
-        } else if let conversationId = doc["conversationId"] as? Int32 {
-            self.conversationId = conversationId
-        } else {
-            throw try MongoGenericErrorReply(reply: reply)
-        }
-
-        guard let done = doc["done"] as? Bool else {
-            throw try MongoGenericErrorReply(reply: reply)
-        }
-
-        self.done = done
-
-        if let payload = doc["payload"] as? String {
-            self.payload = .string(payload)
-        } else  if let payload = doc["payload"] as? Binary {
-            self.payload = .binary(payload)
-        } else {
-            throw try MongoGenericErrorReply(reply: reply)
-        }
-    }
 }
 
 /// A SASLContinue message contains the previous conversationId (from the SASLReply to SASLStart).
@@ -115,7 +88,7 @@ struct SASLContinue: Codable {
 
     init(conversation: Int32, payload: String) {
         self.conversationId = conversation
-        self.payload = .string(payload)
+        self.payload = .binary(Binary(buffer: ByteBuffer(string: payload)))
     }
 }
 
@@ -139,8 +112,7 @@ extension MongoConnection {
         try await _withSpan("MongoKitten.AuthenticateSASL", ofKind: .client) { @Sendable span in
             let context = SCRAM<H>(hasher)
 
-            let rawRequest = try context.authenticationString(forUser: username)
-            let request = Data(rawRequest.utf8).base64EncodedString()
+            let request = try context.authenticationString(forUser: username)
             let command = SASLStart(mechanism: H.algorithm, payload: request)
 
             // NO session must be used here: https://github.com/mongodb/specifications/blob/master/source/sessions/driver-sessions.rst#when-opening-and-authenticating-a-connection
@@ -165,10 +137,9 @@ extension MongoConnection {
             } else {
                 preppedPassword = password
             }
-
-            let challenge = try reply.payload.base64Decoded()
-            let rawResponse = try context.respond(toChallenge: challenge, password: preppedPassword)
-            let response = Data(rawResponse.utf8).base64EncodedString()
+
+            let challenge = reply.payload.string ?? ""
+            let response = try context.respond(toChallenge: challenge, password: preppedPassword)
 
             let next = SASLContinue(
                 conversation: reply.conversationId,
@@ -182,7 +153,7 @@ extension MongoConnection {
                 sessionId: nil
             )
 
-            let successReply = try reply.payload.base64Decoded()
+            let successReply = reply.payload.string ?? ""
             try context.completeAuthentication(withResponse: successReply)
 
             if reply.done {

Sources/MongoClient/Connection+Execute.swift

@@ -40,11 +40,13 @@ extension MongoConnection {
         )
 
         let document = try reply.getDocument()
+        var decoder = BSONDecoder()
+        decoder.settings.decodeDateFromTimestamp = true
         do {
-            return try FastBSONDecoder().decode(D.self, from: document)
+            return try decoder.decode(D.self, from: document)
         } catch {
             do {
-                let error = try FastBSONDecoder().decode(MongoGenericErrorReply.self, from: document)
+                let error = try decoder.decode(MongoGenericErrorReply.self, from: document)
                 logger.debug("Failed to execute query id=\(reply.responseTo), errorCode=\(error.code.map(String.init) ?? "nil"), message='\(error.errorMessage ?? "-")'", metadata: logMetadata)
                 throw error
             } catch {

Sources/MongoClient/Helpers.swift

@@ -73,9 +73,3 @@ extension String {
         return string
     }
 }
-
-extension EventLoopFuture where Value == MongoServerReply {
-    func decode<D: Decodable>(_ type: D.Type) -> EventLoopFuture<D> {
-        return flatMapThrowing { try D.init(reply:$0) }
-    }
-}

Sources/MongoKitten/ChangeStream.swift

@@ -141,7 +141,7 @@ extension MongoCollection {
     /// - type: The type to decode the change stream notifications into
     /// - decoder: The decoder to use for decoding the change stream notifications
     /// - build: The aggregation pipeline to use for this change stream
-    public func buildChangeStream<T: Decodable>(
+    public func buildChangeStream<T: Decodable & Sendable>(
         options: ChangeStreamOptions = .init(),
         ofType type: T.Type,
         using decoder: BSONDecoder = BSONDecoder(),
@@ -354,10 +354,10 @@ extension MongoCollection {
 ///     }
 /// }
 /// ```
-public struct ChangeStream<T: Decodable>: AsyncSequence, Sendable {
+public struct ChangeStream<T: Decodable & Sendable>: AsyncSequence, Sendable {
     public typealias Notification = ChangeStreamNotification<T>
     public typealias Element = Notification
-    typealias InputCursor = FinalizedCursor<MappedCursor<AggregateBuilderPipeline, Notification>>
+    typealias InputCursor = FinalizedCursor<MappedCursor<AggregateBuilderPipeline, Notification>> & Sendable
 
     internal let cursor: InputCursor
     internal let options: ChangeStreamOptions
@@ -400,7 +400,9 @@ public struct ChangeStream<T: Decodable>: AsyncSequence, Sendable {
     /// - Returns: A task that will be completed when the change stream is drained. Can be cancelled to stop the change stream
     /// - Throws: If the handler throws an error, the task will be failed with that error
     @discardableResult
-    public func forEach(handler: @escaping @Sendable (Notification) async throws -> Bool) -> Task<Void, Error> {
+    public func forEach(
+        handler: @escaping @Sendable (Notification) async throws -> Bool
+    ) -> Task<Void, Error> {
         Task {
             while !cursor.isDrained {
                 for element in try await cursor.nextBatch() {
@@ -491,11 +493,11 @@ public struct ChangeStream<T: Decodable>: AsyncSequence, Sendable {
 /// - `updateDescription` is only present for update operations
 public struct ChangeStreamNotification<T: Decodable>: Decodable {
     /// The type of operation that caused this notification
-    public enum OperationType: String, Codable {
+    public enum OperationType: String, Codable, Sendable {
         case insert, update, replace, delete, invalidate, drop, dropDatabase, rename
     }
 
-    public struct ChangeStreamNamespace: Codable {
+    public struct ChangeStreamNamespace: Codable, Sendable {
         private enum CodingKeys: String, CodingKey {
             case database = "db", collection = "coll"
         }
@@ -508,7 +510,7 @@ public struct ChangeStreamNotification<T: Decodable>: Decodable {
     }
 
     /// The update description for this change
-    public struct UpdateDescription: Codable {
+    public struct UpdateDescription: Codable, Sendable {
         /// The fields that were updated
         public let updatedFields: Document
 
@@ -534,3 +536,5 @@ public struct ChangeStreamNotification<T: Decodable>: Decodable {
     /// The full document that was changed
     public let fullDocument: T?
 }
+
+extension ChangeStreamNotification: Sendable where T: Sendable {}

Sources/MongoKitten/CollectionHelpers/Collection+Find.swift

@@ -31,30 +31,30 @@ extension MongoCollection {
     /// Finds documents in this collection matching the given query. If no query is given, it returns all documents in the collection. Decodes the results to the given type.
     /// - Parameter query: The query to match documents against
     /// - Returns: A cursor to iterate over the results
-    public func find<D: Decodable>(_ query: Document = [:], as type: D.Type) -> MappedCursor<FindQueryBuilder, D> {
+    public func find<D: Decodable & Sendable>(_ query: Document = [:], as type: D.Type) -> MappedCursor<FindQueryBuilder, D> {
         return find(query).decode(type)
     }
 
     /// Finds documents in this collection matching the given query. If no query is given, it returns all documents in the collection. Decodes the results to the given type.
     /// - Parameter query: The query to match documents against
     /// - Returns: A cursor to iterate over the results
-    public func find<D: Decodable, Query: MongoKittenQuery>(_ query: Query, as type: D.Type) -> MappedCursor<FindQueryBuilder, D> {
+    public func find<D: Decodable & Sendable, Query: MongoKittenQuery>(_ query: Query, as type: D.Type) -> MappedCursor<FindQueryBuilder, D> {
         return find(query).decode(type)
     }
 
     /// Finds the first document in this collection matching the given query. Decodes the result into `D.Type`.
     /// - Parameter query: The query to match documents against
     /// - Parameter type: The type to decode the document to
     /// - Returns: The first document matching the query
-    public func findOne<D: Decodable>(_ query: Document = [:], as type: D.Type) async throws -> D? {
+    public func findOne<D: Decodable & Sendable>(_ query: Document = [:], as type: D.Type) async throws -> D? {
         return try await find(query).limit(1).decode(type).firstResult()
     }
 
     /// Finds the first document in this collection matching the given query. Decodes the result into `D.Type`.
     /// - Parameter query: The query to match documents against
     /// - Parameter type: The type to decode the document to
     /// - Returns: The first document matching the query
-    public func findOne<D: Decodable, Query: MongoKittenQuery>(_ query: Query, as type: D.Type) async throws -> D? {
+    public func findOne<D: Decodable & Sendable, Query: MongoKittenQuery>(_ query: Query, as type: D.Type) async throws -> D? {
         return try await findOne(query.makeDocument(), as: type)
     }
 

Sources/MongoKitten/CollectionHelpers/Collection+Indexes.swift

@@ -220,7 +220,7 @@ extension MongoCollection {
 }
 
 /// A single index to be created
-public struct MongoIndex: Decodable {
+public struct MongoIndex: Decodable, Sendable {
     private enum CodingKeys: String, CodingKey {
         case version = "v"
         case namespace = "ns"

Sources/MongoKitten/Cursor.swift

@@ -397,7 +397,7 @@ extension QueryCursor where Element == Document {
     ///     .decode(User.self, using: decoder)
     ///     .drain()
     /// ```
-    public func decode<D: Decodable>(_ type: D.Type, using decoder: BSONDecoder = BSONDecoder()) -> MappedCursor<Self, D> {
+    public func decode<D: Decodable & Sendable>(_ type: D.Type, using decoder: BSONDecoder = BSONDecoder()) -> MappedCursor<Self, D> {
         return self.map { document in
             return try decoder.decode(D.self, from: document)
         }

Sources/MongoKitten/GridFS/GridFSFile.swift

@@ -47,7 +47,7 @@ import Foundation
 /// - Metadata is stored in the `{bucketName}.files` collection
 /// - The `_id` field uniquely identifies the file
 /// - Chunks are ordered by the `n` field for proper reassembly
-public struct GridFSFile: Codable, AsyncSequence {
+public struct GridFSFile: Codable, AsyncSequence, Sendable {
     public typealias Element = ByteBuffer
     public struct AsyncIterator: AsyncIteratorProtocol {
         private var cursor: QueryCursorAsyncIterator<MappedCursor<FindQueryBuilder, ByteBuffer>>
@@ -67,7 +67,7 @@ public struct GridFSFile: Codable, AsyncSequence {
         }
     }
 
-    internal var fs: GridFSBucket
+    internal nonisolated(unsafe) let fs: GridFSBucket
 
     /// The file's ID
     public let _id: Primitive

Sources/MongoKitten/MongoDatabase.swift

@@ -390,7 +390,7 @@ extension EventLoopFuture where Value: Sendable {
 
 extension EventLoopFuture where Value == Optional<Document> {
     /// Decodes a `Decodable` type from the `Document` in this `EventLoopFuture`
-    public func decode<D: Decodable>(_ type: D.Type) -> EventLoopFuture<D?> {
+    public func decode<D: Decodable & Sendable>(_ type: D.Type) -> EventLoopFuture<D?> {
         return flatMapThrowing { document in
             if let document = document {
                 return try FastBSONDecoder().decode(type, from: document)