Merge pull request #117 from CityStructure/cosmosdb-datawritemode-delete

introduce DataWriteMode.Delete to undo inserts to the wrong container

Author: markjbrown

Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/CosmosDataSinkExtension.cs

@@ -162,7 +162,7 @@ private static Task<ItemResult> AddItemAsync(Container container, ExpandoObject
                         return t.Result;
                     }
 
-                    return new ItemResult(null, HttpStatusCode.InternalServerError);
+                    return new ItemResult(null, mode, HttpStatusCode.InternalServerError);
                 }, cancellationToken);
             return task;
         }
@@ -190,14 +190,24 @@ private static async Task<ItemResult> PopulateItem(Container container, ExpandoO
                     var upsertResponse = await container.UpsertItemAsync(item, cancellationToken: cancellationToken);
                     statusCode = upsertResponse.StatusCode;
                     break;
+                case DataWriteMode.DeleteStream:
+                    ArgumentNullException.ThrowIfNull(partitionKeyPath, nameof(partitionKeyPath));
+                    var deleteMessage = await container.DeleteItemStreamAsync(itemId, new PartitionKey(GetPropertyValue(item, partitionKeyPath.TrimStart('/'))), cancellationToken: cancellationToken);
+                    statusCode = deleteMessage.StatusCode;
+                    break;
+                case DataWriteMode.Delete:
+                    ArgumentNullException.ThrowIfNull(partitionKeyPath, nameof(partitionKeyPath));
+                    var deleteResponse = await container.DeleteItemAsync<ExpandoObject>(itemId, new PartitionKey(GetPropertyValue(item, partitionKeyPath.TrimStart('/'))), cancellationToken: cancellationToken);
+                    statusCode = deleteResponse.StatusCode;
+                    break;
             }
 
             if (statusCode == null)
             {
                 throw new ArgumentOutOfRangeException(nameof(mode), $"Invalid data write mode specified: {mode}");
             }
 
-            return new ItemResult(itemId, statusCode.Value);
+            return new ItemResult(itemId, mode, statusCode.Value);
         }
 
         private static MemoryStream CreateItemStream(ExpandoObject item)
@@ -216,9 +226,10 @@ public IEnumerable<IDataExtensionSettings> GetSettings()
             yield return new CosmosSinkSettings();
         }
 
-        public record ItemResult(string? Id, HttpStatusCode StatusCode)
+        public record ItemResult(string? Id, DataWriteMode DataWriteMode, HttpStatusCode StatusCode)
         {
-            public bool IsSuccess => StatusCode is HttpStatusCode.OK or HttpStatusCode.Created;
+            public bool IsSuccess => StatusCode is HttpStatusCode.OK or HttpStatusCode.Created ||
+                (StatusCode is HttpStatusCode.NoContent or HttpStatusCode.NotFound && DataWriteMode is DataWriteMode.Delete or DataWriteMode.DeleteStream);
             public int ItemCount => IsSuccess ? 1 : 0;
         }
     }

Extensions/Cosmos/Cosmos.DataTransfer.CosmosExtension/DataWriteMode.cs

@@ -6,5 +6,7 @@ public enum DataWriteMode
         Insert,
         UpsertStream,
         Upsert,
+        DeleteStream,
+        Delete
     }
 }