Atomizer.EntityFrameworkCore 0.6.2

Atomizer

Break down complexity. Scale with confidence. Atomizer transforms large-scale job processing into atomic, reliable, and easily managed units—making distributed systems simple, robust, and a joy to work with.

Overview

Atomizer is a modern, high-performance job scheduling and queueing framework for ASP.NET Core. Built for cloud-native, distributed applications as well as smaller setups and local development, Atomizer provides a powerful yet easy-to-use solution for processing background jobs, handling complex workflows, and scaling your applications effortlessly.

• Effortless distributed scaling: Atomizer works seamlessly in clustered setups, letting you process jobs across multiple servers for true horizontal scalability.
• Flexible architecture: Plug in your preferred storage backend, configure multiple queues, and extend with custom drivers or handlers.
• Reliable and robust: Enjoy graceful shutdowns, automatic retries, and job visibility timeouts to ensure jobs are never lost or duplicated.
• Developer-friendly: Atomizer integrates with ASP.NET Core DI, logging, and modern C# features, so you can focus on your business logic.

Features

• ⏰ Recurring Scheduled Jobs — Cron-like recurring execution for time-based workflows.
• 🚀 Distributed Processing — Scale out to as many servers as your storage backend supports; Atomizer coordinates job execution across the cluster.
• 🗄️ Multiple Storage Backends — Use Entity Framework Core for durable, database-backed queues; Redis for distributed low-latency queues; in-memory for fast local development & testing.
• 🔀 Multiple Queues — Configure independent queues with custom processing options for each workload.
• 🧩 Extensible Drivers & Handlers — Easily add new storage drivers or job handlers; auto-register handlers from assemblies.
• ♻️ Advanced Retry Policies — Automatic, configurable retries to keep your jobs running smoothly—even when things go wrong.
• 🛑 Graceful Shutdown — Ensure in-flight jobs finish and pending batched jobs are safely released for re-processing during shutdowns.
• 📦 Batch Processing — Tune throughput with batch size and parallelism settings per queue.
• ⏳ Visibility Timeout — Prevent job duplication by locking jobs during processing.
• 🕒 FIFO Partitioned Processing — Guarantee strict in-order, one-at-a-time execution per partition key (e.g. per customer, per entity).
• 🧪 In-Memory Driver — Perfect for local development and testing; spin up queues instantly with zero setup.
• 📈 Dashboard — Optional operational dashboard for jobs, schedules, queue statistics, worker heartbeats, and built-in job/schedule actions.
• 🔔 ASP.NET Core Integration — Works with DI, logging, and modern C# idioms.

The dashboard gives teams a focused operational view of Atomizer: inspect queue health, review job and schedule details, watch active workers, and trigger common actions without building a custom admin UI.

Quick Start

Get up and running in minutes:

1. Install the package

dotnet add package Atomizer

# Choose a durable storage backend
dotnet add package Atomizer.EntityFrameworkCore
# or
dotnet add package Atomizer.Redis

# Optional: add the monitoring dashboard
dotnet add package Atomizer.Dashboard

2. Configure Atomizer

Set up Atomizer in your ASP.NET Core project:

builder.Services.AddAtomizer(options =>
{
    // Configure the default queue 
    // (optional, a default queue is created automatically with configuration like below)
    options.AddQueue(QueueKey.Default, queue => 
    {
        queue.DegreeOfParallelism = 4; // Max 4 jobs processed concurrently
        queue.BatchSize = 10; // Retrieve 10 jobs at a time
        queue.VisibilityTimeout = TimeSpan.FromMinutes(5); // Prevent job duplication by "hiding" jobs for 5 minutes while processing
        queue.StorageCheckInterval = TimeSpan.FromSeconds(15); // Poll for new jobs every 15 seconds
    });
    
    // Add more queues as needed
    options.AddQueue("product", queue => 
    {
        queue.DegreeOfParallelism = 2;
        queue.BatchSize = 5;
    });
    
    // Register job handlers automatically
    options.AddHandlersFrom<AssignStockJobHandler>();
    
    // Use EF Core-backed job storage.
    options.UseEntityFrameworkCoreStorage<ExampleDbContext>();
});

// Add Atomizer processing services
builder.Services.AddAtomizerProcessing(options =>
{
    options.StartupDelay = TimeSpan.FromSeconds(5); // Delay startup to allow other services to initialize
    options.GracefulShutdownTimeout = TimeSpan.FromSeconds(30); // Allow up to 30 seconds for jobs to finish on shutdown
});

// Optional: add the Atomizer Dashboard services
builder.Services.AddAtomizerDashboard(options =>
{
    options.Title = "Atomizer Dashboard";
    options.StatsRefreshInterval = TimeSpan.FromSeconds(5);
    options.JobsRefreshInterval = TimeSpan.FromSeconds(30);
});

Inside your DbContext:

protected override void OnModelCreating(ModelBuilder modelBuilder)
{
    modelBuilder.AddAtomizerEntities(schema: "atomizer");
    // ...other model config...
}

Make sure to run migrations to create the necessary tables.

Note:

If using MySql, set schema to 'null' or configure schema behavior in your DbContext as MySql is not compatible with EF Core schemas.

To use Redis-backed storage instead, reference Atomizer.Redis and configure the storage option with a StackExchange.Redis connection string:

using Atomizer.Redis;

builder.Services.AddAtomizer(options =>
{
    options.AddHandlersFrom<AssignStockJobHandler>();
    options.UseRedisStorage(builder.Configuration.GetConnectionString("Redis")!);
});

Redis storage implements the same IAtomizerStorage monitoring methods as the other backends, so it works with Atomizer.Dashboard without referencing the dashboard package from Atomizer.Redis.

3. Define a Job Handler

Create a handler for your job payload:

public record SendNewsletterCommand(Product Product);

public class SendNewsletterJob(INewsletterService newsletterService, IEmailService emailService)
    : IAtomizerJob<SendNewsletterCommand>
{
    public async Task HandleAsync(SendNewsletterCommand payload, JobContext context)
    {
        var subscribers = await newsletterService.GetSubscribersAsync(payload.Product.CategoryId);
        var emails = new List<Email>();
        
        foreach (var subscriber in subscribers)
        {
            emails.Add(new Email { /* ... */ });
        }
        
        await Task.WhenAll(emails.ConvertAll(email => emailService.SendEmailAsync(email)));
    }
}

4. Enqueue, schedule, execute, or dequeue a Job

Add jobs to the queue from your application code:

app.MapPost(
    "/products",
    async ([FromServices] IAtomizerClient atomizerClient, [FromServices] ExampleDbContext dbContext) =>
    {
        var product = new Product { /* ... */, CategoryId = Guid.NewGuid() };
        dbContext.Products.Add(product);
        await dbContext.SaveChangesAsync();
        
        await atomizerClient.EnqueueAsync(new SendNewsletterCommand(product));

        return Results.Created($"/products/{product.Id}", product);
    }
);

Use ExecuteAsync when the job should run immediately in the current process but still be recorded in Atomizer as completed or failed:

var jobId = await atomizerClient.ExecuteAsync(new SendNewsletterCommand(product));

If the handler throws, Atomizer records the job as failed and then rethrows the exception to the caller.

Use DequeueAsync to cancel a job that is still pending:

var dequeued = await atomizerClient.DequeueAsync(jobId);

It returns false when the job does not exist or is no longer pending.

5. FIFO Processing (Partitioned Jobs)

To guarantee jobs for the same entity execute one-at-a-time in enqueue order, assign a PartitionKey:

// All stock events for the same product are processed in strict FIFO order.
await atomizerClient.EnqueueAsync(
    new StockEvent(productId, "restock", delta: 50),
    options => options.PartitionKey = new PartitionKey(productId.ToString())
);

Jobs sharing the same PartitionKey and queue are serialized: the next job in the partition only starts after the previous one completes (or fails and is rescheduled). Unpartitioned jobs in the same queue are unaffected and continue to process in parallel.

6. Schedule Recurring Jobs

in Program.cs:

...
var app = builder.Build();

var atomizer = app.Services.GetRequiredService<IAtomizerClient>();

await atomizer.ScheduleRecurringAsync(
    new LoggerJobPayload("Recurring job started", LogLevel.Information),
    "LoggerJob",
    Schedule.Every(2).Minutes()
);

// Later, when the recurring schedule should stop:
// safe to call even when the schedule has already been deleted or never existed.
await atomizer.DeleteRecurringAsync("LoggerJob");

...

7. Dashboard (Optional)

If you install Atomizer.Dashboard, map the embedded dashboard SPA after building your app:

var app = builder.Build();

app.MapAtomizerDashboard("/atomizer");

Then browse to /atomizer to inspect jobs, job details, schedules, queue statistics, and active worker heartbeats. The dashboard also includes operational actions such as triggering registered job types, retrying failed jobs, cancelling pending jobs, enabling or disabling schedules, and running schedules immediately.

If no authorization filters are configured, dashboard requests are restricted to localhost by default. Add an IAtomizerDashboardAuthorizationFilter through AddAtomizerDashboard(options => options.Authorization.Add(...)) before exposing it outside local development.

For production, prefer integrating with ASP.NET Core authorization:

builder.Services.AddAuthorization(options =>
{
    options.AddPolicy("AtomizerDashboard", policy =>
        policy.RequireAuthenticatedUser().RequireRole("Operations"));
});

builder.Services.AddAtomizerDashboard(options =>
{
    options.Authorization.RequirePolicy("AtomizerDashboard");
});

Make sure the host app runs its authentication middleware before mapping the dashboard:

app.UseAuthentication();
app.UseAuthorization();
app.MapAtomizerDashboard("/atomizer");

The dashboard also provides shortcuts for common requirements:

builder.Services.AddAtomizerDashboard(options =>
{
    options.Authorization.RequireAuthenticatedUser();
    // or: options.Authorization.RequireRoles("Operations", "Admin");
    // or: options.Authorization.RequireClaim("scope", "atomizer.dashboard.read");
});

Multiple dashboard authorization filters are evaluated as alternatives; the first filter that authorizes a request allows it.

Basic authentication is available as an explicit opt-in filter. It requires HTTPS by default:

builder.Services.AddAtomizerDashboard(options =>
{
    options.Authorization.RequireBasicAuthentication(
        builder.Configuration["AtomizerDashboard:Username"]!,
        builder.Configuration["AtomizerDashboard:Password"]!);
});

For credentials stored outside configuration, use the async validator overload:

builder.Services.AddAtomizerDashboard(options =>
{
    options.Authorization.RequireBasicAuthentication(async (context, username, password) =>
    {
        var validator = context.RequestServices.GetRequiredService<IDashboardCredentialValidator>();
        return await validator.ValidateAsync(username, password, context.RequestAborted);
    });
});

Custom authorization filters can also do asynchronous work:

public sealed class DashboardAuthorizationFilter : IAtomizerDashboardAuthorizationFilter
{
    public async ValueTask<DashboardAuthorizationResult> AuthorizeAsync(HttpContext context)
    {
        var access = context.RequestServices.GetRequiredService<IDashboardAccessService>();
        return await access.CanReadDashboardAsync(context.User, context.RequestAborted)
            ? DashboardAuthorizationResult.Authorized
            : DashboardAuthorizationResult.Forbidden;
    }
}

Dashboard API calls are same-origin requests and include same-origin credentials, so cookie, Windows, and other host-level authentication schemes can flow naturally. If your setup needs the embedded frontend to attach an extra request header, configure it when serving the dashboard HTML:

builder.Services.AddAtomizerDashboard(options =>
{
    options.Authorization.RequirePolicy("AtomizerDashboard");
    options.Client.RequestHeaders.Add("X-CSRF-TOKEN", context => context.Request.Cookies["X-CSRF-TOKEN"]);
});

Do not put long-lived shared secrets in Client.RequestHeaders; those values are rendered into the dashboard page and are visible to the browser. Use them for request metadata such as CSRF tokens or short-lived per-user tokens.

Contributing

1. Fork the repository.
2. Create a new branch (feature/xyz).
3. Commit your changes with clear messages.
4. Submit a PR with details of your changes and test coverage.

License

This project is licensed under the MIT License. See the LICENSE file for details