diff --git a/Directory.Packages.props b/Directory.Packages.props
index 890910fc..2ce4b9f3 100644
--- a/Directory.Packages.props
+++ b/Directory.Packages.props
@@ -14,8 +14,18 @@
-
-
+
+
+
+
+
+
+
+
+
diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/IBackgroundJob.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IBackgroundJob.cs
new file mode 100644
index 00000000..eb72223d
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IBackgroundJob.cs
@@ -0,0 +1,13 @@
+using CrispyWaffle.BackgroundJobs.Core;
+
+namespace CrispyWaffle.BackgroundJobs.Abstractions
+{
+ ///
+ /// Low-level job representation (mostly used for in-memory workflows).
+ /// For persisted jobs, prefer IBackgroundJobHandler with registry-based activation.
+ ///
+ public interface IBackgroundJob
+ {
+ Task ExecuteAsync(CancellationToken cancellationToken);
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/IBackgroundJobHandler.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IBackgroundJobHandler.cs
new file mode 100644
index 00000000..fe255907
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IBackgroundJobHandler.cs
@@ -0,0 +1,13 @@
+using CrispyWaffle.BackgroundJobs.Core;
+
+namespace CrispyWaffle.BackgroundJobs.Abstractions
+{
+ ///
+ /// Typed handler used by persisted jobs. Implement this to allow DI-resolved handlers.
+ /// TData is the payload type stored as JSON in the job store.
+ ///
+ public interface IBackgroundJobHandler
+ {
+ Task HandleAsync(TData data, CancellationToken cancellationToken);
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobScheduler.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobScheduler.cs
new file mode 100644
index 00000000..5b4d777a
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobScheduler.cs
@@ -0,0 +1,20 @@
+namespace CrispyWaffle.BackgroundJobs.Abstractions
+{
+ public interface IJobScheduler
+ {
+ Task ScheduleAsync(
+ string handlerName,
+ object payload,
+ TimeSpan delay,
+ int maxAttempts = 3,
+ JobPriority priority = JobPriority.Normal
+ );
+
+ Task EnqueueAsync(
+ string handlerName,
+ object payload,
+ int maxAttempts = 3,
+ JobPriority priority = JobPriority.Normal
+ );
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobStore.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobStore.cs
new file mode 100644
index 00000000..aeb4d420
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/IJobStore.cs
@@ -0,0 +1,28 @@
+namespace CrispyWaffle.BackgroundJobs.Abstractions
+{
+ public interface IJobStore
+ {
+ Task SaveAsync(JobEntity job, CancellationToken cancellationToken = default);
+
+ ///
+ /// Fetch the next available job that is due (ScheduledAt <= UtcNow) and Pending.
+ /// Implementation should atomically mark it as Processing to avoid double pick-up.
+ ///
+ Task FetchNextAsync(CancellationToken cancellationToken = default);
+
+ Task MarkCompletedAsync(Guid jobId, CancellationToken cancellationToken = default);
+
+ Task MarkFailedAsync(
+ Guid jobId,
+ string error,
+ CancellationToken cancellationToken = default
+ );
+
+ Task MarkRetryAsync(
+ Guid jobId,
+ DateTimeOffset? nextAttemptAt,
+ int attemptCount,
+ CancellationToken cancellationToken = default
+ );
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobEntity.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobEntity.cs
new file mode 100644
index 00000000..1d09431b
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobEntity.cs
@@ -0,0 +1,47 @@
+namespace CrispyWaffle.BackgroundJobs.Abstractions
+{
+ using System;
+
+ ///
+ /// Persistent representation of a job.
+ /// HandlerName identifies a registered handler; Payload is JSON.
+ ///
+ public class JobEntity
+ {
+ public Guid Id { get; set; } = Guid.NewGuid();
+
+ ///
+ /// Logical handler name registered in IJobHandlerRegistry.
+ ///
+ public string HandlerName { get; set; } = string.Empty;
+
+ ///
+ /// JSON payload (serialized) for the handler.
+ ///
+ public string Payload { get; set; } = string.Empty;
+
+ public JobPriority Priority { get; set; } = JobPriority.Normal;
+
+ public JobStatus Status { get; set; } = JobStatus.Pending;
+
+ ///
+ /// When the job becomes due for execution. Null means immediately.
+ ///
+ public DateTimeOffset? ScheduledAt { get; set; }
+
+ public int Attempt { get; set; } = 0;
+
+ public int MaxAttempt { get; set; } = 3;
+
+ public string? LastError { get; set; }
+
+ public DateTimeOffset CreatedAt { get; set; } = DateTimeOffset.UtcNow;
+
+ public DateTimeOffset UpdatedAt { get; set; } = DateTimeOffset.UtcNow;
+
+ ///
+ /// Optional delay in seconds between retries (used only as hint).
+ ///
+ public int RetryDelaySeconds { get; set; } = 0;
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobPriority.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobPriority.cs
new file mode 100644
index 00000000..1ac54c2e
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobPriority.cs
@@ -0,0 +1,9 @@
+namespace CrispyWaffle.BackgroundJobs.Abstractions
+{
+ public enum JobPriority
+ {
+ High = 0,
+ Normal = 1,
+ Low = 2,
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobStatus.cs b/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobStatus.cs
new file mode 100644
index 00000000..8854e13a
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Abstractions/JobStatus.cs
@@ -0,0 +1,11 @@
+namespace CrispyWaffle.BackgroundJobs.Abstractions
+{
+ public enum JobStatus
+ {
+ Pending = 0,
+ Processing = 1,
+ Completed = 2,
+ Failed = 3,
+ Dead = 4,
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundJobQueue.cs b/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundJobQueue.cs
new file mode 100644
index 00000000..b735e2d2
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundJobQueue.cs
@@ -0,0 +1,52 @@
+using System.Threading.Channels;
+using CrispyWaffle.BackgroundJobs.Abstractions;
+
+namespace CrispyWaffle.BackgroundJobs.Core
+{
+ ///
+ /// Lightweight in-memory priority queue using Channel for signaling.
+ /// Priorities: lower int = higher priority.
+ /// This queue stores JobEntity (for uniformity with persistent store model).
+ ///
+ public class BackgroundJobQueue
+ {
+ private readonly object _lock = new();
+ private readonly PriorityQueue _pq = new();
+ private readonly Channel _signal = Channel.CreateUnbounded(
+ new UnboundedChannelOptions { SingleReader = false, SingleWriter = false }
+ );
+
+ public void Enqueue(JobEntity job)
+ {
+ lock (_lock)
+ {
+ _pq.Enqueue(job, (int)job.Priority);
+ }
+ // Always write to signal channel to notify consumers.
+ _ = _signal.Writer.WriteAsync(job);
+ }
+
+ public async Task DequeueAsync(CancellationToken cancellationToken)
+ {
+ // Wait until a job is signaled and then pop from our priority queue.
+ try
+ {
+ await _signal.Reader.ReadAsync(cancellationToken);
+ }
+ catch (OperationCanceledException)
+ {
+ return null;
+ }
+
+ lock (_lock)
+ {
+ if (_pq.TryDequeue(out var job, out _))
+ {
+ return job;
+ }
+ }
+
+ return null;
+ }
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs b/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs
new file mode 100644
index 00000000..aeb30985
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Core/BackgroundWorker.cs
@@ -0,0 +1,369 @@
+using System.Text.Json;
+using CrispyWaffle.BackgroundJobs.Abstractions;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace CrispyWaffle.BackgroundJobs.Core
+{
+ public class BackgroundWorker : BackgroundService
+ {
+ private readonly IServiceProvider _provider;
+ private readonly BackgroundJobQueue? _queue;
+ private readonly IJobStore? _store;
+ private readonly IJobHandlerRegistry _registry;
+ private readonly JobOptions _options;
+ private readonly ILogger _logger;
+
+ public BackgroundWorker(
+ IServiceProvider provider,
+ JobOptions options,
+ ILogger logger
+ )
+ {
+ _provider = provider;
+ _options = options;
+ _logger = logger;
+
+ // These may or may not be registered depending on configuration; resolve lazily per scope
+ _queue = provider.GetService();
+ _store = provider.GetService();
+ _registry = provider.GetRequiredService();
+ }
+
+ protected override Task ExecuteAsync(CancellationToken stoppingToken)
+ {
+ _logger.LogInformation(
+ "BackgroundWorker starting with {Workers} workers",
+ _options.WorkerCount
+ );
+
+ var tasks = new Task[_options.WorkerCount];
+ for (int i = 0; i < _options.WorkerCount; i++)
+ tasks[i] = Task.Run(() => WorkerLoopAsync(i, stoppingToken), stoppingToken);
+
+ return Task.WhenAll(tasks);
+ }
+
+ private async Task WorkerLoopAsync(int workerId, CancellationToken stoppingToken)
+ {
+ _logger.LogInformation("Worker {WorkerId} started", workerId);
+
+ while (!stoppingToken.IsCancellationRequested)
+ {
+ try
+ {
+ var job = await FetchNextJobAsync(stoppingToken);
+ if (job == null)
+ continue;
+
+ if (await RescheduleIfNotDueAsync(job, stoppingToken))
+ continue;
+
+ if (!await ProcessJobWithScopeAsync(job, stoppingToken))
+ break; // shutting down
+ }
+ catch (Exception ex)
+ {
+ _logger.LogError(ex, "Worker loop encountered an unexpected error");
+ await Task.Delay(TimeSpan.FromSeconds(1), stoppingToken);
+ }
+ }
+
+ _logger.LogInformation("Worker {WorkerId} stopping", workerId);
+ }
+
+ private async Task FetchNextJobAsync(CancellationToken stoppingToken)
+ {
+ if (_store != null)
+ {
+ var job = await _store.FetchNextAsync(stoppingToken);
+ if (job == null)
+ await Task.Delay(_options.PollingInterval, stoppingToken);
+ return job;
+ }
+
+ if (_queue != null)
+ {
+ var job = await _queue.DequeueAsync(stoppingToken);
+ if (job == null)
+ await Task.Delay(TimeSpan.FromMilliseconds(200), stoppingToken);
+ return job;
+ }
+
+ _logger.LogWarning("No job queue nor store configured. Worker sleeping.");
+ await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
+ return null;
+ }
+
+ private async Task RescheduleIfNotDueAsync(
+ JobEntity job,
+ CancellationToken stoppingToken
+ )
+ {
+ if (!job.ScheduledAt.HasValue || job.ScheduledAt.Value <= DateTimeOffset.UtcNow)
+ return false;
+
+ // For persistence, mark back to pending with same ScheduledAt; for in-memory, just re-enqueue.
+ if (_store != null)
+ {
+ // mark as Pending again (store.FetchNextAsync should have returned only due jobs, this is defensive)
+ await _store.MarkRetryAsync(job.Id, job.ScheduledAt, job.Attempt, stoppingToken);
+ }
+ else
+ {
+ _queue!.Enqueue(job);
+ }
+
+ return true;
+ }
+
+ private async Task ProcessJobWithScopeAsync(
+ JobEntity job,
+ CancellationToken stoppingToken
+ )
+ {
+ // Process job with scope so handlers can have scoped services.
+ using var scope = _provider.CreateScope();
+ var scopedProvider = scope.ServiceProvider;
+
+ bool success;
+ try
+ {
+ success = await ProcessJobAsync(job, scopedProvider, stoppingToken);
+ }
+ catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+ {
+ // If shutting down, put the job back to pending (or leave DB in Processing if store does locking; here we mark retry)
+ if (_store != null)
+ {
+ var next = DateTimeOffset.UtcNow.AddSeconds(5);
+ await _store.MarkRetryAsync(job.Id, next, job.Attempt, stoppingToken);
+ }
+ else
+ {
+ _queue!.Enqueue(job);
+ }
+ return false; // exit loop to allow graceful shutdown
+ }
+
+ if (!success)
+ {
+ _logger.LogWarning(
+ "Job {JobId} failed after processing attempt {Attempt}",
+ job.Id,
+ job.Attempt
+ );
+ }
+
+ return true;
+ }
+
+ private async Task ProcessJobAsync(
+ JobEntity job,
+ IServiceProvider scopedProvider,
+ CancellationToken cancellationToken
+ )
+ {
+ try
+ {
+ if (!_registry.TryGet(job.HandlerName, out var handlerType, out var dataType))
+ {
+ await FailJobAsync(
+ job,
+ $"No handler registered for '{job.HandlerName}'",
+ cancellationToken
+ );
+ return false;
+ }
+
+ // Deserialize payload into dataType
+ var data = DeserializePayload(job, dataType);
+
+ // Resolve handler instance
+ var handler = scopedProvider.GetService(handlerType!);
+ if (handler == null)
+ {
+ await FailJobAsync(
+ job,
+ $"Handler type {handlerType} not resolved from DI",
+ cancellationToken
+ );
+ return false;
+ }
+
+ // Invoke handler using reflection and await the Task
+ var result = await InvokeHandleAsync(handler, data, cancellationToken);
+
+ if (result.Success)
+ {
+ if (_store != null)
+ await _store.MarkCompletedAsync(job.Id, cancellationToken);
+ return true;
+ }
+
+ await HandleJobFailureAsync(job, result, cancellationToken);
+ return false;
+ }
+ catch (Exception ex)
+ {
+ _logger.LogError(ex, "Exception when processing job {JobId}", job.Id);
+ await ScheduleRetryAsync(job, cancellationToken);
+ return false;
+ }
+ }
+
+ private async Task FailJobAsync(
+ JobEntity job,
+ string message,
+ CancellationToken cancellationToken
+ )
+ {
+ _logger.LogError(message);
+ if (_store != null)
+ await _store.MarkFailedAsync(job.Id, message, cancellationToken);
+ }
+
+ private static object? DeserializePayload(JobEntity job, Type? dataType)
+ {
+ if (string.IsNullOrWhiteSpace(job.Payload))
+ return null;
+ return JsonSerializer.Deserialize(job.Payload, dataType ?? typeof(object));
+ }
+
+ private async Task HandleJobFailureAsync(
+ JobEntity job,
+ JobResult result,
+ CancellationToken cancellationToken
+ )
+ {
+ job.Attempt++;
+ var shouldRetry = result.Retry && job.Attempt < job.MaxAttempt;
+ if (shouldRetry)
+ {
+ var backoffSeconds = ComputeBackoffSeconds(job.Attempt);
+ var nextAttempt = DateTimeOffset.UtcNow.AddSeconds(backoffSeconds);
+ _logger.LogInformation(
+ "Scheduling retry for job {JobId} after {Delay}s (attempt {Attempt}/{MaxAttempt})",
+ job.Id,
+ backoffSeconds,
+ job.Attempt,
+ job.MaxAttempt
+ );
+
+ if (_store != null)
+ {
+ await _store.MarkRetryAsync(
+ job.Id,
+ nextAttempt,
+ job.Attempt,
+ cancellationToken
+ );
+ }
+ else
+ {
+ job.ScheduledAt = nextAttempt;
+ _queue!.Enqueue(job);
+ }
+ }
+ else
+ {
+ // exhaust
+ var err = result.ErrorMessage ?? "Job failed";
+ if (_store != null)
+ await _store.MarkFailedAsync(job.Id, err, cancellationToken);
+ else
+ _logger.LogError(
+ "Job {JobId} failed and will not be retried: {Error}",
+ job.Id,
+ err
+ );
+ }
+ }
+
+ private async Task ScheduleRetryAsync(JobEntity job, CancellationToken cancellationToken)
+ {
+ job.Attempt++;
+ var nextAttempt = DateTimeOffset.UtcNow.AddSeconds(ComputeBackoffSeconds(job.Attempt));
+
+ if (_store != null)
+ {
+ await _store.MarkRetryAsync(job.Id, nextAttempt, job.Attempt, cancellationToken);
+ }
+ else
+ {
+ job.ScheduledAt = nextAttempt;
+ _queue!.Enqueue(job);
+ }
+ }
+
+ private static int ComputeBackoffSeconds(int attempt)
+ {
+ // exponential backoff with cap
+ var seconds = (int)Math.Pow(2, Math.Min(attempt, 10));
+ return Math.Min(seconds, 300);
+ }
+
+ private static async Task InvokeHandleAsync(
+ object handler,
+ object? data,
+ CancellationToken cancellationToken
+ )
+ {
+ // Find HandleAsync method (TData, CancellationToken)
+ var method = handler.GetType().GetMethod("HandleAsync");
+ if (method == null)
+ throw new InvalidOperationException("Handler does not implement HandleAsync");
+
+ var parameters = method.GetParameters();
+ object? invokeArg1 = null;
+ object? invokeArg2 = null;
+ if (parameters.Length == 2)
+ {
+ invokeArg1 = data;
+ invokeArg2 = cancellationToken;
+ }
+ else if (parameters.Length == 1)
+ {
+ invokeArg1 = data ?? cancellationToken;
+ if (invokeArg1 is CancellationToken)
+ {
+ invokeArg2 = null;
+ }
+ }
+
+ var taskObj = method.Invoke(
+ handler,
+ invokeArg2 == null ? new[] { invokeArg1 } : new[] { invokeArg1, invokeArg2 }
+ );
+ if (taskObj == null)
+ throw new InvalidOperationException(
+ "Handler returned null instead of Task"
+ );
+
+ // support Task and Task
+ if (taskObj is Task typed)
+ {
+ return await typed;
+ }
+
+ if (taskObj is Task task)
+ {
+ await task;
+ // attempt to read Result property (in case of Task)
+ var resultProperty = task.GetType().GetProperty("Result");
+ if (resultProperty != null)
+ {
+ var res = resultProperty.GetValue(task);
+ if (res is JobResult jr)
+ return jr;
+ }
+
+ // else assume success
+ return JobResult.Ok();
+ }
+
+ throw new InvalidOperationException("Handler.HandleAsync returned unsupported type");
+ }
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/IJobHandlerRegistry.cs b/Src/CrispyWaffle.BackgroundJobs/Core/IJobHandlerRegistry.cs
new file mode 100644
index 00000000..e3e27b8d
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Core/IJobHandlerRegistry.cs
@@ -0,0 +1,11 @@
+namespace CrispyWaffle.BackgroundJobs.Core
+{
+ public interface IJobHandlerRegistry
+ {
+ void Register(string handlerName)
+ where THandler : class,
+ CrispyWaffle.BackgroundJobs.Abstractions.IBackgroundJobHandler;
+
+ bool TryGet(string handlerName, out Type? handlerType, out Type? dataType);
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/JobDispatcher.cs b/Src/CrispyWaffle.BackgroundJobs/Core/JobDispatcher.cs
new file mode 100644
index 00000000..6a8f407a
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Core/JobDispatcher.cs
@@ -0,0 +1,95 @@
+using System.Text.Json;
+using CrispyWaffle.BackgroundJobs.Abstractions;
+
+namespace CrispyWaffle.BackgroundJobs.Core
+{
+ public class JobDispatcher
+ {
+ private readonly IServiceProvider _provider;
+ private readonly BackgroundJobQueue? _queue;
+ private readonly IJobStore? _store;
+ private readonly JobOptions _options;
+
+ public JobDispatcher(
+ IServiceProvider provider,
+ BackgroundJobQueue? queue,
+ IJobStore? store,
+ JobOptions options
+ )
+ {
+ _provider = provider;
+ _queue = queue;
+ _store = store;
+ _options = options;
+ }
+
+ public async Task EnqueueAsync(
+ string handlerName,
+ object payload,
+ int? maxAttempts = null,
+ JobPriority priority = JobPriority.Normal
+ )
+ {
+ var job = new JobEntity
+ {
+ HandlerName = handlerName,
+ Payload = JsonSerializer.Serialize(payload),
+ Priority = priority,
+ MaxAttempt = maxAttempts ?? _options.DefaultMaxAttempt,
+ Status = JobStatus.Pending,
+ CreatedAt = DateTimeOffset.UtcNow,
+ };
+
+ if (_store != null)
+ {
+ await _store.SaveAsync(job);
+ }
+ else
+ {
+ _queue?.Enqueue(job);
+ }
+
+ return job.Id;
+ }
+
+ public async Task ScheduleAsync(
+ string handlerName,
+ object payload,
+ TimeSpan delay,
+ int? maxAttempts = null,
+ JobPriority priority = JobPriority.Normal
+ )
+ {
+ var job = new JobEntity
+ {
+ HandlerName = handlerName,
+ Payload = JsonSerializer.Serialize(payload),
+ Priority = priority,
+ MaxAttempt = maxAttempts ?? _options.DefaultMaxAttempt,
+ Status = JobStatus.Pending,
+ ScheduledAt = DateTimeOffset.UtcNow.Add(delay),
+ CreatedAt = DateTimeOffset.UtcNow,
+ };
+
+ if (_store != null)
+ {
+ await _store.SaveAsync(job);
+ }
+ else
+ {
+ // schedule in-memory using Task.Delay
+ _ = Task.Run(async () =>
+ {
+ try
+ {
+ await Task.Delay(delay);
+ _queue?.Enqueue(job);
+ }
+ catch
+ { /* suppressed */
+ }
+ });
+ }
+ }
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/JobHandlerRegistry.cs b/Src/CrispyWaffle.BackgroundJobs/Core/JobHandlerRegistry.cs
new file mode 100644
index 00000000..606027a9
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Core/JobHandlerRegistry.cs
@@ -0,0 +1,33 @@
+using System.Collections.Concurrent;
+
+namespace CrispyWaffle.BackgroundJobs.Core
+{
+ public class JobHandlerRegistry : IJobHandlerRegistry
+ {
+ private readonly ConcurrentDictionary _map =
+ new();
+
+ public void Register(string handlerName)
+ where THandler : class,
+ CrispyWaffle.BackgroundJobs.Abstractions.IBackgroundJobHandler
+ {
+ if (string.IsNullOrWhiteSpace(handlerName))
+ throw new ArgumentException("handlerName required", nameof(handlerName));
+ _map[handlerName] = (typeof(THandler), typeof(TData));
+ }
+
+ public bool TryGet(string handlerName, out Type? handlerType, out Type? dataType)
+ {
+ if (handlerName != null && _map.TryGetValue(handlerName, out var pair))
+ {
+ handlerType = pair.handlerType;
+ dataType = pair.dataType;
+ return true;
+ }
+
+ handlerType = null;
+ dataType = null;
+ return false;
+ }
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/JobOptions.cs b/Src/CrispyWaffle.BackgroundJobs/Core/JobOptions.cs
new file mode 100644
index 00000000..d3e3fd2c
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Core/JobOptions.cs
@@ -0,0 +1,25 @@
+namespace CrispyWaffle.BackgroundJobs.Core
+{
+ public class JobOptions
+ {
+ ///
+ /// Number of concurrent worker loops.
+ ///
+ public int WorkerCount { get; set; } = Environment.ProcessorCount;
+
+ ///
+ /// Default maximum retry attempts for jobs enqueued without explicit MaxAttempt.
+ ///
+ public int DefaultMaxAttempt { get; set; } = 3;
+
+ ///
+ /// Base polling delay (when using persistent store and no job found).
+ ///
+ public TimeSpan PollingInterval { get; set; } = TimeSpan.FromSeconds(2);
+
+ ///
+ /// Maximum exponential backoff seconds for retry.
+ ///
+ public int MaxBackoffSeconds { get; set; } = 300; // 5 minutes
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Core/JobResult.cs b/Src/CrispyWaffle.BackgroundJobs/Core/JobResult.cs
new file mode 100644
index 00000000..d9a8531d
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Core/JobResult.cs
@@ -0,0 +1,10 @@
+namespace CrispyWaffle.BackgroundJobs.Core
+{
+ public sealed record JobResult(bool Success, bool Retry = false, string? ErrorMessage = null)
+ {
+ public static JobResult Ok() => new(true);
+
+ public static JobResult Fail(string message, bool retry = false) =>
+ new(false, retry, message);
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/CrispyWaffle.BackgroundJobs.csproj b/Src/CrispyWaffle.BackgroundJobs/CrispyWaffle.BackgroundJobs.csproj
new file mode 100644
index 00000000..6a747f0b
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/CrispyWaffle.BackgroundJobs.csproj
@@ -0,0 +1,18 @@
+
+
+ net8.0
+ enable
+ enable
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/Src/CrispyWaffle.BackgroundJobs/Exceptions/JobFailedException.cs b/Src/CrispyWaffle.BackgroundJobs/Exceptions/JobFailedException.cs
new file mode 100644
index 00000000..210e7f87
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Exceptions/JobFailedException.cs
@@ -0,0 +1,8 @@
+namespace CrispyWaffle.BackgroundJobs.Exceptions
+{
+ public class JobFailedException : Exception
+ {
+ public JobFailedException(string message)
+ : base(message) { }
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Extensions/ServiceCollectionExtensions.cs b/Src/CrispyWaffle.BackgroundJobs/Extensions/ServiceCollectionExtensions.cs
new file mode 100644
index 00000000..55132b38
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Extensions/ServiceCollectionExtensions.cs
@@ -0,0 +1,118 @@
+using CrispyWaffle.BackgroundJobs.Abstractions;
+using CrispyWaffle.BackgroundJobs.Core;
+using CrispyWaffle.BackgroundJobs.Monitoring;
+using CrispyWaffle.BackgroundJobs.Persistence;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection.Extensions;
+
+namespace CrispyWaffle.BackgroundJobs.Extensions
+{
+ public static class ServiceCollectionExtensions
+ {
+ ///
+ /// Register core background job services. By default uses InMemoryJobStore. To use EF Core, call AddDbContext before this and pass useEfCore=true.
+ ///
+ public static IServiceCollection AddCrispyBackgroundJobs(
+ this IServiceCollection services,
+ Action? configureOptions = null,
+ bool useEfCoreStore = false
+ )
+ {
+ var options = new JobOptions();
+ configureOptions?.Invoke(options);
+
+ services.AddSingleton(options);
+ services.AddSingleton();
+ services.AddSingleton();
+ services.AddSingleton();
+
+ // Register appropriate store
+ if (useEfCoreStore)
+ {
+ // JobDbContext must be registered by caller
+ services.TryAddScoped();
+ }
+ else
+ {
+ services.TryAddSingleton();
+ services.TryAddSingleton();
+ }
+
+ // Scheduler & worker
+ services.AddSingleton<
+ IJobScheduler,
+ CrispyWaffle.BackgroundJobs.Scheduling.DelayedJobScheduler
+ >();
+
+ // Hosted service (worker)
+ services.AddHostedService();
+
+ return services;
+ }
+
+ ///
+ /// Register a typed job handler where handler is resolved from DI and payload is serialized as JSON.
+ /// handlerName is the logical identifier used when enqueueing jobs.
+ ///
+ public static IServiceCollection AddJobHandler(
+ this IServiceCollection services,
+ string handlerName
+ )
+ where THandler : class, IBackgroundJobHandler
+ {
+ services.AddScoped();
+ // The registry registration happens at runtime; to ensure the registry has the entry we add a post-processor
+ services.AddSingleton(sp => new JobHandlerStartupFilter(
+ sp,
+ handlerName,
+ typeof(THandler),
+ typeof(TData)
+ ));
+ return services;
+ }
+
+ // StartupFilter ensures the registry is populated after DI container is built
+ private class JobHandlerStartupFilter : Microsoft.AspNetCore.Hosting.IStartupFilter
+ {
+ private readonly IServiceProvider _sp;
+ private readonly string _handlerName;
+ private readonly Type _handlerType;
+ private readonly Type _dataType;
+
+ public JobHandlerStartupFilter(
+ IServiceProvider sp,
+ string handlerName,
+ Type handlerType,
+ Type dataType
+ )
+ {
+ _sp = sp;
+ _handlerName = handlerName;
+ _handlerType = handlerType;
+ _dataType = dataType;
+ }
+
+ public Action Configure(
+ Action next
+ )
+ {
+ return app =>
+ {
+ var registry = _sp.GetRequiredService();
+ var registerMethod = registry
+ .GetType()
+ .GetMethod("Register")
+ ?.MakeGenericMethod(_handlerType, _dataType);
+ // If Register(string) exists we call via reflection
+ if (registerMethod != null)
+ {
+ registerMethod.Invoke(registry, new object[] { _handlerName });
+ }
+
+ next(app);
+ };
+ }
+ }
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Monitoring/JobMetrics.cs b/Src/CrispyWaffle.BackgroundJobs/Monitoring/JobMetrics.cs
new file mode 100644
index 00000000..726db9f5
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Monitoring/JobMetrics.cs
@@ -0,0 +1,19 @@
+using System.Collections.Concurrent;
+
+namespace CrispyWaffle.BackgroundJobs.Monitoring
+{
+ public class JobMetrics
+ {
+ private readonly ConcurrentDictionary _counters = new();
+
+ public void Increment(string key)
+ {
+ _counters.AddOrUpdate(key, 1, (_, v) => v + 1);
+ }
+
+ public long Get(string key) => _counters.TryGetValue(key, out var v) ? v : 0;
+
+ public System.Collections.Generic.IDictionary Snapshot() =>
+ new System.Collections.Generic.Dictionary(_counters);
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Monitoring/MonitoringMiddleware.cs b/Src/CrispyWaffle.BackgroundJobs/Monitoring/MonitoringMiddleware.cs
new file mode 100644
index 00000000..c9920c00
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Monitoring/MonitoringMiddleware.cs
@@ -0,0 +1,29 @@
+using System.Text.Json;
+using Microsoft.AspNetCore.Http;
+
+namespace CrispyWaffle.BackgroundJobs.Monitoring
+{
+ public class MonitoringMiddleware
+ {
+ private readonly RequestDelegate _next;
+ private readonly JobMetrics _metrics;
+
+ public MonitoringMiddleware(RequestDelegate next, JobMetrics metrics)
+ {
+ _next = next;
+ _metrics = metrics;
+ }
+
+ public async Task InvokeAsync(HttpContext context)
+ {
+ if (context.Request.Path.StartsWithSegments("/jobs/metrics"))
+ {
+ context.Response.ContentType = "application/json";
+ await context.Response.WriteAsync(JsonSerializer.Serialize(_metrics.Snapshot()));
+ return;
+ }
+
+ await _next(context);
+ }
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Persistence/EfCoreJobStore.cs b/Src/CrispyWaffle.BackgroundJobs/Persistence/EfCoreJobStore.cs
new file mode 100644
index 00000000..3b1e4816
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Persistence/EfCoreJobStore.cs
@@ -0,0 +1,105 @@
+using CrispyWaffle.BackgroundJobs.Abstractions;
+using Microsoft.EntityFrameworkCore;
+
+namespace CrispyWaffle.BackgroundJobs.Persistence
+{
+ public class EfCoreJobStore : IJobStore
+ {
+ private readonly JobDbContext _db;
+
+ public EfCoreJobStore(JobDbContext db)
+ {
+ _db = db;
+ }
+
+ public async Task SaveAsync(JobEntity job, CancellationToken cancellationToken = default)
+ {
+ job.UpdatedAt = DateTimeOffset.UtcNow;
+ _db.Jobs.Add(job);
+ await _db.SaveChangesAsync(cancellationToken);
+ }
+
+ public async Task FetchNextAsync(CancellationToken cancellationToken = default)
+ {
+ // We attempt to find a pending job that is due, ordered by priority and created time. We then mark it Processing inside a transaction
+ var now = DateTimeOffset.UtcNow;
+
+ // This uses a simple approach: select candidate IDs then try to update one row in a transaction; this reduces race conditions.
+ var candidate = await _db
+ .Jobs.Where(j =>
+ j.Status == JobStatus.Pending && (j.ScheduledAt == null || j.ScheduledAt <= now)
+ )
+ .OrderBy(j => (int)j.Priority)
+ .ThenBy(j => j.CreatedAt)
+ .Select(j => j.Id)
+ .FirstOrDefaultAsync(cancellationToken);
+
+ if (candidate == Guid.Empty)
+ return null;
+
+ // Try to claim the job
+ var job = await _db.Jobs.FirstOrDefaultAsync(j => j.Id == candidate, cancellationToken);
+ if (job == null)
+ return null;
+
+ job.Status = JobStatus.Processing;
+ job.UpdatedAt = DateTimeOffset.UtcNow;
+
+ try
+ {
+ await _db.SaveChangesAsync(cancellationToken);
+ return job;
+ }
+ catch (DbUpdateConcurrencyException)
+ {
+ // Someone else claimed it
+ return null;
+ }
+ }
+
+ public async Task MarkCompletedAsync(
+ Guid jobId,
+ CancellationToken cancellationToken = default
+ )
+ {
+ var job = await _db.Jobs.FirstOrDefaultAsync(j => j.Id == jobId, cancellationToken);
+ if (job == null)
+ return;
+ job.Status = JobStatus.Completed;
+ job.UpdatedAt = DateTimeOffset.UtcNow;
+ await _db.SaveChangesAsync(cancellationToken);
+ }
+
+ public async Task MarkFailedAsync(
+ Guid jobId,
+ string error,
+ CancellationToken cancellationToken = default
+ )
+ {
+ var job = await _db.Jobs.FirstOrDefaultAsync(j => j.Id == jobId, cancellationToken);
+ if (job == null)
+ return;
+ job.Status = JobStatus.Failed;
+ job.LastError = error;
+ job.UpdatedAt = DateTimeOffset.UtcNow;
+ await _db.SaveChangesAsync(cancellationToken);
+ }
+
+ public async Task MarkRetryAsync(
+ Guid jobId,
+ DateTimeOffset? nextAttemptAt,
+ int attemptCount,
+ CancellationToken cancellationToken = default
+ )
+ {
+ var job = await _db.Jobs.FirstOrDefaultAsync(j => j.Id == jobId, cancellationToken);
+ if (job == null)
+ return;
+ job.Attempt = attemptCount;
+ job.Status = JobStatus.Pending;
+ job.ScheduledAt = nextAttemptAt;
+ job.UpdatedAt = DateTimeOffset.UtcNow;
+ await _db.SaveChangesAsync(cancellationToken);
+ }
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Persistence/InMemoryJobStore.cs b/Src/CrispyWaffle.BackgroundJobs/Persistence/InMemoryJobStore.cs
new file mode 100644
index 00000000..ca4bac47
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Persistence/InMemoryJobStore.cs
@@ -0,0 +1,86 @@
+using System.Collections.Concurrent;
+using CrispyWaffle.BackgroundJobs.Abstractions;
+
+namespace CrispyWaffle.BackgroundJobs.Persistence
+{
+ public class InMemoryJobStore : IJobStore
+ {
+ private readonly ConcurrentDictionary _storage = new();
+
+ public Task SaveAsync(JobEntity job, CancellationToken cancellationToken = default)
+ {
+ job.UpdatedAt = DateTimeOffset.UtcNow;
+ _storage[job.Id] = job;
+ return Task.CompletedTask;
+ }
+
+ public Task FetchNextAsync(CancellationToken cancellationToken = default)
+ {
+ var now = DateTimeOffset.UtcNow;
+ var next = _storage
+ .Values.Where(j =>
+ j.Status == JobStatus.Pending
+ && (!j.ScheduledAt.HasValue || j.ScheduledAt.Value <= now)
+ )
+ .OrderBy(j => (int)j.Priority)
+ .ThenBy(j => j.CreatedAt)
+ .FirstOrDefault();
+
+ if (next != null)
+ {
+ // try to transition to Processing; using atomic update
+ next.Status = JobStatus.Processing;
+ next.UpdatedAt = DateTimeOffset.UtcNow;
+ _storage[next.Id] = next;
+ return Task.FromResult(next);
+ }
+
+ return Task.FromResult(null);
+ }
+
+ public Task MarkCompletedAsync(Guid jobId, CancellationToken cancellationToken = default)
+ {
+ if (_storage.TryGetValue(jobId, out var job))
+ {
+ job.Status = JobStatus.Completed;
+ job.UpdatedAt = DateTimeOffset.UtcNow;
+ _storage[jobId] = job;
+ }
+ return Task.CompletedTask;
+ }
+
+ public Task MarkFailedAsync(
+ Guid jobId,
+ string error,
+ CancellationToken cancellationToken = default
+ )
+ {
+ if (_storage.TryGetValue(jobId, out var job))
+ {
+ job.Status = JobStatus.Failed;
+ job.LastError = error;
+ job.UpdatedAt = DateTimeOffset.UtcNow;
+ _storage[jobId] = job;
+ }
+ return Task.CompletedTask;
+ }
+
+ public Task MarkRetryAsync(
+ Guid jobId,
+ DateTimeOffset? nextAttemptAt,
+ int attemptCount,
+ CancellationToken cancellationToken = default
+ )
+ {
+ if (_storage.TryGetValue(jobId, out var job))
+ {
+ job.Attempt = attemptCount;
+ job.Status = JobStatus.Pending;
+ job.ScheduledAt = nextAttemptAt;
+ job.UpdatedAt = DateTimeOffset.UtcNow;
+ _storage[jobId] = job;
+ }
+ return Task.CompletedTask;
+ }
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Persistence/JobDbContext.cs b/Src/CrispyWaffle.BackgroundJobs/Persistence/JobDbContext.cs
new file mode 100644
index 00000000..a81b112b
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Persistence/JobDbContext.cs
@@ -0,0 +1,32 @@
+using System.Reflection.Emit;
+using CrispyWaffle.BackgroundJobs.Abstractions;
+using Microsoft.EntityFrameworkCore;
+
+namespace CrispyWaffle.BackgroundJobs.Persistence
+{
+ public class JobDbContext : DbContext
+ {
+ public JobDbContext(DbContextOptions options)
+ : base(options) { }
+
+ public DbSet Jobs { get; set; } = null!;
+
+ protected override void OnModelCreating(ModelBuilder modelBuilder)
+ {
+ base.OnModelCreating(modelBuilder);
+
+ modelBuilder.Entity(eb =>
+ {
+ eb.HasKey(j => j.Id);
+ eb.Property(j => j.HandlerName).IsRequired().HasMaxLength(256);
+ eb.Property(j => j.Payload).IsRequired(false);
+ eb.Property(j => j.Priority).HasConversion();
+ eb.Property(j => j.Status).HasConversion();
+ eb.Property(j => j.CreatedAt).IsRequired();
+ eb.Property(j => j.UpdatedAt).IsRequired();
+ eb.Property(j => j.MaxAttempt).HasDefaultValue(3);
+ eb.Property(j => j.Attempt).HasDefaultValue(0);
+ });
+ }
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Scheduling/CronJobScheduler.cs b/Src/CrispyWaffle.BackgroundJobs/Scheduling/CronJobScheduler.cs
new file mode 100644
index 00000000..828f16e9
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Scheduling/CronJobScheduler.cs
@@ -0,0 +1,40 @@
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace CrispyWaffle.BackgroundJobs.Scheduling
+{
+ ///
+ /// CronJobScheduler is a lightweight helper that triggers an action on a CRON expression.
+ /// Note: to support CRON syntax you'd typically add a dependency such as NCrontab or Cronos.
+ /// This is a stub showing how to wire a hosted service cron runner.
+ ///
+ public abstract class CronJobService : BackgroundService
+ {
+ private readonly ILogger _logger;
+
+ protected CronJobService(ILogger logger)
+ {
+ _logger = logger;
+ }
+
+ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+ {
+ // Example stub: run every minute
+ while (!stoppingToken.IsCancellationRequested)
+ {
+ try
+ {
+ await ExecuteOnce(stoppingToken);
+ }
+ catch (Exception ex)
+ {
+ _logger.LogError(ex, "Cron job execution error");
+ }
+
+ await Task.Delay(TimeSpan.FromMinutes(1), stoppingToken);
+ }
+ }
+
+ protected abstract Task ExecuteOnce(CancellationToken stoppingToken);
+ }
+}
diff --git a/Src/CrispyWaffle.BackgroundJobs/Scheduling/DelayedJobScheduler.cs b/Src/CrispyWaffle.BackgroundJobs/Scheduling/DelayedJobScheduler.cs
new file mode 100644
index 00000000..54201717
--- /dev/null
+++ b/Src/CrispyWaffle.BackgroundJobs/Scheduling/DelayedJobScheduler.cs
@@ -0,0 +1,42 @@
+using CrispyWaffle.BackgroundJobs.Abstractions;
+using CrispyWaffle.BackgroundJobs.Core;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace CrispyWaffle.BackgroundJobs.Scheduling
+{
+ ///
+ /// Schedules jobs by either creating a persisted scheduled job (if store available) or using in-memory Task.Delay.
+ ///
+ public class DelayedJobScheduler : IJobScheduler
+ {
+ private readonly IServiceProvider _provider;
+
+ public DelayedJobScheduler(IServiceProvider provider)
+ {
+ _provider = provider;
+ }
+
+ public Task EnqueueAsync(
+ string handlerName,
+ object payload,
+ int maxAttempts = 3,
+ JobPriority priority = JobPriority.Normal
+ )
+ {
+ var dispatcher = _provider.GetRequiredService();
+ return dispatcher.EnqueueAsync(handlerName, payload, maxAttempts, priority);
+ }
+
+ public Task ScheduleAsync(
+ string handlerName,
+ object payload,
+ TimeSpan delay,
+ int maxAttempts = 3,
+ JobPriority priority = JobPriority.Normal
+ )
+ {
+ var dispatcher = _provider.GetRequiredService();
+ return dispatcher.ScheduleAsync(handlerName, payload, delay, maxAttempts, priority);
+ }
+ }
+}
diff --git a/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs b/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs
index 2ad342d0..46a28511 100644
--- a/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs
+++ b/Src/CrispyWaffle.HttpClient/HttpClientWrapper.cs
@@ -45,29 +45,45 @@ private System.Net.Http.HttpClient CreateClient(HttpRequestOptions? perRequest)
? _httpClientFactory.CreateClient()
: _httpClientFactory.CreateClient(name);
+ ConfigureBaseAddress(client, perRequest);
+ ConfigureTimeout(client, perRequest);
+ ApplyDefaultHeaders(client, perRequest);
+ EnsureAcceptHeader(client);
+
+ return client;
+ }
+
+ private void ConfigureBaseAddress(
+ System.Net.Http.HttpClient client,
+ HttpRequestOptions? perRequest
+ )
+ {
var baseAddr = perRequest?.BaseAddress ?? _defaults.BaseAddress;
if (!string.IsNullOrEmpty(baseAddr) && client.BaseAddress == null)
{
client.BaseAddress = new Uri(baseAddr);
}
+ }
+ private void ConfigureTimeout(
+ System.Net.Http.HttpClient client,
+ HttpRequestOptions? perRequest
+ )
+ {
var timeoutSeconds = perRequest?.TimeoutSeconds ?? _defaults.TimeoutSeconds;
client.Timeout = TimeSpan.FromSeconds(Math.Max(1, timeoutSeconds));
+ }
- // Merge default headers (global) and per-request headers
- var headers = new Dictionary();
- if (_defaults.DefaultRequestHeaders != null)
- {
- foreach (var kv in _defaults.DefaultRequestHeaders)
- headers[kv.Key] = kv.Value;
- }
- if (perRequest?.DefaultRequestHeaders != null)
- {
- foreach (var kv in perRequest.DefaultRequestHeaders)
- headers[kv.Key] = kv.Value;
- }
+ private void ApplyDefaultHeaders(
+ System.Net.Http.HttpClient client,
+ HttpRequestOptions? perRequest
+ )
+ {
+ var headers = MergeRequestHeaders(
+ _defaults.DefaultRequestHeaders,
+ perRequest?.DefaultRequestHeaders
+ );
- // Set headers (be careful to not duplicate)
foreach (var kv in headers)
{
if (!client.DefaultRequestHeaders.Contains(kv.Key))
@@ -82,16 +98,35 @@ private System.Net.Http.HttpClient CreateClient(HttpRequestOptions? perRequest)
}
}
}
+ }
- // Ensure Accept header
+ private static void EnsureAcceptHeader(System.Net.Http.HttpClient client)
+ {
if (client.DefaultRequestHeaders.Accept.Count == 0)
{
client.DefaultRequestHeaders.Accept.Add(
new System.Net.Http.Headers.MediaTypeWithQualityHeaderValue("application/json")
);
}
+ }
- return client;
+ private static Dictionary MergeRequestHeaders(
+ IDictionary? defaults,
+ IDictionary? perRequest
+ )
+ {
+ var headers = new Dictionary();
+ if (defaults != null)
+ {
+ foreach (var kv in defaults)
+ headers[kv.Key] = kv.Value;
+ }
+ if (perRequest != null)
+ {
+ foreach (var kv in perRequest)
+ headers[kv.Key] = kv.Value;
+ }
+ return headers;
}
#region Public Async Methods
@@ -168,7 +203,59 @@ CancellationToken cancellationToken
var sw = Stopwatch.StartNew();
// The action will create a NEW HttpRequestMessage each time it's invoked (important for retries).
- Func> action = async ct =>
+ var action = BuildRequestAction(
+ client,
+ method,
+ url,
+ body,
+ effectiveOptions,
+ serializer
+ );
+
+ // Execute the action with retry policy. The policy is expected to return the final response (or null + exception).
+ var (resp, ex) = await RetryPolicy
+ .ExecuteAsync(
+ action,
+ effectiveOptions.RetryCount,
+ effectiveOptions.RetryBaseDelayMs,
+ effectiveOptions.MaxRetryDelayMs,
+ cancellationToken
+ )
+ .ConfigureAwait(false);
+
+ sw.Stop();
+
+ // If no response after retries
+ if (resp == null)
+ {
+ return BuildNoResponseResult(effectiveOptions, ex, sw.Elapsed);
+ }
+
+ // IMPORTANT: dispose HttpResponseMessage after we've finished reading headers/body.
+ using (resp)
+ {
+ return await BuildResponseResultAsync(
+ resp,
+ method,
+ url,
+ serializer,
+ effectiveOptions,
+ sw.Elapsed
+ )
+ .ConfigureAwait(false);
+ }
+ }
+
+ private Func> BuildRequestAction(
+ System.Net.Http.HttpClient client,
+ HttpMethod method,
+ string url,
+ object? body,
+ HttpRequestOptions effectiveOptions,
+ IJsonSerializer serializer
+ )
+ {
+ return async ct =>
{
// Create a new request per attempt - ensures headers/content are fresh for retries.
using var req = new HttpRequestMessage(method, url);
@@ -200,157 +287,167 @@ CancellationToken cancellationToken
.SendAsync(req, HttpCompletionOption.ResponseHeadersRead, ct)
.ConfigureAwait(false);
};
+ }
- // Execute the action with retry policy. The policy is expected to return the final response (or null + exception).
- var (resp, ex) = await RetryPolicy
- .ExecuteAsync(
- action,
- effectiveOptions.RetryCount,
- effectiveOptions.RetryBaseDelayMs,
- effectiveOptions.MaxRetryDelayMs,
- cancellationToken
- )
- .ConfigureAwait(false);
+ private static HttpResponse BuildNoResponseResult(
+ HttpRequestOptions effectiveOptions,
+ Exception? ex,
+ TimeSpan elapsed
+ )
+ {
+ if (effectiveOptions.ThrowOnError)
+ {
+ throw new HttpClientException("HTTP request failed after retries", null, null, ex);
+ }
- sw.Stop();
+ var errors = new List { ex?.Message ?? "Unknown error" };
+ return HttpResponse.Failure(null, null, errors, null, elapsed);
+ }
- // If no response after retries
- if (resp == null)
+ private async Task> BuildResponseResultAsync(
+ HttpResponseMessage resp,
+ HttpMethod method,
+ string url,
+ IJsonSerializer serializer,
+ HttpRequestOptions effectiveOptions,
+ TimeSpan elapsed
+ )
+ {
+ string? raw;
+ try
{
+ // If there is content, read it as string. Keep it simple and synchronous from serializer's perspective.
+ raw =
+ resp.Content != null
+ ? await resp.Content.ReadAsStringAsync().ConfigureAwait(false)
+ : null;
+ }
+ catch (Exception readEx)
+ {
+ // Failed to read content.
+ _logger?.LogError(
+ readEx,
+ "Failed to read response content for {Method} {Url}",
+ method,
+ url
+ );
if (effectiveOptions.ThrowOnError)
{
throw new HttpClientException(
- "HTTP request failed after retries",
- null,
+ "Failed to read response content",
+ resp.StatusCode,
null,
- ex
+ readEx
);
}
- var errors = new List { ex?.Message ?? "Unknown error" };
- return HttpResponse.Failure(null, null, errors, null, sw.Elapsed);
+ var readErrors = new List { readEx.Message };
+ return HttpResponse.Failure(
+ resp.StatusCode,
+ null,
+ readErrors,
+ resp.Headers.ToDictionary(h => h.Key, h => h.Value.AsEnumerable()),
+ elapsed
+ );
}
- // IMPORTANT: dispose HttpResponseMessage after we've finished reading headers/body.
- using (resp)
+ // Gather headers (including content headers).
+ var headers = resp.Headers.ToDictionary(h => h.Key, h => h.Value.AsEnumerable());
+ if (resp.Content?.Headers != null)
{
- string? raw = null;
- try
+ foreach (var h in resp.Content.Headers)
+ headers[h.Key] = h.Value.AsEnumerable();
+ }
+
+ return resp.IsSuccessStatusCode
+ ? BuildSuccessResult(
+ resp,
+ raw,
+ method,
+ url,
+ serializer,
+ effectiveOptions,
+ headers,
+ elapsed
+ )
+ : BuildFailureResult(resp, raw, headers, effectiveOptions, elapsed);
+ }
+
+ private HttpResponse BuildSuccessResult(
+ HttpResponseMessage resp,
+ string? raw,
+ HttpMethod method,
+ string url,
+ IJsonSerializer serializer,
+ HttpRequestOptions effectiveOptions,
+ IDictionary> headers,
+ TimeSpan elapsed
+ )
+ {
+ T? data = default;
+ try
+ {
+ // If raw body is present, try to deserialize; otherwise default(T).
+ if (!string.IsNullOrWhiteSpace(raw))
{
- // If there is content, read it as string. Keep it simple and synchronous from serializer's perspective.
- if (resp.Content != null)
- {
- raw = await resp.Content.ReadAsStringAsync().ConfigureAwait(false);
- }
+ data = serializer.Deserialize(raw);
}
- catch (Exception readEx)
+ }
+ catch (Exception deserEx)
+ {
+ // Deserialization failed - log and either throw or return failure depending on options.
+ _logger?.LogError(
+ deserEx,
+ "Failed to deserialize response body to {Type} for {Method} {Url}",
+ typeof(T).FullName,
+ method,
+ url
+ );
+ if (effectiveOptions.ThrowOnError)
{
- // Failed to read content.
- _logger?.LogError(
- readEx,
- "Failed to read response content for {Method} {Url}",
- method,
- url
- );
- if (effectiveOptions.ThrowOnError)
- {
- throw new HttpClientException(
- "Failed to read response content",
- resp.StatusCode,
- null,
- readEx
- );
- }
-
- var readErrors = new List { readEx.Message };
- return HttpResponse.Failure(
+ throw new HttpClientException(
+ $"Failed to deserialize response to {typeof(T)}",
resp.StatusCode,
- null,
- readErrors,
- resp.Headers.ToDictionary(h => h.Key, h => h.Value.AsEnumerable()),
- sw.Elapsed
+ raw,
+ deserEx
);
}
- // Gather headers (including content headers).
- var headers = resp.Headers.ToDictionary(h => h.Key, h => h.Value.AsEnumerable());
- if (resp.Content?.Headers != null)
- {
- foreach (var h in resp.Content.Headers)
- headers[h.Key] = h.Value.AsEnumerable();
- }
-
- // Successful HTTP status
- if (resp.IsSuccessStatusCode)
- {
- T? data = default;
- try
- {
- // If raw body is present, try to deserialize; otherwise default(T).
- if (!string.IsNullOrWhiteSpace(raw))
- {
- data = serializer.Deserialize(raw);
- }
- }
- catch (Exception deserEx)
- {
- // Deserialization failed - log and either throw or return failure depending on options.
- _logger?.LogError(
- deserEx,
- "Failed to deserialize response body to {Type} for {Method} {Url}",
- typeof(T).FullName,
- method,
- url
- );
- if (effectiveOptions.ThrowOnError)
- {
- throw new HttpClientException(
- $"Failed to deserialize response to {typeof(T)}",
- resp.StatusCode,
- raw,
- deserEx
- );
- }
-
- return HttpResponse.Failure(
- resp.StatusCode,
- raw,
- new List { "DeserializationFailed: " + deserEx.Message },
- headers,
- sw.Elapsed
- );
- }
+ return HttpResponse.Failure(
+ resp.StatusCode,
+ raw,
+ new List { "DeserializationFailed: " + deserEx.Message },
+ headers,
+ elapsed
+ );
+ }
- return HttpResponse.Success(data, resp.StatusCode, raw, headers, sw.Elapsed);
- }
- else
- {
- // Non-success HTTP status
- var errors = new List
- {
- $"HTTP {(int)resp.StatusCode} {resp.ReasonPhrase}",
- };
- if (!string.IsNullOrWhiteSpace(raw))
- errors.Add(raw);
+ return HttpResponse.Success(data, resp.StatusCode, raw, headers, elapsed);
+ }
- if (effectiveOptions.ThrowOnError)
- {
- throw new HttpClientException(
- $"HTTP request failed with {(int)resp.StatusCode}",
- resp.StatusCode,
- raw
- );
- }
+ private static HttpResponse BuildFailureResult(
+ HttpResponseMessage resp,
+ string? raw,
+ IDictionary> headers,
+ HttpRequestOptions effectiveOptions,
+ TimeSpan elapsed
+ )
+ {
+ // Non-success HTTP status
+ var errors = new List { $"HTTP {(int)resp.StatusCode} {resp.ReasonPhrase}" };
+ if (!string.IsNullOrWhiteSpace(raw))
+ errors.Add(raw);
- return HttpResponse.Failure(
- resp.StatusCode,
- raw,
- errors,
- headers,
- sw.Elapsed
- );
- }
+ if (effectiveOptions.ThrowOnError)
+ {
+ throw new HttpClientException(
+ $"HTTP request failed with {(int)resp.StatusCode}",
+ resp.StatusCode,
+ raw
+ );
}
+
+ return HttpResponse.Failure(resp.StatusCode, raw, errors, headers, elapsed);
}
#endregion