release: templating, messaging pub/sub, and the workers system

README.md

@@ -86,6 +86,7 @@ block until cancellation for long-running hosts.
 | `SquidStd.Storage` | Local file storage backend (`AddFileStorage`). | [![readme](https://img.shields.io/badge/readme-1390A3.svg)](src/SquidStd.Storage/README.md) · [![NuGet](https://img.shields.io/nuget/v/SquidStd.Storage.svg)](https://www.nuget.org/packages/SquidStd.Storage/) |
 | `SquidStd.Storage.S3` | S3/MinIO storage backend (`AddS3Storage`). | [![readme](https://img.shields.io/badge/readme-1390A3.svg)](src/SquidStd.Storage.S3/README.md) · [![NuGet](https://img.shields.io/nuget/v/SquidStd.Storage.S3.svg)](https://www.nuget.org/packages/SquidStd.Storage.S3/) |
 | `SquidStd.Scripting.Lua` | Lua scripting engine with attribute-based modules and event bridging. | [![readme](https://img.shields.io/badge/readme-1390A3.svg)](src/SquidStd.Scripting.Lua/README.md) · [![NuGet](https://img.shields.io/nuget/v/SquidStd.Scripting.Lua.svg)](https://www.nuget.org/packages/SquidStd.Scripting.Lua/) |
+| `SquidStd.Templating` | Scriban templating with a named-template registry and `templates/*.tmpl` auto-load (`AddTemplating`). | [![readme](https://img.shields.io/badge/readme-1390A3.svg)](src/SquidStd.Templating/README.md) · [![NuGet](https://img.shields.io/nuget/v/SquidStd.Templating.svg)](https://www.nuget.org/packages/SquidStd.Templating/) |
 
 ## Architecture
 

SquidStd.slnx

@@ -18,6 +18,10 @@
         <Project Path="src/SquidStd.Storage.Abstractions/SquidStd.Storage.Abstractions.csproj" />
         <Project Path="src/SquidStd.Storage.S3/SquidStd.Storage.S3.csproj" />
         <Project Path="src/SquidStd.Storage/SquidStd.Storage.csproj" />
+        <Project Path="src/SquidStd.Templating/SquidStd.Templating.csproj" />
+        <Project Path="src/SquidStd.Workers.Abstractions/SquidStd.Workers.Abstractions.csproj" />
+        <Project Path="src/SquidStd.Workers/SquidStd.Workers.csproj" />
+        <Project Path="src/SquidStd.Workers.Manager/SquidStd.Workers.Manager.csproj" />
     </Folder>
     <Folder Name="/tests/">
         <Project Path="tests/SquidStd.Tests/SquidStd.Tests.csproj" />

docs/articles/templating.md

@@ -0,0 +1 @@
+[!include[](../../src/SquidStd.Templating/README.md)]

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

docs/articles/workers-abstractions.md

@@ -0,0 +1 @@
+[!include[](../../src/SquidStd.Workers.Abstractions/README.md)]

docs/articles/workers-manager.md

@@ -0,0 +1 @@
+[!include[](../../src/SquidStd.Workers.Manager/README.md)]

docs/articles/workers.md

@@ -0,0 +1 @@
+[!include[](../../src/SquidStd.Workers/README.md)]

src/SquidStd.Messaging.Abstractions/Data/Events/TopicMessageEvent.cs

@@ -0,0 +1,9 @@
+using SquidStd.Core.Interfaces.Events;
+
+namespace SquidStd.Messaging.Abstractions.Data.Events;
+
+/// <summary>
+/// Event published on the in-process event bus when a topic message is bridged. <see cref="Data" /> is the
+/// deserialized message.
+/// </summary>
+public sealed record TopicMessageEvent(string Topic, object Data) : IEvent;

src/SquidStd.Messaging.Abstractions/Interfaces/IMessageTopic.cs

@@ -0,0 +1,13 @@
+namespace SquidStd.Messaging.Abstractions.Interfaces;
+
+/// <summary>
+/// Typed facade for publishing to and subscribing to topics (fan-out).
+/// </summary>
+public interface IMessageTopic
+{
+    /// <summary>Publishes a typed message to a topic.</summary>
+    Task PublishAsync<TMessage>(string topic, TMessage message, CancellationToken cancellationToken = default);
+
+    /// <summary>Subscribes a typed handler to a topic. Dispose to unsubscribe.</summary>
+    IDisposable Subscribe<TMessage>(string topic, Func<TMessage, CancellationToken, Task> handler);
+}

src/SquidStd.Messaging.Abstractions/Interfaces/ITopicEventBridge.cs

@@ -0,0 +1,11 @@
+namespace SquidStd.Messaging.Abstractions.Interfaces;
+
+/// <summary>
+/// Bridges a topic into the in-process event bus: each message of type <c>T</c> on the topic is republished
+/// as a <c>TopicMessageEvent</c>.
+/// </summary>
+public interface ITopicEventBridge
+{
+    /// <summary>Starts bridging the given topic; dispose the result to stop.</summary>
+    IDisposable Bridge<T>(string topic);
+}

src/SquidStd.Messaging.Abstractions/Interfaces/ITopicProvider.cs

@@ -0,0 +1,16 @@
+using SquidStd.Abstractions.Interfaces.Services;
+
+namespace SquidStd.Messaging.Abstractions.Interfaces;
+
+/// <summary>
+/// Byte-level publish/subscribe transport: every current subscriber of a topic receives every message
+/// (transient, at-most-once fan-out).
+/// </summary>
+public interface ITopicProvider : ISquidStdService, IAsyncDisposable
+{
+    /// <summary>Publishes a raw payload to a topic (fan-out to all current subscribers).</summary>
+    Task PublishAsync(string topic, ReadOnlyMemory<byte> payload, CancellationToken cancellationToken = default);
+
+    /// <summary>Subscribes a raw handler to a topic. Dispose to unsubscribe.</summary>
+    IDisposable Subscribe(string topic, Func<ReadOnlyMemory<byte>, CancellationToken, Task> handler);
+}

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;
+
+/// <summary>
+/// Typed facade over an <see cref="ITopicProvider" />: serializes outgoing messages and deserializes
+/// incoming payloads.
+/// </summary>
+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;
+    }
+
+    /// <inheritdoc />
+    public Task PublishAsync<TMessage>(string topic, TMessage message, CancellationToken cancellationToken = default)
+        => _provider.PublishAsync(topic, _serializer.Serialize(message), cancellationToken);
+
+    /// <inheritdoc />
+    public IDisposable Subscribe<TMessage>(string topic, Func<TMessage, CancellationToken, Task> handler)
+    {
+        ArgumentNullException.ThrowIfNull(handler);
+
+        return _provider.Subscribe(topic, (payload, cancellationToken) => handler(_deserializer.Deserialize<TMessage>(payload), cancellationToken));
+    }
+}

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;
+
+/// <summary>
+/// One-way Topic → EventBus bridge. Subscribes a topic and republishes each message as a
+/// <see cref="TopicMessageEvent" /> on the <see cref="IEventBus" />.
+/// </summary>
+public sealed class TopicEventBridge : ITopicEventBridge
+{
+    private readonly IMessageTopic _topic;
+    private readonly IEventBus _eventBus;
+
+    public TopicEventBridge(IMessageTopic topic, IEventBus eventBus)
+    {
+        _topic = topic;
+        _eventBus = eventBus;
+    }
+
+    /// <inheritdoc />
+    public IDisposable Bridge<T>(string topic)
+        => _topic.Subscribe<T>(topic, (data, cancellationToken) => _eventBus.PublishAsync(new TopicMessageEvent(topic, data!), cancellationToken));
+}

src/SquidStd.Messaging.RabbitMq/Extensions/RabbitMqMessagingRegistrationExtensions.cs

@@ -49,6 +49,10 @@ public static IContainer AddRabbitMqMessaging(
         );
         container.Register<IMessageQueue, MessageQueue>(Reuse.Singleton);
 
+        container.Register<ITopicProvider, RabbitMqTopicProvider>(Reuse.Singleton);
+        container.Register<IMessageTopic, MessageTopic>(Reuse.Singleton);
+        container.Register<ITopicEventBridge, TopicEventBridge>(Reuse.Singleton);
+
         return container;
     }
 

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;
+
+/// <summary>
+/// RabbitMQ <see cref="ITopicProvider" />: topics map to fanout exchanges; each subscriber binds an
+/// exclusive auto-delete queue and consumes with auto-ack (transient, at-most-once fan-out).
+/// </summary>
+public sealed class RabbitMqTopicProvider : ITopicProvider
+{
+    private readonly ILogger _logger = Log.ForContext<RabbitMqTopicProvider>();
+    private readonly RabbitMqOptions _options;
+    private readonly SemaphoreSlim _publishLock = new(1, 1);
+    private readonly Lock _exchangeSync = new();
+    private readonly HashSet<string> _declared = new(StringComparer.Ordinal);
+    private IConnection? _connection;
+    private IChannel? _publishChannel;
+    private int _disposed;
+
+    public RabbitMqTopicProvider(RabbitMqOptions options)
+    {
+        _options = options;
+    }
+
+    /// <inheritdoc />
+    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);
+    }
+
+    /// <inheritdoc />
+    public ValueTask StopAsync(CancellationToken cancellationToken = default)
+        => DisposeAsync();
+
+    /// <inheritdoc />
+    public async Task PublishAsync(string topic, ReadOnlyMemory<byte> 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();
+        }
+    }
+
+    /// <inheritdoc />
+    public IDisposable Subscribe(string topic, Func<ReadOnlyMemory<byte>, 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
+        );
+    }
+
+    /// <inheritdoc />
+    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<ReadOnlyMemory<byte>, CancellationToken, Task> _handler;
+        private IChannel? _channel;
+        private string? _consumerTag;
+        private int _disposed;
+
+        public Subscription(
+            RabbitMqTopicProvider provider,
+            IConnection connection,
+            string topic,
+            Func<ReadOnlyMemory<byte>, 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.
+                }
+            }
+        }
+    }
+}

src/SquidStd.Messaging/Extensions/MessagingRegistrationExtensions.cs

@@ -37,6 +37,10 @@ public static IContainer AddInMemoryMessaging(this IContainer container, Messagi
         container.Register<IQueueProvider, InMemoryQueueProvider>(Reuse.Singleton);
         container.Register<IMessageQueue, MessageQueue>(Reuse.Singleton);
 
+        container.Register<ITopicProvider, InMemoryTopicProvider>(Reuse.Singleton);
+        container.Register<IMessageTopic, MessageTopic>(Reuse.Singleton);
+        container.Register<ITopicEventBridge, TopicEventBridge>(Reuse.Singleton);
+
         return container;
     }