feat: add UpsertAsync/UpsertBulkAsync

README.md

@@ -702,9 +702,14 @@ public partial class MyDbContext : DocumentDbContext
 using var db = new MyDbContext("mydb.db");
 
 // Operations are tracked automatically
-await db.Users.InsertAsync(new User { Name = "Alice" });
+var aliceId = await db.Users.InsertAsync(new User { Name = "Alice" });
 await db.Users.InsertAsync(new User { Name = "Bob" });
 
+// Upsert: inserts if the id is unset/absent, otherwise replaces the existing document.
+// Resolved with a single primary-index lookup — not a Find followed by Insert/Update.
+UpsertResult<ObjectId> result = await db.Users.UpsertAsync(new User { Id = aliceId, Name = "Alice Updated" });
+bool wasInserted = result.Inserted; // false: aliceId already existed, so this replaced it
+
 // Commit all changes at once
 await db.SaveChangesAsync();
 
@@ -813,6 +818,26 @@ int u = await engine.UpdateBulkAsync("orders", [(id1, d1), (id2, d2)], ct);
 int d = await engine.DeleteBulkAsync("orders", [id1, id2], ct);
 ```
 
+### Upsert
+
+Inserts the document if its `_id` is unset/absent or not present in the collection, otherwise
+replaces the existing document with that id. Unlike a naive "find, then insert-or-update"
+wrapper, this is resolved with a **single primary-index lookup per document** inside one
+transaction — no double lookups, no window for another writer to race the check.
+
+```csharp
+// New document (no _id, or an _id not yet in the collection) → inserted
+UpsertResult<BsonId> inserted = await orders.UpsertAsync(newDoc, ct);
+Console.WriteLine(inserted.Inserted); // true
+
+// Existing document → replaced in place, no "Duplicate key" exception
+UpsertResult<BsonId> replaced = await orders.UpsertAsync(existingDoc, ct);
+Console.WriteLine(replaced.Inserted); // false
+
+// Bulk (single transaction) — each document resolved independently, in order
+List<UpsertResult<BsonId>> results = await orders.UpsertBulkAsync([doc1, doc2, doc3], ct);
+```
+
 ### Index Management
 
 ```csharp

src/BLite.Core/Collections/DocumentCollection.cs

@@ -2628,14 +2628,25 @@ private async Task<bool> UpdateCore(T entity, ITransaction transaction)
 
     private async Task<bool> UpdateDataCore(TId id, T entity, byte[] docData, ITransaction transaction, int docLength = -1)
     {
-        if (docLength >= 0 && docLength < docData.Length)
-            docData = docData[..docLength]; // trim to actual serialized size
         var key = _mapper.ToIndexKey(id);
-        var bytesWritten = docData.Length;
-
         if (!_primaryIndex.TryFind(key, out var oldLocation, transaction.TransactionId))
             return false;
 
+        return await UpdateAtLocationCore(id, entity, docData, key, oldLocation, transaction, docLength);
+    }
+
+    /// <summary>
+    /// Replaces the document at an already-resolved primary-index location.
+    /// Callers that already hold <paramref name="oldLocation"/> from their own
+    /// <c>_primaryIndex.TryFind</c> (e.g. <see cref="UpsertCore"/>) use this to avoid a
+    /// second lookup for the same key.
+    /// </summary>
+    private async Task<bool> UpdateAtLocationCore(TId id, T entity, byte[] docData, IndexKey key, DocumentLocation oldLocation, ITransaction transaction, int docLength = -1)
+    {
+        if (docLength >= 0 && docLength < docData.Length)
+            docData = docData[..docLength]; // trim to actual serialized size
+        var bytesWritten = docData.Length;
+
         // Retrieve old version for index updates
         var oldEntity = await FindByLocation(oldLocation, transaction);
         if (oldEntity == null) return false;
@@ -2700,6 +2711,165 @@ private async Task<bool> UpdateDataCore(TId id, T entity, byte[] docData, ITrans
         }
     }
 
+    /// <summary>
+    /// Inserts <paramref name="entity"/> if its id is unset or not present in the collection,
+    /// otherwise replaces the existing document with the same id. Resolved with a single
+    /// primary-index lookup per document — not a separate Find followed by Insert/Update.
+    /// </summary>
+    public ValueTask<UpsertResult<TId>> UpsertAsync(T entity, CancellationToken ct = default)
+        => UpsertAsync(entity, null, ct);
+
+    /// <inheritdoc cref="UpsertAsync(T, CancellationToken)"/>
+    public async ValueTask<UpsertResult<TId>> UpsertAsync(T entity, ITransaction? transaction, CancellationToken ct = default)
+    {
+        if (entity == null) throw new ArgumentNullException(nameof(entity));
+
+        var sw = _storage.MetricsDispatcher != null ? ValueStopwatch.StartNew() : default;
+        bool success = false;
+        bool inserted = false;
+        bool autoCommit = transaction == null;
+
+        if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct))
+            throw new TimeoutException("Timed out acquiring collection lock (Upsert).");
+
+        transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted);
+        try
+        {
+            try
+            {
+                var result = await UpsertCore(entity, transaction);
+                inserted = result.Inserted;
+                if (autoCommit)
+                {
+                    await transaction.CommitAsync(ct);
+                    // ── OnInsert retention trigger — only fires when this call actually inserted ──
+                    if (inserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0)
+                        await ApplyRetentionPolicyCoreAsync(ct);
+                }
+                else if (inserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0
+                         && transaction is Transaction concreteTx)
+                {
+                    concreteTx.OnCommit += () => _ = RunScheduledRetentionAsync();
+                }
+                success = true;
+                return result;
+            }
+            catch
+            {
+                await transaction.RollbackAsync();
+                throw;
+            }
+        }
+        finally
+        {
+            _collectionLock.Release();
+            if (sw.IsActive)
+                _storage.MetricsDispatcher?.Publish(new MetricEvent
+                {
+                    Timestamp      = sw.StartTimestamp,
+                    Type           = inserted ? MetricEventType.CollectionInsert : MetricEventType.CollectionUpdate,
+                    ElapsedMicros  = sw.GetElapsedMicros(),
+                    CollectionName = _collectionName,
+                    Success        = success,
+                });
+        }
+    }
+
+    /// <summary>
+    /// Upserts multiple documents in a single transaction. Each entity is resolved with its
+    /// own single primary-index lookup, in list order.
+    /// </summary>
+    public ValueTask<List<UpsertResult<TId>>> UpsertBulkAsync(IEnumerable<T> entities, CancellationToken ct = default)
+        => UpsertBulkAsync(entities, null, ct);
+
+    /// <inheritdoc cref="UpsertBulkAsync(IEnumerable{T}, CancellationToken)"/>
+    public async ValueTask<List<UpsertResult<TId>>> UpsertBulkAsync(IEnumerable<T> entities, ITransaction? transaction, CancellationToken ct = default)
+    {
+        if (entities == null) throw new ArgumentNullException(nameof(entities));
+
+        var entityList = entities.ToList();
+        var results = new List<UpsertResult<TId>>(entityList.Count);
+        bool autoCommit = transaction == null;
+
+        if (!await _collectionLock.WaitAsync(WriteLockTimeoutMs, ct))
+            throw new TimeoutException("Timed out acquiring collection lock (UpsertBulk).");
+
+        transaction ??= _storage.BeginTransaction(IsolationLevel.ReadCommitted);
+        try
+        {
+            try
+            {
+                bool anyInserted = false;
+                foreach (var entity in entityList)
+                {
+                    var result = await UpsertCore(entity, transaction);
+                    anyInserted |= result.Inserted;
+                    results.Add(result);
+                }
+
+                if (autoCommit)
+                {
+                    await transaction.CommitAsync(ct);
+                    if (anyInserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0)
+                        await ApplyRetentionPolicyCoreAsync(ct);
+                }
+                else if (anyInserted && _retentionPolicy != null && (_retentionPolicy.Triggers & RetentionTrigger.OnInsert) != 0
+                         && transaction is Transaction concreteTx)
+                {
+                    concreteTx.OnCommit += () => _ = RunScheduledRetentionAsync();
+                }
+                return results;
+            }
+            catch
+            {
+                await transaction.RollbackAsync();
+                throw;
+            }
+        }
+        finally
+        {
+            _collectionLock.Release();
+        }
+    }
+
+    /// <summary>
+    /// Resolves the entity's id, then commits it with a single primary-index lookup:
+    /// found → in-place/relocated replace via <see cref="UpdateAtLocationCore"/>;
+    /// not found → insert via <see cref="InsertDataCore"/>. When the id is unset
+    /// (default), the lookup is skipped entirely since no existing document could match it.
+    /// </summary>
+    private async Task<UpsertResult<TId>> UpsertCore(T entity, ITransaction transaction)
+    {
+        var id = _mapper.GetId(entity);
+        if (EqualityComparer<TId>.Default.Equals(id, default!))
+        {
+            var newId = await InsertCore(entity, transaction);
+            return new UpsertResult<TId>(newId, Inserted: true);
+        }
+
+        var key = _mapper.ToIndexKey(id);
+        var length = SerializeWithRetry(entity, out var buffer);
+        try
+        {
+            if (_primaryIndex.TryFind(key, out var oldLocation, transaction.TransactionId))
+            {
+                if (!await UpdateAtLocationCore(id, entity, buffer, key, oldLocation, transaction, length))
+                    throw new InvalidOperationException(
+                        $"Upsert failed for id '{id}': the primary index resolved a location whose document could not be read (stale or corrupted index entry).");
+                return new UpsertResult<TId>(id, Inserted: false);
+            }
+            else
+            {
+                await InsertDataCore(id, entity, buffer, transaction, length);
+                return new UpsertResult<TId>(id, Inserted: true);
+            }
+        }
+        finally
+        {
+            ArrayPool<byte>.Shared.Return(buffer);
+        }
+    }
+
     /// <summary>
     /// Asynchronously deletes a document by its primary key.
     /// </summary>

src/BLite.Core/Collections/IDocumentCollection.cs

@@ -77,6 +77,17 @@ public interface IDocumentCollection<TId, T> where T : class
     ValueTask<int> UpdateBulkAsync(IEnumerable<T> entities, CancellationToken ct = default);
     ValueTask<int> UpdateBulkAsync(IEnumerable<T> entities, ITransaction? transaction, CancellationToken ct = default);
 
+    // ── Upsert ────────────────────────────────────────────────────────────────
+    // Inserts the document if its id is unset or absent from the collection, otherwise
+    // replaces the existing document. Resolves this with a single primary-index lookup
+    // per document — not a Find followed by a separate Insert/Update call.
+
+    ValueTask<UpsertResult<TId>> UpsertAsync(T entity, CancellationToken ct = default);
+    ValueTask<UpsertResult<TId>> UpsertAsync(T entity, ITransaction? transaction, CancellationToken ct = default);
+
+    ValueTask<List<UpsertResult<TId>>> UpsertBulkAsync(IEnumerable<T> entities, CancellationToken ct = default);
+    ValueTask<List<UpsertResult<TId>>> UpsertBulkAsync(IEnumerable<T> entities, ITransaction? transaction, CancellationToken ct = default);
+
     // ── Delete ────────────────────────────────────────────────────────────────
 
     ValueTask<bool> DeleteAsync(TId id, CancellationToken ct = default);

src/BLite.Core/Collections/UpsertResult.cs

@@ -0,0 +1,7 @@
+namespace BLite.Core.Collections;
+
+/// <summary>
+/// Outcome of an <c>UpsertAsync</c> call: the resolved primary key, and whether the
+/// document was newly inserted (<c>true</c>) or an existing document was replaced (<c>false</c>).
+/// </summary>
+public readonly record struct UpsertResult<TId>(TId Id, bool Inserted);