WorkflowForge.Extensions.Persistence 2.0.0

WorkflowForge.Extensions.Persistence

<p align="center"> <img src="https://raw.githubusercontent.com/animatlabs/workflow-forge/main/icon.png" alt="WorkflowForge" width="120" height="120"> </p>

Workflow state persistence extension for WorkflowForge with in-memory and SQLite providers for checkpointing and recovery.

Dependencies

InMemory Provider: ZERO external dependencies - Pure WorkflowForge
SQLite Provider: Uses Microsoft.Data.Sqlite as an external dependency (runtime unification)

Installation

dotnet add package WorkflowForge.Extensions.Persistence

Requires: .NET Standard 2.0 or later

Quick Start

In-Memory (Development/Testing)

using WorkflowForge.Extensions.Persistence.InMemory;

var provider = new InMemoryPersistenceProvider();

// Save workflow state
var state = new WorkflowState
{
    ExecutionId = foundry.ExecutionId,
    WorkflowName = workflow.Name,
    Properties = foundry.Properties.ToDictionary(x => x.Key, x => x.Value),
    CompletedOperations = new List<string> { "Op1", "Op2" }
};

await provider.SaveWorkflowStateAsync(foundry.ExecutionId, state);

// Load workflow state
var loadedState = await provider.LoadWorkflowStateAsync(foundry.ExecutionId);

SQLite (Production)

using WorkflowForge.Extensions.Persistence.SQLite;

var provider = new SQLitePersistenceProvider("workflows.db");

// Same API as InMemory
await provider.SaveWorkflowStateAsync(foundry.ExecutionId, state);
var loadedState = await provider.LoadWorkflowStateAsync(foundry.ExecutionId);

Key Features

• Two Providers: InMemory (testing) and SQLite (production)
• Checkpoint Support: Save workflow state at any point
• Resume Workflows: Continue from last checkpoint
• Pluggable Architecture: IWorkflowPersistenceProvider interface
• Thread-Safe: Concurrent workflow support
• Transactional: SQLite provider uses transactions

Configuration

Via appsettings.json

{
  "WorkflowForge": {
    "Extensions": {
      "Persistence": {
        "Enabled": true,
        "PersistOnOperationComplete": true,
        "PersistOnWorkflowComplete": true,
        "PersistOnFailure": true,
        "MaxVersions": 10,
        "InstanceId": "my-instance-id",
        "WorkflowKey": "my-workflow-key"
      }
    }
  }
}

Via Code

using WorkflowForge.Extensions.Persistence;

var options = new PersistenceOptions
{
    Enabled = true,
    PersistOnOperationComplete = true,
    PersistOnWorkflowComplete = true,
    PersistOnFailure = true,
    MaxVersions = 10,
    InstanceId = "my-instance-id",
    WorkflowKey = "my-workflow-key"
};

foundry.UsePersistence(provider, options);

Via Dependency Injection

using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using WorkflowForge.Extensions.Persistence;

services.AddPersistenceConfiguration(configuration);
var options = serviceProvider.GetRequiredService<IOptions<PersistenceOptions>>().Value;

See Configuration Guide for complete options.

Workflow State Structure

public class WorkflowState
{
    public Guid ExecutionId { get; set; }
    public string WorkflowName { get; set; }
    public Dictionary<string, object> Properties { get; set; }
    public List<string> CompletedOperations { get; set; }
    public int NextOperationIndex { get; set; }
    public DateTimeOffset CreatedAt { get; set; }
    public DateTimeOffset? UpdatedAt { get; set; }
}

Custom Persistence Provider

public class CosmosDbPersistenceProvider : IWorkflowPersistenceProvider
{
    private readonly CosmosClient _client;
    private readonly Container _container;
    
    public async Task SaveWorkflowStateAsync(
        Guid executionId,
        WorkflowState state,
        CancellationToken cancellationToken = default)
    {
        await _container.UpsertItemAsync(state, new PartitionKey(executionId.ToString()), cancellationToken: cancellationToken);
    }
    
    public async Task<WorkflowState?> LoadWorkflowStateAsync(
        Guid executionId,
        CancellationToken cancellationToken = default)
    {
        try
        {
            var response = await _container.ReadItemAsync<WorkflowState>(
                executionId.ToString(),
                new PartitionKey(executionId.ToString()),
                cancellationToken: cancellationToken);
            return response.Resource;
        }
        catch (CosmosException ex) when (ex.StatusCode == HttpStatusCode.NotFound)
        {
            return null;
        }
    }
}

Usage Patterns

Save Checkpoint at Each Operation

foundry.OperationCompleted += async (s, e) =>
{
    var state = CreateStateFromFoundry(foundry);
    await provider.SaveWorkflowStateAsync(foundry.ExecutionId, state);
};

Resume from Checkpoint

var state = await provider.LoadWorkflowStateAsync(executionId);
if (state != null)
{
    // Restore foundry state
    foreach (var prop in state.Properties)
    {
        foundry.SetProperty(prop.Key, prop.Value);
    }
    
    // Skip completed operations
    foundry.SetProperty("NextOperationIndex", state.NextOperationIndex);
}

Documentation

• Getting Started
• Configuration Guide
• Extensions Overview
• Sample 18: Persistence
• Recovery Extension - Recovery orchestration on top of persistence