Add Azure Table Storage persistence provider to WorkflowCore.Providers.Azure

.github/workflows/dotnet.yml

@@ -83,6 +83,7 @@ jobs:
             8.0.x
             9.0.x
             10.0.x
+      - name: Restore dependencies
         run: dotnet restore
       - name: Build
         run: dotnet build --no-restore
@@ -294,3 +295,33 @@ jobs:
         with:
           name: oracle-test-results
           path: test-results/
+  Azure-Tests:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Setup .NET
+        uses: actions/setup-dotnet@v1
+        with:
+          dotnet-version: | 
+            8.0.x
+            9.0.x
+      - name: Restore dependencies
+        run: dotnet restore
+      - name: Build
+        run: dotnet build --no-restore
+      - name: Azure Tests
+        run: dotnet test test/WorkflowCore.Tests.Azure --no-build --verbosity detailed --logger "trx;LogFileName=AzureTests.trx" --logger "console;verbosity=detailed" --results-directory ./test-results -p:ParallelizeTestCollections=false
+      - name: Publish Test Results
+        uses: dorny/test-reporter@v1
+        if: success() || failure()
+        with:
+          name: Azure Test Results
+          path: test-results/*.trx
+          reporter: dotnet-trx
+          fail-on-error: false
+      - name: Upload Test Results
+        uses: actions/upload-artifact@v4
+        if: always()
+        with:
+          name: azure-test-results
+          path: test-results/

WorkflowCore.sln

@@ -158,6 +158,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "WorkflowCore.Persistence.Or
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "WorkflowCore.Tests.Oracle", "test\WorkflowCore.Tests.Oracle\WorkflowCore.Tests.Oracle.csproj", "{A2837F1C-3740-4375-9069-81AE32C867CA}"
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "WorkflowCore.Tests.Azure", "test\WorkflowCore.Tests.Azure\WorkflowCore.Tests.Azure.csproj", "{D4E5F6A7-B8C9-4D0E-1F2A-3B4C5D6E7F80}"
+EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "WorkflowCore.AI.AzureFoundry", "src\extensions\WorkflowCore.AI.AzureFoundry\WorkflowCore.AI.AzureFoundry.csproj", "{A74FFAE5-9788-4C0D-9F77-AF6F6468A1A1}"
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "WorkflowCore.AI.AzureFoundry.Tests", "test\WorkflowCore.AI.AzureFoundry.Tests\WorkflowCore.AI.AzureFoundry.Tests.csproj", "{AAB9DDFF-0A0A-43BE-BF00-2BA13ED526C8}"
@@ -846,6 +848,10 @@ Global
 		{A2837F1C-3740-4375-9069-81AE32C867CA}.Release|x64.Build.0 = Release|Any CPU
 		{A2837F1C-3740-4375-9069-81AE32C867CA}.Release|x86.ActiveCfg = Release|Any CPU
 		{A2837F1C-3740-4375-9069-81AE32C867CA}.Release|x86.Build.0 = Release|Any CPU
+		{D4E5F6A7-B8C9-4D0E-1F2A-3B4C5D6E7F80}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{D4E5F6A7-B8C9-4D0E-1F2A-3B4C5D6E7F80}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{D4E5F6A7-B8C9-4D0E-1F2A-3B4C5D6E7F80}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{D4E5F6A7-B8C9-4D0E-1F2A-3B4C5D6E7F80}.Release|Any CPU.Build.0 = Release|Any CPU
 		{A74FFAE5-9788-4C0D-9F77-AF6F6468A1A1}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
 		{A74FFAE5-9788-4C0D-9F77-AF6F6468A1A1}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{A74FFAE5-9788-4C0D-9F77-AF6F6468A1A1}.Debug|x64.ActiveCfg = Debug|Any CPU
@@ -946,6 +952,7 @@ Global
 		{AF205715-C8B7-42EF-BF14-AFC9E7F27242} = {2EEE6ABD-EE9B-473F-AF2D-6DABB85D7BA2}
 		{635629BC-9D5C-40C6-BBD0-060550ECE290} = {2EEE6ABD-EE9B-473F-AF2D-6DABB85D7BA2}
 		{A2837F1C-3740-4375-9069-81AE32C867CA} = {E6CEAD8D-F565-471E-A0DC-676F54EAEDEB}
+		{D4E5F6A7-B8C9-4D0E-1F2A-3B4C5D6E7F80} = {E6CEAD8D-F565-471E-A0DC-676F54EAEDEB}
 		{A74FFAE5-9788-4C0D-9F77-AF6F6468A1A1} = {6803696C-B19A-4B27-9193-082A02B6F205}
 		{AAB9DDFF-0A0A-43BE-BF00-2BA13ED526C8} = {E6CEAD8D-F565-471E-A0DC-676F54EAEDEB}
 		{D75C8112-6A4D-4A13-BB79-2D23DF66E4CB} = {5080DB09-CBE8-4C45-9957-C3BB7651755E}

docs/persistence.md

@@ -10,5 +10,56 @@ There are several persistence providers available as separate Nuget packages.
 * [Sqlite](https://github.com/danielgerlag/workflow-core/tree/master/src/providers/WorkflowCore.Persistence.Sqlite)
 * [Amazon DynamoDB](https://github.com/danielgerlag/workflow-core/tree/master/src/providers/WorkflowCore.Providers.AWS)
 * [Cosmos DB](https://github.com/danielgerlag/workflow-core/tree/master/src/providers/WorkflowCore.Providers.Azure)
+* [Azure Table Storage](https://github.com/danielgerlag/workflow-core/tree/master/src/providers/WorkflowCore.Providers.Azure)
 * [Redis](https://github.com/danielgerlag/workflow-core/tree/master/src/providers/WorkflowCore.Providers.Redis)
-* [Oracle](https://github.com/danielgerlag/workflow-core/tree/master/src/providers/WorkflowCore.Persistence.Oracle)
+* [Oracle](https://github.com/danielgerlag/workflow-core/tree/master/src/providers/WorkflowCore.Persistence.Oracle)
+
+## Implementing a custom persistence provider
+
+To implement a custom persistence provider, create a class that implements `IPersistenceProvider` interface:
+
+```csharp
+public interface IPersistenceProvider : IWorkflowRepository, ISubscriptionRepository, IEventRepository, IScheduledCommandRepository
+{        
+    Task PersistErrors(IEnumerable<ExecutionError> errors, CancellationToken cancellationToken = default);
+    void EnsureStoreExists();
+}
+```
+
+The `IPersistenceProvider` interface combines four repository interfaces:
+
+### IWorkflowRepository
+Handles workflow instance storage and retrieval:
+- `CreateNewWorkflow` - Create and store a new workflow instance
+- `PersistWorkflow` - Update an existing workflow instance
+- `GetWorkflowInstance` - Retrieve a specific workflow instance
+- `GetRunnableInstances` - Get workflow instances ready for execution
+
+### IEventRepository  
+Manages workflow events:
+- `CreateEvent` - Store a new event
+- `GetEvent` - Retrieve a specific event
+- `GetRunnableEvents` - Get events ready for processing
+- `MarkEventProcessed/Unprocessed` - Update event status
+
+### ISubscriptionRepository
+Handles event subscriptions:
+- `CreateEventSubscription` - Create new event subscription
+- `GetSubscriptions` - Query subscriptions for events
+- `TerminateSubscription` - Remove a subscription
+- `SetSubscriptionToken/ClearSubscriptionToken` - Manage subscription locking
+
+### IScheduledCommandRepository
+For future command scheduling (optional):
+- `ScheduleCommand` - Schedule a command for future execution
+- `ProcessCommands` - Execute scheduled commands
+- `SupportsScheduledCommands` - Indicates if provider supports this feature
+
+Once implemented, register your provider:
+
+```csharp
+services.AddWorkflow(options => 
+{
+    options.UsePersistence(sp => new MyCustomPersistenceProvider());
+});
+```

src/WorkflowCore/Services/LifeCycleEventPublisher.cs

@@ -26,10 +26,16 @@ public LifeCycleEventPublisher(ILifeCycleEventHub eventHub, WorkflowOptions work
 
         public void PublishNotification(LifeCycleEvent evt)
         {
-            if (_outbox.IsAddingCompleted || !_workflowOptions.EnableLifeCycleEventsPublisher)
+            if (!_workflowOptions.EnableLifeCycleEventsPublisher)
                 return;
 
-            _outbox.Add(evt);
+            try
+            {
+                _outbox.TryAdd(evt);
+            }
+            catch (ObjectDisposedException)
+            {
+            }
         }
 
         public void Start()
@@ -44,8 +50,7 @@ public void Start()
                 _outbox = new BlockingCollection<LifeCycleEvent>();
             }
 
-            _dispatchTask = new Task(Execute);
-            _dispatchTask.Start();
+            _dispatchTask = Task.Run(Execute);
         }
 
         public void Stop()
@@ -57,22 +62,36 @@ public void Stop()
 
         public void Dispose()
         {
+            if (_dispatchTask != null)
+            {
+                if (!_outbox.IsAddingCompleted)
+                    _outbox.CompleteAdding();
+                if (!_dispatchTask.Wait(TimeSpan.FromSeconds(30)))
+                    _logger.LogWarning("Lifecycle event publisher did not stop within timeout");
+                _dispatchTask = null;
+            }
             _outbox.Dispose();
         }
 
-        private async void Execute()
+        private async Task Execute()
         {
-            foreach (var evt in _outbox.GetConsumingEnumerable())
+            try
             {
-                try
-                {
-                    await _eventHub.PublishNotification(evt);
-                }
-                catch (Exception ex)
+                foreach (var evt in _outbox.GetConsumingEnumerable())
                 {
-                    _logger.LogError(default(EventId), ex, ex.Message);
+                    try
+                    {
+                        await _eventHub.PublishNotification(evt);
+                    }
+                    catch (Exception ex)
+                    {
+                        _logger.LogError(default(EventId), ex, ex.Message);
+                    }
                 }
             }
+            catch (ObjectDisposedException)
+            {
+            }
         }
     }
 }

src/providers/WorkflowCore.Providers.Azure/Models/EventTableEntity.cs

@@ -0,0 +1,54 @@
+using System;
+using Azure;
+using Azure.Data.Tables;
+using Newtonsoft.Json;
+using WorkflowCore.Models;
+
+namespace WorkflowCore.Providers.Azure.Models
+{
+    public class EventTableEntity : ITableEntity
+    {
+        public string PartitionKey { get; set; }
+        public string RowKey { get; set; }
+        public DateTimeOffset? Timestamp { get; set; }
+        public ETag ETag { get; set; }
+
+        public string EventName { get; set; }
+        public string EventKey { get; set; }
+        public string EventData { get; set; }
+        public DateTime EventTime { get; set; }
+        public bool IsProcessed { get; set; }
+
+        private static JsonSerializerSettings SerializerSettings = new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All };
+
+        public static EventTableEntity FromInstance(Event instance)
+        {
+            return new EventTableEntity
+            {
+                PartitionKey = "event",
+                RowKey = instance.Id,
+                EventName = instance.EventName,
+                EventKey = instance.EventKey,
+                EventTime = EnsureUtc(instance.EventTime),
+                IsProcessed = instance.IsProcessed,
+                EventData = JsonConvert.SerializeObject(instance.EventData, SerializerSettings),
+            };
+        }
+
+        private static DateTime EnsureUtc(DateTime dt) =>
+            dt.Kind == DateTimeKind.Utc ? dt : DateTime.SpecifyKind(dt, DateTimeKind.Utc);
+
+        public static Event ToInstance(EventTableEntity entity)
+        {
+            return new Event
+            {
+                Id = entity.RowKey,
+                EventName = entity.EventName,
+                EventKey = entity.EventKey,
+                EventTime = entity.EventTime,
+                IsProcessed = entity.IsProcessed,
+                EventData = JsonConvert.DeserializeObject(entity.EventData, SerializerSettings),
+            };
+        }
+    }
+}

src/providers/WorkflowCore.Providers.Azure/Models/ScheduledCommandTableEntity.cs

@@ -0,0 +1,44 @@
+using System;
+using Azure;
+using Azure.Data.Tables;
+using Newtonsoft.Json;
+using WorkflowCore.Models;
+
+namespace WorkflowCore.Providers.Azure.Models
+{
+    public class ScheduledCommandTableEntity : ITableEntity
+    {
+        public string PartitionKey { get; set; }
+        public string RowKey { get; set; }
+        public DateTimeOffset? Timestamp { get; set; }
+        public ETag ETag { get; set; }
+
+        public string CommandName { get; set; }
+        public string Data { get; set; }
+        public long ExecuteTime { get; set; }
+
+        private static JsonSerializerSettings SerializerSettings = new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All };
+
+        public static ScheduledCommandTableEntity FromInstance(ScheduledCommand instance, string rowKey = null)
+        {
+            return new ScheduledCommandTableEntity
+            {
+                PartitionKey = "command",
+                RowKey = rowKey ?? Guid.NewGuid().ToString(),
+                CommandName = instance.CommandName,
+                Data = instance.Data,
+                ExecuteTime = instance.ExecuteTime,
+            };
+        }
+
+        public static ScheduledCommand ToInstance(ScheduledCommandTableEntity entity)
+        {
+            return new ScheduledCommand
+            {
+                CommandName = entity.CommandName,
+                Data = entity.Data,
+                ExecuteTime = entity.ExecuteTime,
+            };
+        }
+    }
+}

src/providers/WorkflowCore.Providers.Azure/Models/SubscriptionTableEntity.cs

@@ -0,0 +1,69 @@
+using System;
+using Azure;
+using Azure.Data.Tables;
+using Newtonsoft.Json;
+using WorkflowCore.Models;
+
+namespace WorkflowCore.Providers.Azure.Models
+{
+    public class SubscriptionTableEntity : ITableEntity
+    {
+        public string PartitionKey { get; set; }
+        public string RowKey { get; set; }
+        public DateTimeOffset? Timestamp { get; set; }
+        public ETag ETag { get; set; }
+
+        public string WorkflowId { get; set; }
+        public int StepId { get; set; }
+        public string ExecutionPointerId { get; set; }
+        public string EventName { get; set; }
+        public string EventKey { get; set; }
+        public DateTime SubscribeAsOf { get; set; }
+        public string SubscriptionData { get; set; }
+        public string ExternalToken { get; set; }
+        public string ExternalWorkerId { get; set; }
+        public DateTime? ExternalTokenExpiry { get; set; }
+
+        private static JsonSerializerSettings SerializerSettings = new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All };
+
+        public static SubscriptionTableEntity FromInstance(EventSubscription instance)
+        {
+            return new SubscriptionTableEntity
+            {
+                PartitionKey = "subscription",
+                RowKey = instance.Id,
+                WorkflowId = instance.WorkflowId,
+                StepId = instance.StepId,
+                ExecutionPointerId = instance.ExecutionPointerId,
+                EventName = instance.EventName,
+                EventKey = instance.EventKey,
+                SubscribeAsOf = EnsureUtc(instance.SubscribeAsOf),
+                ExternalToken = instance.ExternalToken,
+                ExternalWorkerId = instance.ExternalWorkerId,
+                ExternalTokenExpiry = instance.ExternalTokenExpiry.HasValue ? EnsureUtc(instance.ExternalTokenExpiry.Value) : (DateTime?)null,
+                SubscriptionData = JsonConvert.SerializeObject(instance.SubscriptionData, SerializerSettings),
+            };
+        }
+
+        private static DateTime EnsureUtc(DateTime dt) =>
+            dt.Kind == DateTimeKind.Utc ? dt : DateTime.SpecifyKind(dt, DateTimeKind.Utc);
+
+        public static EventSubscription ToInstance(SubscriptionTableEntity entity)
+        {
+            return new EventSubscription
+            {
+                Id = entity.RowKey,
+                WorkflowId = entity.WorkflowId,
+                StepId = entity.StepId,
+                ExecutionPointerId = entity.ExecutionPointerId,
+                EventName = entity.EventName,
+                EventKey = entity.EventKey,
+                SubscribeAsOf = entity.SubscribeAsOf,
+                ExternalToken = entity.ExternalToken,
+                ExternalWorkerId = entity.ExternalWorkerId,
+                ExternalTokenExpiry = entity.ExternalTokenExpiry,
+                SubscriptionData = JsonConvert.DeserializeObject(entity.SubscriptionData, SerializerSettings),
+            };
+        }
+    }
+}