diff --git a/README.md b/README.md
index a879589d..306d9ab4 100644
--- a/README.md
+++ b/README.md
@@ -86,6 +86,7 @@ block until cancellation for long-running hosts.
| `SquidStd.Storage` | Local file storage backend (`AddFileStorage`). | [](src/SquidStd.Storage/README.md) · [](https://www.nuget.org/packages/SquidStd.Storage/) |
| `SquidStd.Storage.S3` | S3/MinIO storage backend (`AddS3Storage`). | [](src/SquidStd.Storage.S3/README.md) · [](https://www.nuget.org/packages/SquidStd.Storage.S3/) |
| `SquidStd.Scripting.Lua` | Lua scripting engine with attribute-based modules and event bridging. | [](src/SquidStd.Scripting.Lua/README.md) · [](https://www.nuget.org/packages/SquidStd.Scripting.Lua/) |
+| `SquidStd.Templating` | Scriban templating with a named-template registry and `templates/*.tmpl` auto-load (`AddTemplating`). | [](src/SquidStd.Templating/README.md) · [](https://www.nuget.org/packages/SquidStd.Templating/) |
## Architecture
diff --git a/SquidStd.slnx b/SquidStd.slnx
index 12d68a83..6d73a395 100644
--- a/SquidStd.slnx
+++ b/SquidStd.slnx
@@ -18,6 +18,10 @@
+
+
+
+
diff --git a/docs/articles/templating.md b/docs/articles/templating.md
new file mode 100644
index 00000000..3bcd926d
--- /dev/null
+++ b/docs/articles/templating.md
@@ -0,0 +1 @@
+[!include[](../../src/SquidStd.Templating/README.md)]
diff --git a/docs/articles/toc.yml b/docs/articles/toc.yml
index 229340cd..280a0eb9 100644
--- a/docs/articles/toc.yml
+++ b/docs/articles/toc.yml
@@ -36,6 +36,14 @@
href: storage-s3.md
- name: SquidStd.Scripting.Lua
href: scripting-lua.md
+- name: SquidStd.Templating
+ href: templating.md
+- name: SquidStd.Workers.Abstractions
+ href: workers-abstractions.md
+- name: SquidStd.Workers
+ href: workers.md
+- name: SquidStd.Workers.Manager
+ href: workers-manager.md
- name: Serialization
href: serialization.md
- name: Scheduler
diff --git a/docs/articles/workers-abstractions.md b/docs/articles/workers-abstractions.md
new file mode 100644
index 00000000..274c4be9
--- /dev/null
+++ b/docs/articles/workers-abstractions.md
@@ -0,0 +1 @@
+[!include[](../../src/SquidStd.Workers.Abstractions/README.md)]
diff --git a/docs/articles/workers-manager.md b/docs/articles/workers-manager.md
new file mode 100644
index 00000000..2d372ba9
--- /dev/null
+++ b/docs/articles/workers-manager.md
@@ -0,0 +1 @@
+[!include[](../../src/SquidStd.Workers.Manager/README.md)]
diff --git a/docs/articles/workers.md b/docs/articles/workers.md
new file mode 100644
index 00000000..22565bdb
--- /dev/null
+++ b/docs/articles/workers.md
@@ -0,0 +1 @@
+[!include[](../../src/SquidStd.Workers/README.md)]
diff --git a/src/SquidStd.Messaging.Abstractions/Data/Events/TopicMessageEvent.cs b/src/SquidStd.Messaging.Abstractions/Data/Events/TopicMessageEvent.cs
new file mode 100644
index 00000000..b3a43554
--- /dev/null
+++ b/src/SquidStd.Messaging.Abstractions/Data/Events/TopicMessageEvent.cs
@@ -0,0 +1,9 @@
+using SquidStd.Core.Interfaces.Events;
+
+namespace SquidStd.Messaging.Abstractions.Data.Events;
+
+///
+/// Event published on the in-process event bus when a topic message is bridged. is the
+/// deserialized message.
+///
+public sealed record TopicMessageEvent(string Topic, object Data) : IEvent;
diff --git a/src/SquidStd.Messaging.Abstractions/Interfaces/IMessageTopic.cs b/src/SquidStd.Messaging.Abstractions/Interfaces/IMessageTopic.cs
new file mode 100644
index 00000000..e1b1c2be
--- /dev/null
+++ b/src/SquidStd.Messaging.Abstractions/Interfaces/IMessageTopic.cs
@@ -0,0 +1,13 @@
+namespace SquidStd.Messaging.Abstractions.Interfaces;
+
+///
+/// Typed facade for publishing to and subscribing to topics (fan-out).
+///
+public interface IMessageTopic
+{
+ /// Publishes a typed message to a topic.
+ Task PublishAsync(string topic, TMessage message, CancellationToken cancellationToken = default);
+
+ /// Subscribes a typed handler to a topic. Dispose to unsubscribe.
+ IDisposable Subscribe(string topic, Func handler);
+}
diff --git a/src/SquidStd.Messaging.Abstractions/Interfaces/ITopicEventBridge.cs b/src/SquidStd.Messaging.Abstractions/Interfaces/ITopicEventBridge.cs
new file mode 100644
index 00000000..2ba8483c
--- /dev/null
+++ b/src/SquidStd.Messaging.Abstractions/Interfaces/ITopicEventBridge.cs
@@ -0,0 +1,11 @@
+namespace SquidStd.Messaging.Abstractions.Interfaces;
+
+///
+/// Bridges a topic into the in-process event bus: each message of type T on the topic is republished
+/// as a TopicMessageEvent.
+///
+public interface ITopicEventBridge
+{
+ /// Starts bridging the given topic; dispose the result to stop.
+ IDisposable Bridge(string topic);
+}
diff --git a/src/SquidStd.Messaging.Abstractions/Interfaces/ITopicProvider.cs b/src/SquidStd.Messaging.Abstractions/Interfaces/ITopicProvider.cs
new file mode 100644
index 00000000..c5882dc4
--- /dev/null
+++ b/src/SquidStd.Messaging.Abstractions/Interfaces/ITopicProvider.cs
@@ -0,0 +1,16 @@
+using SquidStd.Abstractions.Interfaces.Services;
+
+namespace SquidStd.Messaging.Abstractions.Interfaces;
+
+///
+/// Byte-level publish/subscribe transport: every current subscriber of a topic receives every message
+/// (transient, at-most-once fan-out).
+///
+public interface ITopicProvider : ISquidStdService, IAsyncDisposable
+{
+ /// Publishes a raw payload to a topic (fan-out to all current subscribers).
+ Task PublishAsync(string topic, ReadOnlyMemory payload, CancellationToken cancellationToken = default);
+
+ /// Subscribes a raw handler to a topic. Dispose to unsubscribe.
+ IDisposable Subscribe(string topic, Func, CancellationToken, Task> handler);
+}
diff --git a/src/SquidStd.Messaging.Abstractions/Services/MessageTopic.cs b/src/SquidStd.Messaging.Abstractions/Services/MessageTopic.cs
new file mode 100644
index 00000000..f6c4a6cc
--- /dev/null
+++ b/src/SquidStd.Messaging.Abstractions/Services/MessageTopic.cs
@@ -0,0 +1,34 @@
+using SquidStd.Core.Interfaces.Serialization;
+using SquidStd.Messaging.Abstractions.Interfaces;
+
+namespace SquidStd.Messaging.Abstractions.Services;
+
+///
+/// Typed facade over an : serializes outgoing messages and deserializes
+/// incoming payloads.
+///
+public sealed class MessageTopic : IMessageTopic
+{
+ private readonly ITopicProvider _provider;
+ private readonly IDataSerializer _serializer;
+ private readonly IDataDeserializer _deserializer;
+
+ public MessageTopic(ITopicProvider provider, IDataSerializer serializer, IDataDeserializer deserializer)
+ {
+ _provider = provider;
+ _serializer = serializer;
+ _deserializer = deserializer;
+ }
+
+ ///
+ public Task PublishAsync(string topic, TMessage message, CancellationToken cancellationToken = default)
+ => _provider.PublishAsync(topic, _serializer.Serialize(message), cancellationToken);
+
+ ///
+ public IDisposable Subscribe(string topic, Func handler)
+ {
+ ArgumentNullException.ThrowIfNull(handler);
+
+ return _provider.Subscribe(topic, (payload, cancellationToken) => handler(_deserializer.Deserialize(payload), cancellationToken));
+ }
+}
diff --git a/src/SquidStd.Messaging.Abstractions/Services/TopicEventBridge.cs b/src/SquidStd.Messaging.Abstractions/Services/TopicEventBridge.cs
new file mode 100644
index 00000000..9cebd1f8
--- /dev/null
+++ b/src/SquidStd.Messaging.Abstractions/Services/TopicEventBridge.cs
@@ -0,0 +1,25 @@
+using SquidStd.Core.Interfaces.Events;
+using SquidStd.Messaging.Abstractions.Data.Events;
+using SquidStd.Messaging.Abstractions.Interfaces;
+
+namespace SquidStd.Messaging.Abstractions.Services;
+
+///
+/// One-way Topic → EventBus bridge. Subscribes a topic and republishes each message as a
+/// on the .
+///
+public sealed class TopicEventBridge : ITopicEventBridge
+{
+ private readonly IMessageTopic _topic;
+ private readonly IEventBus _eventBus;
+
+ public TopicEventBridge(IMessageTopic topic, IEventBus eventBus)
+ {
+ _topic = topic;
+ _eventBus = eventBus;
+ }
+
+ ///
+ public IDisposable Bridge(string topic)
+ => _topic.Subscribe(topic, (data, cancellationToken) => _eventBus.PublishAsync(new TopicMessageEvent(topic, data!), cancellationToken));
+}
diff --git a/src/SquidStd.Messaging.RabbitMq/Extensions/RabbitMqMessagingRegistrationExtensions.cs b/src/SquidStd.Messaging.RabbitMq/Extensions/RabbitMqMessagingRegistrationExtensions.cs
index b56cd98c..839e8ccf 100644
--- a/src/SquidStd.Messaging.RabbitMq/Extensions/RabbitMqMessagingRegistrationExtensions.cs
+++ b/src/SquidStd.Messaging.RabbitMq/Extensions/RabbitMqMessagingRegistrationExtensions.cs
@@ -49,6 +49,10 @@ public static IContainer AddRabbitMqMessaging(
);
container.Register(Reuse.Singleton);
+ container.Register(Reuse.Singleton);
+ container.Register(Reuse.Singleton);
+ container.Register(Reuse.Singleton);
+
return container;
}
diff --git a/src/SquidStd.Messaging.RabbitMq/Services/RabbitMqTopicProvider.cs b/src/SquidStd.Messaging.RabbitMq/Services/RabbitMqTopicProvider.cs
new file mode 100644
index 00000000..ccb60449
--- /dev/null
+++ b/src/SquidStd.Messaging.RabbitMq/Services/RabbitMqTopicProvider.cs
@@ -0,0 +1,218 @@
+using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
+using Serilog;
+using SquidStd.Messaging.Abstractions.Interfaces;
+using SquidStd.Messaging.RabbitMq.Data.Config;
+
+namespace SquidStd.Messaging.RabbitMq.Services;
+
+///
+/// RabbitMQ : topics map to fanout exchanges; each subscriber binds an
+/// exclusive auto-delete queue and consumes with auto-ack (transient, at-most-once fan-out).
+///
+public sealed class RabbitMqTopicProvider : ITopicProvider
+{
+ private readonly ILogger _logger = Log.ForContext();
+ private readonly RabbitMqOptions _options;
+ private readonly SemaphoreSlim _publishLock = new(1, 1);
+ private readonly Lock _exchangeSync = new();
+ private readonly HashSet _declared = new(StringComparer.Ordinal);
+ private IConnection? _connection;
+ private IChannel? _publishChannel;
+ private int _disposed;
+
+ public RabbitMqTopicProvider(RabbitMqOptions options)
+ {
+ _options = options;
+ }
+
+ ///
+ public async ValueTask StartAsync(CancellationToken cancellationToken = default)
+ {
+ var factory = new ConnectionFactory { AutomaticRecoveryEnabled = true };
+
+ if (_options.Uri is not null)
+ {
+ factory.Uri = _options.Uri;
+ }
+ else
+ {
+ factory.HostName = _options.HostName;
+ factory.Port = _options.Port;
+ factory.VirtualHost = _options.VirtualHost;
+ factory.UserName = _options.UserName;
+ factory.Password = _options.Password;
+ }
+
+ _connection = await factory.CreateConnectionAsync(cancellationToken);
+ _publishChannel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken);
+ }
+
+ ///
+ public ValueTask StopAsync(CancellationToken cancellationToken = default)
+ => DisposeAsync();
+
+ ///
+ public async Task PublishAsync(string topic, ReadOnlyMemory payload, CancellationToken cancellationToken = default)
+ {
+ ArgumentException.ThrowIfNullOrWhiteSpace(topic);
+ var channel = _publishChannel ?? throw new InvalidOperationException("Provider not started.");
+
+ await EnsureExchangeAsync(channel, topic, cancellationToken);
+
+ var properties = new BasicProperties();
+
+ await _publishLock.WaitAsync(cancellationToken);
+
+ try
+ {
+ await channel.BasicPublishAsync(
+ exchange: topic,
+ routingKey: string.Empty,
+ mandatory: false,
+ basicProperties: properties,
+ body: payload,
+ cancellationToken: cancellationToken
+ );
+ }
+ finally
+ {
+ _publishLock.Release();
+ }
+ }
+
+ ///
+ public IDisposable Subscribe(string topic, Func, CancellationToken, Task> handler)
+ {
+ ArgumentException.ThrowIfNullOrWhiteSpace(topic);
+ ArgumentNullException.ThrowIfNull(handler);
+
+ var connection = _connection ?? throw new InvalidOperationException("Provider not started.");
+ var subscription = new Subscription(this, connection, topic, handler);
+ subscription.Start();
+
+ return subscription;
+ }
+
+ private async Task EnsureExchangeAsync(IChannel channel, string topic, CancellationToken cancellationToken)
+ {
+ lock (_exchangeSync)
+ {
+ if (!_declared.Add(topic))
+ {
+ return;
+ }
+ }
+
+ await channel.ExchangeDeclareAsync(
+ exchange: topic,
+ type: ExchangeType.Fanout,
+ durable: false,
+ autoDelete: false,
+ cancellationToken: cancellationToken
+ );
+ }
+
+ ///
+ public async ValueTask DisposeAsync()
+ {
+ if (Interlocked.Exchange(ref _disposed, 1) != 0)
+ {
+ return;
+ }
+
+ if (_publishChannel is not null)
+ {
+ await _publishChannel.CloseAsync();
+ await _publishChannel.DisposeAsync();
+ }
+
+ if (_connection is not null)
+ {
+ await _connection.CloseAsync();
+ await _connection.DisposeAsync();
+ }
+
+ _publishLock.Dispose();
+ }
+
+ private sealed class Subscription : IDisposable
+ {
+ private readonly RabbitMqTopicProvider _provider;
+ private readonly IConnection _connection;
+ private readonly string _topic;
+ private readonly Func, CancellationToken, Task> _handler;
+ private IChannel? _channel;
+ private string? _consumerTag;
+ private int _disposed;
+
+ public Subscription(
+ RabbitMqTopicProvider provider,
+ IConnection connection,
+ string topic,
+ Func, CancellationToken, Task> handler
+ )
+ {
+ _provider = provider;
+ _connection = connection;
+ _topic = topic;
+ _handler = handler;
+ }
+
+ public void Start()
+ => StartAsync().GetAwaiter().GetResult();
+
+ private async Task StartAsync()
+ {
+ _channel = await _connection.CreateChannelAsync();
+
+ await _channel.ExchangeDeclareAsync(_topic, ExchangeType.Fanout, durable: false, autoDelete: false);
+
+ var queue = await _channel.QueueDeclareAsync(queue: string.Empty, durable: false, exclusive: true, autoDelete: true);
+ await _channel.QueueBindAsync(queue: queue.QueueName, exchange: _topic, routingKey: string.Empty);
+
+ var consumer = new AsyncEventingBasicConsumer(_channel);
+ consumer.ReceivedAsync += OnReceivedAsync;
+
+ _consumerTag = await _channel.BasicConsumeAsync(queue.QueueName, autoAck: true, consumer: consumer);
+ }
+
+ private async Task OnReceivedAsync(object sender, BasicDeliverEventArgs args)
+ {
+ try
+ {
+ await _handler(args.Body, CancellationToken.None);
+ }
+ catch (Exception ex)
+ {
+ _provider._logger.Warning(ex, "RabbitMq topic '{Topic}' handler failed", _topic);
+ }
+ }
+
+ public void Dispose()
+ {
+ if (Interlocked.Exchange(ref _disposed, 1) != 0)
+ {
+ return;
+ }
+
+ if (_channel is not null)
+ {
+ try
+ {
+ if (_consumerTag is not null)
+ {
+ _channel.BasicCancelAsync(_consumerTag).GetAwaiter().GetResult();
+ }
+
+ _channel.CloseAsync().GetAwaiter().GetResult();
+ _channel.DisposeAsync().AsTask().GetAwaiter().GetResult();
+ }
+ catch
+ {
+ // Best-effort teardown.
+ }
+ }
+ }
+ }
+}
diff --git a/src/SquidStd.Messaging/Extensions/MessagingRegistrationExtensions.cs b/src/SquidStd.Messaging/Extensions/MessagingRegistrationExtensions.cs
index 92a3299a..a2ca9fee 100644
--- a/src/SquidStd.Messaging/Extensions/MessagingRegistrationExtensions.cs
+++ b/src/SquidStd.Messaging/Extensions/MessagingRegistrationExtensions.cs
@@ -37,6 +37,10 @@ public static IContainer AddInMemoryMessaging(this IContainer container, Messagi
container.Register(Reuse.Singleton);
container.Register(Reuse.Singleton);
+ container.Register(Reuse.Singleton);
+ container.Register(Reuse.Singleton);
+ container.Register(Reuse.Singleton);
+
return container;
}
diff --git a/src/SquidStd.Messaging/Services/InMemoryTopicProvider.cs b/src/SquidStd.Messaging/Services/InMemoryTopicProvider.cs
new file mode 100644
index 00000000..3a950261
--- /dev/null
+++ b/src/SquidStd.Messaging/Services/InMemoryTopicProvider.cs
@@ -0,0 +1,96 @@
+using System.Collections.Concurrent;
+using Serilog;
+using SquidStd.Messaging.Abstractions.Interfaces;
+
+namespace SquidStd.Messaging.Services;
+
+///
+/// In-memory : fan-out delivery to all current subscribers of a topic.
+/// Transient and at-most-once; exceptions in one subscriber are isolated.
+///
+public sealed class InMemoryTopicProvider : ITopicProvider
+{
+ private readonly ILogger _logger = Log.ForContext();
+ private readonly ConcurrentDictionary, CancellationToken, Task>>> _topics = new(StringComparer.Ordinal);
+ private int _disposed;
+
+ ///
+ public ValueTask StartAsync(CancellationToken cancellationToken = default)
+ => ValueTask.CompletedTask;
+
+ ///
+ public ValueTask StopAsync(CancellationToken cancellationToken = default)
+ => DisposeAsync();
+
+ ///
+ public async Task PublishAsync(string topic, ReadOnlyMemory payload, CancellationToken cancellationToken = default)
+ {
+ ArgumentException.ThrowIfNullOrWhiteSpace(topic);
+
+ if (!_topics.TryGetValue(topic, out var handlers))
+ {
+ return;
+ }
+
+ foreach (var handler in handlers.Values)
+ {
+ try
+ {
+ await handler(payload, cancellationToken);
+ }
+ catch (Exception ex)
+ {
+ _logger.Error(ex, "Topic '{Topic}' subscriber failed", topic);
+ }
+ }
+ }
+
+ ///
+ public IDisposable Subscribe(string topic, Func, CancellationToken, Task> handler)
+ {
+ ArgumentException.ThrowIfNullOrWhiteSpace(topic);
+ ArgumentNullException.ThrowIfNull(handler);
+
+ var handlers = _topics.GetOrAdd(topic, static _ => new());
+ var id = Guid.NewGuid();
+ handlers[id] = handler;
+
+ return new Subscription(handlers, id);
+ }
+
+ ///
+ public ValueTask DisposeAsync()
+ {
+ if (Interlocked.Exchange(ref _disposed, 1) != 0)
+ {
+ return ValueTask.CompletedTask;
+ }
+
+ _topics.Clear();
+
+ return ValueTask.CompletedTask;
+ }
+
+ private sealed class Subscription : IDisposable
+ {
+ private readonly ConcurrentDictionary, CancellationToken, Task>> _handlers;
+ private readonly Guid _id;
+ private int _disposed;
+
+ public Subscription(ConcurrentDictionary, CancellationToken, Task>> handlers, Guid id)
+ {
+ _handlers = handlers;
+ _id = id;
+ }
+
+ public void Dispose()
+ {
+ if (Interlocked.Exchange(ref _disposed, 1) != 0)
+ {
+ return;
+ }
+
+ _handlers.TryRemove(_id, out _);
+ }
+ }
+}
diff --git a/src/SquidStd.Services.Core/Extensions/RegisterDefaultServicesExtensions.cs b/src/SquidStd.Services.Core/Extensions/RegisterDefaultServicesExtensions.cs
index 1ce76469..148ff3a6 100644
--- a/src/SquidStd.Services.Core/Extensions/RegisterDefaultServicesExtensions.cs
+++ b/src/SquidStd.Services.Core/Extensions/RegisterDefaultServicesExtensions.cs
@@ -4,6 +4,7 @@
using SquidStd.Abstractions.Extensions.Container;
using SquidStd.Abstractions.Extensions.Services;
using SquidStd.Core.Data.Bootstrap;
+using SquidStd.Core.Directories;
using SquidStd.Core.Data.Jobs;
using SquidStd.Core.Data.Metrics;
using SquidStd.Core.Data.Storage;
@@ -79,6 +80,7 @@ public IContainer RegisterCoreServices()
/// The same container for chaining.
public IContainer RegisterCoreServices(string configName, string configDirectory)
{
+ container.RegisterInstance(new DirectoriesConfig(configDirectory, []));
container.RegisterDataSerializer();
container.RegisterDefaultCoreConfigSections();
container.RegisterConfigManagerService(configName, configDirectory);
diff --git a/src/SquidStd.Templating/Extensions/TemplatingRegistrationExtensions.cs b/src/SquidStd.Templating/Extensions/TemplatingRegistrationExtensions.cs
new file mode 100644
index 00000000..5ea4f59c
--- /dev/null
+++ b/src/SquidStd.Templating/Extensions/TemplatingRegistrationExtensions.cs
@@ -0,0 +1,23 @@
+using DryIoc;
+using SquidStd.Abstractions.Extensions.Services;
+using SquidStd.Templating.Interfaces;
+using SquidStd.Templating.Services;
+
+namespace SquidStd.Templating.Extensions;
+
+///
+/// DryIoc registration helpers for the templating module.
+///
+public static class TemplatingRegistrationExtensions
+{
+ ///
+ /// Registers the Scriban template renderer as a singleton SquidStd service (so its startup
+ /// auto-load of templates/*.tmpl runs with the host). Requires a registered DirectoriesConfig.
+ ///
+ public static IContainer AddTemplating(this IContainer container)
+ {
+ ArgumentNullException.ThrowIfNull(container);
+
+ return container.RegisterStdService(-1);
+ }
+}
diff --git a/src/SquidStd.Templating/Interfaces/ITemplateRenderer.cs b/src/SquidStd.Templating/Interfaces/ITemplateRenderer.cs
new file mode 100644
index 00000000..8fa4ee8c
--- /dev/null
+++ b/src/SquidStd.Templating/Interfaces/ITemplateRenderer.cs
@@ -0,0 +1,16 @@
+namespace SquidStd.Templating.Interfaces;
+
+///
+/// Renders Scriban templates from strings or registered named templates.
+///
+public interface ITemplateRenderer
+{
+ /// Renders an ad-hoc template string against a model.
+ ValueTask RenderAsync(string template, object? model, CancellationToken cancellationToken = default);
+
+ /// Compiles and registers a named template (replacing any existing one with the same name).
+ void Register(string name, string template);
+
+ /// Renders a previously registered template by name.
+ ValueTask RenderByNameAsync(string name, object? model, CancellationToken cancellationToken = default);
+}
diff --git a/src/SquidStd.Templating/README.md b/src/SquidStd.Templating/README.md
new file mode 100644
index 00000000..6cc99300
--- /dev/null
+++ b/src/SquidStd.Templating/README.md
@@ -0,0 +1,62 @@
+
+ 
+
+
+SquidStd.Templating
+
+
+ 
+ 
+ 
+ 
+
+
+Templating for SquidStd, backed by [Scriban](https://github.com/scriban/scriban). `ITemplateRenderer`
+renders ad-hoc template strings and named templates (compiled and cached), and auto-loads
+`*.tmpl` files from the `templates/` directory at startup.
+
+## Install
+
+```bash
+dotnet add package SquidStd.Templating
+```
+
+## Features
+
+- `ITemplateRenderer.RenderAsync(template, model)` — render an ad-hoc template string.
+- `Register(name, template)` + `RenderByNameAsync(name, model)` — compiled, cached named templates.
+- Startup auto-load: every `templates/**/*.tmpl` is registered by relative path without extension (`emails/welcome.tmpl` → `emails/welcome`).
+- Scriban default member naming (`snake_case`): a `UserName` property is `{{ user.user_name }}`.
+- Parse/render failures surface as `TemplateException`; unknown names as `InvalidOperationException`.
+
+## Usage
+
+```csharp
+using DryIoc;
+using SquidStd.Templating.Extensions;
+using SquidStd.Templating.Interfaces;
+
+container.AddTemplating(); // after RegisterCoreServices (a shared DirectoriesConfig is registered there)
+
+var renderer = container.Resolve();
+
+// ad-hoc
+var hi = await renderer.RenderAsync("Hi {{ user.name }}!", new { User = new { Name = "squid" } });
+
+// named (registered manually or auto-loaded from templates/)
+renderer.Register("welcome", "Welcome {{ user.name }}");
+var welcome = await renderer.RenderByNameAsync("welcome", new { User = new { Name = "squid" } });
+```
+
+## Key types
+
+| Type | Purpose |
+|------|---------|
+| `ITemplateRenderer` | Render ad-hoc and named templates. |
+| `ScribanTemplateRenderer` | Scriban implementation; compiles/caches named templates and auto-loads `templates/*.tmpl`. |
+| `TemplateException` | Raised on parse/render failures. |
+| `TemplatingRegistrationExtensions` | `AddTemplating(...)` registration. |
+
+## License
+
+MIT — part of [SquidStd](https://github.com/tgiachi/squid-std).
diff --git a/src/SquidStd.Templating/Services/ScribanTemplateRenderer.cs b/src/SquidStd.Templating/Services/ScribanTemplateRenderer.cs
new file mode 100644
index 00000000..c3b703b6
--- /dev/null
+++ b/src/SquidStd.Templating/Services/ScribanTemplateRenderer.cs
@@ -0,0 +1,103 @@
+using System.Collections.Concurrent;
+using Scriban;
+using Scriban.Syntax;
+using SquidStd.Abstractions.Interfaces.Services;
+using SquidStd.Core.Directories;
+using SquidStd.Templating.Interfaces;
+
+namespace SquidStd.Templating.Services;
+
+///
+/// Scriban-backed . Named templates are compiled and cached; ad-hoc
+/// strings are parsed per call. On start it auto-loads templates/**/*.tmpl.
+///
+public sealed class ScribanTemplateRenderer : ITemplateRenderer, ISquidStdService
+{
+ private const string TemplateExtension = ".tmpl";
+
+ private readonly DirectoriesConfig _directories;
+ private readonly ConcurrentDictionary _named = new(StringComparer.Ordinal);
+
+ public ScribanTemplateRenderer(DirectoriesConfig directories)
+ {
+ _directories = directories;
+ }
+
+ ///
+ public async ValueTask RenderAsync(string template, object? model, CancellationToken cancellationToken = default)
+ {
+ ArgumentException.ThrowIfNullOrWhiteSpace(template);
+
+ return await RenderCompiledAsync(Parse(template, "(inline)"), model, "(inline)");
+ }
+
+ ///
+ public void Register(string name, string template)
+ {
+ ArgumentException.ThrowIfNullOrWhiteSpace(name);
+ ArgumentException.ThrowIfNullOrWhiteSpace(template);
+
+ _named[name] = Parse(template, name);
+ }
+
+ ///
+ public async ValueTask RenderByNameAsync(string name, object? model, CancellationToken cancellationToken = default)
+ {
+ ArgumentException.ThrowIfNullOrWhiteSpace(name);
+
+ if (!_named.TryGetValue(name, out var compiled))
+ {
+ throw new InvalidOperationException($"No template registered with name '{name}'.");
+ }
+
+ return await RenderCompiledAsync(compiled, model, name);
+ }
+
+ ///
+ public async ValueTask StartAsync(CancellationToken cancellationToken = default)
+ {
+ var directory = _directories["templates"];
+
+ if (!Directory.Exists(directory))
+ {
+ return;
+ }
+
+ foreach (var file in Directory.EnumerateFiles(directory, "*" + TemplateExtension, SearchOption.AllDirectories))
+ {
+ var relative = Path.GetRelativePath(directory, file).Replace('\\', '/');
+ var name = relative[..^TemplateExtension.Length];
+ var content = await File.ReadAllTextAsync(file, cancellationToken);
+
+ Register(name, content);
+ }
+ }
+
+ ///
+ public ValueTask StopAsync(CancellationToken cancellationToken = default)
+ => ValueTask.CompletedTask;
+
+ private static Template Parse(string text, string name)
+ {
+ var template = Template.Parse(text);
+
+ if (template.HasErrors)
+ {
+ throw new TemplateException($"Template '{name}' has parse errors: {string.Join("; ", template.Messages)}");
+ }
+
+ return template;
+ }
+
+ private static async ValueTask RenderCompiledAsync(Template template, object? model, string name)
+ {
+ try
+ {
+ return await template.RenderAsync(model);
+ }
+ catch (ScriptRuntimeException ex)
+ {
+ throw new TemplateException($"Template '{name}' failed to render: {ex.Message}", ex);
+ }
+ }
+}
diff --git a/src/SquidStd.Templating/SquidStd.Templating.csproj b/src/SquidStd.Templating/SquidStd.Templating.csproj
new file mode 100644
index 00000000..c088a852
--- /dev/null
+++ b/src/SquidStd.Templating/SquidStd.Templating.csproj
@@ -0,0 +1,20 @@
+
+
+
+ net10.0
+ enable
+ enable
+ true
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/src/SquidStd.Templating/TemplateException.cs b/src/SquidStd.Templating/TemplateException.cs
new file mode 100644
index 00000000..73f2bd57
--- /dev/null
+++ b/src/SquidStd.Templating/TemplateException.cs
@@ -0,0 +1,17 @@
+namespace SquidStd.Templating;
+
+///
+/// Raised when a template fails to parse or render.
+///
+public sealed class TemplateException : Exception
+{
+ public TemplateException(string message)
+ : base(message)
+ {
+ }
+
+ public TemplateException(string message, Exception innerException)
+ : base(message, innerException)
+ {
+ }
+}
diff --git a/src/SquidStd.Workers.Abstractions/Data/JobRequest.cs b/src/SquidStd.Workers.Abstractions/Data/JobRequest.cs
new file mode 100644
index 00000000..0effc8c4
--- /dev/null
+++ b/src/SquidStd.Workers.Abstractions/Data/JobRequest.cs
@@ -0,0 +1,8 @@
+namespace SquidStd.Workers.Abstractions.Data;
+
+///
+/// A unit of work the manager enqueues on the jobs queue for a worker to execute.
+///
+/// Logical name of the job, used by the worker to pick a handler.
+/// Opaque string key/value arguments for the job.
+public sealed record JobRequest(string JobName, IReadOnlyDictionary Parameters);
diff --git a/src/SquidStd.Workers.Abstractions/Data/WorkerHeartbeat.cs b/src/SquidStd.Workers.Abstractions/Data/WorkerHeartbeat.cs
new file mode 100644
index 00000000..1213b09a
--- /dev/null
+++ b/src/SquidStd.Workers.Abstractions/Data/WorkerHeartbeat.cs
@@ -0,0 +1,18 @@
+using SquidStd.Workers.Abstractions.Types;
+
+namespace SquidStd.Workers.Abstractions.Data;
+
+///
+/// A liveness signal a worker publishes on the heartbeat topic at a fixed interval.
+///
+/// Stable identity of the reporting worker.
+/// UTC time the heartbeat was produced.
+/// Self-reported status ( when no job is running, else ).
+/// Number of jobs currently in progress on this worker.
+/// Maximum number of jobs the worker runs at once.
+public sealed record WorkerHeartbeat(
+ string WorkerId,
+ DateTime TimestampUtc,
+ WorkerStatusType Status,
+ int ActiveJobs,
+ int MaxConcurrency);
diff --git a/src/SquidStd.Workers.Abstractions/Data/WorkerInfo.cs b/src/SquidStd.Workers.Abstractions/Data/WorkerInfo.cs
new file mode 100644
index 00000000..22eac1ab
--- /dev/null
+++ b/src/SquidStd.Workers.Abstractions/Data/WorkerInfo.cs
@@ -0,0 +1,20 @@
+using SquidStd.Workers.Abstractions.Types;
+
+namespace SquidStd.Workers.Abstractions.Data;
+
+///
+/// The manager-side view of a worker, folded from incoming heartbeats.
+///
+/// Stable identity of the worker.
+/// Last known status; the manager sets on missed heartbeats.
+/// Jobs in progress as of the most recent heartbeat.
+/// Maximum number of jobs the worker runs at once.
+/// UTC time the manager first saw this worker.
+/// UTC time of the most recent heartbeat.
+public sealed record WorkerInfo(
+ string WorkerId,
+ WorkerStatusType Status,
+ int ActiveJobs,
+ int MaxConcurrency,
+ DateTime FirstSeenUtc,
+ DateTime LastSeenUtc);
diff --git a/src/SquidStd.Workers.Abstractions/README.md b/src/SquidStd.Workers.Abstractions/README.md
new file mode 100644
index 00000000..32063e9f
--- /dev/null
+++ b/src/SquidStd.Workers.Abstractions/README.md
@@ -0,0 +1,35 @@
+
+
+
+
+SquidStd.Workers.Abstractions
+
+
+ 
+ 
+
+
+
+
+Shared, dependency-free contracts between the SquidStd worker runtime and manager: the job descriptor, the
+worker heartbeat, the manager-side worker view, the worker status enum, and the conventional channel names.
+
+## Install
+
+```bash
+dotnet add package SquidStd.Workers.Abstractions
+```
+
+## Key types
+
+| Type | Purpose |
+|------|---------|
+| `JobRequest` | A job (name + string parameters) enqueued by the manager and consumed by a worker. |
+| `WorkerHeartbeat` | Liveness signal a worker publishes (status, active jobs, max concurrency). |
+| `WorkerInfo` | The manager-side view of a worker, folded from heartbeats. |
+| `WorkerStatusType` | `Idle` / `Busy` / `Offline`. |
+| `WorkerChannels` | Default jobs-queue and heartbeat-topic names. |
+
+## License
+
+MIT — part of [SquidStd](https://github.com/tgiachi/squid-std).
diff --git a/src/SquidStd.Workers.Abstractions/SquidStd.Workers.Abstractions.csproj b/src/SquidStd.Workers.Abstractions/SquidStd.Workers.Abstractions.csproj
new file mode 100644
index 00000000..2f0e9b5b
--- /dev/null
+++ b/src/SquidStd.Workers.Abstractions/SquidStd.Workers.Abstractions.csproj
@@ -0,0 +1,10 @@
+
+
+
+ net10.0
+ enable
+ enable
+ true
+
+
+
diff --git a/src/SquidStd.Workers.Abstractions/Types/WorkerStatusType.cs b/src/SquidStd.Workers.Abstractions/Types/WorkerStatusType.cs
new file mode 100644
index 00000000..d3aa947c
--- /dev/null
+++ b/src/SquidStd.Workers.Abstractions/Types/WorkerStatusType.cs
@@ -0,0 +1,17 @@
+namespace SquidStd.Workers.Abstractions.Types;
+
+///
+/// Lifecycle status of a worker. Workers report or ;
+/// the manager assigns when a worker's heartbeats stop arriving.
+///
+public enum WorkerStatusType
+{
+ /// The worker is running and has no job in progress.
+ Idle,
+
+ /// The worker is currently processing a job.
+ Busy,
+
+ /// The manager has not seen a heartbeat within the expected window.
+ Offline
+}
diff --git a/src/SquidStd.Workers.Abstractions/WorkerChannels.cs b/src/SquidStd.Workers.Abstractions/WorkerChannels.cs
new file mode 100644
index 00000000..6514a147
--- /dev/null
+++ b/src/SquidStd.Workers.Abstractions/WorkerChannels.cs
@@ -0,0 +1,14 @@
+namespace SquidStd.Workers.Abstractions;
+
+///
+/// Default names of the messaging channels the workers system uses. Shared so the manager
+/// and workers agree out of the box; either side may override them via its own configuration.
+///
+public static class WorkerChannels
+{
+ /// Competing-consumers queue the manager publishes JobRequests onto.
+ public const string JobQueue = "squidstd.workers.jobs";
+
+ /// Fan-out topic workers publish WorkerHeartbeats onto.
+ public const string HeartbeatTopic = "squidstd.workers.heartbeat";
+}
diff --git a/src/SquidStd.Workers.Manager/Data/Config/WorkerManagerConfig.cs b/src/SquidStd.Workers.Manager/Data/Config/WorkerManagerConfig.cs
new file mode 100644
index 00000000..beef18cd
--- /dev/null
+++ b/src/SquidStd.Workers.Manager/Data/Config/WorkerManagerConfig.cs
@@ -0,0 +1,21 @@
+using SquidStd.Workers.Abstractions;
+
+namespace SquidStd.Workers.Manager.Data.Config;
+
+///
+/// Configuration for the worker manager (config section "workerManager").
+///
+public sealed class WorkerManagerConfig
+{
+ /// Seconds without a heartbeat before a worker is considered Offline. Falls back to 30 when not positive.
+ public int OfflineTimeoutSeconds { get; set; } = 30;
+
+ /// Seconds between offline sweeps. Falls back to 10 when not positive.
+ public int SweepIntervalSeconds { get; set; } = 10;
+
+ /// Queue jobs are enqueued onto.
+ public string JobQueue { get; set; } = WorkerChannels.JobQueue;
+
+ /// Topic heartbeats are collected from.
+ public string HeartbeatTopic { get; set; } = WorkerChannels.HeartbeatTopic;
+}
diff --git a/src/SquidStd.Workers.Manager/Data/EnqueueJobRequest.cs b/src/SquidStd.Workers.Manager/Data/EnqueueJobRequest.cs
new file mode 100644
index 00000000..8869f5f7
--- /dev/null
+++ b/src/SquidStd.Workers.Manager/Data/EnqueueJobRequest.cs
@@ -0,0 +1,8 @@
+namespace SquidStd.Workers.Manager.Data;
+
+///
+/// Request body for enqueuing a job via the manager's HTTP surface.
+///
+/// Logical name of the job.
+/// Optional string key/value arguments; treated as empty when null.
+public sealed record EnqueueJobRequest(string JobName, IReadOnlyDictionary? Parameters);
diff --git a/src/SquidStd.Workers.Manager/Data/Events/WorkerStatusChangedEvent.cs b/src/SquidStd.Workers.Manager/Data/Events/WorkerStatusChangedEvent.cs
new file mode 100644
index 00000000..9b15b767
--- /dev/null
+++ b/src/SquidStd.Workers.Manager/Data/Events/WorkerStatusChangedEvent.cs
@@ -0,0 +1,13 @@
+using SquidStd.Core.Interfaces.Events;
+using SquidStd.Workers.Abstractions.Types;
+
+namespace SquidStd.Workers.Manager.Data.Events;
+
+///
+/// Published on the event bus when a worker's status changes: discovered ( null),
+/// gone Offline (via the sweep), or returned (Offline → Idle/Busy).
+///
+/// The worker whose status changed.
+/// Previous status, or null when the worker was just discovered.
+/// The new status.
+public sealed record WorkerStatusChangedEvent(string WorkerId, WorkerStatusType? OldStatus, WorkerStatusType NewStatus) : IEvent;
diff --git a/src/SquidStd.Workers.Manager/Endpoints/WorkerManagerEndpoints.cs b/src/SquidStd.Workers.Manager/Endpoints/WorkerManagerEndpoints.cs
new file mode 100644
index 00000000..95765ec7
--- /dev/null
+++ b/src/SquidStd.Workers.Manager/Endpoints/WorkerManagerEndpoints.cs
@@ -0,0 +1,48 @@
+using Microsoft.AspNetCore.Http;
+using Microsoft.AspNetCore.Http.HttpResults;
+using SquidStd.Workers.Abstractions.Data;
+using SquidStd.Workers.Manager.Data;
+using SquidStd.Workers.Manager.Interfaces;
+
+namespace SquidStd.Workers.Manager.Endpoints;
+
+///
+/// Typed minimal-API handlers for the worker manager surface. Static so they can be unit-tested directly.
+///
+public static class WorkerManagerEndpoints
+{
+ /// Lists all known workers.
+ public static Ok> GetWorkers(IWorkerRegistry registry)
+ => TypedResults.Ok(registry.GetAll());
+
+ /// Returns a single worker, or 404 when unknown.
+ public static Results, NotFound> GetWorker(string id, IWorkerRegistry registry)
+ {
+ var info = registry.Get(id);
+
+ return info is null ? TypedResults.NotFound() : TypedResults.Ok(info);
+ }
+
+ /// Enqueues a job; 400 on a blank name, 503 when the queue is unavailable.
+ public static async Task, ProblemHttpResult>> EnqueueJob(
+ EnqueueJobRequest request,
+ IJobScheduler scheduler,
+ CancellationToken cancellationToken)
+ {
+ if (string.IsNullOrWhiteSpace(request.JobName))
+ {
+ return TypedResults.BadRequest("JobName is required.");
+ }
+
+ try
+ {
+ await scheduler.EnqueueAsync(request.JobName, request.Parameters ?? new Dictionary(), cancellationToken);
+
+ return TypedResults.Accepted((string?)null);
+ }
+ catch (Exception)
+ {
+ return TypedResults.Problem(statusCode: StatusCodes.Status503ServiceUnavailable, detail: "Job queue unavailable.");
+ }
+ }
+}
diff --git a/src/SquidStd.Workers.Manager/Extensions/WorkerManagerEndpointsExtensions.cs b/src/SquidStd.Workers.Manager/Extensions/WorkerManagerEndpointsExtensions.cs
new file mode 100644
index 00000000..6290d917
--- /dev/null
+++ b/src/SquidStd.Workers.Manager/Extensions/WorkerManagerEndpointsExtensions.cs
@@ -0,0 +1,25 @@
+using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Routing;
+using SquidStd.Workers.Manager.Endpoints;
+
+namespace SquidStd.Workers.Manager.Extensions;
+
+///
+/// Maps the opt-in worker manager HTTP endpoints.
+///
+public static class WorkerManagerEndpointsExtensions
+{
+ ///
+ /// Maps GET /workers, GET /workers/{id}, and POST /jobs.
+ ///
+ public static IEndpointRouteBuilder MapWorkerManagerEndpoints(this IEndpointRouteBuilder endpoints)
+ {
+ ArgumentNullException.ThrowIfNull(endpoints);
+
+ endpoints.MapGet("/workers", WorkerManagerEndpoints.GetWorkers);
+ endpoints.MapGet("/workers/{id}", WorkerManagerEndpoints.GetWorker);
+ endpoints.MapPost("/jobs", WorkerManagerEndpoints.EnqueueJob);
+
+ return endpoints;
+ }
+}
diff --git a/src/SquidStd.Workers.Manager/Extensions/WorkerManagerRegistrationExtensions.cs b/src/SquidStd.Workers.Manager/Extensions/WorkerManagerRegistrationExtensions.cs
new file mode 100644
index 00000000..4b825bab
--- /dev/null
+++ b/src/SquidStd.Workers.Manager/Extensions/WorkerManagerRegistrationExtensions.cs
@@ -0,0 +1,42 @@
+using DryIoc;
+using SquidStd.Abstractions.Extensions.Config;
+using SquidStd.Abstractions.Extensions.Services;
+using SquidStd.Core.Data.Timing;
+using SquidStd.Services.Core.Services.Scheduling;
+using SquidStd.Workers.Manager.Data.Config;
+using SquidStd.Workers.Manager.Interfaces;
+using SquidStd.Workers.Manager.Services;
+
+namespace SquidStd.Workers.Manager.Extensions;
+
+///
+/// DryIoc registration helpers for the worker manager.
+///
+public static class WorkerManagerRegistrationExtensions
+{
+ ///
+ /// Registers the worker manager: config section, registry, job scheduler, the collector and sweep
+ /// lifecycle services, and the timer-wheel pump (only if it is not already registered).
+ ///
+ public static IContainer AddWorkerManager(this IContainer container)
+ {
+ ArgumentNullException.ThrowIfNull(container);
+
+ container.RegisterConfigSection("workerManager", static () => new WorkerManagerConfig(), -50);
+
+ container.Register(Reuse.Singleton);
+ container.RegisterMapping();
+ container.Register(Reuse.Singleton);
+
+ container.RegisterStdService(100);
+ container.RegisterStdService(110);
+
+ if (!container.IsRegistered())
+ {
+ container.RegisterConfigSection("timerWheelPump", static () => new TimerWheelPumpConfig(), -90);
+ container.RegisterStdService(-1);
+ }
+
+ return container;
+ }
+}
diff --git a/src/SquidStd.Workers.Manager/Interfaces/IJobScheduler.cs b/src/SquidStd.Workers.Manager/Interfaces/IJobScheduler.cs
new file mode 100644
index 00000000..25efa737
--- /dev/null
+++ b/src/SquidStd.Workers.Manager/Interfaces/IJobScheduler.cs
@@ -0,0 +1,13 @@
+namespace SquidStd.Workers.Manager.Interfaces;
+
+///
+/// Enqueues jobs for workers to consume.
+///
+public interface IJobScheduler
+{
+ /// Enqueues a job with parameters.
+ Task EnqueueAsync(string jobName, IReadOnlyDictionary parameters, CancellationToken cancellationToken = default);
+
+ /// Enqueues a job with no parameters.
+ Task EnqueueAsync(string jobName, CancellationToken cancellationToken = default);
+}
diff --git a/src/SquidStd.Workers.Manager/Interfaces/IWorkerRegistry.cs b/src/SquidStd.Workers.Manager/Interfaces/IWorkerRegistry.cs
new file mode 100644
index 00000000..1859d6ca
--- /dev/null
+++ b/src/SquidStd.Workers.Manager/Interfaces/IWorkerRegistry.cs
@@ -0,0 +1,15 @@
+using SquidStd.Workers.Abstractions.Data;
+
+namespace SquidStd.Workers.Manager.Interfaces;
+
+///
+/// Read access to the manager's in-memory view of known workers.
+///
+public interface IWorkerRegistry
+{
+ /// Returns a snapshot of all known workers.
+ IReadOnlyCollection GetAll();
+
+ /// Returns the worker with the given id, or null when unknown.
+ WorkerInfo? Get(string workerId);
+}
diff --git a/src/SquidStd.Workers.Manager/README.md b/src/SquidStd.Workers.Manager/README.md
new file mode 100644
index 00000000..a1e33a74
--- /dev/null
+++ b/src/SquidStd.Workers.Manager/README.md
@@ -0,0 +1,50 @@
+
+
+
+
+SquidStd.Workers.Manager
+
+
+ 
+ 
+
+
+
+
+The SquidStd worker manager. Enqueues jobs, collects worker heartbeats into an in-memory registry, marks
+workers `Offline` on a periodic sweep, publishes status-transition events, and exposes an opt-in ASP.NET
+minimal-API surface.
+
+## Install
+
+```bash
+dotnet add package SquidStd.Workers.Manager
+```
+
+## Usage
+
+```csharp
+using DryIoc;
+using SquidStd.Workers.Manager.Extensions;
+
+container.AddInMemoryMessaging(); // or AddRabbitMqMessaging(...)
+container.AddWorkerManager();
+
+// In an ASP.NET app, map the opt-in HTTP surface:
+app.MapWorkerManagerEndpoints(); // GET /workers, GET /workers/{id}, POST /jobs
+```
+
+## Key types
+
+| Type | Purpose |
+|------|---------|
+| `IJobScheduler` | `EnqueueAsync(jobName, parameters)` onto the jobs queue. |
+| `IWorkerRegistry` | `GetAll()` / `Get(id)` over the live worker view. |
+| `WorkerStatusChangedEvent` | Published on the event bus on discover / offline / return. |
+| `WorkerManagerConfig` | `OfflineTimeoutSeconds`, `SweepIntervalSeconds`, queue/topic names. |
+| `WorkerManagerRegistrationExtensions` | `AddWorkerManager()`. |
+| `WorkerManagerEndpointsExtensions` | `MapWorkerManagerEndpoints()`. |
+
+## License
+
+MIT — part of [SquidStd](https://github.com/tgiachi/squid-std).
diff --git a/src/SquidStd.Workers.Manager/Services/HeartbeatCollectorService.cs b/src/SquidStd.Workers.Manager/Services/HeartbeatCollectorService.cs
new file mode 100644
index 00000000..91317e0a
--- /dev/null
+++ b/src/SquidStd.Workers.Manager/Services/HeartbeatCollectorService.cs
@@ -0,0 +1,65 @@
+using Serilog;
+using SquidStd.Abstractions.Interfaces.Services;
+using SquidStd.Core.Interfaces.Events;
+using SquidStd.Messaging.Abstractions.Interfaces;
+using SquidStd.Workers.Abstractions;
+using SquidStd.Workers.Abstractions.Data;
+using SquidStd.Workers.Manager.Data.Config;
+
+namespace SquidStd.Workers.Manager.Services;
+
+///
+/// Subscribes to the heartbeat topic and folds each into the registry,
+/// publishing any resulting status transition on the event bus.
+///
+public sealed class HeartbeatCollectorService : ISquidStdService
+{
+ private readonly ILogger _logger = Log.ForContext();
+ private readonly IMessageTopic _topic;
+ private readonly WorkerRegistry _registry;
+ private readonly IEventBus _eventBus;
+ private readonly string _topicName;
+ private IDisposable? _subscription;
+
+ public HeartbeatCollectorService(IMessageTopic topic, WorkerRegistry registry, IEventBus eventBus, WorkerManagerConfig config)
+ {
+ _topic = topic;
+ _registry = registry;
+ _eventBus = eventBus;
+ _topicName = string.IsNullOrWhiteSpace(config.HeartbeatTopic) ? WorkerChannels.HeartbeatTopic : config.HeartbeatTopic;
+ }
+
+ ///
+ public ValueTask StartAsync(CancellationToken cancellationToken = default)
+ {
+ _subscription = _topic.Subscribe(_topicName, OnHeartbeatAsync);
+ _logger.Information("Worker manager collecting heartbeats from topic '{Topic}'.", _topicName);
+
+ return ValueTask.CompletedTask;
+ }
+
+ ///
+ public ValueTask StopAsync(CancellationToken cancellationToken = default)
+ {
+ _subscription?.Dispose();
+ _subscription = null;
+
+ return ValueTask.CompletedTask;
+ }
+
+ private async Task OnHeartbeatAsync(WorkerHeartbeat heartbeat, CancellationToken cancellationToken)
+ {
+ try
+ {
+ var change = _registry.Record(heartbeat);
+ if (change is not null)
+ {
+ await _eventBus.PublishAsync(change, cancellationToken);
+ }
+ }
+ catch (Exception ex)
+ {
+ _logger.Error(ex, "Failed to process heartbeat from '{WorkerId}'.", heartbeat.WorkerId);
+ }
+ }
+}
diff --git a/src/SquidStd.Workers.Manager/Services/JobScheduler.cs b/src/SquidStd.Workers.Manager/Services/JobScheduler.cs
new file mode 100644
index 00000000..82336a01
--- /dev/null
+++ b/src/SquidStd.Workers.Manager/Services/JobScheduler.cs
@@ -0,0 +1,30 @@
+using SquidStd.Messaging.Abstractions.Interfaces;
+using SquidStd.Workers.Abstractions;
+using SquidStd.Workers.Abstractions.Data;
+using SquidStd.Workers.Manager.Data.Config;
+using SquidStd.Workers.Manager.Interfaces;
+
+namespace SquidStd.Workers.Manager.Services;
+
+///
+/// Default : publishes s onto the configured jobs queue.
+///
+public sealed class JobScheduler : IJobScheduler
+{
+ private readonly IMessageQueue _queue;
+ private readonly string _queueName;
+
+ public JobScheduler(IMessageQueue queue, WorkerManagerConfig config)
+ {
+ _queue = queue;
+ _queueName = string.IsNullOrWhiteSpace(config.JobQueue) ? WorkerChannels.JobQueue : config.JobQueue;
+ }
+
+ ///
+ public Task EnqueueAsync(string jobName, IReadOnlyDictionary parameters, CancellationToken cancellationToken = default)
+ => _queue.PublishAsync(_queueName, new JobRequest(jobName, parameters), cancellationToken);
+
+ ///
+ public Task EnqueueAsync(string jobName, CancellationToken cancellationToken = default)
+ => EnqueueAsync(jobName, new Dictionary(), cancellationToken);
+}
diff --git a/src/SquidStd.Workers.Manager/Services/WorkerOfflineSweepService.cs b/src/SquidStd.Workers.Manager/Services/WorkerOfflineSweepService.cs
new file mode 100644
index 00000000..34d63a8d
--- /dev/null
+++ b/src/SquidStd.Workers.Manager/Services/WorkerOfflineSweepService.cs
@@ -0,0 +1,80 @@
+using Serilog;
+using SquidStd.Abstractions.Interfaces.Services;
+using SquidStd.Core.Interfaces.Events;
+using SquidStd.Core.Interfaces.Timing;
+using SquidStd.Workers.Manager.Data.Config;
+
+namespace SquidStd.Workers.Manager.Services;
+
+///
+/// Periodically marks workers Offline whose heartbeats have stopped, using the timer wheel
+/// (), and publishes the resulting transitions.
+///
+public sealed class WorkerOfflineSweepService : ISquidStdService
+{
+ private const int DefaultIntervalSeconds = 10;
+
+ private readonly ILogger _logger = Log.ForContext();
+ private readonly ITimerService _timer;
+ private readonly WorkerRegistry _registry;
+ private readonly IEventBus _eventBus;
+ private readonly TimeSpan _interval;
+ private string? _timerId;
+
+ public WorkerOfflineSweepService(ITimerService timer, WorkerRegistry registry, IEventBus eventBus, WorkerManagerConfig config)
+ {
+ _timer = timer;
+ _registry = registry;
+ _eventBus = eventBus;
+
+ var seconds = config.SweepIntervalSeconds > 0 ? config.SweepIntervalSeconds : DefaultIntervalSeconds;
+ if (config.SweepIntervalSeconds <= 0)
+ {
+ _logger.Warning(
+ "SweepIntervalSeconds was {Value}; falling back to {Default}s.",
+ config.SweepIntervalSeconds,
+ DefaultIntervalSeconds);
+ }
+
+ _interval = TimeSpan.FromSeconds(seconds);
+ }
+
+ ///
+ public ValueTask StartAsync(CancellationToken cancellationToken = default)
+ {
+ _timerId = _timer.RegisterTimer("worker-offline-sweep", _interval, OnTick, delay: _interval, repeat: true);
+
+ return ValueTask.CompletedTask;
+ }
+
+ ///
+ public ValueTask StopAsync(CancellationToken cancellationToken = default)
+ {
+ if (_timerId is not null)
+ {
+ _timer.UnregisterTimer(_timerId);
+ _timerId = null;
+ }
+
+ return ValueTask.CompletedTask;
+ }
+
+ /// Runs one sweep and publishes the transitions. Public so it can be driven directly in tests.
+ public async Task RunSweepAsync()
+ {
+ try
+ {
+ foreach (var change in _registry.Sweep(DateTime.UtcNow))
+ {
+ await _eventBus.PublishAsync(change, CancellationToken.None);
+ }
+ }
+ catch (Exception ex)
+ {
+ _logger.Error(ex, "Worker offline sweep failed.");
+ }
+ }
+
+ private void OnTick()
+ => _ = RunSweepAsync();
+}
diff --git a/src/SquidStd.Workers.Manager/Services/WorkerRegistry.cs b/src/SquidStd.Workers.Manager/Services/WorkerRegistry.cs
new file mode 100644
index 00000000..d7804ae8
--- /dev/null
+++ b/src/SquidStd.Workers.Manager/Services/WorkerRegistry.cs
@@ -0,0 +1,105 @@
+using SquidStd.Workers.Abstractions.Data;
+using SquidStd.Workers.Abstractions.Types;
+using SquidStd.Workers.Manager.Data.Config;
+using SquidStd.Workers.Manager.Data.Events;
+using SquidStd.Workers.Manager.Interfaces;
+
+namespace SquidStd.Workers.Manager.Services;
+
+///
+/// In-memory registry of workers, folded from heartbeats. Pure: it returns status transitions for the
+/// caller to publish, and never touches the event bus or a real clock (the sweep takes the time as a parameter).
+///
+public sealed class WorkerRegistry : IWorkerRegistry
+{
+ private readonly Lock _sync = new();
+ private readonly Dictionary _workers = new(StringComparer.Ordinal);
+ private readonly TimeSpan _offlineTimeout;
+
+ public WorkerRegistry(WorkerManagerConfig config)
+ {
+ var seconds = config.OfflineTimeoutSeconds > 0 ? config.OfflineTimeoutSeconds : 30;
+ _offlineTimeout = TimeSpan.FromSeconds(seconds);
+ }
+
+ ///
+ public IReadOnlyCollection GetAll()
+ {
+ lock (_sync)
+ {
+ return _workers.Values.ToArray();
+ }
+ }
+
+ ///
+ public WorkerInfo? Get(string workerId)
+ {
+ lock (_sync)
+ {
+ return _workers.GetValueOrDefault(workerId);
+ }
+ }
+
+ ///
+ /// Folds a heartbeat into the registry. Returns a transition only when the worker is newly discovered
+ /// or has returned from ; otherwise null.
+ ///
+ public WorkerStatusChangedEvent? Record(WorkerHeartbeat heartbeat)
+ {
+ var now = DateTime.UtcNow;
+
+ lock (_sync)
+ {
+ if (!_workers.TryGetValue(heartbeat.WorkerId, out var existing))
+ {
+ _workers[heartbeat.WorkerId] = new WorkerInfo(
+ heartbeat.WorkerId,
+ heartbeat.Status,
+ heartbeat.ActiveJobs,
+ heartbeat.MaxConcurrency,
+ now,
+ now);
+
+ return new WorkerStatusChangedEvent(heartbeat.WorkerId, null, heartbeat.Status);
+ }
+
+ var old = existing.Status;
+ _workers[heartbeat.WorkerId] = existing with
+ {
+ Status = heartbeat.Status,
+ ActiveJobs = heartbeat.ActiveJobs,
+ MaxConcurrency = heartbeat.MaxConcurrency,
+ LastSeenUtc = now
+ };
+
+ return old == WorkerStatusType.Offline && heartbeat.Status != WorkerStatusType.Offline
+ ? new WorkerStatusChangedEvent(heartbeat.WorkerId, old, heartbeat.Status)
+ : null;
+ }
+ }
+
+ ///
+ /// Marks workers Offline whose last heartbeat is older than the configured timeout relative to
+ /// . Returns the resulting transitions.
+ ///
+ public IReadOnlyList Sweep(DateTime nowUtc)
+ {
+ var changes = new List();
+
+ lock (_sync)
+ {
+ foreach (var (id, info) in _workers.ToArray())
+ {
+ if (info.Status == WorkerStatusType.Offline || nowUtc - info.LastSeenUtc <= _offlineTimeout)
+ {
+ continue;
+ }
+
+ _workers[id] = info with { Status = WorkerStatusType.Offline };
+ changes.Add(new WorkerStatusChangedEvent(id, info.Status, WorkerStatusType.Offline));
+ }
+ }
+
+ return changes;
+ }
+}
diff --git a/src/SquidStd.Workers.Manager/SquidStd.Workers.Manager.csproj b/src/SquidStd.Workers.Manager/SquidStd.Workers.Manager.csproj
new file mode 100644
index 00000000..6b1b3dcc
--- /dev/null
+++ b/src/SquidStd.Workers.Manager/SquidStd.Workers.Manager.csproj
@@ -0,0 +1,20 @@
+
+
+
+ net10.0
+ enable
+ enable
+ true
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/src/SquidStd.Workers/Data/Config/WorkersConfig.cs b/src/SquidStd.Workers/Data/Config/WorkersConfig.cs
new file mode 100644
index 00000000..d9bf2b4f
--- /dev/null
+++ b/src/SquidStd.Workers/Data/Config/WorkersConfig.cs
@@ -0,0 +1,27 @@
+using SquidStd.Workers.Abstractions;
+
+namespace SquidStd.Workers.Data.Config;
+
+///
+/// Configuration for the worker runtime (config section "workers").
+///
+public sealed class WorkersConfig
+{
+ ///
+ /// Stable worker identity. When blank, the runtime falls back to the machine name
+ /// (in Docker, the container hostname).
+ ///
+ public string WorkerId { get; set; } = string.Empty;
+
+ /// Seconds between heartbeats. Falls back to 10 when not positive.
+ public int HeartbeatIntervalSeconds { get; set; } = 10;
+
+ /// Maximum jobs processed in parallel. Falls back to the processor count when not positive.
+ public int MaxConcurrency { get; set; }
+
+ /// Queue the worker consumes jobs from.
+ public string JobQueue { get; set; } = WorkerChannels.JobQueue;
+
+ /// Topic the worker publishes heartbeats to.
+ public string HeartbeatTopic { get; set; } = WorkerChannels.HeartbeatTopic;
+}
diff --git a/src/SquidStd.Workers/Exceptions/JobHandlerNotFoundException.cs b/src/SquidStd.Workers/Exceptions/JobHandlerNotFoundException.cs
new file mode 100644
index 00000000..61a4d92d
--- /dev/null
+++ b/src/SquidStd.Workers/Exceptions/JobHandlerNotFoundException.cs
@@ -0,0 +1,17 @@
+namespace SquidStd.Workers.Exceptions;
+
+///
+/// Thrown when a job arrives whose name has no registered .
+///
+public sealed class JobHandlerNotFoundException : Exception
+{
+ /// The job name that had no handler.
+ public string JobName { get; }
+
+ /// Initializes the exception for the given job name.
+ public JobHandlerNotFoundException(string jobName)
+ : base($"No job handler is registered for job '{jobName}'.")
+ {
+ JobName = jobName;
+ }
+}
diff --git a/src/SquidStd.Workers/Extensions/WorkersRegistrationExtensions.cs b/src/SquidStd.Workers/Extensions/WorkersRegistrationExtensions.cs
new file mode 100644
index 00000000..18eb37ba
--- /dev/null
+++ b/src/SquidStd.Workers/Extensions/WorkersRegistrationExtensions.cs
@@ -0,0 +1,46 @@
+using DryIoc;
+using SquidStd.Abstractions.Extensions.Config;
+using SquidStd.Abstractions.Extensions.Services;
+using SquidStd.Workers.Data.Config;
+using SquidStd.Workers.Interfaces;
+using SquidStd.Workers.Services;
+
+namespace SquidStd.Workers.Extensions;
+
+///
+/// DryIoc registration helpers for the worker runtime.
+///
+public static class WorkersRegistrationExtensions
+{
+ ///
+ /// Registers the worker runtime: the "workers" config section, shared state, job dispatcher, and the
+ /// consumer + heartbeat lifecycle services.
+ ///
+ public static IContainer AddWorkers(this IContainer container)
+ {
+ ArgumentNullException.ThrowIfNull(container);
+
+ container.RegisterConfigSection("workers", static () => new WorkersConfig(), -50);
+
+ container.Register(Reuse.Singleton);
+ container.Register(Reuse.Singleton);
+
+ container.RegisterStdService(100);
+ container.RegisterStdService(110);
+
+ return container;
+ }
+
+ ///
+ /// Registers a job handler so the dispatcher can route jobs to it by name.
+ ///
+ public static IContainer AddJobHandler(this IContainer container)
+ where THandler : class, IJobHandler
+ {
+ ArgumentNullException.ThrowIfNull(container);
+
+ container.Register(Reuse.Singleton);
+
+ return container;
+ }
+}
diff --git a/src/SquidStd.Workers/Interfaces/IJobDispatcher.cs b/src/SquidStd.Workers/Interfaces/IJobDispatcher.cs
new file mode 100644
index 00000000..fdae86b5
--- /dev/null
+++ b/src/SquidStd.Workers/Interfaces/IJobDispatcher.cs
@@ -0,0 +1,15 @@
+using SquidStd.Workers.Abstractions.Data;
+
+namespace SquidStd.Workers.Interfaces;
+
+///
+/// Routes a to the registered for its job name.
+///
+public interface IJobDispatcher
+{
+ ///
+ /// Dispatches the job to its handler.
+ ///
+ /// No handler matches the job name.
+ Task DispatchAsync(JobRequest job, CancellationToken cancellationToken);
+}
diff --git a/src/SquidStd.Workers/Interfaces/IJobHandler.cs b/src/SquidStd.Workers/Interfaces/IJobHandler.cs
new file mode 100644
index 00000000..d64a706e
--- /dev/null
+++ b/src/SquidStd.Workers/Interfaces/IJobHandler.cs
@@ -0,0 +1,15 @@
+using SquidStd.Workers.Abstractions.Data;
+
+namespace SquidStd.Workers.Interfaces;
+
+///
+/// Handles jobs of a single named kind. Implemented by consumers of the worker library.
+///
+public interface IJobHandler
+{
+ /// The this handler processes.
+ string JobName { get; }
+
+ /// Executes the job.
+ Task HandleAsync(JobRequest job, CancellationToken cancellationToken);
+}
diff --git a/src/SquidStd.Workers/Interfaces/IWorkerState.cs b/src/SquidStd.Workers/Interfaces/IWorkerState.cs
new file mode 100644
index 00000000..685e510c
--- /dev/null
+++ b/src/SquidStd.Workers/Interfaces/IWorkerState.cs
@@ -0,0 +1,27 @@
+using SquidStd.Workers.Abstractions.Types;
+
+namespace SquidStd.Workers.Interfaces;
+
+///
+/// Shared runtime state of a worker, read by the heartbeat service and mutated by the consumer.
+///
+public interface IWorkerState
+{
+ /// Stable worker identity.
+ string WorkerId { get; }
+
+ /// Maximum jobs processed in parallel.
+ int MaxConcurrency { get; }
+
+ /// Jobs currently in progress.
+ int ActiveJobs { get; }
+
+ /// while any job is active, otherwise .
+ WorkerStatusType Status { get; }
+
+ /// Records that a job started (increments ).
+ void JobStarted();
+
+ /// Records that a job finished (decrements ).
+ void JobFinished();
+}
diff --git a/src/SquidStd.Workers/README.md b/src/SquidStd.Workers/README.md
new file mode 100644
index 00000000..753a2a65
--- /dev/null
+++ b/src/SquidStd.Workers/README.md
@@ -0,0 +1,58 @@
+
+
+
+
+SquidStd.Workers
+
+
+ 
+ 
+
+
+
+
+The SquidStd worker runtime. Consumes `JobRequest`s from the jobs queue, dispatches each to a named
+`IJobHandler` (up to `MaxConcurrency` in parallel), and publishes a `WorkerHeartbeat` on the heartbeat topic
+every few seconds.
+
+## Install
+
+```bash
+dotnet add package SquidStd.Workers
+```
+
+## Usage
+
+```csharp
+using DryIoc;
+using SquidStd.Workers.Extensions;
+
+container.AddInMemoryMessaging(); // or AddRabbitMqMessaging(...)
+container.AddWorkers();
+container.AddJobHandler();
+```
+
+```csharp
+public sealed class ResizeImageHandler : IJobHandler
+{
+ public string JobName => "resize-image";
+
+ public Task HandleAsync(JobRequest job, CancellationToken cancellationToken)
+ {
+ // job.Parameters["width"], ...
+ return Task.CompletedTask;
+ }
+}
+```
+
+## Key types
+
+| Type | Purpose |
+|------|---------|
+| `IJobHandler` | Handles jobs of one named kind. |
+| `WorkersConfig` | `WorkerId`, `HeartbeatIntervalSeconds`, `MaxConcurrency`, queue/topic names. |
+| `WorkersRegistrationExtensions` | `AddWorkers()` and `AddJobHandler()`. |
+
+## License
+
+MIT — part of [SquidStd](https://github.com/tgiachi/squid-std).
diff --git a/src/SquidStd.Workers/Services/JobDispatcher.cs b/src/SquidStd.Workers/Services/JobDispatcher.cs
new file mode 100644
index 00000000..9d481073
--- /dev/null
+++ b/src/SquidStd.Workers/Services/JobDispatcher.cs
@@ -0,0 +1,41 @@
+using Serilog;
+using SquidStd.Workers.Abstractions.Data;
+using SquidStd.Workers.Exceptions;
+using SquidStd.Workers.Interfaces;
+
+namespace SquidStd.Workers.Services;
+
+///
+/// Default : indexes the registered handlers by job name.
+///
+public sealed class JobDispatcher : IJobDispatcher
+{
+ private readonly ILogger _logger = Log.ForContext();
+ private readonly Dictionary _handlers = new(StringComparer.Ordinal);
+
+ public JobDispatcher(IEnumerable handlers)
+ {
+ foreach (var handler in handlers)
+ {
+ if (_handlers.ContainsKey(handler.JobName))
+ {
+ _logger.Warning(
+ "Duplicate job handler for '{JobName}'; the last registration wins.",
+ handler.JobName);
+ }
+
+ _handlers[handler.JobName] = handler;
+ }
+ }
+
+ ///
+ public Task DispatchAsync(JobRequest job, CancellationToken cancellationToken)
+ {
+ if (!_handlers.TryGetValue(job.JobName, out var handler))
+ {
+ throw new JobHandlerNotFoundException(job.JobName);
+ }
+
+ return handler.HandleAsync(job, cancellationToken);
+ }
+}
diff --git a/src/SquidStd.Workers/Services/WorkerConsumerService.cs b/src/SquidStd.Workers/Services/WorkerConsumerService.cs
new file mode 100644
index 00000000..e598cde5
--- /dev/null
+++ b/src/SquidStd.Workers/Services/WorkerConsumerService.cs
@@ -0,0 +1,86 @@
+using Serilog;
+using SquidStd.Abstractions.Interfaces.Services;
+using SquidStd.Messaging.Abstractions.Interfaces;
+using SquidStd.Workers.Abstractions;
+using SquidStd.Workers.Abstractions.Data;
+using SquidStd.Workers.Data.Config;
+using SquidStd.Workers.Exceptions;
+using SquidStd.Workers.Interfaces;
+
+namespace SquidStd.Workers.Services;
+
+///
+/// Subscribes to the jobs queue and dispatches each to its handler,
+/// bounded by .
+///
+public sealed class WorkerConsumerService : ISquidStdService, IQueueMessageListenerAsync
+{
+ private readonly ILogger _logger = Log.ForContext();
+ private readonly IMessageQueue _queue;
+ private readonly IJobDispatcher _dispatcher;
+ private readonly IWorkerState _state;
+ private readonly SemaphoreSlim _semaphore;
+ private readonly string _queueName;
+ private IDisposable? _subscription;
+
+ /// Constructs the consumer wired to the messaging queue and worker state.
+ public WorkerConsumerService(IMessageQueue queue, IJobDispatcher dispatcher, IWorkerState state, WorkersConfig config)
+ {
+ _queue = queue;
+ _dispatcher = dispatcher;
+ _state = state;
+ _semaphore = new SemaphoreSlim(state.MaxConcurrency);
+ _queueName = string.IsNullOrWhiteSpace(config.JobQueue) ? WorkerChannels.JobQueue : config.JobQueue;
+ }
+
+ ///
+ public ValueTask StartAsync(CancellationToken cancellationToken = default)
+ {
+ _subscription = _queue.Subscribe(_queueName, this);
+ _logger.Information("Worker consuming jobs from queue '{Queue}'.", _queueName);
+
+ return ValueTask.CompletedTask;
+ }
+
+ ///
+ public ValueTask StopAsync(CancellationToken cancellationToken = default)
+ {
+ _subscription?.Dispose();
+ _subscription = null;
+
+ return ValueTask.CompletedTask;
+ }
+
+ ///
+ public async Task HandleAsync(JobRequest message, CancellationToken cancellationToken)
+ {
+ await _semaphore.WaitAsync(cancellationToken);
+ _state.JobStarted();
+
+ try
+ {
+ await _dispatcher.DispatchAsync(message, cancellationToken);
+ }
+ catch (JobHandlerNotFoundException)
+ {
+ _logger.Warning("No handler for job '{JobName}'; dropping.", message.JobName);
+ }
+ catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+ {
+ _logger.Debug("Job '{JobName}' cancelled during shutdown; will be redelivered.", message.JobName);
+
+ throw;
+ }
+ catch (Exception ex)
+ {
+ _logger.Error(ex, "Job '{JobName}' failed; it will be retried.", message.JobName);
+
+ throw;
+ }
+ finally
+ {
+ _state.JobFinished();
+ _semaphore.Release();
+ }
+ }
+}
diff --git a/src/SquidStd.Workers/Services/WorkerHeartbeatService.cs b/src/SquidStd.Workers/Services/WorkerHeartbeatService.cs
new file mode 100644
index 00000000..c5fa548a
--- /dev/null
+++ b/src/SquidStd.Workers/Services/WorkerHeartbeatService.cs
@@ -0,0 +1,117 @@
+using Serilog;
+using SquidStd.Abstractions.Interfaces.Services;
+using SquidStd.Messaging.Abstractions.Interfaces;
+using SquidStd.Workers.Abstractions;
+using SquidStd.Workers.Abstractions.Data;
+using SquidStd.Workers.Data.Config;
+using SquidStd.Workers.Interfaces;
+
+namespace SquidStd.Workers.Services;
+
+///
+/// Publishes a on the heartbeat topic immediately on start and then once
+/// per configured interval.
+///
+public sealed class WorkerHeartbeatService : ISquidStdService
+{
+ private const int DefaultIntervalSeconds = 10;
+
+ private readonly ILogger _logger = Log.ForContext();
+ private readonly IMessageTopic _topic;
+ private readonly IWorkerState _state;
+ private readonly TimeSpan _interval;
+ private readonly string _topicName;
+ private CancellationTokenSource? _loopCts;
+ private Task? _loopTask;
+
+ public WorkerHeartbeatService(IMessageTopic topic, IWorkerState state, WorkersConfig config)
+ {
+ _topic = topic;
+ _state = state;
+
+ var seconds = config.HeartbeatIntervalSeconds > 0 ? config.HeartbeatIntervalSeconds : DefaultIntervalSeconds;
+ if (config.HeartbeatIntervalSeconds <= 0)
+ {
+ _logger.Warning(
+ "HeartbeatIntervalSeconds was {Value}; falling back to {Default}s.",
+ config.HeartbeatIntervalSeconds,
+ DefaultIntervalSeconds);
+ }
+
+ _interval = TimeSpan.FromSeconds(seconds);
+ _topicName = string.IsNullOrWhiteSpace(config.HeartbeatTopic) ? WorkerChannels.HeartbeatTopic : config.HeartbeatTopic;
+ }
+
+ ///
+ public ValueTask StartAsync(CancellationToken cancellationToken = default)
+ {
+ _loopCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+ _loopTask = Task.Run(() => RunLoopAsync(_loopCts.Token), CancellationToken.None);
+
+ return ValueTask.CompletedTask;
+ }
+
+ ///
+ public async ValueTask StopAsync(CancellationToken cancellationToken = default)
+ {
+ if (_loopCts is null)
+ {
+ return;
+ }
+
+ await _loopCts.CancelAsync();
+
+ try
+ {
+ if (_loopTask is not null)
+ {
+ await _loopTask;
+ }
+ }
+ catch (OperationCanceledException)
+ {
+ // Expected on shutdown.
+ }
+ finally
+ {
+ _loopCts.Dispose();
+ _loopCts = null;
+ _loopTask = null;
+ }
+ }
+
+ private async Task RunLoopAsync(CancellationToken cancellationToken)
+ {
+ using var timer = new PeriodicTimer(_interval);
+
+ await PublishAsync(cancellationToken);
+
+ while (await timer.WaitForNextTickAsync(cancellationToken))
+ {
+ await PublishAsync(cancellationToken);
+ }
+ }
+
+ private async Task PublishAsync(CancellationToken cancellationToken)
+ {
+ try
+ {
+ var heartbeat = new WorkerHeartbeat(
+ _state.WorkerId,
+ DateTime.UtcNow,
+ _state.Status,
+ _state.ActiveJobs,
+ _state.MaxConcurrency);
+
+ await _topic.PublishAsync(_topicName, heartbeat, cancellationToken);
+ }
+ catch (OperationCanceledException)
+ {
+ throw;
+ }
+ catch (Exception ex)
+ {
+ _logger.Error(ex, "Failed to publish heartbeat; will retry on the next tick.");
+ }
+ }
+}
diff --git a/src/SquidStd.Workers/Services/WorkerState.cs b/src/SquidStd.Workers/Services/WorkerState.cs
new file mode 100644
index 00000000..8b96046b
--- /dev/null
+++ b/src/SquidStd.Workers/Services/WorkerState.cs
@@ -0,0 +1,40 @@
+using SquidStd.Workers.Abstractions.Types;
+using SquidStd.Workers.Data.Config;
+using SquidStd.Workers.Interfaces;
+
+namespace SquidStd.Workers.Services;
+
+///
+/// Default backed by an interlocked counter; resolves identity and
+/// concurrency from at construction.
+///
+public sealed class WorkerState : IWorkerState
+{
+ private int _activeJobs;
+
+ ///
+ public string WorkerId { get; }
+
+ ///
+ public int MaxConcurrency { get; }
+
+ public WorkerState(WorkersConfig config)
+ {
+ WorkerId = string.IsNullOrWhiteSpace(config.WorkerId) ? Environment.MachineName : config.WorkerId;
+ MaxConcurrency = config.MaxConcurrency > 0 ? config.MaxConcurrency : Environment.ProcessorCount;
+ }
+
+ ///
+ public int ActiveJobs => Volatile.Read(ref _activeJobs);
+
+ ///
+ public WorkerStatusType Status => ActiveJobs == 0 ? WorkerStatusType.Idle : WorkerStatusType.Busy;
+
+ ///
+ public void JobStarted()
+ => Interlocked.Increment(ref _activeJobs);
+
+ ///
+ public void JobFinished()
+ => Interlocked.Decrement(ref _activeJobs);
+}
diff --git a/src/SquidStd.Workers/SquidStd.Workers.csproj b/src/SquidStd.Workers/SquidStd.Workers.csproj
new file mode 100644
index 00000000..ff5daeb4
--- /dev/null
+++ b/src/SquidStd.Workers/SquidStd.Workers.csproj
@@ -0,0 +1,15 @@
+
+
+
+ net10.0
+ enable
+ enable
+ true
+
+
+
+
+
+
+
+
diff --git a/tests/SquidStd.Tests/Integration/Workers/WorkerSystemIntegrationTests.cs b/tests/SquidStd.Tests/Integration/Workers/WorkerSystemIntegrationTests.cs
new file mode 100644
index 00000000..d02ed74e
--- /dev/null
+++ b/tests/SquidStd.Tests/Integration/Workers/WorkerSystemIntegrationTests.cs
@@ -0,0 +1,151 @@
+using DryIoc;
+using SquidStd.Core.Interfaces.Events;
+using SquidStd.Messaging.Abstractions.Interfaces;
+using SquidStd.Messaging.RabbitMq.Data.Config;
+using SquidStd.Messaging.RabbitMq.Extensions;
+using SquidStd.Services.Core.Services;
+using SquidStd.Tests.Messaging.RabbitMq;
+using SquidStd.Workers.Abstractions.Data;
+using SquidStd.Workers.Data.Config;
+using SquidStd.Workers.Interfaces;
+using SquidStd.Workers.Services;
+using SquidStd.Workers.Manager.Data.Config;
+using SquidStd.Workers.Manager.Services;
+
+namespace SquidStd.Tests.Integration.Workers;
+
+///
+/// End-to-end test of the workers system over a real RabbitMQ broker (Testcontainers): the manager
+/// enqueues a job that the worker consumes and runs (real queue), and the worker's heartbeat reaches the
+/// manager's registry (real fan-out topic).
+///
+[Collection(RabbitMqCollection.Name)]
+public class WorkerSystemIntegrationTests
+{
+ private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(30);
+
+ private readonly RabbitMqContainerFixture _fixture;
+
+ public WorkerSystemIntegrationTests(RabbitMqContainerFixture fixture)
+ {
+ _fixture = fixture;
+ }
+
+ [Fact]
+ public async Task Manager_EnqueuesJob_WorkerRunsIt_AndHeartbeatReachesRegistry()
+ {
+ using var container = new Container();
+ container.RegisterInstance(new EventBusService());
+ container.AddRabbitMqMessaging(new RabbitMqOptions { Uri = new Uri(_fixture.AmqpUri) });
+
+ // Unique channel names per run so parallel/other tests on the shared broker do not interfere.
+ var suffix = Guid.NewGuid().ToString("N");
+ var jobQueue = $"it.jobs.{suffix}";
+ var heartbeatTopic = $"it.hb.{suffix}";
+
+ var workersConfig = new WorkersConfig
+ {
+ WorkerId = "it-worker",
+ MaxConcurrency = 2,
+ HeartbeatIntervalSeconds = 1,
+ JobQueue = jobQueue,
+ HeartbeatTopic = heartbeatTopic
+ };
+ var managerConfig = new WorkerManagerConfig
+ {
+ JobQueue = jobQueue,
+ HeartbeatTopic = heartbeatTopic
+ };
+
+ // Open the RabbitMQ connections.
+ var queueProvider = container.Resolve();
+ var topicProvider = container.Resolve();
+ await queueProvider.StartAsync();
+ await topicProvider.StartAsync();
+
+ var queue = container.Resolve();
+ var topic = container.Resolve();
+ var eventBus = container.Resolve();
+
+ // Worker side.
+ var jobRan = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var handler = new CapturingJobHandler("greet", jobRan);
+ var workerState = new WorkerState(workersConfig);
+ var dispatcher = new JobDispatcher([handler]);
+ var consumer = new WorkerConsumerService(queue, dispatcher, workerState, workersConfig);
+ var heartbeatService = new WorkerHeartbeatService(topic, workerState, workersConfig);
+
+ // Manager side.
+ var registry = new WorkerRegistry(managerConfig);
+ var collector = new HeartbeatCollectorService(topic, registry, eventBus, managerConfig);
+ var scheduler = new JobScheduler(queue, managerConfig);
+
+ try
+ {
+ await collector.StartAsync();
+ await consumer.StartAsync();
+ await heartbeatService.StartAsync();
+
+ // Let the broker register the queue consumer and the topic binding before publishing.
+ await Task.Delay(500);
+
+ await scheduler.EnqueueAsync("greet", new Dictionary { ["name"] = "squid" });
+
+ // The worker consumed the job from the real queue and ran the handler.
+ var ranJob = await jobRan.Task.WaitAsync(Timeout);
+ Assert.Equal("greet", ranJob.JobName);
+ Assert.Equal("squid", ranJob.Parameters["name"]);
+
+ // The worker's heartbeat reached the manager's registry over the real topic.
+ var seen = await PollForWorker(registry, "it-worker");
+ Assert.NotNull(seen);
+ Assert.Equal("it-worker", seen!.WorkerId);
+ Assert.Equal(2, seen.MaxConcurrency);
+ }
+ finally
+ {
+ await heartbeatService.StopAsync();
+ await consumer.StopAsync();
+ await collector.StopAsync();
+ await topicProvider.StopAsync();
+ await queueProvider.StopAsync();
+ }
+ }
+
+ private static async Task PollForWorker(WorkerRegistry registry, string workerId)
+ {
+ var deadline = DateTime.UtcNow + Timeout;
+ while (DateTime.UtcNow < deadline)
+ {
+ var info = registry.Get(workerId);
+ if (info is not null)
+ {
+ return info;
+ }
+
+ await Task.Delay(200);
+ }
+
+ return null;
+ }
+
+ private sealed class CapturingJobHandler : IJobHandler
+ {
+ private readonly TaskCompletionSource _completion;
+
+ public CapturingJobHandler(string jobName, TaskCompletionSource completion)
+ {
+ JobName = jobName;
+ _completion = completion;
+ }
+
+ public string JobName { get; }
+
+ public Task HandleAsync(JobRequest job, CancellationToken cancellationToken)
+ {
+ _completion.TrySetResult(job);
+
+ return Task.CompletedTask;
+ }
+ }
+}
diff --git a/tests/SquidStd.Tests/Manager/HeartbeatCollectorServiceTests.cs b/tests/SquidStd.Tests/Manager/HeartbeatCollectorServiceTests.cs
new file mode 100644
index 00000000..736e52c1
--- /dev/null
+++ b/tests/SquidStd.Tests/Manager/HeartbeatCollectorServiceTests.cs
@@ -0,0 +1,62 @@
+using DryIoc;
+using SquidStd.Core.Interfaces.Events;
+using SquidStd.Messaging.Abstractions.Interfaces;
+using SquidStd.Messaging.Extensions;
+using SquidStd.Services.Core.Services;
+using SquidStd.Workers.Abstractions;
+using SquidStd.Workers.Abstractions.Data;
+using SquidStd.Workers.Abstractions.Types;
+using SquidStd.Workers.Manager.Data.Config;
+using SquidStd.Workers.Manager.Data.Events;
+using SquidStd.Workers.Manager.Services;
+
+namespace SquidStd.Tests.Manager;
+
+public class HeartbeatCollectorServiceTests
+{
+ [Fact]
+ public async Task Collector_RecordsHeartbeat_AndPublishesDiscoveredEvent()
+ {
+ var container = new Container();
+ var eventBus = new EventBusService();
+ container.RegisterInstance(eventBus);
+ container.AddInMemoryMessaging();
+
+ var topic = container.Resolve();
+ var registry = new WorkerRegistry(new WorkerManagerConfig());
+
+ var discovered = new TaskCompletionSource();
+ eventBus.RegisterAsyncListener(new DelegateListener(e => discovered.TrySetResult(e)));
+
+ var service = new HeartbeatCollectorService(topic, registry, eventBus, new WorkerManagerConfig());
+ await service.StartAsync();
+
+ await topic.PublishAsync(
+ WorkerChannels.HeartbeatTopic,
+ new WorkerHeartbeat("w1", DateTime.UtcNow, WorkerStatusType.Idle, 0, 8));
+
+ var change = await discovered.Task.WaitAsync(TimeSpan.FromSeconds(5));
+ await service.StopAsync();
+
+ Assert.Equal("w1", change.WorkerId);
+ Assert.Null(change.OldStatus);
+ Assert.NotNull(registry.Get("w1"));
+ }
+
+ private sealed class DelegateListener : IAsyncEventListener
+ {
+ private readonly Action _onEvent;
+
+ public DelegateListener(Action onEvent)
+ {
+ _onEvent = onEvent;
+ }
+
+ public Task HandleAsync(WorkerStatusChangedEvent eventData, CancellationToken cancellationToken)
+ {
+ _onEvent(eventData);
+
+ return Task.CompletedTask;
+ }
+ }
+}
diff --git a/tests/SquidStd.Tests/Manager/JobSchedulerTests.cs b/tests/SquidStd.Tests/Manager/JobSchedulerTests.cs
new file mode 100644
index 00000000..7da319c6
--- /dev/null
+++ b/tests/SquidStd.Tests/Manager/JobSchedulerTests.cs
@@ -0,0 +1,52 @@
+using DryIoc;
+using SquidStd.Core.Interfaces.Events;
+using SquidStd.Messaging.Abstractions.Interfaces;
+using SquidStd.Messaging.Extensions;
+using SquidStd.Services.Core.Services;
+using SquidStd.Workers.Abstractions;
+using SquidStd.Workers.Abstractions.Data;
+using SquidStd.Workers.Manager.Data.Config;
+using SquidStd.Workers.Manager.Services;
+
+namespace SquidStd.Tests.Manager;
+
+public class JobSchedulerTests
+{
+ [Fact]
+ public async Task EnqueueAsync_PublishesJobRequestToConfiguredQueue()
+ {
+ var container = new Container();
+ container.RegisterInstance(new EventBusService());
+ container.AddInMemoryMessaging();
+ var queue = container.Resolve();
+
+ var received = new TaskCompletionSource();
+ using var _ = queue.Subscribe(
+ WorkerChannels.JobQueue,
+ new DelegateListener(job => received.TrySetResult(job)));
+
+ var scheduler = new JobScheduler(queue, new WorkerManagerConfig());
+ await scheduler.EnqueueAsync("resize", new Dictionary { ["w"] = "100" });
+
+ var job = await received.Task.WaitAsync(TimeSpan.FromSeconds(5));
+ Assert.Equal("resize", job.JobName);
+ Assert.Equal("100", job.Parameters["w"]);
+ }
+
+ private sealed class DelegateListener : IQueueMessageListenerAsync
+ {
+ private readonly Action _onMessage;
+
+ public DelegateListener(Action onMessage)
+ {
+ _onMessage = onMessage;
+ }
+
+ public Task HandleAsync(JobRequest message, CancellationToken cancellationToken)
+ {
+ _onMessage(message);
+
+ return Task.CompletedTask;
+ }
+ }
+}
diff --git a/tests/SquidStd.Tests/Manager/Support/FakeTimerService.cs b/tests/SquidStd.Tests/Manager/Support/FakeTimerService.cs
new file mode 100644
index 00000000..03857243
--- /dev/null
+++ b/tests/SquidStd.Tests/Manager/Support/FakeTimerService.cs
@@ -0,0 +1,41 @@
+using SquidStd.Core.Interfaces.Timing;
+
+namespace SquidStd.Tests.Manager.Support;
+
+/// Captures timer registration for tests without running a real timer wheel.
+public sealed class FakeTimerService : ITimerService
+{
+ public string? RegisteredName { get; private set; }
+
+ public TimeSpan RegisteredInterval { get; private set; }
+
+ public bool RegisteredRepeat { get; private set; }
+
+ public string? UnregisteredId { get; private set; }
+
+ public string RegisterTimer(string name, TimeSpan interval, Action callback, TimeSpan? delay = null, bool repeat = false)
+ {
+ RegisteredName = name;
+ RegisteredInterval = interval;
+ RegisteredRepeat = repeat;
+
+ return "timer-1";
+ }
+
+ public void UnregisterAllTimers()
+ {
+ }
+
+ public bool UnregisterTimer(string timerId)
+ {
+ UnregisteredId = timerId;
+
+ return true;
+ }
+
+ public int UnregisterTimersByName(string name)
+ => 0;
+
+ public int UpdateTicksDelta(long timestampMilliseconds)
+ => 0;
+}
diff --git a/tests/SquidStd.Tests/Manager/WorkerManagerEndpointsTests.cs b/tests/SquidStd.Tests/Manager/WorkerManagerEndpointsTests.cs
new file mode 100644
index 00000000..589e2324
--- /dev/null
+++ b/tests/SquidStd.Tests/Manager/WorkerManagerEndpointsTests.cs
@@ -0,0 +1,83 @@
+using DryIoc;
+using Microsoft.AspNetCore.Http.HttpResults;
+using SquidStd.Core.Interfaces.Events;
+using SquidStd.Messaging.Abstractions.Interfaces;
+using SquidStd.Messaging.Extensions;
+using SquidStd.Services.Core.Services;
+using SquidStd.Workers.Abstractions.Data;
+using SquidStd.Workers.Abstractions.Types;
+using SquidStd.Workers.Manager.Data;
+using SquidStd.Workers.Manager.Data.Config;
+using SquidStd.Workers.Manager.Endpoints;
+using SquidStd.Workers.Manager.Services;
+
+namespace SquidStd.Tests.Manager;
+
+public class WorkerManagerEndpointsTests
+{
+ private static WorkerRegistry RegistryWith(params string[] workerIds)
+ {
+ var registry = new WorkerRegistry(new WorkerManagerConfig());
+ foreach (var id in workerIds)
+ {
+ registry.Record(new WorkerHeartbeat(id, DateTime.UtcNow, WorkerStatusType.Idle, 0, 8));
+ }
+
+ return registry;
+ }
+
+ private static JobScheduler NewScheduler()
+ {
+ var container = new Container();
+ container.RegisterInstance(new EventBusService());
+ container.AddInMemoryMessaging();
+
+ return new JobScheduler(container.Resolve(), new WorkerManagerConfig());
+ }
+
+ [Fact]
+ public void GetWorkers_ReturnsOkWithAll()
+ {
+ var result = WorkerManagerEndpoints.GetWorkers(RegistryWith("w1", "w2"));
+
+ Assert.Equal(2, result.Value!.Count);
+ }
+
+ [Fact]
+ public void GetWorker_ReturnsOk_WhenPresent()
+ {
+ var result = WorkerManagerEndpoints.GetWorker("w1", RegistryWith("w1"));
+
+ Assert.IsType>(result.Result);
+ }
+
+ [Fact]
+ public void GetWorker_ReturnsNotFound_WhenAbsent()
+ {
+ var result = WorkerManagerEndpoints.GetWorker("missing", RegistryWith("w1"));
+
+ Assert.IsType(result.Result);
+ }
+
+ [Fact]
+ public async Task EnqueueJob_ReturnsAccepted_AndSchedulesJob()
+ {
+ var result = await WorkerManagerEndpoints.EnqueueJob(
+ new EnqueueJobRequest("resize", new Dictionary()),
+ NewScheduler(),
+ CancellationToken.None);
+
+ Assert.IsType(result.Result);
+ }
+
+ [Fact]
+ public async Task EnqueueJob_ReturnsBadRequest_WhenJobNameBlank()
+ {
+ var result = await WorkerManagerEndpoints.EnqueueJob(
+ new EnqueueJobRequest(" ", null),
+ NewScheduler(),
+ CancellationToken.None);
+
+ Assert.IsType>(result.Result);
+ }
+}
diff --git a/tests/SquidStd.Tests/Manager/WorkerManagerRegistrationExtensionsTests.cs b/tests/SquidStd.Tests/Manager/WorkerManagerRegistrationExtensionsTests.cs
new file mode 100644
index 00000000..56c6c7c2
--- /dev/null
+++ b/tests/SquidStd.Tests/Manager/WorkerManagerRegistrationExtensionsTests.cs
@@ -0,0 +1,62 @@
+using DryIoc;
+using SquidStd.Core.Interfaces.Events;
+using SquidStd.Core.Interfaces.Timing;
+using SquidStd.Messaging.Extensions;
+using SquidStd.Services.Core.Services;
+using SquidStd.Services.Core.Services.Scheduling;
+using SquidStd.Tests.Manager.Support;
+using SquidStd.Workers.Manager.Data.Config;
+using SquidStd.Workers.Manager.Extensions;
+using SquidStd.Workers.Manager.Interfaces;
+using SquidStd.Workers.Manager.Services;
+
+namespace SquidStd.Tests.Manager;
+
+public class WorkerManagerRegistrationExtensionsTests
+{
+ private static Container NewContainer()
+ {
+ var container = new Container();
+ container.RegisterInstance(new EventBusService());
+ container.AddInMemoryMessaging();
+ // RegisterCoreServices normally provides ITimerService; supply a fake so the sweep service resolves.
+ container.RegisterInstance(new FakeTimerService());
+ // ConfigManager normally registers config instances; register one directly for the test.
+ container.RegisterInstance(new WorkerManagerConfig());
+
+ return container;
+ }
+
+ [Fact]
+ public void AddWorkerManager_RegistersResolvableServices()
+ {
+ using var container = NewContainer();
+
+ container.AddWorkerManager();
+
+ Assert.NotNull(container.Resolve());
+ Assert.NotNull(container.Resolve());
+ Assert.NotNull(container.Resolve());
+ Assert.NotNull(container.Resolve());
+ }
+
+ [Fact]
+ public void AddWorkerManager_RegistersTimerWheelPump_WhenAbsent()
+ {
+ using var container = NewContainer();
+
+ container.AddWorkerManager();
+
+ Assert.True(container.IsRegistered());
+ }
+
+ [Fact]
+ public void AddWorkerManager_SharesRegistryInstanceAcrossInterfaceAndConcrete()
+ {
+ using var container = NewContainer();
+
+ container.AddWorkerManager();
+
+ Assert.Same(container.Resolve(), container.Resolve());
+ }
+}
diff --git a/tests/SquidStd.Tests/Manager/WorkerOfflineSweepServiceTests.cs b/tests/SquidStd.Tests/Manager/WorkerOfflineSweepServiceTests.cs
new file mode 100644
index 00000000..d1a0dfec
--- /dev/null
+++ b/tests/SquidStd.Tests/Manager/WorkerOfflineSweepServiceTests.cs
@@ -0,0 +1,67 @@
+using SquidStd.Services.Core.Services;
+using SquidStd.Tests.Manager.Support;
+using SquidStd.Workers.Abstractions.Data;
+using SquidStd.Workers.Abstractions.Types;
+using SquidStd.Workers.Manager.Data.Config;
+using SquidStd.Workers.Manager.Data.Events;
+using SquidStd.Workers.Manager.Services;
+using SquidStd.Core.Interfaces.Events;
+
+namespace SquidStd.Tests.Manager;
+
+public class WorkerOfflineSweepServiceTests
+{
+ [Fact]
+ public async Task StartAsync_RegistersRepeatingTimer_StopAsyncUnregisters()
+ {
+ var timer = new FakeTimerService();
+ var registry = new WorkerRegistry(new WorkerManagerConfig());
+ var service = new WorkerOfflineSweepService(timer, registry, new EventBusService(), new WorkerManagerConfig { SweepIntervalSeconds = 5 });
+
+ await service.StartAsync();
+
+ Assert.Equal("worker-offline-sweep", timer.RegisteredName);
+ Assert.Equal(TimeSpan.FromSeconds(5), timer.RegisteredInterval);
+ Assert.True(timer.RegisteredRepeat);
+
+ await service.StopAsync();
+ Assert.Equal("timer-1", timer.UnregisteredId);
+ }
+
+ [Fact]
+ public async Task RunSweepAsync_MarksOverdueWorkerOffline_AndPublishesTransition()
+ {
+ var registry = new WorkerRegistry(new WorkerManagerConfig { OfflineTimeoutSeconds = 1 });
+ registry.Record(new WorkerHeartbeat("w1", DateTime.UtcNow, WorkerStatusType.Idle, 0, 8));
+ await Task.Delay(1100);
+
+ var eventBus = new EventBusService();
+ var offline = new TaskCompletionSource();
+ eventBus.RegisterAsyncListener(new DelegateListener(e => offline.TrySetResult(e)));
+
+ var service = new WorkerOfflineSweepService(new FakeTimerService(), registry, eventBus, new WorkerManagerConfig());
+ await service.RunSweepAsync();
+
+ var change = await offline.Task.WaitAsync(TimeSpan.FromSeconds(5));
+ Assert.Equal("w1", change.WorkerId);
+ Assert.Equal(WorkerStatusType.Offline, change.NewStatus);
+ Assert.Equal(WorkerStatusType.Offline, registry.Get("w1")!.Status);
+ }
+
+ private sealed class DelegateListener : IAsyncEventListener
+ {
+ private readonly Action _onEvent;
+
+ public DelegateListener(Action onEvent)
+ {
+ _onEvent = onEvent;
+ }
+
+ public Task HandleAsync(WorkerStatusChangedEvent eventData, CancellationToken cancellationToken)
+ {
+ _onEvent(eventData);
+
+ return Task.CompletedTask;
+ }
+ }
+}
diff --git a/tests/SquidStd.Tests/Manager/WorkerRegistryTests.cs b/tests/SquidStd.Tests/Manager/WorkerRegistryTests.cs
new file mode 100644
index 00000000..1cf24a7d
--- /dev/null
+++ b/tests/SquidStd.Tests/Manager/WorkerRegistryTests.cs
@@ -0,0 +1,97 @@
+using SquidStd.Workers.Abstractions.Data;
+using SquidStd.Workers.Abstractions.Types;
+using SquidStd.Workers.Manager.Data.Config;
+using SquidStd.Workers.Manager.Services;
+
+namespace SquidStd.Tests.Manager;
+
+public class WorkerRegistryTests
+{
+ private static WorkerRegistry NewRegistry(int offlineTimeoutSeconds = 30)
+ => new(new WorkerManagerConfig { OfflineTimeoutSeconds = offlineTimeoutSeconds });
+
+ private static WorkerHeartbeat Heartbeat(string id, WorkerStatusType status, int activeJobs = 0)
+ => new(id, DateTime.UtcNow, status, activeJobs, 8);
+
+ [Fact]
+ public void Record_NewWorker_ReturnsDiscoveredTransitionAndStores()
+ {
+ var registry = NewRegistry();
+
+ var change = registry.Record(Heartbeat("w1", WorkerStatusType.Idle));
+
+ Assert.NotNull(change);
+ Assert.Equal("w1", change!.WorkerId);
+ Assert.Null(change.OldStatus);
+ Assert.Equal(WorkerStatusType.Idle, change.NewStatus);
+
+ var info = registry.Get("w1");
+ Assert.NotNull(info);
+ Assert.Equal(WorkerStatusType.Idle, info!.Status);
+ Assert.Equal(8, info.MaxConcurrency);
+ }
+
+ [Fact]
+ public void Record_ExistingWorker_IdleToBusy_NoTransitionButStateUpdated()
+ {
+ var registry = NewRegistry();
+ registry.Record(Heartbeat("w1", WorkerStatusType.Idle));
+
+ var change = registry.Record(Heartbeat("w1", WorkerStatusType.Busy, activeJobs: 2));
+
+ Assert.Null(change);
+ Assert.Equal(WorkerStatusType.Busy, registry.Get("w1")!.Status);
+ Assert.Equal(2, registry.Get("w1")!.ActiveJobs);
+ }
+
+ [Fact]
+ public void Sweep_MarksOverdueWorkerOffline_AndReturnsTransition()
+ {
+ var registry = NewRegistry(offlineTimeoutSeconds: 30);
+ registry.Record(Heartbeat("w1", WorkerStatusType.Idle));
+
+ var changes = registry.Sweep(DateTime.UtcNow.AddSeconds(31));
+
+ var change = Assert.Single(changes);
+ Assert.Equal("w1", change.WorkerId);
+ Assert.Equal(WorkerStatusType.Offline, change.NewStatus);
+ Assert.Equal(WorkerStatusType.Offline, registry.Get("w1")!.Status);
+ }
+
+ [Fact]
+ public void Sweep_DoesNotReEmitAlreadyOfflineWorker()
+ {
+ var registry = NewRegistry(offlineTimeoutSeconds: 30);
+ registry.Record(Heartbeat("w1", WorkerStatusType.Idle));
+ registry.Sweep(DateTime.UtcNow.AddSeconds(31));
+
+ var changes = registry.Sweep(DateTime.UtcNow.AddSeconds(62));
+
+ Assert.Empty(changes);
+ }
+
+ [Fact]
+ public void Record_AfterOffline_ReturnsReturnTransition()
+ {
+ var registry = NewRegistry(offlineTimeoutSeconds: 30);
+ registry.Record(Heartbeat("w1", WorkerStatusType.Idle));
+ registry.Sweep(DateTime.UtcNow.AddSeconds(31));
+
+ var change = registry.Record(Heartbeat("w1", WorkerStatusType.Busy));
+
+ Assert.NotNull(change);
+ Assert.Equal(WorkerStatusType.Offline, change!.OldStatus);
+ Assert.Equal(WorkerStatusType.Busy, change.NewStatus);
+ }
+
+ [Fact]
+ public void GetAll_ReturnsAllKnownWorkers()
+ {
+ var registry = NewRegistry();
+ registry.Record(Heartbeat("w1", WorkerStatusType.Idle));
+ registry.Record(Heartbeat("w2", WorkerStatusType.Busy));
+
+ Assert.Equal(2, registry.GetAll().Count);
+ Assert.Null(registry.Get("missing"));
+ }
+}
diff --git a/tests/SquidStd.Tests/Messaging/InMemoryTopicProviderTests.cs b/tests/SquidStd.Tests/Messaging/InMemoryTopicProviderTests.cs
new file mode 100644
index 00000000..7757d639
--- /dev/null
+++ b/tests/SquidStd.Tests/Messaging/InMemoryTopicProviderTests.cs
@@ -0,0 +1,58 @@
+using System.Text;
+using SquidStd.Messaging.Services;
+
+namespace SquidStd.Tests.Messaging;
+
+public class InMemoryTopicProviderTests
+{
+ private static ReadOnlyMemory Bytes(string s) => Encoding.UTF8.GetBytes(s);
+
+ [Fact]
+ public async Task Publish_FansOutToAllSubscribers()
+ {
+ await using var provider = new InMemoryTopicProvider();
+ var a = 0;
+ var b = 0;
+ provider.Subscribe("t", (_, _) => { Interlocked.Increment(ref a); return Task.CompletedTask; });
+ provider.Subscribe("t", (_, _) => { Interlocked.Increment(ref b); return Task.CompletedTask; });
+
+ await provider.PublishAsync("t", Bytes("x"));
+
+ Assert.Equal(1, a);
+ Assert.Equal(1, b);
+ }
+
+ [Fact]
+ public async Task Dispose_StopsDelivery()
+ {
+ await using var provider = new InMemoryTopicProvider();
+ var count = 0;
+ var subscription = provider.Subscribe("t", (_, _) => { Interlocked.Increment(ref count); return Task.CompletedTask; });
+
+ subscription.Dispose();
+ await provider.PublishAsync("t", Bytes("x"));
+
+ Assert.Equal(0, count);
+ }
+
+ [Fact]
+ public async Task Publish_NoSubscribers_IsNoOp()
+ {
+ await using var provider = new InMemoryTopicProvider();
+
+ await provider.PublishAsync("t", Bytes("x"));
+ }
+
+ [Fact]
+ public async Task FailingSubscriber_IsIsolated()
+ {
+ await using var provider = new InMemoryTopicProvider();
+ var received = 0;
+ provider.Subscribe("t", (_, _) => throw new InvalidOperationException("boom"));
+ provider.Subscribe("t", (_, _) => { Interlocked.Increment(ref received); return Task.CompletedTask; });
+
+ await provider.PublishAsync("t", Bytes("x"));
+
+ Assert.Equal(1, received);
+ }
+}
diff --git a/tests/SquidStd.Tests/Messaging/MessageTopicTests.cs b/tests/SquidStd.Tests/Messaging/MessageTopicTests.cs
new file mode 100644
index 00000000..dcbcafa2
--- /dev/null
+++ b/tests/SquidStd.Tests/Messaging/MessageTopicTests.cs
@@ -0,0 +1,29 @@
+using SquidStd.Core.Json;
+using SquidStd.Messaging.Abstractions.Interfaces;
+using SquidStd.Messaging.Abstractions.Services;
+using SquidStd.Messaging.Services;
+
+namespace SquidStd.Tests.Messaging;
+
+public class MessageTopicTests
+{
+ private sealed class Ping
+ {
+ public string Source { get; set; } = "";
+ }
+
+ [Fact]
+ public async Task PublishSubscribe_RoundTripsTypedMessage()
+ {
+ await using var provider = new InMemoryTopicProvider();
+ var serializer = new JsonDataSerializer();
+ IMessageTopic topic = new MessageTopic(provider, serializer, serializer);
+ var received = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ topic.Subscribe("pings", (ping, _) => { received.TrySetResult(ping); return Task.CompletedTask; });
+
+ await topic.PublishAsync("pings", new Ping { Source = "w1" });
+
+ var got = await received.Task.WaitAsync(TimeSpan.FromSeconds(5));
+ Assert.Equal("w1", got.Source);
+ }
+}
diff --git a/tests/SquidStd.Tests/Messaging/MessagingRegistrationExtensionsTests.cs b/tests/SquidStd.Tests/Messaging/MessagingRegistrationExtensionsTests.cs
index 140608b0..e06330f0 100644
--- a/tests/SquidStd.Tests/Messaging/MessagingRegistrationExtensionsTests.cs
+++ b/tests/SquidStd.Tests/Messaging/MessagingRegistrationExtensionsTests.cs
@@ -1,9 +1,11 @@
using DryIoc;
+using SquidStd.Core.Interfaces.Events;
using SquidStd.Core.Interfaces.Metrics;
using SquidStd.Core.Interfaces.Serialization;
using SquidStd.Messaging.Extensions;
using SquidStd.Messaging.Services;
using SquidStd.Messaging.Abstractions.Interfaces;
+using SquidStd.Services.Core.Services;
namespace SquidStd.Tests.Messaging;
@@ -34,4 +36,17 @@ public void AddInMemoryMessaging_MetricsAndProviderShareInstance()
Assert.Same(container.Resolve(), container.Resolve());
}
+
+ [Fact]
+ public void AddInMemoryMessaging_RegistersTopicServices()
+ {
+ using var container = new Container();
+ container.RegisterInstance(new EventBusService());
+
+ container.AddInMemoryMessaging();
+
+ Assert.NotNull(container.Resolve());
+ Assert.NotNull(container.Resolve());
+ Assert.NotNull(container.Resolve());
+ }
}
diff --git a/tests/SquidStd.Tests/Messaging/RabbitMq/RabbitMqTopicProviderTests.cs b/tests/SquidStd.Tests/Messaging/RabbitMq/RabbitMqTopicProviderTests.cs
new file mode 100644
index 00000000..28775802
--- /dev/null
+++ b/tests/SquidStd.Tests/Messaging/RabbitMq/RabbitMqTopicProviderTests.cs
@@ -0,0 +1,59 @@
+using System.Text;
+using SquidStd.Messaging.RabbitMq.Data.Config;
+using SquidStd.Messaging.RabbitMq.Services;
+
+namespace SquidStd.Tests.Messaging.RabbitMq;
+
+[Collection(RabbitMqCollection.Name)]
+public class RabbitMqTopicProviderTests
+{
+ private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(30);
+
+ private readonly RabbitMqContainerFixture _fixture;
+
+ public RabbitMqTopicProviderTests(RabbitMqContainerFixture fixture)
+ {
+ _fixture = fixture;
+ }
+
+ private RabbitMqTopicProvider NewProvider()
+ => new(new RabbitMqOptions { Uri = new Uri(_fixture.AmqpUri) });
+
+ private static ReadOnlyMemory Bytes(string s) => Encoding.UTF8.GetBytes(s);
+ private static string Topic() => "topic-" + Guid.NewGuid().ToString("N");
+
+ [Fact]
+ public async Task Publish_FansOutToAllSubscribers()
+ {
+ await using var provider = NewProvider();
+ await provider.StartAsync();
+ var topic = Topic();
+ var a = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var b = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ provider.Subscribe(topic, (payload, _) => { a.TrySetResult(Encoding.UTF8.GetString(payload.Span)); return Task.CompletedTask; });
+ provider.Subscribe(topic, (payload, _) => { b.TrySetResult(Encoding.UTF8.GetString(payload.Span)); return Task.CompletedTask; });
+
+ // Allow the broker to register the bindings before publishing.
+ await Task.Delay(300);
+ await provider.PublishAsync(topic, Bytes("hello"));
+
+ Assert.Equal("hello", await a.Task.WaitAsync(Timeout));
+ Assert.Equal("hello", await b.Task.WaitAsync(Timeout));
+ }
+
+ [Fact]
+ public async Task Publish_BeforeAnySubscriber_IsNotReceived()
+ {
+ await using var provider = NewProvider();
+ await provider.StartAsync();
+ var topic = Topic();
+
+ await provider.PublishAsync(topic, Bytes("lost"));
+
+ var received = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ provider.Subscribe(topic, (payload, _) => { received.TrySetResult(Encoding.UTF8.GetString(payload.Span)); return Task.CompletedTask; });
+ await Task.Delay(300);
+
+ await Assert.ThrowsAsync(async () => await received.Task.WaitAsync(TimeSpan.FromSeconds(1)));
+ }
+}
diff --git a/tests/SquidStd.Tests/Messaging/TopicEventBridgeTests.cs b/tests/SquidStd.Tests/Messaging/TopicEventBridgeTests.cs
new file mode 100644
index 00000000..d68276d7
--- /dev/null
+++ b/tests/SquidStd.Tests/Messaging/TopicEventBridgeTests.cs
@@ -0,0 +1,49 @@
+using SquidStd.Core.Interfaces.Events;
+using SquidStd.Core.Json;
+using SquidStd.Messaging.Abstractions.Data.Events;
+using SquidStd.Messaging.Abstractions.Interfaces;
+using SquidStd.Messaging.Abstractions.Services;
+using SquidStd.Messaging.Services;
+using SquidStd.Services.Core.Services;
+
+namespace SquidStd.Tests.Messaging;
+
+public class TopicEventBridgeTests
+{
+ private sealed class Beat
+ {
+ public string Worker { get; set; } = "";
+ }
+
+ private sealed class CapturingListener : IAsyncEventListener
+ {
+ public TaskCompletionSource Received { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ public Task HandleAsync(TopicMessageEvent eventData, CancellationToken cancellationToken)
+ {
+ Received.TrySetResult(eventData);
+
+ return Task.CompletedTask;
+ }
+ }
+
+ [Fact]
+ public async Task Bridge_RepublishesTopicMessageOnEventBus()
+ {
+ await using var provider = new InMemoryTopicProvider();
+ var serializer = new JsonDataSerializer();
+ IMessageTopic topic = new MessageTopic(provider, serializer, serializer);
+ var eventBus = new EventBusService();
+ ITopicEventBridge bridge = new TopicEventBridge(topic, eventBus);
+
+ var listener = new CapturingListener();
+ eventBus.RegisterAsyncListener(listener);
+ using var subscription = bridge.Bridge("workers.heartbeat");
+
+ await topic.PublishAsync("workers.heartbeat", new Beat { Worker = "w1" });
+
+ var evt = await listener.Received.Task.WaitAsync(TimeSpan.FromSeconds(5));
+ Assert.Equal("workers.heartbeat", evt.Topic);
+ Assert.Equal("w1", Assert.IsType(evt.Data).Worker);
+ }
+}
diff --git a/tests/SquidStd.Tests/Services/Core/RegisterDefaultServicesExtensionsTests.cs b/tests/SquidStd.Tests/Services/Core/RegisterDefaultServicesExtensionsTests.cs
index 74e259a7..e8d2ee14 100644
--- a/tests/SquidStd.Tests/Services/Core/RegisterDefaultServicesExtensionsTests.cs
+++ b/tests/SquidStd.Tests/Services/Core/RegisterDefaultServicesExtensionsTests.cs
@@ -2,6 +2,7 @@
using SquidStd.Abstractions.Data.Internal.Config;
using SquidStd.Abstractions.Data.Internal.Services;
using SquidStd.Core.Data.Bootstrap;
+using SquidStd.Core.Directories;
using SquidStd.Core.Data.Jobs;
using SquidStd.Core.Data.Metrics;
using SquidStd.Core.Data.Storage;
@@ -180,6 +181,17 @@ public void RegisterCoreServices_RegistersSecretServices()
Assert.True(container.IsRegistered());
}
+ [Fact]
+ public void RegisterCoreServices_RegistersDirectoriesConfig()
+ {
+ using var temp = new TempDirectory();
+ using var container = new Container();
+
+ container.RegisterCoreServices("app", temp.Path);
+
+ Assert.True(container.IsRegistered());
+ }
+
[Fact]
public void RegisterSecretServices_RegistersSecretProtectorAndStore()
{
diff --git a/tests/SquidStd.Tests/SquidStd.Tests.csproj b/tests/SquidStd.Tests/SquidStd.Tests.csproj
index 59802f79..7e8df98e 100644
--- a/tests/SquidStd.Tests/SquidStd.Tests.csproj
+++ b/tests/SquidStd.Tests/SquidStd.Tests.csproj
@@ -48,6 +48,10 @@
+
+
+
+
diff --git a/tests/SquidStd.Tests/Templating/ScribanTemplateRendererTests.cs b/tests/SquidStd.Tests/Templating/ScribanTemplateRendererTests.cs
new file mode 100644
index 00000000..414f63ba
--- /dev/null
+++ b/tests/SquidStd.Tests/Templating/ScribanTemplateRendererTests.cs
@@ -0,0 +1,87 @@
+using SquidStd.Core.Directories;
+using SquidStd.Templating;
+using SquidStd.Templating.Services;
+using SquidStd.Tests.Support;
+
+namespace SquidStd.Tests.Templating;
+
+public class ScribanTemplateRendererTests
+{
+ private static ScribanTemplateRenderer NewRenderer(string root)
+ => new(new DirectoriesConfig(root, []));
+
+ [Fact]
+ public async Task RenderAsync_RendersModel()
+ {
+ using var temp = new TempDirectory();
+ var renderer = NewRenderer(temp.Path);
+
+ var result = await renderer.RenderAsync("Hi {{ user.name }}", new { User = new { Name = "squid" } });
+
+ Assert.Equal("Hi squid", result);
+ }
+
+ [Fact]
+ public async Task RegisterThenRenderByName_Works()
+ {
+ using var temp = new TempDirectory();
+ var renderer = NewRenderer(temp.Path);
+ renderer.Register("greet", "Hi {{ user.name }}");
+
+ var result = await renderer.RenderByNameAsync("greet", new { User = new { Name = "squid" } });
+
+ Assert.Equal("Hi squid", result);
+ }
+
+ [Fact]
+ public async Task RenderByNameAsync_UnknownName_Throws()
+ {
+ using var temp = new TempDirectory();
+ var renderer = NewRenderer(temp.Path);
+
+ await Assert.ThrowsAsync(async () => await renderer.RenderByNameAsync("nope", null));
+ }
+
+ [Fact]
+ public async Task RenderAsync_MalformedTemplate_Throws()
+ {
+ using var temp = new TempDirectory();
+ var renderer = NewRenderer(temp.Path);
+
+ await Assert.ThrowsAsync(async () => await renderer.RenderAsync("{{ for x in }}", null));
+ }
+
+ [Fact]
+ public void Register_MalformedTemplate_Throws()
+ {
+ using var temp = new TempDirectory();
+ var renderer = NewRenderer(temp.Path);
+
+ Assert.Throws(() => renderer.Register("bad", "{{ for x in }}"));
+ }
+
+ [Fact]
+ public async Task RenderAsync_EmptyTemplate_Throws()
+ {
+ using var temp = new TempDirectory();
+ var renderer = NewRenderer(temp.Path);
+
+ await Assert.ThrowsAsync(async () => await renderer.RenderAsync(string.Empty, null));
+ }
+
+ [Fact]
+ public async Task StartAsync_AutoLoadsTemplatesFromDirectory()
+ {
+ using var temp = new TempDirectory();
+ var emails = Path.Combine(temp.Path, "templates", "emails");
+ Directory.CreateDirectory(emails);
+ await File.WriteAllTextAsync(Path.Combine(emails, "welcome.tmpl"), "Welcome {{ user.name }}");
+
+ var renderer = NewRenderer(temp.Path);
+ await renderer.StartAsync();
+
+ var result = await renderer.RenderByNameAsync("emails/welcome", new { User = new { Name = "squid" } });
+
+ Assert.Equal("Welcome squid", result);
+ }
+}
diff --git a/tests/SquidStd.Tests/Templating/TemplatingRegistrationTests.cs b/tests/SquidStd.Tests/Templating/TemplatingRegistrationTests.cs
new file mode 100644
index 00000000..7046a789
--- /dev/null
+++ b/tests/SquidStd.Tests/Templating/TemplatingRegistrationTests.cs
@@ -0,0 +1,25 @@
+using DryIoc;
+using SquidStd.Core.Directories;
+using SquidStd.Templating.Extensions;
+using SquidStd.Templating.Interfaces;
+using SquidStd.Tests.Support;
+
+namespace SquidStd.Tests.Templating;
+
+public class TemplatingRegistrationTests
+{
+ [Fact]
+ public async Task AddTemplating_ResolvesRendererAndRenders()
+ {
+ using var temp = new TempDirectory();
+ var container = new Container();
+ container.RegisterInstance(new DirectoriesConfig(temp.Path, []));
+
+ container.AddTemplating();
+
+ var renderer = container.Resolve();
+ var result = await renderer.RenderAsync("Hi {{ user.name }}", new { User = new { Name = "squid" } });
+
+ Assert.Equal("Hi squid", result);
+ }
+}
diff --git a/tests/SquidStd.Tests/Workers/JobDispatcherTests.cs b/tests/SquidStd.Tests/Workers/JobDispatcherTests.cs
new file mode 100644
index 00000000..6114d4dc
--- /dev/null
+++ b/tests/SquidStd.Tests/Workers/JobDispatcherTests.cs
@@ -0,0 +1,46 @@
+using SquidStd.Tests.Workers.Support;
+using SquidStd.Workers.Abstractions.Data;
+using SquidStd.Workers.Exceptions;
+using SquidStd.Workers.Services;
+
+namespace SquidStd.Tests.Workers;
+
+public class JobDispatcherTests
+{
+ private static JobRequest Job(string name)
+ => new(name, new Dictionary());
+
+ [Fact]
+ public async Task DispatchAsync_InvokesHandlerMatchingJobName()
+ {
+ var resize = new RecordingJobHandler("resize");
+ var encode = new RecordingJobHandler("encode");
+ var dispatcher = new JobDispatcher([resize, encode]);
+
+ await dispatcher.DispatchAsync(Job("encode"), CancellationToken.None);
+
+ Assert.Empty(resize.Received);
+ Assert.Single(encode.Received);
+ }
+
+ [Fact]
+ public async Task DispatchAsync_ThrowsWhenNoHandlerMatches()
+ {
+ var dispatcher = new JobDispatcher([new RecordingJobHandler("resize")]);
+
+ var ex = await Assert.ThrowsAsync(
+ () => dispatcher.DispatchAsync(Job("unknown"), CancellationToken.None));
+
+ Assert.Equal("unknown", ex.JobName);
+ }
+
+ [Fact]
+ public async Task DispatchAsync_PropagatesHandlerException()
+ {
+ var boom = new RecordingJobHandler("resize") { ThrowOnHandle = new InvalidOperationException("boom") };
+ var dispatcher = new JobDispatcher([boom]);
+
+ await Assert.ThrowsAsync(
+ () => dispatcher.DispatchAsync(Job("resize"), CancellationToken.None));
+ }
+}
diff --git a/tests/SquidStd.Tests/Workers/Support/FakeMessageQueue.cs b/tests/SquidStd.Tests/Workers/Support/FakeMessageQueue.cs
new file mode 100644
index 00000000..1a1812d2
--- /dev/null
+++ b/tests/SquidStd.Tests/Workers/Support/FakeMessageQueue.cs
@@ -0,0 +1,23 @@
+using SquidStd.Messaging.Abstractions.Interfaces;
+
+namespace SquidStd.Tests.Workers.Support;
+
+/// Inert for unit tests that drive the consumer's HandleAsync directly.
+public sealed class FakeMessageQueue : IMessageQueue
+{
+ public Task PublishAsync(string queueName, TMessage message, CancellationToken cancellationToken = default)
+ => Task.CompletedTask;
+
+ public IDisposable Subscribe(string queueName, IQueueMessageListener listener)
+ => new Subscription();
+
+ public IDisposable Subscribe(string queueName, IQueueMessageListenerAsync listener)
+ => new Subscription();
+
+ private sealed class Subscription : IDisposable
+ {
+ public void Dispose()
+ {
+ }
+ }
+}
diff --git a/tests/SquidStd.Tests/Workers/Support/RecordingJobHandler.cs b/tests/SquidStd.Tests/Workers/Support/RecordingJobHandler.cs
new file mode 100644
index 00000000..75fef96a
--- /dev/null
+++ b/tests/SquidStd.Tests/Workers/Support/RecordingJobHandler.cs
@@ -0,0 +1,55 @@
+using SquidStd.Workers.Abstractions.Data;
+using SquidStd.Workers.Interfaces;
+
+namespace SquidStd.Tests.Workers.Support;
+
+///
+/// Configurable for tests: records the jobs it received and can be told
+/// to throw, or to block until released, so concurrency and error paths can be exercised.
+///
+public sealed class RecordingJobHandler : IJobHandler
+{
+ private readonly List _received = [];
+ private readonly Lock _sync = new();
+
+ public RecordingJobHandler(string jobName)
+ {
+ JobName = jobName;
+ }
+
+ public string JobName { get; }
+
+ public Exception? ThrowOnHandle { get; set; }
+
+ /// When set, awaits this before returning.
+ public TaskCompletionSource? Gate { get; set; }
+
+ public IReadOnlyList Received
+ {
+ get
+ {
+ lock (_sync)
+ {
+ return _received.ToArray();
+ }
+ }
+ }
+
+ public async Task HandleAsync(JobRequest job, CancellationToken cancellationToken)
+ {
+ lock (_sync)
+ {
+ _received.Add(job);
+ }
+
+ if (ThrowOnHandle is not null)
+ {
+ throw ThrowOnHandle;
+ }
+
+ if (Gate is not null)
+ {
+ await Gate.Task.WaitAsync(cancellationToken);
+ }
+ }
+}
diff --git a/tests/SquidStd.Tests/Workers/WorkerConsumerServiceTests.cs b/tests/SquidStd.Tests/Workers/WorkerConsumerServiceTests.cs
new file mode 100644
index 00000000..7f04c1e3
--- /dev/null
+++ b/tests/SquidStd.Tests/Workers/WorkerConsumerServiceTests.cs
@@ -0,0 +1,76 @@
+using SquidStd.Tests.Workers.Support;
+using SquidStd.Workers.Abstractions.Data;
+using SquidStd.Workers.Data.Config;
+using SquidStd.Workers.Services;
+
+namespace SquidStd.Tests.Workers;
+
+public class WorkerConsumerServiceTests
+{
+ private static JobRequest Job(string name)
+ => new(name, new Dictionary());
+
+ private static WorkerConsumerService Build(WorkerState state, params RecordingJobHandler[] handlers)
+ => new(new FakeMessageQueue(), new JobDispatcher(handlers), state, new WorkersConfig());
+
+ [Fact]
+ public async Task HandleAsync_RunsMatchingHandler()
+ {
+ var handler = new RecordingJobHandler("resize");
+ var state = new WorkerState(new() { WorkerId = "w1", MaxConcurrency = 2 });
+ var consumer = Build(state, handler);
+
+ await consumer.HandleAsync(Job("resize"), CancellationToken.None);
+
+ Assert.Single(handler.Received);
+ Assert.Equal(0, state.ActiveJobs);
+ }
+
+ [Fact]
+ public async Task HandleAsync_DropsUnknownJobWithoutThrowing()
+ {
+ var state = new WorkerState(new() { WorkerId = "w1", MaxConcurrency = 2 });
+ var consumer = Build(state, new RecordingJobHandler("resize"));
+
+ var ex = await Record.ExceptionAsync(() => consumer.HandleAsync(Job("unknown"), CancellationToken.None));
+
+ Assert.Null(ex);
+ Assert.Equal(0, state.ActiveJobs);
+ }
+
+ [Fact]
+ public async Task HandleAsync_RethrowsHandlerExceptionForRequeue()
+ {
+ var handler = new RecordingJobHandler("resize") { ThrowOnHandle = new InvalidOperationException("boom") };
+ var state = new WorkerState(new() { WorkerId = "w1", MaxConcurrency = 2 });
+ var consumer = Build(state, handler);
+
+ await Assert.ThrowsAsync(
+ () => consumer.HandleAsync(Job("resize"), CancellationToken.None));
+
+ Assert.Equal(0, state.ActiveJobs);
+ }
+
+ [Fact]
+ public async Task HandleAsync_NeverExceedsMaxConcurrency()
+ {
+ var handler = new RecordingJobHandler("resize") { Gate = new TaskCompletionSource() };
+ var state = new WorkerState(new() { WorkerId = "w1", MaxConcurrency = 2 });
+ var consumer = Build(state, handler);
+
+ // Launch 5 concurrent dispatches against a handler that blocks on its gate.
+ var inFlight = Enumerable.Range(0, 5)
+ .Select(_ => consumer.HandleAsync(Job("resize"), CancellationToken.None))
+ .ToArray();
+
+ // Give the semaphore time to admit as many as it will, then assert the cap held.
+ await Task.Delay(200);
+ Assert.True(state.ActiveJobs <= 2, $"ActiveJobs was {state.ActiveJobs}, expected <= 2");
+
+ // Release the gate so everything drains.
+ handler.Gate!.SetResult();
+ await Task.WhenAll(inFlight);
+ Assert.Equal(0, state.ActiveJobs);
+ Assert.Equal(5, handler.Received.Count);
+ }
+}
diff --git a/tests/SquidStd.Tests/Workers/WorkerContractsTests.cs b/tests/SquidStd.Tests/Workers/WorkerContractsTests.cs
new file mode 100644
index 00000000..8e9df30b
--- /dev/null
+++ b/tests/SquidStd.Tests/Workers/WorkerContractsTests.cs
@@ -0,0 +1,83 @@
+using SquidStd.Core.Json;
+using SquidStd.Workers.Abstractions;
+using SquidStd.Workers.Abstractions.Data;
+using SquidStd.Workers.Abstractions.Types;
+
+namespace SquidStd.Tests.Workers;
+
+public class WorkerContractsTests
+{
+ private static readonly JsonDataSerializer Serializer = new();
+
+ [Fact]
+ public void JobRequest_RoundTrips_PreservingNameAndParameters()
+ {
+ var original = new JobRequest(
+ "resize-image",
+ new Dictionary { ["width"] = "800", ["height"] = "600" });
+
+ var bytes = Serializer.Serialize(original);
+ var restored = Serializer.Deserialize(bytes);
+
+ Assert.Equal("resize-image", restored.JobName);
+ Assert.Equal("800", restored.Parameters["width"]);
+ Assert.Equal("600", restored.Parameters["height"]);
+ }
+
+ [Fact]
+ public void WorkerHeartbeat_RoundTrips_PreservingAllFields()
+ {
+ var original = new WorkerHeartbeat(
+ "worker-1",
+ new DateTime(2026, 6, 23, 10, 0, 0, DateTimeKind.Utc),
+ WorkerStatusType.Busy,
+ 3,
+ 8);
+
+ var restored = Serializer.Deserialize(Serializer.Serialize(original));
+
+ Assert.Equal(original, restored);
+ }
+
+ [Theory]
+ [InlineData(WorkerStatusType.Idle)]
+ [InlineData(WorkerStatusType.Busy)]
+ [InlineData(WorkerStatusType.Offline)]
+ public void WorkerHeartbeat_RoundTrips_EveryStatus(WorkerStatusType status)
+ {
+ var original = new WorkerHeartbeat(
+ "worker-3",
+ new DateTime(2026, 6, 23, 11, 0, 0, DateTimeKind.Utc),
+ status,
+ 0,
+ 4);
+
+ var restored = Serializer.Deserialize(Serializer.Serialize(original));
+
+ Assert.Equal(status, restored.Status);
+ }
+
+ [Fact]
+ public void WorkerInfo_RoundTrips_PreservingAllFields()
+ {
+ var original = new WorkerInfo(
+ "worker-4",
+ WorkerStatusType.Offline,
+ 2,
+ 8,
+ new DateTime(2026, 6, 23, 9, 0, 0, DateTimeKind.Utc),
+ new DateTime(2026, 6, 23, 9, 30, 0, DateTimeKind.Utc));
+
+ var restored = Serializer.Deserialize(Serializer.Serialize(original));
+
+ Assert.Equal(original, restored);
+ }
+
+ [Fact]
+ public void WorkerChannels_NamesAreNonEmptyAndDistinct()
+ {
+ Assert.False(string.IsNullOrWhiteSpace(WorkerChannels.JobQueue));
+ Assert.False(string.IsNullOrWhiteSpace(WorkerChannels.HeartbeatTopic));
+ Assert.NotEqual(WorkerChannels.JobQueue, WorkerChannels.HeartbeatTopic);
+ }
+}
diff --git a/tests/SquidStd.Tests/Workers/WorkerHeartbeatServiceTests.cs b/tests/SquidStd.Tests/Workers/WorkerHeartbeatServiceTests.cs
new file mode 100644
index 00000000..6efcf659
--- /dev/null
+++ b/tests/SquidStd.Tests/Workers/WorkerHeartbeatServiceTests.cs
@@ -0,0 +1,69 @@
+using DryIoc;
+using SquidStd.Core.Interfaces.Events;
+using SquidStd.Messaging.Abstractions.Interfaces;
+using SquidStd.Messaging.Extensions;
+using SquidStd.Services.Core.Services;
+using SquidStd.Workers.Abstractions;
+using SquidStd.Workers.Abstractions.Data;
+using SquidStd.Workers.Abstractions.Types;
+using SquidStd.Workers.Data.Config;
+using SquidStd.Workers.Services;
+
+namespace SquidStd.Tests.Workers;
+
+public class WorkerHeartbeatServiceTests
+{
+ private static (IMessageTopic topic, WorkerState state) NewMessaging(WorkersConfig config)
+ {
+ var container = new Container();
+ container.RegisterInstance(new EventBusService());
+ container.AddInMemoryMessaging();
+
+ return (container.Resolve(), new WorkerState(config));
+ }
+
+ [Fact]
+ public async Task StartAsync_PublishesHeartbeatImmediately()
+ {
+ var config = new WorkersConfig { WorkerId = "w1", MaxConcurrency = 8, HeartbeatIntervalSeconds = 60 };
+ var (topic, state) = NewMessaging(config);
+
+ var received = new TaskCompletionSource();
+ using var _ = topic.Subscribe(
+ WorkerChannels.HeartbeatTopic,
+ (hb, _) => { received.TrySetResult(hb); return Task.CompletedTask; });
+
+ var service = new WorkerHeartbeatService(topic, state, config);
+ await service.StartAsync();
+
+ var heartbeat = await received.Task.WaitAsync(TimeSpan.FromSeconds(5));
+ await service.StopAsync();
+
+ Assert.Equal("w1", heartbeat.WorkerId);
+ Assert.Equal(8, heartbeat.MaxConcurrency);
+ Assert.Equal(0, heartbeat.ActiveJobs);
+ Assert.Equal(WorkerStatusType.Idle, heartbeat.Status);
+ }
+
+ [Fact]
+ public async Task Heartbeat_ReflectsBusyStateWithActiveJobs()
+ {
+ var config = new WorkersConfig { WorkerId = "w1", MaxConcurrency = 8, HeartbeatIntervalSeconds = 60 };
+ var (topic, state) = NewMessaging(config);
+ state.JobStarted();
+
+ var received = new TaskCompletionSource();
+ using var _ = topic.Subscribe(
+ WorkerChannels.HeartbeatTopic,
+ (hb, _) => { received.TrySetResult(hb); return Task.CompletedTask; });
+
+ var service = new WorkerHeartbeatService(topic, state, config);
+ await service.StartAsync();
+
+ var heartbeat = await received.Task.WaitAsync(TimeSpan.FromSeconds(5));
+ await service.StopAsync();
+
+ Assert.Equal(1, heartbeat.ActiveJobs);
+ Assert.Equal(WorkerStatusType.Busy, heartbeat.Status);
+ }
+}
diff --git a/tests/SquidStd.Tests/Workers/WorkerStateTests.cs b/tests/SquidStd.Tests/Workers/WorkerStateTests.cs
new file mode 100644
index 00000000..8e8b7942
--- /dev/null
+++ b/tests/SquidStd.Tests/Workers/WorkerStateTests.cs
@@ -0,0 +1,51 @@
+using SquidStd.Workers.Data.Config;
+using SquidStd.Workers.Services;
+using SquidStd.Workers.Abstractions.Types;
+
+namespace SquidStd.Tests.Workers;
+
+public class WorkerStateTests
+{
+ [Fact]
+ public void Status_IsIdleWhenNoActiveJobs()
+ {
+ var state = new WorkerState(new WorkersConfig { WorkerId = "w1", MaxConcurrency = 4 });
+
+ Assert.Equal(WorkerStatusType.Idle, state.Status);
+ Assert.Equal(0, state.ActiveJobs);
+ }
+
+ [Fact]
+ public void Status_IsBusyWhileJobsActive()
+ {
+ var state = new WorkerState(new WorkersConfig { WorkerId = "w1", MaxConcurrency = 4 });
+
+ state.JobStarted();
+ state.JobStarted();
+
+ Assert.Equal(WorkerStatusType.Busy, state.Status);
+ Assert.Equal(2, state.ActiveJobs);
+
+ state.JobFinished();
+ state.JobFinished();
+
+ Assert.Equal(WorkerStatusType.Idle, state.Status);
+ Assert.Equal(0, state.ActiveJobs);
+ }
+
+ [Fact]
+ public void WorkerId_FallsBackToMachineNameWhenBlank()
+ {
+ var state = new WorkerState(new WorkersConfig { WorkerId = " ", MaxConcurrency = 4 });
+
+ Assert.Equal(Environment.MachineName, state.WorkerId);
+ }
+
+ [Fact]
+ public void MaxConcurrency_FallsBackToProcessorCountWhenNotPositive()
+ {
+ var state = new WorkerState(new WorkersConfig { WorkerId = "w1", MaxConcurrency = 0 });
+
+ Assert.Equal(Environment.ProcessorCount, state.MaxConcurrency);
+ }
+}
diff --git a/tests/SquidStd.Tests/Workers/WorkersRegistrationExtensionsTests.cs b/tests/SquidStd.Tests/Workers/WorkersRegistrationExtensionsTests.cs
new file mode 100644
index 00000000..5d4b73b4
--- /dev/null
+++ b/tests/SquidStd.Tests/Workers/WorkersRegistrationExtensionsTests.cs
@@ -0,0 +1,65 @@
+using DryIoc;
+using SquidStd.Core.Interfaces.Events;
+using SquidStd.Messaging.Extensions;
+using SquidStd.Services.Core.Services;
+using SquidStd.Workers.Data.Config;
+using SquidStd.Workers.Extensions;
+using SquidStd.Workers.Interfaces;
+using SquidStd.Workers.Services;
+
+namespace SquidStd.Tests.Workers;
+
+public class WorkersRegistrationExtensionsTests
+{
+ private static Container NewContainer()
+ {
+ var container = new Container();
+ container.RegisterInstance(new EventBusService());
+ container.AddInMemoryMessaging();
+ // ConfigManager normally registers config instances; register one directly for the test.
+ container.RegisterInstance(new WorkersConfig { WorkerId = "w1", MaxConcurrency = 4 });
+
+ return container;
+ }
+
+ [Fact]
+ public void AddWorkers_RegistersResolvableServices()
+ {
+ using var container = NewContainer();
+
+ container.AddWorkers();
+
+ Assert.NotNull(container.Resolve());
+ Assert.NotNull(container.Resolve());
+ Assert.NotNull(container.Resolve());
+ Assert.NotNull(container.Resolve());
+ }
+
+ [Fact]
+ public async Task AddJobHandler_MakesHandlerReachableFromDispatcher()
+ {
+ using var container = NewContainer();
+
+ container.AddWorkers();
+ container.AddJobHandler();
+
+ var dispatcher = container.Resolve();
+ await dispatcher.DispatchAsync(new("echo", new Dictionary()), CancellationToken.None);
+
+ Assert.Equal(1, EchoJobHandler.Calls);
+ }
+
+ private sealed class EchoJobHandler : IJobHandler
+ {
+ public static int Calls;
+
+ public string JobName => "echo";
+
+ public Task HandleAsync(SquidStd.Workers.Abstractions.Data.JobRequest job, CancellationToken cancellationToken)
+ {
+ Interlocked.Increment(ref Calls);
+
+ return Task.CompletedTask;
+ }
+ }
+}