[Internal] DTS: Adds logic to build request and parse response

Microsoft.Azure.Cosmos/src/CosmosClient.cs

@@ -1173,7 +1173,7 @@ public virtual FeedIterator GetDatabaseQueryStreamIterator(
 #endif
         virtual DistributedWriteTransaction CreateDistributedWriteTransaction()
         {
-            return new DistributedWriteTransactionCore();
+            return new DistributedWriteTransactionCore(this.ClientContext);
         }
 
         /// <summary>

Microsoft.Azure.Cosmos/src/DistributedTransaction/DistributedTransactionCommitter.cs

@@ -0,0 +1,106 @@
+// ------------------------------------------------------------
+// Copyright (c) Microsoft Corporation.  All rights reserved.
+// ------------------------------------------------------------
+
+namespace Microsoft.Azure.Cosmos
+{
+    using System;
+    using System.Collections.Generic;
+    using System.IO;
+    using System.Threading;
+    using System.Threading.Tasks;
+    using Microsoft.Azure.Cosmos.Core.Trace;
+    using Microsoft.Azure.Cosmos.Tracing;
+    using Microsoft.Azure.Documents;
+
+    internal class DistributedTransactionCommitter
+    {
+        private readonly IReadOnlyList<DistributedTransactionOperation> operations;
+        private readonly CosmosClientContext clientContext;
+
+        public DistributedTransactionCommitter(
+            IReadOnlyList<DistributedTransactionOperation> operations,
+            CosmosClientContext clientContext)
+        {
+            this.operations = operations ?? throw new ArgumentNullException(nameof(operations));
+            this.clientContext = clientContext ?? throw new ArgumentNullException(nameof(clientContext));
+        }
+
+        public async Task<DistributedTransactionResponse> CommitTransactionAsync(CancellationToken cancellationToken)
+        {
+            try
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                await DistributedTransactionCommitterUtils.ResolveCollectionRidsAsync(
+                    this.operations,
+                    this.clientContext,
+                    cancellationToken);
+
+                DistributedTransactionServerRequest serverRequest = await DistributedTransactionServerRequest.CreateAsync(
+                    this.operations,
+                    this.clientContext.SerializerCore,
+                    cancellationToken);
+
+                return await this.ExecuteCommitAsync(serverRequest, cancellationToken);
+            }
+            catch (Exception ex)
+            {
+                DefaultTrace.TraceError($"Distributed transaction failed: {ex.Message}");
+                await this.AbortTransactionAsync(cancellationToken);
+                throw;
+            }
+        }
+
+        private async Task<DistributedTransactionResponse> ExecuteCommitAsync(
+            DistributedTransactionServerRequest serverRequest,
+            CancellationToken cancellationToken)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+            using (ITrace trace = Trace.GetRootTrace("Execute Distributed Transaction Commit", TraceComponent.Batch, Tracing.TraceLevel.Info))
+            {
+                DistributedTransactionRequest transactionRequest = new DistributedTransactionRequest(
+                    serverRequest.Operations,
+                    OperationType.Batch,
+                    ResourceType.Document);
+
+                using (MemoryStream bodyStream = serverRequest.TransferBodyStream())
+                {
+                    ResponseMessage responseMessage = await this.clientContext.ProcessResourceOperationStreamAsync(
+                        resourceUri: null,
+                        resourceType: transactionRequest.ResourceType,
+                        operationType: transactionRequest.OperationType,
+                        requestOptions: null,
+                        cosmosContainerCore: null,
+                        partitionKey: null,
+                        itemId: null,
+                        streamPayload: bodyStream,
+                        requestEnricher: requestMessage => this.EnrichRequestMessage(requestMessage, transactionRequest),
+                        trace: trace,
+                        cancellationToken: cancellationToken);
+
+                    cancellationToken.ThrowIfCancellationRequested();
+
+                    return await DistributedTransactionResponse.FromResponseMessageAsync(
+                        responseMessage,
+                        serverRequest,
+                        this.clientContext.SerializerCore,
+                        transactionRequest.IdempotencyToken,
+                        trace,
+                        cancellationToken);
+                }
+            }
+        }
+
+        private void EnrichRequestMessage(RequestMessage requestMessage, DistributedTransactionRequest transactionRequest)
+        {
+            // Set DTC-specific headers
+            requestMessage.Headers.Add("x-ms-dtc-operation-id", transactionRequest.IdempotencyToken.ToString());
+            requestMessage.UseGatewayMode = true;
+        }
+
+        private Task AbortTransactionAsync(CancellationToken cancellationToken)
+        {
+            throw new NotImplementedException();
+        }
+    }
+}

Microsoft.Azure.Cosmos/src/DistributedTransaction/DistributedTransactionCommitterUtils.cs

@@ -0,0 +1,57 @@
+// ------------------------------------------------------------
+// Copyright (c) Microsoft Corporation.  All rights reserved.
+// ------------------------------------------------------------
+
+namespace Microsoft.Azure.Cosmos
+{
+    using System;
+    using System.Collections.Generic;
+    using System.Linq;
+    using System.Threading;
+    using System.Threading.Tasks;
+    using Microsoft.Azure.Cosmos.Common;
+    using Microsoft.Azure.Cosmos.Core.Trace;
+    using Microsoft.Azure.Cosmos.Tracing;
+    using Microsoft.Azure.Documents;
+
+    internal class DistributedTransactionCommitterUtils
+    {
+        public static async Task ResolveCollectionRidsAsync(
+            IReadOnlyList<DistributedTransactionOperation> operations,
+            CosmosClientContext clientContext,
+            CancellationToken cancellationToken)
+        {
+            CollectionCache collectionCache = await clientContext.DocumentClient.GetCollectionCacheAsync(NoOpTrace.Singleton);
+            IEnumerable<Task> ridResolutionTasks = operations
+               .GroupBy(op => $"/dbs/{op.Database}/colls/{op.Container}")
+               .Select(async group =>
+               {
+                   string collectionPath = group.Key;
+                   try
+                   {
+                       ContainerProperties containerProperties = await clientContext.GetCachedContainerPropertiesAsync(
+                           collectionPath,
+                           NoOpTrace.Singleton,
+                           cancellationToken);
+
+                       string containerResourceId = containerProperties.ResourceId;
+                       ResourceId resourceId = ResourceId.Parse(containerResourceId);
+                       string databaseResourceId = resourceId.DatabaseId.ToString();
+
+                       foreach (DistributedTransactionOperation operation in group)
+                       {
+                           operation.CollectionResourceId = containerResourceId;
+                           operation.DatabaseResourceId = databaseResourceId;
+                       }
+                   }
+                   catch (Exception ex)
+                   {
+                       DefaultTrace.TraceError($"Failed to resolve RID for {collectionPath}: {ex.Message}");
+                       throw;
+                   }
+               });
+            await Task.WhenAll(ridResolutionTasks);
+        }
+
+    }
+}

Microsoft.Azure.Cosmos/src/DistributedTransaction/DistributedTransactionOperation.cs

@@ -6,16 +6,19 @@ namespace Microsoft.Azure.Cosmos
 {
     using System;
     using System.IO;
-
-    //using Microsoft.Azure.Documents;
+    using System.Threading;
+    using System.Threading.Tasks;
+    using Microsoft.Azure.Documents;
 
     /// <summary>
-    /// Represents an operation on a document whichwill be executed as a part of a distributed transaction.
+    /// Represents an operation on a document which will be executed as a part of a distributed transaction.
     /// </summary>
     internal class DistributedTransactionOperation
     {
+        protected Memory<byte> body;
+
         public DistributedTransactionOperation(
-            Documents.OperationType operationType,
+            OperationType operationType,
             int operationIndex,
             string database,
             string container,
@@ -36,11 +39,37 @@ public DistributedTransactionOperation(
 
         public string Container { get; internal set; }
 
-        public Documents.OperationType OperationType { get; internal set; }
+        public OperationType OperationType { get; internal set; }
 
-        public int OperationIndex { get; internal set; } 
+        public int OperationIndex { get; internal set; }
 
         public string Id { get; internal set; }
+
+        public string CollectionResourceId { get; internal set; }
+
+        public string DatabaseResourceId { get; internal set; }
+
+        internal string PartitionKeyJson { get; set; }
+
+        internal string SessionToken { get; set; }
+
+        internal string ETag { get; set; }
+
+        internal Stream ResourceStream { get; set; }
+
+        internal Memory<byte> ResourceBody
+        {
+            get => this.body;
+            set => this.body = value;
+        }
+
+        internal virtual async Task MaterializeResourceAsync(CosmosSerializerCore serializerCore, CancellationToken cancellationToken)
+        {
+            if (this.body.IsEmpty && this.ResourceStream != null)
+            {
+                this.body = await BatchExecUtils.StreamToMemoryAsync(this.ResourceStream, cancellationToken);
+            }
+        }
     }
 
     internal class DistributedTransactionOperation<T> : DistributedTransactionOperation
@@ -69,6 +98,18 @@ public DistributedTransactionOperation(
         {
             this.Resource = resource;
         }
+
         public T Resource { get; internal set; }
+
+        internal override Task MaterializeResourceAsync(CosmosSerializerCore serializerCore, CancellationToken cancellationToken)
+        {
+            if (this.body.IsEmpty && this.Resource != null)
+            {
+                this.ResourceStream = serializerCore.ToStream(this.Resource);
+                return base.MaterializeResourceAsync(serializerCore, cancellationToken);
+            }
+
+            return Task.CompletedTask;
+        }
     }
 }

Microsoft.Azure.Cosmos/src/DistributedTransaction/DistributedTransactionOperationResult.cs

@@ -7,6 +7,7 @@ namespace Microsoft.Azure.Cosmos
     using System;
     using System.IO;
     using System.Net;
+    using System.Text.Json;
     using Microsoft.Azure.Cosmos.Tracing;
     using Microsoft.Azure.Documents;
 
@@ -27,11 +28,14 @@ internal DistributedTransactionOperationResult(HttpStatusCode statusCode)
 
         internal DistributedTransactionOperationResult(DistributedTransactionOperationResult other)
         {
+            this.Index = other.Index;
             this.StatusCode = other.StatusCode;
             this.SubStatusCode = other.SubStatusCode;
             this.ETag = other.ETag;
             this.ResourceStream = other.ResourceStream;
             this.SessionToken = other.SessionToken;
+            this.RequestCharge = other.RequestCharge;
+            this.ActivityId = other.ActivityId;
             this.Trace = other.Trace;
         }
 
@@ -43,6 +47,11 @@ protected DistributedTransactionOperationResult()
         {
         }
 
+        /// <summary>
+        /// Gets the index of this operation within the distributed transaction.
+        /// </summary>
+        public virtual int Index { get; internal set; }
+
         /// <summary>
         /// Gets the HTTP status code returned by the operation.
         /// </summary>
@@ -51,7 +60,7 @@ protected DistributedTransactionOperationResult()
         /// <summary>
         /// Gets a value indicating whether the HTTP status code returned by the operation indicates success.
         /// </summary>
-        public virtual bool IsSuccessStatusCode => ((int)this.StatusCode >= 200) && ((int)this.StatusCode <= 299);
+        public virtual bool IsSuccessStatusCode => (int)this.StatusCode >= 200 && (int)this.StatusCode <= 299;
 
         /// <summary>
         /// Gets the entity tag (ETag) associated with the operation result.
@@ -70,8 +79,70 @@ protected DistributedTransactionOperationResult()
         /// </summary>
         public virtual Stream ResourceStream { get; internal set; }
 
+        /// <summary>
+        /// Request charge in request units for the operation.
+        /// </summary>
+        internal virtual double RequestCharge { get; set; }
+
         internal virtual SubStatusCodes SubStatusCode { get; set; }
 
+        /// <summary>
+        /// ActivityId related to the operation.
+        /// </summary>
+        internal virtual string ActivityId { get; set; }
+
         internal ITrace Trace { get; set; }
+
+        /// <summary>
+        /// Creates a <see cref="DistributedTransactionOperationResult"/> from a JSON element.
+        /// </summary>
+        /// <param name="json">The JSON element containing the operation result.</param>
+        /// <returns>The deserialized operation result.</returns>
+        internal static DistributedTransactionOperationResult FromJson(JsonElement json)
+        {
+            DistributedTransactionOperationResult operationResult = new DistributedTransactionOperationResult();
+
+            if (json.TryGetProperty("index", out JsonElement indexElement))
+            {
+                operationResult.Index = indexElement.GetInt32();
+            }
+
+            if (json.TryGetProperty("statusCode", out JsonElement statusCodeElement))
+            {
+                operationResult.StatusCode = (HttpStatusCode)statusCodeElement.GetInt32();
+            }
+
+            if (json.TryGetProperty("substatuscode", out JsonElement subStatusCodeElement))
+            {
+                operationResult.SubStatusCode = (SubStatusCodes)subStatusCodeElement.GetInt32();
+            }
+
+            if (json.TryGetProperty("etag", out JsonElement eTagElement) && eTagElement.ValueKind != JsonValueKind.Null)
+            {
+                operationResult.ETag = eTagElement.GetString();
+            }
+
+            if (json.TryGetProperty("resourcebody", out JsonElement resourceBodyElement) && resourceBodyElement.ValueKind != JsonValueKind.Null)
+            {
+                string resourceBodyBase64 = resourceBodyElement.GetString();
+                if (!string.IsNullOrEmpty(resourceBodyBase64))
+                {
+                    byte[] resourceBody = Convert.FromBase64String(resourceBodyBase64);
+                    operationResult.ResourceStream = new MemoryStream(resourceBody, 0, resourceBody.Length, writable: false, publiclyVisible: true);
+                }
+            }
+
+            if (json.TryGetProperty("sessionToken", out JsonElement sessionTokenElement) && sessionTokenElement.ValueKind != JsonValueKind.Null)
+            {
+                operationResult.SessionToken = sessionTokenElement.GetString();
+            }
+
+            if (json.TryGetProperty("requestCharge", out JsonElement requestChargeElement))
+            {
+                operationResult.RequestCharge = Math.Round(requestChargeElement.GetDouble(), 2);
+            }
+
+            return operationResult;
+        }
     }
 }