add conveyo

Author: ftechmax

README.md

@@ -7,10 +7,6 @@ In this repository, **MSA** stands for **Message-driven Service Architecture**:
 [![Release](https://github.com/ftechmax/msa-templates/actions/workflows/release.yml/badge.svg)](https://github.com/ftechmax/msa-templates/actions/workflows/release.yml)
 [![codecov](https://codecov.io/gh/ftechmax/msa-templates/graph/badge.svg?token=I4QI609IIQ)](https://codecov.io/gh/ftechmax/msa-templates)
 
-> [!NOTE]
-> The templates currently use **MassTransit**, but this will be removed in a future update and replaced by my own library **Conveyo**: https://github.com/ftechmax/conveyo. Conveyo will also support workers written in Go.
-> MassTransit's licensing model changed, and I want the messaging stack to remain fully open and free.
-
 > [!NOTE]
 > If you previously installed the now-retired `MSA.Templates` NuGet package, you can remove the stale local install with `dotnet new uninstall MSA.Templates`. The new flow does not use `dotnet new`.
 
@@ -37,7 +33,7 @@ These templates are the accumulation of those lessons, packaged as a starting po
 - A clear split between the **HTTP edge** in the API and **asynchronous domain work** in the worker
 - A default stack for **messaging, persistence, caching, and telemetry** picked from what I reach for on every project
 - **Kubernetes** manifests included for local, self-hosted, and cloud clusters
-- Opinionated defaults you can rip out: every choice (MassTransit, PostgreSQL, Valkey, SignalR) is isolated behind its own wiring so you can swap one without touching the rest
+- Opinionated defaults you can rip out: every choice (Conveyo, PostgreSQL, Valkey, SignalR) is isolated behind its own wiring so you can swap one without touching the rest
 
 This is not meant to be the only way to do things. It's the default I've shipped to production more than once.
 
@@ -54,7 +50,7 @@ These templates are intended to be used together as a "vertical slice" of a mess
 
 The templates come wired up for:
 
-- **Messaging** via MassTransit (Conveyo coming soon!) on top of RabbitMQ
+- **Messaging** via Conveyo on top of RabbitMQ
 - **Persistence** via the Npgsql package, storing documents as PostgreSQL JSONB (an in-cluster Postgres StatefulSet in the generated Kubernetes setup)
 - **Caching** via Valkey using the StackExchange.Redis package
 - **Real-time updates** via SignalR

docs/api.md

@@ -8,7 +8,7 @@ The API should not contain domain logic. Its main jobs are:
 
 - **Own the HTTP contract**: controllers, DTOs, a versioning strategy when you need one, and input validation.
 - **Validate early**: reject bad requests before they become messages that bounce around the system.
-- **Send commands** to the worker via the bus using MassTransit. The API should not need direct DB access to do writes.
+- **Send commands** to the worker via the bus using Conveyo. The API should not need direct DB access to do writes.
 - **Serve reads efficiently**: the template uses projections and caches them in Valkey/Redis so read paths stay fast without coupling reads to the write model.
 
 The API also bridges worker events back to connected clients. It consumes published events from the worker and uses them to:

docs/worker.md

@@ -1,6 +1,6 @@
 # Worker project
 
-The worker service is responsible for consuming commands and external events, applying business logic, persisting state, and publishing events that represent domain outcomes. It runs as a background service and does not expose HTTP endpoints. The template wires up MassTransit on RabbitMQ for messaging, Npgsql for persistence (PostgreSQL JSONB, an in-cluster Postgres StatefulSet in the default Kubernetes setup), and OpenTelemetry for traces, metrics, and logs.
+The worker service is responsible for consuming commands and external events, applying business logic, persisting state, and publishing events that represent domain outcomes. It runs as a background service and does not expose HTTP endpoints. The template wires up Conveyo on RabbitMQ for messaging, Npgsql for persistence (PostgreSQL JSONB, an in-cluster Postgres StatefulSet in the default Kubernetes setup), and OpenTelemetry for traces, metrics, and logs.
 
 ## Responsibilities
 
@@ -16,7 +16,7 @@ In this stack, the worker owns the domain work:
 
 - Handlers must be **idempotent** because RabbitMQ can redeliver the same message after a worker restart or ack timeout.
 - OpenTelemetry exports traces, metrics, and logs, so you can answer "what happened to message X?" from a single trace ID.
-- Transient failures go through MassTransit's retry pipeline, and unrecoverable ones land in `_error` queues. The template ships defaults; you set the retry counts and backoff per consumer.
+- Transient failures go through Conveyo's retry pipeline (exponential backoff), and unrecoverable ones are published as a `Fault` and moved to a `{queue}_error` queue. The template ships defaults; you set the retry counts and backoff per consumer.
 - Background jobs belong here too. The template shows a hosted service for cache invalidation.
 
 ## Command handling and event publication

templates/api/ApplicationName.Api.Application.Test/Services/ExampleServiceTest.cs

@@ -9,7 +9,7 @@
 using FakeItEasy;
 using Mapster;
 using MapsterMapper;
-using MassTransit;
+using Conveyo;
 using NUnit.Framework;
 using Shouldly;
 
@@ -25,8 +25,6 @@ public class ExampleServiceTest
 
     private IBus _bus;
 
-    private ISendEndpoint _sendEndpoint;
-
     private ExampleService _subjectUnderTest;
 
     [SetUp]
@@ -36,13 +34,7 @@ public void Setup()
 
         _protoCacheRepository = _fixture.Freeze<IProtoCacheRepository>();
 
-        var queue = new Uri("queue:example");
-        EndpointConvention.Map<CreateExampleCommand>(queue);
-        EndpointConvention.Map<UpdateExampleCommand>(queue);
-
         _bus = _fixture.Freeze<IBus>();
-        _sendEndpoint = _fixture.Freeze<ISendEndpoint>();
-        A.CallTo(() => _bus.GetSendEndpoint(A<Uri>._)).Returns(_sendEndpoint);
 
         var mapperConfig = new TypeAdapterConfig();
         mapperConfig.Scan(typeof(MappingProfile).Assembly);
@@ -220,7 +212,7 @@ public async Task HandleAsync_CreateExampleDto()
         var dto = _fixture.Create<CreateExampleDto>();
 
         var capturedCommand = default(CreateExampleCommand);
-        A.CallTo(() => _sendEndpoint.Send(A<CreateExampleCommand>._, A<CancellationToken>._)).Invokes(
+        A.CallTo(() => _bus.Send(A<CreateExampleCommand>._, A<CancellationToken>._)).Invokes(
             (CreateExampleCommand arg0, CancellationToken _) =>
             {
                 capturedCommand = arg0;
@@ -230,7 +222,7 @@ public async Task HandleAsync_CreateExampleDto()
         await _subjectUnderTest.HandleAsync(dto);
 
         // Assert
-        A.CallTo(() => _sendEndpoint.Send(A<CreateExampleCommand>._, A<CancellationToken>._)).MustHaveHappenedOnceExactly();
+        A.CallTo(() => _bus.Send(A<CreateExampleCommand>._, A<CancellationToken>._)).MustHaveHappenedOnceExactly();
 
         capturedCommand.ShouldNotBeNull();
         capturedCommand.Name.ShouldSatisfyAllConditions(
@@ -246,7 +238,7 @@ public async Task HandleAsync_UpdateExampleDto()
         var dto = _fixture.Create<UpdateExampleDto>();
 
         var capturedCommand = default(UpdateExampleCommand);
-        A.CallTo(() => _sendEndpoint.Send(A<UpdateExampleCommand>._, A<CancellationToken>._)).Invokes(
+        A.CallTo(() => _bus.Send(A<UpdateExampleCommand>._, A<CancellationToken>._)).Invokes(
             (UpdateExampleCommand arg0, CancellationToken _) =>
             {
                 capturedCommand = arg0;
@@ -256,7 +248,7 @@ public async Task HandleAsync_UpdateExampleDto()
         await _subjectUnderTest.HandleAsync(id, dto);
 
         // Assert
-        A.CallTo(() => _sendEndpoint.Send(A<UpdateExampleCommand>._, A<CancellationToken>._)).MustHaveHappenedOnceExactly();
+        A.CallTo(() => _bus.Send(A<UpdateExampleCommand>._, A<CancellationToken>._)).MustHaveHappenedOnceExactly();
 
         capturedCommand.ShouldNotBeNull();
         capturedCommand.CorrelationId.ShouldSatisfyAllConditions(

templates/api/ApplicationName.Api.Application/ApplicationName.Api.Application.csproj

@@ -9,7 +9,7 @@
   <ItemGroup>
     <PackageReference Include="ArgDefender" Version="2.1.0" />
     <PackageReference Include="Mapster.DependencyInjection" Version="10.0.7" />
-    <PackageReference Include="MassTransit" Version="8.5.7" />
+    <PackageReference Include="Conveyo" Version="0.1.0-preview.*" />
     <PackageReference Include="Microsoft.Extensions.Caching.Abstractions" Version="10.0.8" />
     <PackageReference Include="protobuf-net" Version="3.2.56" />
   </ItemGroup>

templates/api/ApplicationName.Api.Application/Services/ExampleService.cs

@@ -5,7 +5,7 @@
 using ApplicationName.Shared.Projections;
 using ArgDefender;
 using MapsterMapper;
-using MassTransit;
+using Conveyo;
 
 namespace ApplicationName.Api.Application.Services;
 

templates/api/ApplicationName.Api.Test/Consumers/LocalEventHandlerTest.cs

@@ -5,7 +5,7 @@
 using AutoFixture;
 using AutoFixture.AutoFakeItEasy;
 using FakeItEasy;
-using MassTransit;
+using Conveyo;
 using NUnit.Framework;
 
 namespace ApplicationName.Api.Test.Consumers

templates/api/ApplicationName.Api/ApplicationName.Api.csproj

@@ -13,8 +13,8 @@
     <PackageReference Include="FluentValidation" Version="12.1.1" />
     <PackageReference Include="FluentValidation.DependencyInjectionExtensions" Version="12.1.1" />
     <PackageReference Include="Mapster" Version="10.0.7" />
-    <PackageReference Include="MassTransit" Version="8.5.9" />
-    <PackageReference Include="MassTransit.RabbitMQ" Version="8.5.9" />
+    <PackageReference Include="Conveyo" Version="0.1.0-preview.*" />
+    <PackageReference Include="Conveyo.RabbitMQ" Version="0.1.0-preview.*" />
     <PackageReference Include="Microsoft.Extensions.Caching.StackExchangeRedis" Version="10.0.8" />
     <PackageReference Include="Microsoft.VisualStudio.Azure.Containers.Tools.Targets" Version="1.23.0" />
     <PackageReference Include="OpenTelemetry.Instrumentation.Runtime" Version="1.15.1" />

templates/api/ApplicationName.Api/Consumers/LocalEventHandler.cs

@@ -3,7 +3,7 @@
 using ApplicationName.Api.Contracts;
 using ApplicationName.Shared.Commands;
 using ApplicationName.Shared.Events;
-using MassTransit;
+using Conveyo;
 using Microsoft.AspNetCore.SignalR;
 
 namespace ApplicationName.Api.Consumers;

templates/api/ApplicationName.Api/Program.cs

@@ -7,10 +7,12 @@
 using ApplicationName.Api.Infrastructure;
 using ApplicationName.Api.Validators;
 using ApplicationName.Shared.Commands;
+using ApplicationName.Shared.Events;
 using FluentValidation;
 using Mapster;
-using MassTransit;
-using MassTransit.Logging;
+using Conveyo;
+using Conveyo.Diagnostics;
+using Conveyo.RabbitMQ;
 using OpenTelemetry.Logs;
 using OpenTelemetry.Metrics;
 using OpenTelemetry.Resources;
@@ -56,18 +58,25 @@ private static void ConfigureServices(IServiceCollection services, Configuration
         services.AddMapster();
         TypeAdapterConfig.GlobalSettings.Scan(Assembly.GetExecutingAssembly());
 
-        // MassTransit + RabbitMQ
-        services.AddMassTransit(i =>
+        // Conveyo + RabbitMQ
+        services.AddConveyo(bus =>
         {
+            // Message type URNs, must match the worker mappings
+            bus.Map<CreateExampleCommand>("applicationname:CreateExampleCommand.v1");
+            bus.Map<UpdateExampleCommand>("applicationname:UpdateExampleCommand.v1");
+            bus.Map<ExampleCreatedEvent>("applicationname:ExampleCreatedEvent.v1");
+            bus.Map<ExampleUpdatedEvent>("applicationname:ExampleUpdatedEvent.v1");
+            bus.Map<ExampleRemoteCodeSetEvent>("applicationname:ExampleRemoteCodeSetEvent.v1");
+
             var uri = new Uri("queue:ApplicationName.Worker");
-            EndpointConvention.Map<CreateExampleCommand>(uri);
-            EndpointConvention.Map<UpdateExampleCommand>(uri);
+            bus.MapEndpointConvention<CreateExampleCommand>(uri);
+            bus.MapEndpointConvention<UpdateExampleCommand>(uri);
 
-            i.AddConsumer<LocalEventHandler>();
+            bus.AddConsumer<LocalEventHandler>();
 
-            i.UsingRabbitMq((ctx, cfg) =>
+            bus.UsingRabbitMq((ctx, cfg) =>
             {
-                cfg.Host(configuration["rabbitmq:host"], configuration["rabbitmq:vhost"], h =>
+                cfg.Host(configuration["rabbitmq:host"]!, configuration["rabbitmq:vhost"]!, h =>
                 {
                     h.Username(configuration["rabbitmq:username"]!);
                     h.Password(configuration["rabbitmq:password"]!);
@@ -128,7 +137,7 @@ private static void ConfigureServices(IServiceCollection services, Configuration
                 {
                     options.RecordException = true;
                 })
-                .AddSource(DiagnosticHeaders.DefaultListenerName) // MassTransit
+                .AddSource(DiagnosticHeaders.DefaultListenerName) // Conveyo
                 .AddRedisInstrumentation()
                 .AddOtlpExporter(configure =>
                 {