[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,111 @@
+// ------------------------------------------------------------
+// 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).ConfigureAwait(false);
+
+                DistributedTransactionServerRequest serverRequest = await DistributedTransactionServerRequest.CreateAsync(
+                    this.operations,
+                    this.clientContext.SerializerCore,
+                    cancellationToken).ConfigureAwait(false);
+
+                return await this.ExecuteCommitAsync(serverRequest, cancellationToken).ConfigureAwait(false);
+            }
+            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 = Tracing.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: ResourceType.Document,
+                        operationType: OperationType.Batch,
+                        requestOptions: null,
+                        cosmosContainerCore: null,
+                        partitionKey: null,
+                        itemId: null,
+                        streamPayload: bodyStream,
+                        requestEnricher: requestMessage => this.EnrichRequestMessage(requestMessage, transactionRequest),
+                        trace: trace,
+                        cancellationToken: cancellationToken).ConfigureAwait(false);
+
+                    cancellationToken.ThrowIfCancellationRequested();
+
+                    return await DistributedTransactionResponse.FromResponseMessageAsync(
+                        responseMessage,
+                        serverRequest,
+                        this.clientContext.SerializerCore,
+                        transactionRequest.IdempotencyToken,
+                        trace,
+                        cancellationToken).ConfigureAwait(false);
+                }
+            }
+        }
+
+        private void EnrichRequestMessage(RequestMessage requestMessage, DistributedTransactionRequest transactionRequest)
+        {
+            // Set DTC-specific headers
+            requestMessage.Headers.Add("x-ms-dtc-operation-id", transactionRequest.IdempotencyToken.ToString());
+        }
+
+        private Task AbortTransactionAsync(CancellationToken cancellationToken)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+
+            // TODO: Implement abort logic to clean up any partial state
+            // This may involve sending an abort request to the coordinator
+            DefaultTrace.TraceWarning("AbortTransactionAsync called but not yet implemented");
+            return Task.CompletedTask;
+        }
+    }
+}

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,22 @@ namespace Microsoft.Azure.Cosmos
 {
     using System;
     using System.IO;
-
-    //using Microsoft.Azure.Documents;
+    using System.Threading;
+    using System.Threading.Tasks;
+    using Microsoft.Azure.Cosmos.Serialization.HybridRow;
+    using Microsoft.Azure.Cosmos.Serialization.HybridRow.IO;
+    using Microsoft.Azure.Cosmos.Serialization.HybridRow.Layouts;
+    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 +42,148 @@ 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 int GetApproximateSerializedLength()
+        {
+            int length = 0;
+
+            if (this.PartitionKeyJson != null)
+            {
+                length += this.PartitionKeyJson.Length;
+            }
+
+            if (this.Id != null)
+            {
+                length += this.Id.Length;
+            }
+
+            if (this.CollectionResourceId != null)
+            {
+                length += this.CollectionResourceId.Length;
+            }
+
+            if (this.DatabaseResourceId != null)
+            {
+                length += this.DatabaseResourceId.Length;
+            }
+
+            if (this.SessionToken != null)
+            {
+                length += this.SessionToken.Length;
+            }
+
+            length += this.body.Length;
+
+            return length;
+        }
+
+        internal static Result WriteOperation(ref RowWriter writer, TypeArgument typeArg, DistributedTransactionOperation operation)
+        {
+            Result r = writer.WriteInt32("index", operation.OperationIndex);
+            if (r != Result.Success)
+            {
+                return r;
+            }
+
+            if (operation.CollectionResourceId != null)
+            {
+                r = writer.WriteString("collectionResourceId", operation.CollectionResourceId);
+                if (r != Result.Success)
+                {
+                    return r;
+                }
+            }
+
+            if (operation.DatabaseResourceId != null)
+            {
+                r = writer.WriteString("databaseResourceId", operation.DatabaseResourceId);
+                if (r != Result.Success)
+                {
+                    return r;
+                }
+            }
+
+            if (operation.PartitionKeyJson != null)
+            {
+                r = writer.WriteString("partitionKey", operation.PartitionKeyJson);
+                if (r != Result.Success)
+                {
+                    return r;
+                }
+            }
+
+            r = writer.WriteInt32("operation", (int)operation.OperationType);
+            if (r != Result.Success)
+            {
+                return r;
+            }
+
+            if (!operation.ResourceBody.IsEmpty)
+            {
+                r = writer.WriteBinary("resourceBody", operation.ResourceBody.Span);
+                if (r != Result.Success)
+                {
+                    return r;
+                }
+            }
+
+            if (operation.SessionToken != null)
+            {
+                r = writer.WriteString("sessionToken", operation.SessionToken);
+                if (r != Result.Success)
+                {
+                    return r;
+                }
+            }
+
+            if (operation.ETag != null)
+            {
+                r = writer.WriteString("etag", operation.ETag);
+                if (r != Result.Success)
+                {
+                    return r;
+                }
+            }
+
+            r = writer.WriteInt32("resourceType", (int)ResourceType.Document);
+            if (r != Result.Success)
+            {
+                return r;
+            }
+
+            return Result.Success;
+        }
     }
 
     internal class DistributedTransactionOperation<T> : DistributedTransactionOperation
@@ -69,6 +212,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;
+        }
     }
 }