Transactional Batch Operations

type TransactionalBatch struct {
    // Has unexported fields.
}

func (c *ContainerClient) NewTransactionalBatch(partitionKey PartitionKey) TransactionalBatch

func (c *ContainerClient) ExecuteTransactionalBatch(
    ctx context.Context,
    b TransactionalBatch,
    o *TransactionalBatchOptions) (TransactionalBatchResponse, error)

func (b *TransactionalBatch) CreateItem(item []byte, o *TransactionalBatchItemOptions)

func (b *TransactionalBatch) ReadItem(itemID string, o *TransactionalBatchItemOptions)

func (b *TransactionalBatch) ReplaceItem(itemID string, item []byte, o *TransactionalBatchItemOptions)

func (b *TransactionalBatch) UpsertItem(item []byte, o *TransactionalBatchItemOptions)

func (b *TransactionalBatch) DeleteItem(itemID string, o *TransactionalBatchItemOptions)

func (b *TransactionalBatch) PatchItem(itemID string, p PatchOperations, o *TransactionalBatchItemOptions)

type TransactionalBatchOptions struct {
    // SessionToken to be used when using Session consistency on the account.
    SessionToken string
    // ConsistencyLevel overrides the account defined consistency level for this operation.
    // Consistency can only be relaxed.
    ConsistencyLevel *ConsistencyLevel
    // When EnableContentResponseOnWrite is false, the operations in the batch response will have no body, except when they are Read operations.
    // The default is false.
    EnableContentResponseOnWrite bool
}

type TransactionalBatchItemOptions struct {
    // IfMatchETag is used to ensure optimistic concurrency control.
    // https://docs.microsoft.com/azure/cosmos-db/sql/database-transactions-optimistic-concurrency#optimistic-concurrency-control
    IfMatchETag *azcore.ETag
}

type TransactionalBatchResponse struct {
    Response
    // SessionToken contains the value from the session token header to be used on session consistency.
    SessionToken string
    // OperationResults contains the individual batch operation results.
    // The order of the results is the same as the order of the operations in the batch.
    OperationResults []TransactionalBatchResult
    // Success indicates if the transaction was successfully committed.
    // If false, one of the operations in the batch failed.
    // Inspect the OperationResults, any operation with status code http.StatusFailedDependency is a dependency failure.
    // The cause of the batch failure is the first operation with status code different from http.StatusFailedDependency.
    Success bool
}

type TransactionalBatchResult struct {
    // StatusCode contains the status code of the operation.
    StatusCode int32
    // RequestCharge contains the request charge for the operation.
    RequestCharge float32
    // ResourceBody contains the body response of the operation.
    // This property is available depending on the EnableContentResponseOnWrite option.
    ResourceBody []byte
    // ETag contains the ETag of the operation.
    ETag azcore.ETag
}

func (or *TransactionalBatchResult) UnmarshalJSON(b []byte) error

import (
    "context"
    "encoding/json"
    "fmt"
    "github.com/Azure/azure-sdk-for-go/sdk/data/azcosmos"
)

container, _ := database.NewContainer("items")

pk := azcosmos.NewPartitionKeyString("electronics")
batch := container.NewTransactionalBatch(pk)

// Add operations to batch
item1 := map[string]interface{}{
    "id":       "item1",
    "category": "electronics",
    "name":     "Laptop",
    "price":    999.99,
}
marshalled1, _ := json.Marshal(item1)
batch.CreateItem(marshalled1, nil)

item2 := map[string]interface{}{
    "id":       "item2",
    "category": "electronics",
    "name":     "Mouse",
    "price":    29.99,
}
marshalled2, _ := json.Marshal(item2)
batch.CreateItem(marshalled2, nil)

batch.ReadItem("existingItem", nil)

// Execute batch
response, err := container.ExecuteTransactionalBatch(context.Background(), batch, nil)
if err != nil {
    panic(err)
}

if response.Success {
    fmt.Printf("Batch succeeded! Total RU charge: %f\n", response.RequestCharge)
    for i, result := range response.OperationResults {
        fmt.Printf("Operation %d: Status %d, RU: %f\n", 
            i, result.StatusCode, result.RequestCharge)
    }
} else {
    fmt.Printf("Batch failed!\n")
    for i, result := range response.OperationResults {
        if result.StatusCode != 424 { // 424 = FailedDependency
            fmt.Printf("Operation %d failed with status %d\n", i, result.StatusCode)
        }
    }
}

container, _ := database.NewContainer("items")

pk := azcosmos.NewPartitionKeyString("electronics")
batch := container.NewTransactionalBatch(pk)

// Create new item
newItem := map[string]interface{}{
    "id":       "newItem",
    "category": "electronics",
    "name":     "Keyboard",
    "price":    79.99,
}
marshalled, _ := json.Marshal(newItem)
batch.CreateItem(marshalled, nil)

// Update existing item
updatedItem := map[string]interface{}{
    "id":       "existingItem",
    "category": "electronics",
    "name":     "Updated Name",
    "price":    89.99,
}
marshalledUpdate, _ := json.Marshal(updatedItem)
batch.ReplaceItem("existingItem", marshalledUpdate, nil)

// Delete item
batch.DeleteItem("oldItem", nil)

// Execute
response, err := container.ExecuteTransactionalBatch(context.Background(), batch, nil)
if err != nil {
    panic(err)
}

fmt.Printf("Batch success: %v\n", response.Success)

container, _ := database.NewContainer("items")

pk := azcosmos.NewPartitionKeyString("electronics")

// Read item to get ETag
readResponse, _ := container.ReadItem(context.Background(), pk, "item1", nil)
etag := readResponse.ETag

// Create batch with ETag check
batch := container.NewTransactionalBatch(pk)

updatedItem := map[string]interface{}{
    "id":       "item1",
    "category": "electronics",
    "name":     "Updated Laptop",
    "price":    899.99,
}
marshalled, _ := json.Marshal(updatedItem)

options := &azcosmos.TransactionalBatchItemOptions{
    IfMatchETag: &etag,
}

batch.ReplaceItem("item1", marshalled, options)

// Execute - will fail if item was modified
response, err := container.ExecuteTransactionalBatch(context.Background(), batch, nil)
if !response.Success {
    fmt.Printf("Batch failed - item was modified by another process\n")
}

container, _ := database.NewContainer("items")

pk := azcosmos.NewPartitionKeyString("electronics")
batch := container.NewTransactionalBatch(pk)

// Add multiple read operations
batch.ReadItem("item1", nil)
batch.ReadItem("item2", nil)
batch.ReadItem("item3", nil)

// Execute batch
response, err := container.ExecuteTransactionalBatch(context.Background(), batch, nil)
if err != nil {
    panic(err)
}

if response.Success {
    for i, result := range response.OperationResults {
        if result.ResourceBody != nil {
            var item map[string]interface{}
            json.Unmarshal(result.ResourceBody, &item)
            fmt.Printf("Item %d: %s\n", i, item["name"])
        }
    }
}

container, _ := database.NewContainer("items")

pk := azcosmos.NewPartitionKeyString("electronics")
batch := container.NewTransactionalBatch(pk)

// Create patch operations
patchOps := azcosmos.PatchOperations{}
patchOps.AppendReplace("/price", 899.99)
patchOps.AppendAdd("/tags", []string{"sale", "featured"})

batch.PatchItem("item1", patchOps, nil)

// Execute
response, err := container.ExecuteTransactionalBatch(context.Background(), batch, nil)
if err != nil {
    panic(err)
}

fmt.Printf("Batch success: %v\n", response.Success)

container, _ := database.NewContainer("items")

pk := azcosmos.NewPartitionKeyString("electronics")
batch := container.NewTransactionalBatch(pk)

item := map[string]interface{}{
    "id":       "item1",
    "category": "electronics",
    "name":     "Laptop",
}
marshalled, _ := json.Marshal(item)
batch.CreateItem(marshalled, nil)

// Disable content response to reduce network overhead
options := &azcosmos.TransactionalBatchOptions{
    EnableContentResponseOnWrite: false,
}

response, err := container.ExecuteTransactionalBatch(context.Background(), batch, options)
if err != nil {
    panic(err)
}

// ResourceBody will be nil for all write operations (except reads)
fmt.Printf("Batch executed with minimal response\n")

container, _ := database.NewContainer("items")

pk := azcosmos.NewPartitionKeyString("electronics")
batch := container.NewTransactionalBatch(pk)

// Add operations that might fail
batch.CreateItem([]byte(`{"id":"item1","category":"electronics"}`), nil)
batch.CreateItem([]byte(`{"id":"item1","category":"electronics"}`), nil) // Duplicate ID - will fail
batch.DeleteItem("item3", nil)

response, err := container.ExecuteTransactionalBatch(context.Background(), batch, nil)
if err != nil {
    panic(err)
}

if !response.Success {
    fmt.Printf("Batch failed. Finding cause...\n")
    
    for i, result := range response.OperationResults {
        if result.StatusCode != 424 { // Not a dependency failure
            fmt.Printf("Operation %d failed with status %d\n", i, result.StatusCode)
            fmt.Printf("This is the root cause of the batch failure\n")
            break
        }
    }
    
    // All other operations will have status 424 (FailedDependency)
    for i, result := range response.OperationResults {
        if result.StatusCode == 424 {
            fmt.Printf("Operation %d was rolled back due to batch failure\n", i)
        }
    }
}

container, _ := database.NewContainer("orders")

pk := azcosmos.NewPartitionKeyString("customer123")
batch := container.NewTransactionalBatch(pk)

// Create new order
order := map[string]interface{}{
    "id":         "order789",
    "customerId": "customer123",
    "total":      150.00,
    "status":     "pending",
}
marshalledOrder, _ := json.Marshal(order)
batch.CreateItem(marshalledOrder, nil)

// Update customer record
customerUpdate := map[string]interface{}{
    "id":         "customer123",
    "customerId": "customer123",
    "lastOrder":  "order789",
    "orderCount": 5,
}
marshalledCustomer, _ := json.Marshal(customerUpdate)
batch.ReplaceItem("customer123", marshalledCustomer, nil)

// Update inventory using patch
inventoryPatch := azcosmos.PatchOperations{}
inventoryPatch.AppendIncrement("/stock", -1)
inventoryPatch.AppendSet("/lastSold", "2024-01-15")
batch.PatchItem("inventory123", inventoryPatch, nil)

// Execute atomically
response, err := container.ExecuteTransactionalBatch(context.Background(), batch, nil)
if err != nil {
    panic(err)
}

if response.Success {
    fmt.Printf("Order created, customer updated, inventory decremented - all in one transaction\n")
    fmt.Printf("Total RU charge: %f\n", response.RequestCharge)
} else {
    fmt.Printf("Transaction failed - all changes rolled back\n")
}

// Important limitations:
// 1. All operations must target the same partition key
// 2. Maximum 100 operations per batch
// 3. Maximum 2MB total request size
// 4. Operations execute in order
// 5. All succeed or all fail (atomicity)

container, _ := database.NewContainer("items")

pk := azcosmos.NewPartitionKeyString("electronics")
batch := container.NewTransactionalBatch(pk)

// Add up to 100 operations
for i := 0; i < 100; i++ {
    item := map[string]interface{}{
        "id":       fmt.Sprintf("item%d", i),
        "category": "electronics",
        "index":    i,
    }
    marshalled, _ := json.Marshal(item)
    batch.CreateItem(marshalled, nil)
}

response, err := container.ExecuteTransactionalBatch(context.Background(), batch, nil)
if err != nil {
    panic(err)
}

fmt.Printf("Batch of 100 operations: %v\n", response.Success)