akka-hosting-actor-patterns

using Akka.Actor;
using Akka.Cluster.Sharding;
 
/// <summary>
/// A generic "child per entity" parent actor.
/// </summary>
/// <remarks>
/// Reuses Akka.Cluster.Sharding's IMessageExtractor for consistent routing.
/// Ideal for unit tests where clustering overhead is unnecessary.
/// </remarks>
public sealed class GenericChildPerEntityParent : ReceiveActor
{
    public static Props CreateProps(
        IMessageExtractor extractor,
        Func<string, Props> propsFactory)
    {
        return Props.Create(() =>
            new GenericChildPerEntityParent(extractor, propsFactory));
    }
 
    private readonly IMessageExtractor _extractor;
    private readonly Func<string, Props> _propsFactory;
 
    public GenericChildPerEntityParent(
        IMessageExtractor extractor,
        Func<string, Props> propsFactory)
    {
        _extractor = extractor;
        _propsFactory = propsFactory;
 
        ReceiveAny(message =>
        {
            var entityId = _extractor.EntityId(message);
            if (entityId is null) return;
 
            // Get existing child or create new one
            Context.Child(entityId)
                .GetOrElse(() => Context.ActorOf(_propsFactory(entityId), entityId))
                .Forward(_extractor.EntityMessage(message));
        });
    }
}

using Akka.Cluster.Sharding;
 
/// <summary>
/// Routes messages to entity actors based on a strongly-typed ID.
/// </summary>
public sealed class OrderMessageExtractor : HashCodeMessageExtractor
{
    public const int DefaultShardCount = 40;
 
    public OrderMessageExtractor(int maxNumberOfShards = DefaultShardCount)
        : base(maxNumberOfShards)
    {
    }
 
    public override string? EntityId(object message)
    {
        return message switch
        {
            IWithOrderId msg => msg.OrderId.Value.ToString(),
            _ => null
        };
    }
}
 
// Define an interface for messages that target a specific entity
public interface IWithOrderId
{
    OrderId OrderId { get; }
}
 
// Use strongly-typed IDs
public readonly record struct OrderId(Guid Value)
{
    public static OrderId New() => new(Guid.NewGuid());
    public override string ToString() => Value.ToString();
}

using Akka.Cluster.Hosting;
using Akka.Cluster.Sharding;
using Akka.Hosting;
 
public static class OrderActorHostingExtensions
{
    /// <summary>
    /// Adds OrderActor with support for both local and clustered modes.
    /// </summary>
    public static AkkaConfigurationBuilder WithOrderActor(
        this AkkaConfigurationBuilder builder,
        AkkaExecutionMode executionMode = AkkaExecutionMode.Clustered,
        string? clusterRole = null)
    {
        if (executionMode == AkkaExecutionMode.LocalTest)
        {
            // Non-clustered mode: Use GenericChildPerEntityParent
            builder.WithActors((system, registry, resolver) =>
            {
                var parent = system.ActorOf(
                    GenericChildPerEntityParent.CreateProps(
                        new OrderMessageExtractor(),
                        entityId => resolver.Props<OrderActor>(entityId)),
                    "orders");
 
                registry.Register<OrderActor>(parent);
            });
        }
        else
        {
            // Clustered mode: Use ShardRegion
            builder.WithShardRegion<OrderActor>(
                "orders",
                (system, registry, resolver) =>
                    entityId => resolver.Props<OrderActor>(entityId),
                new OrderMessageExtractor(),
                new ShardOptions
                {
                    StateStoreMode = StateStoreMode.DData,
                    Role = clusterRole
                });
        }
 
        return builder;
    }
}

public static class DomainActorHostingExtensions
{
    /// <summary>
    /// Adds all order domain actors with sharding support.
    /// </summary>
    public static AkkaConfigurationBuilder WithOrderDomainActors(
        this AkkaConfigurationBuilder builder,
        AkkaExecutionMode executionMode = AkkaExecutionMode.Clustered,
        string? clusterRole = null)
    {
        return builder
            .WithOrderActor(executionMode, clusterRole)
            .WithPaymentActor(executionMode, clusterRole)
            .WithShipmentActor(executionMode, clusterRole)
            .WithNotificationActor(); // Singleton, no sharding needed
    }
}

public static class SharedAkkaHostingExtensions
{
    public static IServiceCollection AddAkkaWithTimeProvider(
        this IServiceCollection services,
        Action<AkkaConfigurationBuilder, IServiceProvider> configure)
    {
        // Register ITimeProvider using the ActorSystem's Scheduler
        services.AddSingleton<ITimeProvider>(sp =>
            sp.GetRequiredService<ActorSystem>().Scheduler);
 
        return services.ConfigureAkka((builder, sp) =>
        {
            configure(builder, sp);
        });
    }
}
 
// In your actor, inject ITimeProvider
public class SubscriptionActor : ReceiveActor
{
    private readonly ITimeProvider _timeProvider;
 
    public SubscriptionActor(ITimeProvider timeProvider)
    {
        _timeProvider = timeProvider;
 
        // Use _timeProvider.GetUtcNow() instead of DateTime.UtcNow
        // This allows tests to control time
    }
}

using Akka.Reminders;
using Akka.Reminders.Sql;
using Akka.Reminders.Sql.Configuration;
using Akka.Reminders.Storage;
 
public static class ReminderHostingExtensions
{
    /// <summary>
    /// Configures akka-reminders with PostgreSQL storage.
    /// </summary>
    public static AkkaConfigurationBuilder WithPostgresReminders(
        this AkkaConfigurationBuilder builder,
        string connectionString,
        string schemaName = "reminders",
        string tableName = "scheduled_reminders",
        bool autoInitialize = true)
    {
        return builder.WithLocalReminders(reminders => reminders
            .WithResolver(sys => new GenericChildPerEntityResolver(sys))
            .WithStorage(system =>
            {
                var settings = SqlReminderStorageSettings.CreatePostgreSql(
                    connectionString,
                    schemaName,
                    tableName,
                    autoInitialize);
                return new SqlReminderStorage(settings, system);
            })
            .WithSettings(new ReminderSettings
            {
                MaxSlippage = TimeSpan.FromSeconds(30),
                MaxDeliveryAttempts = 3,
                RetryBackoffBase = TimeSpan.FromSeconds(10)
            }));
    }
 
    /// <summary>
    /// Configures akka-reminders with in-memory storage for testing.
    /// </summary>
    public static AkkaConfigurationBuilder WithInMemoryReminders(
        this AkkaConfigurationBuilder builder)
    {
        return builder.WithLocalReminders(reminders => reminders
            .WithResolver(sys => new GenericChildPerEntityResolver(sys))
            .WithStorage(system => new InMemoryReminderStorage())
            .WithSettings(new ReminderSettings
            {
                MaxSlippage = TimeSpan.FromSeconds(1),
                MaxDeliveryAttempts = 3,
                RetryBackoffBase = TimeSpan.FromMilliseconds(100)
            }));
    }
}

using Akka.Actor;
using Akka.Hosting;
using Akka.Reminders;
 
/// <summary>
/// Resolves reminder targets to GenericChildPerEntityParent actors.
/// </summary>
public sealed class GenericChildPerEntityResolver : IReminderActorResolver
{
    private readonly ActorSystem _system;
 
    public GenericChildPerEntityResolver(ActorSystem system)
    {
        _system = system;
    }
 
    public IActorRef ResolveActorRef(ReminderEntry entry)
    {
        var registry = ActorRegistry.For(_system);
 
        return entry.Key switch
        {
            var k when k.StartsWith("order-") =>
                registry.Get<OrderActor>(),
            var k when k.StartsWith("subscription-") =>
                registry.Get<SubscriptionActor>(),
            _ => throw new InvalidOperationException(
                $"Unknown reminder key format: {entry.Key}")
        };
    }
}

public static AkkaConfigurationBuilder WithEmailSenderActor(
    this AkkaConfigurationBuilder builder)
{
    return builder.WithActors((system, registry, resolver) =>
    {
        var actor = system.ActorOf(
            resolver.Props<EmailSenderActor>(),
            "email-sender");
        registry.Register<EmailSenderActor>(actor);
    });
}

/// <summary>
/// Marker type for ActorRegistry to retrieve the order manager
/// (GenericChildPerEntityParent for OrderActors).
/// </summary>
public sealed class OrderManagerActor;
 
// Usage in extension method
registry.Register<OrderManagerActor>(parent);
 
// Usage in controller/service
public class OrderService
{
    private readonly IActorRef _orderManager;
 
    public OrderService(IRequiredActor<OrderManagerActor> orderManager)
    {
        _orderManager = orderManager.ActorRef;
    }
 
    public async Task<OrderResponse> CreateOrder(CreateOrderCommand cmd)
    {
        return await _orderManager.Ask<OrderResponse>(cmd);
    }
}

public sealed class OrderProcessingActor : ReceiveActor
{
    private readonly IServiceProvider _serviceProvider;
    private readonly IActorRef _notificationActor;
 
    public OrderProcessingActor(
        IServiceProvider serviceProvider,
        IRequiredActor<NotificationActor> notificationActor)
    {
        _serviceProvider = serviceProvider;
        _notificationActor = notificationActor.ActorRef;
 
        ReceiveAsync<ProcessOrder>(HandleProcessOrder);
    }
 
    private async Task HandleProcessOrder(ProcessOrder msg)
    {
        // Create scope for this message - disposed after processing
        using var scope = _serviceProvider.CreateScope();
 
        // Resolve scoped services within the scope
        var orderRepository = scope.ServiceProvider.GetRequiredService<IOrderRepository>();
        var paymentService = scope.ServiceProvider.GetRequiredService<IPaymentService>();
        var emailComposer = scope.ServiceProvider.GetRequiredService<IOrderEmailComposer>();
 
        // Do work with scoped services
        var order = await orderRepository.GetByIdAsync(msg.OrderId);
        var payment = await paymentService.ProcessAsync(order);
 
        // DbContext changes committed when scope disposes
    }
}

public sealed class NotificationActor : ReceiveActor
{
    private readonly IEmailLinkGenerator _linkGenerator;  // Singleton - OK!
    private readonly IMjmlTemplateRenderer _renderer;     // Singleton - OK!
 
    public NotificationActor(
        IEmailLinkGenerator linkGenerator,
        IMjmlTemplateRenderer renderer)
    {
        _linkGenerator = linkGenerator;
        _renderer = renderer;
 
        Receive<SendWelcomeEmail>(Handle);
    }
}

// BAD: Scoped service injected into long-lived actor
public sealed class BadActor : ReceiveActor
{
    private readonly IOrderRepository _repo;  // Scoped! DbContext lives forever!
 
    public BadActor(IOrderRepository repo)  // Captured at actor creation
    {
        _repo = repo;  // This DbContext will become stale
    }
}
 
// GOOD: Inject IServiceProvider, create scope per message
public sealed class GoodActor : ReceiveActor
{
    private readonly IServiceProvider _sp;
 
    public GoodActor(IServiceProvider sp)
    {
        _sp = sp;
        ReceiveAsync<ProcessOrder>(async msg =>
        {
            using var scope = _sp.CreateScope();
            var repo = scope.ServiceProvider.GetRequiredService<IOrderRepository>();
            // Fresh DbContext for this message
        });
    }
}

builder.WithShardRegion<OrderActor>(
    "orders",
    (system, registry, resolver) => entityId => resolver.Props<OrderActor>(entityId),
    new OrderMessageExtractor(),
    new ShardOptions
    {
        StateStoreMode = StateStoreMode.DData,
        RememberEntities = false,  // DEFAULT - almost always correct
        Role = clusterRole
    });

/// <summary>
/// Marker type for ActorRegistry. Use this to retrieve the OrderActor shard region.
/// </summary>
public sealed class OrderActorRegion;
 
// Registration - use marker type as generic parameter
builder.WithShardRegion<OrderActorRegion>(
    "orders",
    (system, registry, resolver) => entityId => resolver.Props<OrderActor>(entityId),
    new OrderMessageExtractor(),
    new ShardOptions { StateStoreMode = StateStoreMode.DData });
 
// Retrieval - same marker type
var orderRegion = ActorRegistry.Get<OrderActorRegion>();
orderRegion.Tell(new CreateOrder(orderId, amount));

// BAD - redundant registration
builder.WithShardRegion<OrderActorRegion>("orders", ...)
    .WithActors((system, registry, resolver) =>
    {
        var region = registry.Get<OrderActorRegion>();
        registry.Register<OrderActorRegion>(region);  // UNNECESSARY!
    });
 
// GOOD - WithShardRegion already registers
builder.WithShardRegion<OrderActorRegion>("orders", ...);
// That's it - OrderActorRegion is now in the registry