diff --git a/README.md b/README.md index 6fc74e9..3929ca8 100644 --- a/README.md +++ b/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 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 inserted = await orders.UpsertAsync(newDoc, ct); +Console.WriteLine(inserted.Inserted); // true + +// Existing document → replaced in place, no "Duplicate key" exception +UpsertResult replaced = await orders.UpsertAsync(existingDoc, ct); +Console.WriteLine(replaced.Inserted); // false + +// Bulk (single transaction) — each document resolved independently, in order +List> results = await orders.UpsertBulkAsync([doc1, doc2, doc3], ct); +``` + ### Index Management ```csharp diff --git a/src/BLite.Core/Collections/DocumentCollection.cs b/src/BLite.Core/Collections/DocumentCollection.cs index 7d0f6c3..7bb3e16 100644 --- a/src/BLite.Core/Collections/DocumentCollection.cs +++ b/src/BLite.Core/Collections/DocumentCollection.cs @@ -2628,14 +2628,25 @@ private async Task UpdateCore(T entity, ITransaction transaction) private async Task 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); + } + + /// + /// Replaces the document at an already-resolved primary-index location. + /// Callers that already hold from their own + /// _primaryIndex.TryFind (e.g. ) use this to avoid a + /// second lookup for the same key. + /// + private async Task 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 UpdateDataCore(TId id, T entity, byte[] docData, ITrans } } + /// + /// Inserts 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. + /// + public ValueTask> UpsertAsync(T entity, CancellationToken ct = default) + => UpsertAsync(entity, null, ct); + + /// + public async ValueTask> 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, + }); + } + } + + /// + /// Upserts multiple documents in a single transaction. Each entity is resolved with its + /// own single primary-index lookup, in list order. + /// + public ValueTask>> UpsertBulkAsync(IEnumerable entities, CancellationToken ct = default) + => UpsertBulkAsync(entities, null, ct); + + /// + public async ValueTask>> UpsertBulkAsync(IEnumerable entities, ITransaction? transaction, CancellationToken ct = default) + { + if (entities == null) throw new ArgumentNullException(nameof(entities)); + + var entityList = entities.ToList(); + var results = new List>(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(); + } + } + + /// + /// Resolves the entity's id, then commits it with a single primary-index lookup: + /// found → in-place/relocated replace via ; + /// not found → insert via . When the id is unset + /// (default), the lookup is skipped entirely since no existing document could match it. + /// + private async Task> UpsertCore(T entity, ITransaction transaction) + { + var id = _mapper.GetId(entity); + if (EqualityComparer.Default.Equals(id, default!)) + { + var newId = await InsertCore(entity, transaction); + return new UpsertResult(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(id, Inserted: false); + } + else + { + await InsertDataCore(id, entity, buffer, transaction, length); + return new UpsertResult(id, Inserted: true); + } + } + finally + { + ArrayPool.Shared.Return(buffer); + } + } + /// /// Asynchronously deletes a document by its primary key. /// diff --git a/src/BLite.Core/Collections/IDocumentCollection.cs b/src/BLite.Core/Collections/IDocumentCollection.cs index 25a8610..ad615cf 100644 --- a/src/BLite.Core/Collections/IDocumentCollection.cs +++ b/src/BLite.Core/Collections/IDocumentCollection.cs @@ -77,6 +77,17 @@ public interface IDocumentCollection where T : class ValueTask UpdateBulkAsync(IEnumerable entities, CancellationToken ct = default); ValueTask UpdateBulkAsync(IEnumerable 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> UpsertAsync(T entity, CancellationToken ct = default); + ValueTask> UpsertAsync(T entity, ITransaction? transaction, CancellationToken ct = default); + + ValueTask>> UpsertBulkAsync(IEnumerable entities, CancellationToken ct = default); + ValueTask>> UpsertBulkAsync(IEnumerable entities, ITransaction? transaction, CancellationToken ct = default); + // ── Delete ──────────────────────────────────────────────────────────────── ValueTask DeleteAsync(TId id, CancellationToken ct = default); diff --git a/src/BLite.Core/Collections/UpsertResult.cs b/src/BLite.Core/Collections/UpsertResult.cs new file mode 100644 index 0000000..ffd7b28 --- /dev/null +++ b/src/BLite.Core/Collections/UpsertResult.cs @@ -0,0 +1,7 @@ +namespace BLite.Core.Collections; + +/// +/// Outcome of an UpsertAsync call: the resolved primary key, and whether the +/// document was newly inserted (true) or an existing document was replaced (false). +/// +public readonly record struct UpsertResult(TId Id, bool Inserted); diff --git a/src/BLite.Core/DynamicCollection.cs b/src/BLite.Core/DynamicCollection.cs index 81a07a6..c00c616 100644 --- a/src/BLite.Core/DynamicCollection.cs +++ b/src/BLite.Core/DynamicCollection.cs @@ -1016,39 +1016,116 @@ public async ValueTask UpdateAsync(BsonId id, BsonDocument newDocument, IT if (!_primaryIndex.TryFind(key, out var oldLocation, transaction.TransactionId)) return false; - var oldDoc = ReadDocumentAt(oldLocation, transaction.TransactionId); - DeleteSlot(oldLocation, transaction); + await UpdateAtLocationCore(id, key, newDocument, oldLocation, transaction); - if (!newDocument.TryGetId(out _)) - newDocument = PrependId(newDocument, id); + if (autoCommit) await transaction.CommitAsync(ct); + success = true; + return true; + } + catch + { + await transaction.RollbackAsync(); + throw; + } + finally + { + _collectionLock.Release(); + if (sw.IsActive) + _storage.MetricsDispatcher?.Publish(new Metrics.MetricEvent + { + Timestamp = sw.StartTimestamp, + Type = Metrics.MetricEventType.CollectionUpdate, + ElapsedMicros = sw.GetElapsedMicros(), + CollectionName = _collectionName, + Success = success, + }); + } + } - var docData = newDocument.RawData; - DocumentLocation newLocation = default; - if (docData.Length + SlotEntry.Size <= _maxDocumentSizeForSinglePage) - { - var pageId = FindPageWithSpace(docData.Length + SlotEntry.Size, transaction.TransactionId); - if (pageId == 0) pageId = AllocateNewDataPage(transaction); - var slotIndex = InsertIntoPage(pageId, docData, transaction); - newLocation = new DocumentLocation(pageId, slotIndex); - } - else - { - throw new InvalidOperationException("Document too large for single page. Overflow not yet supported in DynamicCollection."); - } + /// + /// Replaces the document at an already-resolved primary-index location. + /// Callers that already hold from their own + /// _primaryIndex.TryFind (e.g. ) use this to avoid a + /// second lookup for the same key. + /// + private async Task UpdateAtLocationCore(BsonId id, IndexKey key, BsonDocument newDocument, DocumentLocation oldLocation, ITransaction transaction) + { + var oldDoc = ReadDocumentAt(oldLocation, transaction.TransactionId); + DeleteSlot(oldLocation, transaction); - _primaryIndex.Delete(key, oldLocation, transaction.TransactionId); - _primaryIndex.Insert(key, newLocation, transaction.TransactionId); + if (!newDocument.TryGetId(out _)) + newDocument = PrependId(newDocument, id); - foreach (var (_, idx) in _secondaryIndexes) + var docData = newDocument.RawData; + DocumentLocation newLocation; + if (docData.Length + SlotEntry.Size <= _maxDocumentSizeForSinglePage) + { + var pageId = FindPageWithSpace(docData.Length + SlotEntry.Size, transaction.TransactionId); + if (pageId == 0) pageId = AllocateNewDataPage(transaction); + var slotIndex = InsertIntoPage(pageId, docData, transaction); + newLocation = new DocumentLocation(pageId, slotIndex); + } + else + { + throw new InvalidOperationException("Document too large for single page. Overflow not yet supported in DynamicCollection."); + } + + _primaryIndex.Delete(key, oldLocation, transaction.TransactionId); + _primaryIndex.Insert(key, newLocation, transaction.TransactionId); + + foreach (var (_, idx) in _secondaryIndexes) + { + if (oldDoc != null) IndexDelete(idx, oldDoc, oldLocation, transaction); + IndexInsert(idx, newDocument, newLocation, transaction); + } + + await NotifyCdcAsync(OperationType.Update, id, transaction, newDocument.RawData); + } + + /// + /// Inserts if it has no _id or that id is not present + /// in the collection, otherwise replaces the existing document with the same id. + /// Resolved with a single primary-index lookup — not a separate Find followed by + /// Insert/Update. + /// + public ValueTask> UpsertAsync(BsonDocument document, CancellationToken ct = default) + => UpsertAsync(document, null, ct); + + /// + public async ValueTask> UpsertAsync(BsonDocument document, ITransaction? transaction, CancellationToken ct = default) + { + if (document == null) throw new ArgumentNullException(nameof(document)); + + var sw = _storage.MetricsDispatcher != null ? Metrics.ValueStopwatch.StartNew() : default; + var auditOpts = _storage.AuditOptions; + var auditVsw = (auditOpts is not null && (auditOpts.Sink is not null || auditOpts.EnableMetrics)) + ? Metrics.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 + { + var result = await UpsertCore(document, transaction); + inserted = result.Inserted; + if (autoCommit) { - if (oldDoc != null) IndexDelete(idx, oldDoc, oldLocation, transaction); - IndexInsert(idx, newDocument, newLocation, transaction); + await transaction.CommitAsync(ct); + 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(); } - - await NotifyCdcAsync(OperationType.Update, id, transaction, newDocument.RawData); - if (autoCommit) await transaction.CommitAsync(ct); success = true; - return true; + return result; } catch { @@ -1062,11 +1139,118 @@ public async ValueTask UpdateAsync(BsonId id, BsonDocument newDocument, IT _storage.MetricsDispatcher?.Publish(new Metrics.MetricEvent { Timestamp = sw.StartTimestamp, - Type = Metrics.MetricEventType.CollectionUpdate, + Type = inserted ? Metrics.MetricEventType.CollectionInsert : Metrics.MetricEventType.CollectionUpdate, ElapsedMicros = sw.GetElapsedMicros(), CollectionName = _collectionName, Success = success, }); + + // ── AUDIT: only when this call actually inserted, matching InsertAsync ── + if (auditVsw.IsActive && success && inserted) + { + var elapsed = auditVsw.GetElapsed(); + var userId = (auditOpts!.ContextProvider ?? AmbientAuditContext.Instance).GetCurrentUserId(); + var txId = transaction?.TransactionId ?? 0UL; + var docBytes = document.RawData.Length; + + var evt = new InsertAuditEvent( + TransactionId: txId, + CollectionName: _collectionName, + DocumentSizeBytes: docBytes, + Elapsed: elapsed, + UserId: userId); + + auditOpts.Sink?.OnInsert(evt); + _storage.AuditMetrics?.RecordInsert(elapsed); + + if (auditOpts.SlowOperationThreshold is { } threshold && elapsed > threshold) + { + auditOpts.Sink?.OnSlowOperation(new SlowOperationEvent( + SlowOperationType.Insert, + CollectionName: _collectionName, + Elapsed: elapsed, + Detail: null)); + } + } + } + } + + /// + /// Upserts multiple documents in a single transaction. Each document is resolved with + /// its own single primary-index lookup, in list order. + /// + public ValueTask>> UpsertBulkAsync(IEnumerable documents, CancellationToken ct = default) + => UpsertBulkAsync(documents, null, ct); + + /// + public async ValueTask>> UpsertBulkAsync(IEnumerable documents, ITransaction? transaction, CancellationToken ct = default) + { + if (documents == null) throw new ArgumentNullException(nameof(documents)); + 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 + { + var results = new List>(); + bool anyInserted = false; + foreach (var doc in documents) + { + ct.ThrowIfCancellationRequested(); + var result = await UpsertCore(doc, 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(); + } + } + + /// + /// Resolves the document's _id, then commits it with a single primary-index + /// lookup: found → in-place/relocated replace via ; + /// not found or absent → insert via . + /// + private async Task> UpsertCore(BsonDocument document, ITransaction transaction) + { + if (!document.TryGetId(out var id) || id.IsEmpty) + { + var newId = await InsertCore(document, transaction); + return new UpsertResult(newId, Inserted: true); + } + + var key = new IndexKey(id.ToBytes()); + if (_primaryIndex.TryFind(key, out var oldLocation, transaction.TransactionId)) + { + await UpdateAtLocationCore(id, key, document, oldLocation, transaction); + return new UpsertResult(id, Inserted: false); + } + else + { + var insertedId = await InsertCore(document, transaction); + return new UpsertResult(insertedId, Inserted: true); } } diff --git a/tests/BLite.Tests/UpsertDynamicCollectionTests.cs b/tests/BLite.Tests/UpsertDynamicCollectionTests.cs new file mode 100644 index 0000000..d44baaf --- /dev/null +++ b/tests/BLite.Tests/UpsertDynamicCollectionTests.cs @@ -0,0 +1,114 @@ +using BLite.Bson; +using BLite.Core; + +namespace BLite.Tests; + +public class UpsertDynamicCollectionTests : IDisposable +{ + private readonly string _dbPath; + private BLiteEngine _engine; + + public UpsertDynamicCollectionTests() + { + _dbPath = Path.Combine(Path.GetTempPath(), $"blite_upsert_dyncol_{Guid.NewGuid():N}.db"); + _engine = new BLiteEngine(_dbPath); + } + + public void Dispose() + { + _engine.Dispose(); + if (File.Exists(_dbPath)) File.Delete(_dbPath); + var wal = Path.ChangeExtension(_dbPath, ".wal"); + if (File.Exists(wal)) File.Delete(wal); + } + + private BsonDocument MakeDoc(string name, int age) + { + var col = _engine.GetOrCreateCollection("tmp_schema"); + return col.CreateDocument(["name", "age"], b => b + .AddString("name", name) + .AddInt32("age", age)); + } + + private BsonDocument MakeDocWithId(BsonId id, string name, int age) + { + var col = _engine.GetOrCreateCollection("tmp_schema"); + return col.CreateDocument(["_id", "name", "age"], b => b + .AddId(id) + .AddString("name", name) + .AddInt32("age", age)); + } + + [Fact] + public async Task Upsert_Without_Id_Inserts_New_Document() + { + var col = _engine.GetOrCreateCollection("users"); + var doc = MakeDoc("Alice", 30); + + var result = await col.UpsertAsync(doc); + await _engine.CommitAsync(); + + Assert.True(result.Inserted); + + var found = await col.FindByIdAsync(result.Id); + Assert.NotNull(found); + found!.TryGetString("name", out var name); + Assert.Equal("Alice", name); + } + + [Fact] + public async Task Upsert_With_Unused_Id_Inserts() + { + var col = _engine.GetOrCreateCollection("users"); + var id = new BsonId(ObjectId.NewObjectId()); + var doc = MakeDocWithId(id, "Bob", 25); + + var result = await col.UpsertAsync(doc); + await _engine.CommitAsync(); + + Assert.True(result.Inserted); + Assert.Equal(id, result.Id); + Assert.Equal(1, await col.CountAsync()); + } + + [Fact] + public async Task Upsert_With_Existing_Id_Replaces_Document() + { + var col = _engine.GetOrCreateCollection("users"); + var id = await col.InsertAsync(MakeDoc("Alice", 30)); + await _engine.CommitAsync(); + + var result = await col.UpsertAsync(MakeDocWithId(id, "Alice", 31)); + await _engine.CommitAsync(); + + Assert.False(result.Inserted); + Assert.Equal(id, result.Id); + Assert.Equal(1, await col.CountAsync()); + + var found = await col.FindByIdAsync(id); + Assert.NotNull(found); + found!.TryGetInt32("age", out var age); + Assert.Equal(31, age); + } + + [Fact] + public async Task UpsertBulk_Mixes_Inserts_And_Updates() + { + var col = _engine.GetOrCreateCollection("users"); + var existingId = await col.InsertAsync(MakeDoc("Alice", 30)); + await _engine.CommitAsync(); + + var newId = new BsonId(ObjectId.NewObjectId()); + var results = await col.UpsertBulkAsync(new[] + { + MakeDocWithId(existingId, "Alice", 31), + MakeDocWithId(newId, "Charlie", 40), + }); + await _engine.CommitAsync(); + + Assert.Equal(2, results.Count); + Assert.False(results[0].Inserted); + Assert.True(results[1].Inserted); + Assert.Equal(2, await col.CountAsync()); + } +} diff --git a/tests/BLite.Tests/UpsertTests.cs b/tests/BLite.Tests/UpsertTests.cs new file mode 100644 index 0000000..9f2af01 --- /dev/null +++ b/tests/BLite.Tests/UpsertTests.cs @@ -0,0 +1,120 @@ +using BLite.Bson; +using BLite.Shared; + +namespace BLite.Tests; + +public class UpsertTests : IDisposable +{ + private readonly string _dbPath; + private readonly string _walPath; + private readonly TestDbContext _db; + + public UpsertTests() + { + _dbPath = Path.Combine(Path.GetTempPath(), $"test_upsert_{Guid.NewGuid()}.db"); + _walPath = Path.Combine(Path.GetTempPath(), $"test_upsert_{Guid.NewGuid()}.wal"); + + _db = new TestDbContext(_dbPath); + } + + [Fact] + public async Task Upsert_With_No_Id_Inserts_New_Document() + { + var user = new User { Name = "Alice", Age = 30 }; + + var result = await _db.Users.UpsertAsync(user); + await _db.SaveChangesAsync(); + + Assert.True(result.Inserted); + Assert.NotEqual(default, result.Id); + + var found = await _db.Users.FindByIdAsync(result.Id); + Assert.NotNull(found); + Assert.Equal("Alice", found.Name); + } + + [Fact] + public async Task Upsert_With_Explicit_Unused_Id_Inserts() + { + var id = 42; + var product = new Product { Id = id, Title = "Widget", Price = 9.99m }; + + var result = await _db.Products.UpsertAsync(product); + await _db.SaveChangesAsync(); + + Assert.True(result.Inserted); + Assert.Equal(id, result.Id); + + var found = await _db.Products.FindByIdAsync(id); + Assert.NotNull(found); + Assert.Equal("Widget", found.Title); + Assert.Equal(1, await _db.Products.CountAsync()); + } + + [Fact] + public async Task Upsert_With_Existing_Id_Replaces_Document() + { + var id = await _db.Products.InsertAsync(new Product { Title = "Widget", Price = 9.99m }); + await _db.SaveChangesAsync(); + + var result = await _db.Products.UpsertAsync(new Product { Id = id, Title = "Widget Pro", Price = 19.99m }); + await _db.SaveChangesAsync(); + + Assert.False(result.Inserted); + Assert.Equal(id, result.Id); + + var found = await _db.Products.FindByIdAsync(id); + Assert.NotNull(found); + Assert.Equal("Widget Pro", found.Title); + Assert.Equal(19.99m, found.Price); + Assert.Equal(1, await _db.Products.CountAsync()); + } + + [Fact] + public async Task Upsert_Does_Not_Throw_Duplicate_Key_Unlike_Insert() + { + var id = await _db.Products.InsertAsync(new Product { Title = "Widget", Price = 9.99m }); + await _db.SaveChangesAsync(); + + // A plain InsertAsync of the same id would throw "Duplicate key violation". + // UpsertAsync must instead replace it. + var result = await _db.Products.UpsertAsync(new Product { Id = id, Title = "Widget v2", Price = 12.0m }); + await _db.SaveChangesAsync(); + + Assert.False(result.Inserted); + Assert.Equal(1, await _db.Products.CountAsync()); + } + + [Fact] + public async Task UpsertBulk_Mixes_Inserts_And_Updates() + { + var existingId = await _db.Products.InsertAsync(new Product { Title = "Widget", Price = 9.99m }); + await _db.SaveChangesAsync(); + + var newId = 999; + var results = await _db.Products.UpsertBulkAsync(new[] + { + new Product { Id = existingId, Title = "Widget Pro", Price = 19.99m }, + new Product { Id = newId, Title = "Gadget", Price = 5.0m }, + }); + await _db.SaveChangesAsync(); + + Assert.Equal(2, results.Count); + Assert.False(results[0].Inserted); + Assert.True(results[1].Inserted); + + Assert.Equal(2, await _db.Products.CountAsync()); + + var updated = await _db.Products.FindByIdAsync(existingId); + Assert.Equal("Widget Pro", updated!.Title); + + var inserted = await _db.Products.FindByIdAsync(newId); + Assert.NotNull(inserted); + Assert.Equal("Gadget", inserted!.Title); + } + + public void Dispose() + { + _db?.Dispose(); + } +}