From 08ec6390ff2f1aad09b71c6041f3d5f5ab641643 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Ivan=20=E2=80=9CCLOVIS=E2=80=9D=20Canet?=
 <ivan.canet@gmail.com>
Date: Mon, 7 Sep 2026 20:39:16 +0200
Subject: [PATCH] feat(driver-multiplatform): Prototype updates with pipeline

---
 .../MultiplatformMongoCollectionImpl.kt       | 75 ++++++++++++++++++-
 1 file changed, 72 insertions(+), 3 deletions(-)

diff --git a/driver-multiplatform/src/commonMain/kotlin/MultiplatformMongoCollectionImpl.kt b/driver-multiplatform/src/commonMain/kotlin/MultiplatformMongoCollectionImpl.kt
index e8b9a946..7111fd2f 100644
--- a/driver-multiplatform/src/commonMain/kotlin/MultiplatformMongoCollectionImpl.kt
+++ b/driver-multiplatform/src/commonMain/kotlin/MultiplatformMongoCollectionImpl.kt
@@ -412,15 +412,84 @@ internal class MultiplatformMongoCollectionImpl<Document : Any>(
 	}
 
 	override suspend fun updateManyWithPipeline(options: UpdateOptions<Document>.() -> Unit, filter: FilterQuery<Document>.() -> Unit, update: UpdateWithPipelineQuery<Document>.() -> Unit): UpdateOperations.UpdateResult {
-		TODO("Not yet implemented")
+		val model = UpdateManyWithPipeline<Document>(
+			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 updateOneWithPipeline(options: UpdateOptions<Document>.() -> Unit, filter: FilterQuery<Document>.() -> Unit, update: UpdateWithPipelineQuery<Document>.() -> Unit): UpdateOperations.UpdateResult {
-		TODO("Not yet implemented")
+		val model = UpdateOneWithPipeline<Document>(
+			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 upsertOneWithPipeline(options: UpdateOptions<Document>.() -> Unit, filter: FilterQuery<Document>.() -> Unit, update: UpdateWithPipelineQuery<Document>.() -> Unit): MultiplatformMongoCollection.UpsertResult {
-		TODO("Not yet implemented")
+		val model = UpsertOneWithPipeline<Document>(
+			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 fun toString(): String =
-- 
2.51.2