From e3d39d7afd9ed2afe7c73acb7e8cc760ad6c844c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ivan=20=E2=80=9CCLOVIS=E2=80=9D=20Canet?= Date: Sun, 6 Sep 2026 09:41:50 +0200 Subject: [PATCH] feat(driver-multiplatform): Prototype updates and upserts --- ...ultiplatformFilteredMongoCollectionImpl.kt | 4 +- .../kotlin/MultiplatformMongoCollection.kt | 27 ++++ .../MultiplatformMongoCollectionImpl.kt | 140 ++++++++++++++++-- .../kotlin/MultiplatformMongoIterableImpl.kt | 2 +- 4 files changed, 159 insertions(+), 14 deletions(-) diff --git a/driver-multiplatform/src/commonMain/kotlin/MultiplatformFilteredMongoCollectionImpl.kt b/driver-multiplatform/src/commonMain/kotlin/MultiplatformFilteredMongoCollectionImpl.kt index 0fcd4dcb..be6859c3 100644 --- a/driver-multiplatform/src/commonMain/kotlin/MultiplatformFilteredMongoCollectionImpl.kt +++ b/driver-multiplatform/src/commonMain/kotlin/MultiplatformFilteredMongoCollectionImpl.kt @@ -62,7 +62,7 @@ private class MultiplatformFilteredMongoCollectionImpl( override val factory: BsonFactory get() = upstream.factory - override suspend fun upsertOne(options: UpdateOptions.() -> Unit, filter: FilterQuery.() -> Unit, update: UpsertQuery.() -> Unit): UpdateOperations.UpsertResult = + override suspend fun upsertOne(options: UpdateOptions.() -> Unit, filter: FilterQuery.() -> Unit, update: UpsertQuery.() -> Unit): MultiplatformMongoCollection.UpsertResult = upstream.upsertOne( options = options, filter = { @@ -72,7 +72,7 @@ private class MultiplatformFilteredMongoCollectionImpl( update = update ) - override suspend fun upsertOneWithPipeline(options: UpdateOptions.() -> Unit, filter: FilterQuery.() -> Unit, update: UpdateWithPipelineQuery.() -> Unit): UpdateOperations.UpsertResult = + override suspend fun upsertOneWithPipeline(options: UpdateOptions.() -> Unit, filter: FilterQuery.() -> Unit, update: UpdateWithPipelineQuery.() -> Unit): MultiplatformMongoCollection.UpsertResult = upstream.upsertOneWithPipeline( options = options, filter = { diff --git a/driver-multiplatform/src/commonMain/kotlin/MultiplatformMongoCollection.kt b/driver-multiplatform/src/commonMain/kotlin/MultiplatformMongoCollection.kt index 1ffd3555..fc546bdd 100644 --- a/driver-multiplatform/src/commonMain/kotlin/MultiplatformMongoCollection.kt +++ b/driver-multiplatform/src/commonMain/kotlin/MultiplatformMongoCollection.kt @@ -17,10 +17,15 @@ package opensavvy.ktmongo.multiplatform import opensavvy.ktmongo.api.MongoCollection +import opensavvy.ktmongo.api.operations.UpdateOperations import opensavvy.ktmongo.bson.multiplatform.BsonFactory +import opensavvy.ktmongo.bson.multiplatform.BsonValue import opensavvy.ktmongo.bson.types.ObjectId import opensavvy.ktmongo.dsl.command.FindOptions +import opensavvy.ktmongo.dsl.command.UpdateOptions import opensavvy.ktmongo.dsl.query.FilterQuery +import opensavvy.ktmongo.dsl.query.UpdateWithPipelineQuery +import opensavvy.ktmongo.dsl.query.UpsertQuery /** * A collection stores related documents together. @@ -73,4 +78,26 @@ interface MultiplatformMongoCollection : MongoCollection.() -> Unit, filter: FilterQuery.() -> Unit): MultiplatformMongoIterable override fun filter(filter: FilterQuery.() -> Unit): MultiplatformMongoCollection + + /** + * The return value of [upsertOne] and [upsertOneWithPipeline]. + */ + interface UpsertResult : UpdateOperations.UpsertResult { + + override val upsertedId: BsonValue? + } + + @IgnorableReturnValue + override suspend fun upsertOne( + options: UpdateOptions.() -> Unit, + filter: FilterQuery.() -> Unit, + update: UpsertQuery.() -> Unit, + ): UpsertResult + + @IgnorableReturnValue + override suspend fun upsertOneWithPipeline( + options: UpdateOptions.() -> Unit, + filter: FilterQuery.() -> Unit, + update: UpdateWithPipelineQuery.() -> Unit, + ): UpsertResult } diff --git a/driver-multiplatform/src/commonMain/kotlin/MultiplatformMongoCollectionImpl.kt b/driver-multiplatform/src/commonMain/kotlin/MultiplatformMongoCollectionImpl.kt index 2f2c4309..bdf7be11 100644 --- a/driver-multiplatform/src/commonMain/kotlin/MultiplatformMongoCollectionImpl.kt +++ b/driver-multiplatform/src/commonMain/kotlin/MultiplatformMongoCollectionImpl.kt @@ -121,12 +121,6 @@ internal class MultiplatformMongoCollectionImpl( TODO("Not yet implemented") } - private fun BsonValue.decodeLong(message: Any?): Long = when (this.type) { - BsonType.Int32 -> decodeInt32().toLong() - BsonType.Int64 -> decodeInt64() - else -> error("Unexpected count type: $type in $message") - } - @OptIn(DangerousMongoApi::class) override suspend fun count(): Long { val result = Field.unsafe("c") @@ -254,15 +248,84 @@ internal class MultiplatformMongoCollectionImpl( ) override suspend fun updateMany(options: UpdateOptions.() -> Unit, filter: FilterQuery.() -> Unit, update: UpdateQuery.() -> Unit): UpdateOperations.UpdateResult { - TODO("Not yet implemented") + val model = UpdateMany( + context = database.client.context, + ).apply { + this.options.options() + this.filter.filter() + this.update.update() + } + + val message = database.client.wire.sendSingle( + database.client.createOpMsg { + document { + writeString("update", name) + writeString($$"$db", database.name) + model.writeTo(this) + } + } + ) + + message as Message.OpMsg + check(message.body.document["ok"]?.decodeDouble() == 1.0) + + // TODO handle unacknowledged updates + + return MultiplatformAcknowledgedUpdateResult(message.body.document) } override suspend fun updateOne(options: UpdateOptions.() -> Unit, filter: FilterQuery.() -> Unit, update: UpdateQuery.() -> Unit): UpdateOperations.UpdateResult { - TODO("Not yet implemented") + val model = UpdateOne( + context = database.client.context, + ).apply { + this.options.options() + this.filter.filter() + this.update.update() + } + + val message = database.client.wire.sendSingle( + database.client.createOpMsg { + document { + writeString("update", name) + writeString($$"$db", database.name) + model.writeTo(this) + } + } + ) + + message as Message.OpMsg + check(message.body.document["ok"]?.decodeDouble() == 1.0) + + // TODO handle unacknowledged updates + + return MultiplatformAcknowledgedUpdateResult(message.body.document) } - override suspend fun upsertOne(options: UpdateOptions.() -> Unit, filter: FilterQuery.() -> Unit, update: UpsertQuery.() -> Unit): UpdateOperations.UpsertResult { - TODO("Not yet implemented") + override suspend fun upsertOne(options: UpdateOptions.() -> Unit, filter: FilterQuery.() -> Unit, update: UpsertQuery.() -> Unit): MultiplatformMongoCollection.UpsertResult { + val model = UpsertOne( + context = database.client.context, + ).apply { + this.options.options() + this.filter.filter() + this.update.update() + } + + val message = database.client.wire.sendSingle( + database.client.createOpMsg { + document { + writeString("update", name) + writeString($$"$db", database.name) + model.writeTo(this) + } + } + ) + + message as Message.OpMsg + check(message.body.document["ok"]?.decodeDouble() == 1.0) + + // TODO handle unacknowledged updates + + return MultiplatformAcknowledgedUpdateResult(message.body.document) } override suspend fun replaceOne(options: ReplaceOptions.() -> Unit, filter: FilterQuery.() -> Unit, document: Document) { @@ -289,10 +352,65 @@ internal class MultiplatformMongoCollectionImpl( TODO("Not yet implemented") } - override suspend fun upsertOneWithPipeline(options: UpdateOptions.() -> Unit, filter: FilterQuery.() -> Unit, update: UpdateWithPipelineQuery.() -> Unit): UpdateOperations.UpsertResult { + override suspend fun upsertOneWithPipeline(options: UpdateOptions.() -> Unit, filter: FilterQuery.() -> Unit, update: UpdateWithPipelineQuery.() -> Unit): MultiplatformMongoCollection.UpsertResult { TODO("Not yet implemented") } override fun toString(): String = "MultiplatformMongoCollection($fullyQualifiedName)" } + +private data class MultiplatformAcknowledgedUpdateResult( + private val doc: BsonDocument, +) : MultiplatformMongoCollection.UpsertResult { + + override val upsertedId: BsonValue? + get() = doc["upserted"]?.decodeArray() + ?.get(0)?.decodeDocument() + ?.get("_id") + + override val upsertedCount: Int + get() = doc["upserted"]?.decodeArray() + ?.size ?: 0 + + override val acknowledged: Boolean + get() = true + + override val matchedCount: Long + get() = doc["n"]?.decodeLong(doc) + ?.minus(upsertedCount) // In the case of an insert, the wire protocol returns 1, but the Java driver reports 0 + ?: 0L + + override val modifiedCount: Long + get() = doc["nModified"]?.decodeLong(doc) ?: 0L + + override fun toString(): String = + "UpdateResult(acknowledged=true, matchedCount=$matchedCount, modifiedCount=$modifiedCount, upsertedCount=$upsertedCount, upsertedId=$upsertedId; decodedFrom=$doc)" +} + +private data object MultiplatformUnacknowledgedUpdateResult : MultiplatformMongoCollection.UpsertResult { + + override val upsertedId: Nothing + get() = throw UnsupportedOperationException("Unacknowledged updates do not provide the upsertedId field") + + override val upsertedCount: Nothing + get() = throw UnsupportedOperationException("Unacknowledged updates do not provide the upsertedCount field") + + override val acknowledged: Boolean + get() = false + + override val matchedCount: Nothing + get() = throw UnsupportedOperationException("Unacknowledged updates do not provide the matchedCount field") + + override val modifiedCount: Nothing + get() = throw UnsupportedOperationException("Unacknowledged updates do not provide the modifiedCount field") + + override fun toString(): String = + "UpdateResult(acknowledged=false)" +} + +private fun BsonValue.decodeLong(message: Any?): Long = when (this.type) { + BsonType.Int32 -> decodeInt32().toLong() + BsonType.Int64 -> decodeInt64() + else -> error("Unexpected count type: $type in $message") +} diff --git a/driver-multiplatform/src/commonMain/kotlin/MultiplatformMongoIterableImpl.kt b/driver-multiplatform/src/commonMain/kotlin/MultiplatformMongoIterableImpl.kt index 56aa862c..62a135ab 100644 --- a/driver-multiplatform/src/commonMain/kotlin/MultiplatformMongoIterableImpl.kt +++ b/driver-multiplatform/src/commonMain/kotlin/MultiplatformMongoIterableImpl.kt @@ -79,7 +79,7 @@ internal class MultiplatformMongoIterableFindImpl( ) firstBatch as Message.OpMsg - check(firstBatch.body.document["ok"]?.decodeDouble() == 1.0) + check(firstBatch.body.document["ok"]?.decodeDouble() == 1.0) { "Response is not OK: $firstBatch" } val cursor = firstBatch.body.document["cursor"]?.decodeDocument() -- 2.51.2