# Azure Table Persistence Using Saga IDs as Partition Keys

<!-- Version variant: astp_6; default: [/samples/azure/azure-table/saga-transactions/index.md](/samples/azure/azure-table/saga-transactions/index.md) -->


This sample demonstrates a client/server scenario using saga and outbox persistences to store records atomically by leveraging transactions. The Saga ID is used as a partition key.

## Projects

### SharedMessages

The shared message contracts used by all endpoints.

### Client

* Sends the `StartOrder` message to `Server`.
* Receives and handles the `OrderCompleted` event.

### Server projects

* Receive the `StartOrder` message and initiate an `OrderSaga`.
* `OrderSaga` requests a timeout with an instance of `CompleteOrder` with the saga data.
* Receive the `OrderShipped` message with a custom header.
* `OrderSaga` publishes an `OrderCompleted` event when the `CompleteOrder` timeout fires.

### Persistence config

Configure the endpoint to use Azure Table Persistence.

<!-- snippet: AzureTableConfig -->

```cs
var endpointConfiguration = new EndpointConfiguration("Samples.AzureTable.Transactions.Server");
endpointConfiguration.EnableOutbox();

var useStorageTable = true;
var persistence = endpointConfiguration.UsePersistence<AzureTablePersistence>();

var connection = useStorageTable ? "UseDevelopmentStorage=true" :
    "TableEndpoint=https://localhost:8081/;AccountName=AzureTableSamples;AccountKey=C2y6yDjf5/R+ob0N8A7Cgv30VRDJIWEHLM+4QDU5DE2nQ9nDuVTqobD4b8mGGyPMbIZnqyMsEcaGQy67XIw/Jw==";

var tableServiceClient = new TableServiceClient(connection);
persistence.UseTableServiceClient(tableServiceClient);

persistence.DefaultTable("Server");
```

<!-- endsnippet -->

## Using Behaviors

Most messages implement `IProvideOrderId`. Since Saga IDs are deterministically derived from saga data (the correlation property name and value), they can be used as a partition key. Then, `IProvidePartitionKeyFromSagaId` can be injected into behaviors in the logical pipeline stage.

<!-- snippet: BehaviorUsingIProvidePartitionKeyFromSagaId -->

```cs
class OrderIdAsPartitionKeyBehavior : Behavior<IIncomingLogicalMessageContext>
{
    public OrderIdAsPartitionKeyBehavior(IProvidePartitionKeyFromSagaId partitionKeyFromSagaId,
        ILogger<OrderIdAsPartitionKeyBehavior> logger)
    {
        partitionKeyFromSagaId1 = partitionKeyFromSagaId;
        this.logger = logger;
    }

    public override async Task Invoke(IIncomingLogicalMessageContext context, Func<Task> next)
    {
        var correlationProperty = SagaCorrelationProperty.None;
        if (context.Message.Instance is IProvideOrderId provideOrderId)
        {
            var partitionKeyValue = provideOrderId.OrderId;
            correlationProperty = new SagaCorrelationProperty("OrderId", partitionKeyValue);
        }

        await partitionKeyFromSagaId1.SetPartitionKey<OrderSagaData>(context, correlationProperty);

        if (context.Headers.TryGetValue(Headers.SagaId, out var sagaIdHeader))
        {
            logger.LogInformation("Saga Id Header: {SagaIdHeader}", sagaIdHeader);
        }

        if (context.Extensions.TryGet<TableInformation>(out var tableInformation))
        {

            logger.LogInformation("Table Information: {TableName}", tableInformation.TableName);
        }

        logger.LogInformation("Found partition key '{PartitionKey}' from '{TypeName}'", context.Extensions.Get<TableEntityPartitionKey>().PartitionKey, nameof(IProvideOrderId));

        await next();
    }

    public class Registration : RegisterStep
    {
        public Registration() :
            base(nameof(OrderIdAsPartitionKeyBehavior),
                typeof(OrderIdAsPartitionKeyBehavior),
                "Determines the PartitionKey from the logical message",
                provider => new OrderIdAsPartitionKeyBehavior(
                    provider.GetRequiredService<IProvidePartitionKeyFromSagaId>(),
                    provider.GetRequiredService<ILogger<OrderIdAsPartitionKeyBehavior>>()
                    ))
        {
            InsertBefore(nameof(LogicalOutboxBehavior));
        }
    }

    IProvidePartitionKeyFromSagaId partitionKeyFromSagaId1;
    private readonly ILogger<OrderIdAsPartitionKeyBehavior> logger;
}
```

<!-- endsnippet -->

Even though one of the handlers replies with a message that does not implement `IProvideOrderId`, transactionality is still maintained because messages that are part of a saga conversation flow automatically have the Saga ID set as a header. In these cases, there is no need to extract correlation property information to derive the Saga ID.

Finally the above behavior is registered in the pipeline.

<!-- snippet: BehaviorRegistration -->

```cs
endpointConfiguration.Pipeline.Register(new OrderIdAsPartitionKeyBehavior.Registration());
```

<!-- endsnippet -->

## Order saga data

<!-- snippet: sagadata -->

```cs
public class OrderSagaData :
    ContainSagaData
{
    public Guid OrderId { get; set; }
    public string OrderDescription { get; set; }
}
```

<!-- endsnippet -->

## Order saga

<!-- snippet: thesaga -->

```cs
public class OrderSaga(ILogger<OrderSaga> logger) :
    Saga<OrderSagaData>,
    IAmStartedByMessages<StartOrder>,
    IHandleMessages<OrderShipped>,
    IHandleTimeouts<CompleteOrder>
{
    protected override void ConfigureHowToFindSaga(SagaPropertyMapper<OrderSagaData> mapper)
    {
        mapper.MapSaga(saga => saga.OrderId)
            .ToMessage<StartOrder>(msg => msg.OrderId);
    }

    public Task Handle(StartOrder message, IMessageHandlerContext context)
    {
        var orderDescription = $"The saga for order {message.OrderId}";
        Data.OrderDescription = orderDescription;

        logger.LogInformation("Received StartOrder message {OrderId}. Starting Saga", Data.OrderId);

        var shipOrder = new ShipOrder
        {
            OrderId = message.OrderId
        };

        logger.LogInformation("Order will complete in 5 seconds");
        var timeoutData = new CompleteOrder
        {
            OrderDescription = orderDescription,
            OrderId = Data.OrderId,
        };

        return Task.WhenAll(
            context.SendLocal(shipOrder),
            RequestTimeout(context, TimeSpan.FromSeconds(5), timeoutData)
        );
    }

    public Task Handle(OrderShipped message, IMessageHandlerContext context)
    {
        logger.LogInformation("Order with OrderId {OrderId} shipped on {ShippingDate}", Data.OrderId, message.ShippingDate);
        return Task.CompletedTask;
    }

    public Task Timeout(CompleteOrder state, IMessageHandlerContext context)
    {
        logger.LogInformation("Saga with OrderId {OrderId} completed", Data.OrderId);
        MarkAsComplete();
        var orderCompleted = new OrderCompleted
        {
            OrderId = Data.OrderId
        };
        return context.Publish(orderCompleted);
    }
}
```

<!-- endsnippet -->