# Admin UI Bundled as embedded resources in the NuGet packages is an admin UI built using [React](https://reactjs.org/) and [Blueprint](https://blueprintjs.com/). It uses the [API](../api/index.md) for all data and requests, so anything that's possible to do in the admin UI is possible to do with the API. It's configured to be used by default when you enable the API but you must also do this in your `Program.cs` before calling `app.Run();` ```cs app.UseNexusAdminUI(); ``` If you deploy Nexus to multiple environments such as dev/test/stage/prod/etc you can add an environment name that is shown in the UI, as well as add an info notice bar with any message you want. Both of these settings are used in the demo environment described below to see it in action. ```cs builder.Services.AddNexus().AddApi(options => { options.EnvironmentName = "Dev"; options.AdminUI.InfoNotice = "This is a note to anyone using the UI"; }); ``` The menu in the Admin UI is generated by which features in Nexus you use. If you use all features then all items are visible. If you want to control the menu you can implement the menu loader to return which menu items you want: ```cs builder.Services.AddNexus().AddApi(options => { options.AdminUI.MenuItemsLoader = httpContext => Task.FromResult(new List { NexusAdminUIMenuItems.HealthChecks, NexusAdminUIMenuItems.Jobs, NexusAdminUIMenuItems.Queues, NexusAdminUIMenuItems.Functions, NexusAdminUIMenuItems.Statistics, }); }); ``` The menu loader is an async lambda that gets an HTTP request passed in that lets you have different menu items for different users. The order in the returned list is respected. ## Experimental new version There's a new version of the admin UI that can be enabled by setting `UseExperimentalNewVersion` to true: ```cs builder.Services.AddNexus().AddApi(options => { options.AdminUI.UseExperimentalNewVersion = true; }); ``` At some point `UseExperimentalNewVersion` will be removed and the new UI will replace the old UI once it's verified. It is already working well and has the same features as the previous admin UI, but to be able to avoid potential problems in production it's opt-in for now. ## Hosting the UI on its own If you want to separate the Admin UI from the API you can host the Admin UI on it's own. Just install the NuGet package `CommerceMind.Nexus.UI` in a separate ASP.NET Web app and in your `Program.cs` you add: ```cs builder.Services .AddNexus() .AddAdminUI(options => { options.Path = "nexus-ui"; // Optional, default is "/" options.ApiBaseUrl = "https://your-nexus-api-url.com"; }) .Build(); var app = builder.Build(); app.UseNexusAdminUI(); ``` This will serve the Admin UI without needing a database connection or any other access to Nexus but the API. ## Demo environment There's a demo environment that's deployed to a free tier Azure Web App here:\ [https://commerce-mind-nexus-a9bsbwgth3cgfnde.swedencentral-01.azurewebsites.net/admin/](https://commerce-mind-nexus-a9bsbwgth3cgfnde.swedencentral-01.azurewebsites.net/admin/) You can click on any buttons you like, start any job or do anything with the queues to get a feel for what the system does. :::note[Demo environment] Since the demo environment is using a free tier Azure Web App the scheduler is paused when nobody is using the API or admin UI which means that the jobs won't run exactly according to schedule. ::: ## Authentication See [API docs about authentication](../api/index.md#authentication). ## Authorization See [API docs about authorization](../api/index.md#authorization). --- # Nexus HTTP API A REST API for all Nexus features are included, and it's what the admin UI uses. The API is not enabled by default but all you need to do to enable it is: ```cs builder.Services.AddNexus().AddApi(options => { // The default options are: options.BasePath = "/"; options.EnvironmentName = null; options.RegisterHealthChecksAsAspNetCoreHealthChecks = true; options.AdminUI.Path = "admin"; options.AdminUI.InfoNotice = null; options.AllowAnonymousApiAccess = true; }); ``` When `BasePath` is `/` it means that the API is accessible on `https://servername/api` and if you set it to `/mybasepath/` the API is accessible on `https://servername/mybasepath/api`. When `BasePath` is `/` it also means that the admin UI is accessible on `https://servername/admin` and if you set the base path it to `/mybasepath/` the admin UI is on `https://servername/mybasepath/admin`. ## Authentication One of the options above is if the system should allow anonymous API access or not. If you configure the API not to allow it the system does not do any authorization though and leaves it up to you to authorize the user through eg a cookie (see more about authorization below). The system will check if `HttpContext.User.Identity.IsAuthenticated` and otherwise respond with HTTP status `401` on any API requests. If you wish to add login to the Admin UI you should setup a middleware on the admin route which checks if the user is signed in or not and redirect to/display a login page. Something like this: ```cs // Program.cs builder.Services.AddNexus().AddApi(options => { options.AllowAnonymousApiAccess = false; }); var app = builder.Build(); app.Use(async (context, next) => { if (context.Request.Path.Value?.StartsWith("/admin") == true && context.User.Identity?.IsAuthenticated == false) { context.Response.Redirect("/login"); await context.Response.CompleteAsync(); } else { await next(); } }); app.UseNexusAdminUI(); ``` If you then implement the `/login` route to show a login form that redirects back to `/admin` when the user has logged in your Nexus admin UI and API is fully protected. ## Authorization Nexus offers a very granular level of authorization on endpoints that perform update/create/delete actions. When initializing the Nexus API you can pass an async lamda to authorize an HTTP request like this: ```cs // Program.cs builder.Services.AddNexus().AddApi(options => { options.IsReadOnly = async (httpContext, nexusEndpoint) => { if (nexusEndpoint == NexusEndpoint.DeleteAllQueueItems) { var user = await httpContext.RequestServices.GetRequiredService().GetCurrentUserAsync(); // Only allow admins to nuke all queue items return !user.IsAdmin; } return false; }; }); ``` The `NexusEndpoint` class contains all endpoints that the Nexus API contains which mutates anything. If you want to build a dynamic registry of which users are allowed to use which endpoints you can use the `NexusEndpoint.AllNexusEndpoints` list to base it on. The list is generated and the static name of an endpoint is the same as the static name. That is, this will always be true: ```cs nameof(NexusEndpoint.Xyz) == NexusEndpoint.Xyz.Name ``` The implementation here is of course made up and your implementation would look differently. Any update/create/delete request sent to the API where the `IsReadOnly` lambda returns true will fail with a `403` status. Note that the admin UI will still display all editing controls but will fail with an error message if a user tries to do something unauthorized. ## Endpoints Besides the `NexusEndpoint` class for mutative endpoints you can see a full list of the API endpoints for the Nexus API in the [API reference](/api/reference/nexus-api/). Nexus also has an optional MCP HTTP server for agents and tools. See [Nexus MCP](./mcp.md) for setup and available tool groups. There's a demo environment that's deployed to a free tier Azure Web App here:\ [https://commerce-mind-nexus-a9bsbwgth3cgfnde.swedencentral-01.azurewebsites.net/swagger/index.html](https://commerce-mind-nexus-a9bsbwgth3cgfnde.swedencentral-01.azurewebsites.net/swagger/index.html) You can send requests to any endpoint you'd like to get a feel for what the system does. Since it's a free tier Azure Web App the scheduler is paused when nobody is using the API or admin UI which means that the jobs won't run exactly according to schedule. --- # Nexus MCP `CommerceMind.Nexus.Mcp` exposes Nexus operations as an MCP HTTP server. It is intended for tools and agents that need to inspect or operate Nexus jobs, queues, functions, health checks, and instances without calling the REST API directly. The MCP package is opt-in. Installing or enabling `CommerceMind.Nexus.Api` does not automatically add the MCP server. ## Setup Install/reference the MCP package in the application that hosts Nexus: ```xml ``` Then register the MCP services and map the endpoint: ```cs using CommerceMind.Nexus; using CommerceMind.Nexus.Api.Extensions; builder.Services .AddNexus() .AddApi() .AddMcp(options => { // This is the default value options.McpEndpoint = "nexus-mcp"; // Only allow agents to read. Set this to false to allow agents to update, delete and run jobs and functions. // The default is true. options.ReadOnly = true; }) .Build(); var app = builder.Build(); app.MapControllers(); app.MapNexusMcp(); ``` ## Security The MCP endpoint is an ASP.NET Core endpoint, not an MVC controller. The Nexus API authorization filter does not automatically protect it. Protect it using normal ASP.NET Core middleware and endpoint conventions: ```cs app.UseAuthentication(); app.UseAuthorization(); app.MapNexusMcp().RequireAuthorization(); ``` Many MCP tools can mutate Nexus state, such as starting jobs or updating/deleting queue items. The typical setup is to require an API key for the MCP endpoint. --- # Nexus vs Function as a Service If you're reading the this documentation you might be thinking "well I can do this with scheduled Azure Functions and Azure Service Bus without any need for something like Nexus". And you're right, you can do that and that setup is not bad at all. But Nexus has features that the Azure Function + Service Bus setup doesn't offer out of the box. Instead of seeing Nexus as a replacement for it you can see it as a complement. If you're already using Azure Functions you can have a function that executes Nexus jobs and functions since Nexus doesn't require you to run it as a [continuous background service](./general/hosting.md). Something like this: ```cs public class MyHttpTrigger { private readonly IScheduledJobExecutorService _scheduledJobExecutorService; private readonly INexusFunctionExecutorService _nexusFunctionExecutorService; public MyHttpTrigger(IScheduledJobExecutorService scheduledJobExecutorService, INexusFunctionExecutorService nexusFunctionExecutorService) { _scheduledJobExecutorService = scheduledJobExecutorService; _nexusFunctionExecutorService = nexusFunctionExecutorService; } [FunctionName("MyHttpTrigger")] public async Task Run( [HttpTrigger(AuthorizationLevel.Function, "get", "post", Route = null)] HttpRequest req, ILogger log ) { var jobResults = await _scheduledJobExecutorService.RunAllJobsPendingSinceLastCallAsync(); var functionResults = await _nexusFunctionExecutorService.RunAllFunctionsPendingSinceLastCallAsync(); return new OkObjectResult($"{jobResults.Count} Nexus jobs and {functionResults.Count} Nexus functions executed"); } } ``` You can schedule this function as often as you like and Nexus will make sure to call the jobs and functions that should run. If you schedule the Azure function to run every five minutes and you have Nexus jobs that are scheduled every minute they will only run every five minutes. You should schedule the Azure function to run as often as you want your most often running Nexus job to run. :::note[Cost efficiency] Running Nexus as a scheduled Azure function can be a cost efficient way of running functions and background jobs in test environments since it doesn't require a server that's always running. ::: ## Benefits with Nexus jobs When everything is running smoothly Azure functions and Nexus are quite equal. But Nexus starts to shine when things aren't. When you get a call from a business stakeholder and she tells you to stop the order export you can quickly go in to the admin UI and [disable the job](./jobs/index.md#disabling-a-job) or even guide the stakeholder how to do it herself. Empowering non-developers with regards to background jobs and queues is one of Nexus strengths. With [health status](./general/monitoring.md) and [job logs](./jobs/logs.md) visible in the admin UI it's possible for almost anyone to get at least a basic understanding of why things are failing and help out in troubleshooting. Another big benefit of Nexus is how easy it makes local development. You're not dependant on accessing cloud functionality to build jobs and with SQLite you don't even need a database server. Minimizing the differences between how code runs locally vs in production is very valuable. ## Benefits with Nexus functions Many of the benefits with Nexus jobs applies to Nexus functions as well but they deserve a special mention since the name can make them seem to do the same thing as Azure Functions. The big difference is how easy it is to pass arguments in a type safe way to a Nexus function compared to an Azure function. ## Nexus queues vs Azure Service Bus queues Azure Service Bus queues are great and have a lot of functionality. One of the strength of Azure Service Bus queues is that it can handle a very large number of messages. It has a throughput of around 2000 messages per second. But for many processes you won't have 2000 messages per second. Instead you want more insight into each and every message for these processes. A trade-off in being able to handle that kind of volume is that Azure doesn't offer the same kind of introspection into queues as Nexus does. You can read, filter, paginate and update Nexus queues either through the UI, the API or in your favorite SQL management application. You can use the built in JSON functionality of your database to find messages by properties in the JSON message. Since Nexus makes a different trade-off than Azure it lets Nexus queues offer features that aren't possible with Azure queues. Such as [virtual queues](./queues/virtual.md), [idempotent messages](./queues/index.md#idempotent-messages) and [to store processed messages](./queues/retention.md). Storing processed messages lets you build sophisticated change tracking since you get access to the previous version of the message to let you diff them. If you want to you can also use the `CommerceMind.Nexus.Azure` NuGet package to read [from an Azure queue and add them to a Nexus queue](./queues/azure.md). --- # Database The system uses a SQL database to store meta data and queue messages. At the time of writing we support SQLite, Postgres and SQL Server. Feel free to reach out to us if you need support for another database engine like MySQL or MariaDB. ## Configuring which database to use The support for the different databases exists in their own NuGet packages: ``` CommerceMind.Nexus.Postgres CommerceMind.Nexus.SqlServer CommerceMind.Nexus.Sqlite ``` Install the package(s) you need before you initialize it below. After you've installed the db package(s) you open your `Program.cs` and go to these lines: ```cs builder.Services .AddNexus() .AddFunctions() .AddScheduledJobs() .AddQueues() // Either: .AddPostgresConnection() // Or: .AddSqlServerConnection() // Or: .AddSqliteConnection() // This should be the last thing you do, after you've finished configuring Nexus .Build(); ``` By default the system will call [`IConfiguration.GetConnectionString("postgres")`](https://docs.microsoft.com/en-us/dotnet/api/microsoft.extensions.configuration.configurationextensions.getconnectionstring?view=dotnet-plat-ext-6.0), [`IConfiguration.GetConnectionString("sqlserver")`](https://docs.microsoft.com/en-us/dotnet/api/microsoft.extensions.configuration.configurationextensions.getconnectionstring?view=dotnet-plat-ext-6.0) or [`IConfiguration.GetConnectionString("sqlite")`](https://docs.microsoft.com/en-us/dotnet/api/microsoft.extensions.configuration.configurationextensions.getconnectionstring?view=dotnet-plat-ext-6.0) to get the connection string to the database. If that doesn't work for you can register a lamda to provide your connection string like this: ```cs builder.Services.AddNexus().AddPostgresConnection(options => { options.ConnectionStringLoader = serviceProvider => { // Find and return your connection string here }; }); // Or: builder.Services.AddNexus().AddSqlServerConnection(options => { options.ConnectionStringLoader = serviceProvider => { // Find and return your connection string here }; }); // Or: builder.Services.AddNexus().AddSqliteConnection(options => { options.ConnectionStringLoader = serviceProvider => { // Find and return your connection string here }; }); ``` :::note[Using a ConnectionStringLoader] When you use `ConnectionStringLoader` you need to make sure that it's a very fast operation as the loader will be called every time Nexus runs a database query. You need to make sure to cache the value if you fetch it from an external store such as Azure KeyVault. ::: ## Controlling the connection By default Nexus will open and close database connections for almost every query, and rely on the underlying connection pooling. In some cases you might want to have more control over the connection such as if you want to include the Nexus queries inside a transaction. One option is to use a `DbConnectionScope` like this: ```cs public async Task RunAsync(IEnqueuer enqueuer) { var dbConnection = CreateDbConnection(); using (var transaction = connection.BeginTransaction()) using (var _ = new DbConnectionScope(connection, transaction)) { // The enqueuer will now use dbConnection to run the database queries to enqueue the message await enqueuer.EnqueueAsync(new SomeMessage()); transaction.Commit(); } // This message will be enqueued using the default way with a new connection await enqueuer.EnqueueAsync(new SomeMessage()); } ``` The effect of this is that any Nexus services inside the `using` block will use the `dbConnection` object rather than creating new connections. Which means that you can start a transaction in `dbConnection` and automatically have all Nexus operations inside the `using` block be a part of that transaction. ## Transient errors Nexus has built-in retries of transient errors using [Polly](https://github.com/App-vNext/Polly) but if you want to you can add custom retry logic by creating your own implementation of the `ITransientDatabaseErrorRetryer` interface. ## Database maintenance Nexus includes two jobs that are enabled by default. One job is called `RebuildDatabaseIndexes` and will rebuild database indexes for Postgres and SQL Server. It's scheduled to run daily at midnight. If you're using SQLite this job won't do anything as it's not needed to rebuild SQLite indexes. By default this job will only rebuild indexes in tables created by Nexus. If you want it to rebuild indexes for all tables in the database you can can pass an option to `AddNexus().AddPostgresConnection()` or `AddNexus().AddSqlServerConnection()`: ```cs builder.Services.AddNexus().AddPostgresConnection(options => { options.LimitDatabaseMaintenanceToNexusTables = false; }); // Or: builder.Services.AddNexus().AddSqlServerConnection(options => { options.LimitDatabaseMaintenanceToNexusTables = false; }); ``` The other job is called `RunDatabaseMaintenanceJob` and is also enabled by default and but not scheduled. For Postgres that will run `VACUUM` and then `ANALYZE`. If you're using SQL Server this job won't do anything as SQL Server has automatic update of query statistics. If you're using SQLite this job will not do anything either. ## Remove database maintenance jobs If you want to completely hide these jobs you can pass a filter lambda when you register the jobs system to exclude these jobs like this: ```cs builder.Services.AddNexus().AddScheduledJobs(options => { options.ScheduledJobFilter = jobType => jobType.Name != "RebuildDatabaseIndexes" && jobType.Name != "RunDatabaseMaintenanceJob"; }); ``` If you want to run the jobs but do something different than the defaults you can create your own implementation of the `IDatabaseMaintenanceService` interface and register it with the service provider like this: ```cs builder.Services.AddSingleton(); ``` ## Using another database engine If you want to use another database engine such as MySQL or MariaDB you can always implement it and open a PR to the Github repository. If you don't have access yet to the repository just ping us! Another option is of course to hire us to implement it for you. --- # Hangfire vs Nexus [Hangfire](https://www.hangfire.io/) is a great library for background processing and there's a lot of similarities between Hangfire and Nexus. Nexus [Functions](../nexus-functions/index.md) was built with Hangfire as an inspiration and it's especially the Functions part of Nexus and Hangfire that have a lot of similarities. Both Nexus Functions and Hangfire support persistence and scaling out to more than one processing server and are both very easy to setup to run some code in the background. Both have a built-in UI to visualize the background tasks, both have automatic retries, etc. Nexus has a slight edge over Hangfire in that Hangfire was created before Nexus and supports .NET Framework. Whereas Nexus is .NET Core only and follows the modern .NET patterns for configuration and initialization which makes it slightly easier to get started with. ### Differences What sets Hangfire and Nexus apart is that Hangfire uses continuations and batches for more advanced scenarios. For such cases Nexus offers [background jobs](../jobs/index.md) and [queues](../queues/index.md) (eg [virtual queues](../queues/virtual.md)) instead to orchestrate complex data flows. A batch in Hangfire corresponds to a queue in Nexus with a [batch processing job](../queues/jobs.md#processing-messages-in-batches). And a continuation in Hangfire corresponds to multiple queues in Nexus where a processing job for one enqueues to another as an effect of processing a message. In Nexus a queue message is a core concept when you design your Nexus application. You design your data flows and background tasks around queue messages and the data they contain. In Hangfire a queue message is instead simply the arguments passed to your background task method. While that's technically quite similar since they're both serialized and stored in a database the difference has some effects on your application. With Hangfire you start with a service method that you want to call in the background. With Nexus you instead often start with what data a queue message contains and add processing of that data after the message has been created. This means that Hangfire might be a better fit when you just want asynchronous and distributed background processing internally in your application and you don't really care about the data as its own entity, you just care about passing arguments to methods. And Nexus becomes a better fit if you want other applications to enqueue messages to your queue since all queues automatically get [API endpoints for that](../queues/enqueueing.md#enqueueing-over-http-using-the-api). Differentiating between the data and the processing of that data can also have benefits in larger application where those are different concerns that you don't want to mix. In Hangfire the piece of code that wants something to happen in the background is tightly coupled with the piece of code that will actually be called in the background. In Nexus the piece of code that enqueues a message for background processing doesn't even have to be in the same application as the piece of code that does the processing of that data. --- # Hosting The service is built to allow both running as a continous background service as well as a service that periodically starts up, executes outstanding jobs and then exits. The default is to run as a continous background service but you can configure it to not use the background service: ```cs builder.Services .AddNexus() .AddFunctions(options => { options.UseBackgroundService = false; }) .AddScheduledJobs(options => { options.UseBackgroundService = false; }); ``` When setting `UseBackgroundService` to `false` you need to call `IScheduledJobExecutorService.RunAllJobsPendingSinceLastCallAsync()` and `INexusFunctionExecutorService.RunAllFunctionsPendingSinceLastCallAsync()` yourself. It will check which jobs have pending runs since the last time the method was called; run them and then return. Which means that you can have the system running in for example an Azure Function or Azure Container Instance that runs periodically. :::note[Running in production] Note that it's recommended to run the background service for the production environment since it offers more granular scheduling, health checks and the admin UI + API. ::: ## Application type The typical use case is to host the service inside an ASP.NET Core application but any type of .NET6+ compatible application can be used to run the scheduler and queue system. The API however requires ASP.NET Core. ## InstanceId In a multi-server/instance environment Nexus keeps track of which server is doing what using `INexusInstance.InstanceId`. By default that id is the value of `Dns.GetHostName()` except if you're running in Azure App Service where a single VM can be running multiple instances of Nexus using deployment slots. In this case Nexus will calculate an instance id using the current process id and the environment variable `WEBSITE_INSTANCE_ID` that's set by the app service. If you for some reason need to run multiple Nexus applications on the same VM you need to implement your own version of `INexusInstance` where `InstanceId` is guaranteed to be unique at any given time. There should never be two applications running at the same time with the same instance id. As a last resort you can set it to a random value at startup but that decreases Nexus ability to track and heal. If there's an application crash Nexus won't be able to update jobs and functions to a non-running state because Nexus is unaware that the specific instance id crashed. When using a stable id and Nexus starts up it will reset all jobs and functions that claims to be running on that instance. If the instance id keeps changing then you need to reset it yourself in case of a crash. ## Graceful shutdown The importance of graceful shutdown isn't specific to Nexus but you should give your application enough time to shutdown gracefully. When you have a Web API that's typically not something you have to think about because a shutdown only needs to finish the current HTTP requests and then it can shutdown without issues. But in the case of background jobs and processing a deploy will trigger a shutdown but you might have an important job running that shouldn't be killed in the middle of execution. All jobs are given a `CancellationToken` which will fire if the application wants to shut down, so it's important for all your jobs to respect that and exit gracefully. Read more about [that here](../jobs/index.md#cancellationtoken). The default shutdown timeout in .NET 6 is 30 seconds: [https://github.com/dotnet/runtime/blob/main/src/libraries/Microsoft.Extensions.Hosting/src/HostOptions.cs](https://github.com/dotnet/runtime/blob/main/src/libraries/Microsoft.Extensions.Hosting/src/HostOptions.cs) This is the time .NET waits after signaling to all background services that a shutdown has initiated until the process terminates. If the process terminates in the middle of running a job you won't know what the status was for that run. Was it done? Can it be restarted? If the process terminated on a server that is then shutdown and replaced with another instance Nexus won't automatically mark it as not running and you'll need to reset the job yourself through the UI or API. The default tiemout of 30 seconds is probably enough for you, but you need to consider it. It might also be that your hosting provider doesn't respect this timeout and has a shorter timeout. Verifying the timeout you have is quite easy. You can create a background service like this: ```cs public class VerifyGracefulShutdownBackgroundService : BackgroundService { private readonly ILogger _logger; public VerifyGracefulShutdownBackgroundService(ILogger logger) { _logger = logger; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { while (!stoppingToken.IsCancellationRequested) { await Task.Delay(500); } _logger.LogInformation("Shutdown initiated"); while (true) { _logger.LogInformation("Still alive..."); await Task.Delay(500); } } } ``` And then in `Program.cs` you do: ```cs serviceCollection.AddHostedService(); ``` Deploy the new code, and then deploy the same code again. After that you check the logs for the timestamps of the `Shutdown initiated` entry and the last `Still alive...` entry. If the period between those are long enough you're good to go. What should be considered long enough is up to you, but 30 seconds is a good starting point. ### Graceful Nexus shutdown For cases when you want Nexus to shutdown before the application starts to shutdown you can use the `IGracefulShutdownService` to tell Nexus to stop all background processes. Calling `IGracefulShutdownService.InitiateShutdown()` will stop all Nexus background processes and Nexus will attempt to cancel all running jobs. You can also initiate a graceful shutdown through the API using `POST api/graceful-shutdown/initiate` but only if you've set `AllowGracefulShutdownThroughApi` to `true` when calling `AddNexus().AddApi()`. Note that in a multi-instance environment you need to call this explicitly for every instance. In a multi-server environment you might want to broadcast an event that all Nexus instances listens to and calls `IGracefulShutdownService.InitiateShutdown()`. Once you've initiated graceful shutdown there's no going back. You need to restart the application for the background processes to start again. ### Soft graceful shutdown In some cases you might want to let any jobs that has started finish before you deploy but you still want to prevent new jobs from starting. In that case you can use the soft graceful shutdown in Nexus which does the same thing as a graceful shutdown except that it won't attempt to cancel running jobs or functions. Just like graceful shutdown this is an unrecoverable state, you need to restart the application for jobs to start again. You initiate a soft graceful shutdown either through the API with `POST api/graceful-shutdown/initiate-soft` or through code with `IGracefulShutdownService.InitiateSoftShutdown()`. # Azure Web Apps [Azure Web Apps](https://azure.microsoft.com/en-us/products/app-service/web/) is a popular and easy way of deploying and managing a .NET application which works great together with Nexus. However since Azure Web Apps are built for a web/API workload and not a worker workload there's some considerations when using it with Nexus. Something to be aware of is that a deployment will start your new application version on the same machines as the current version is running on and it will start the new version before the old version is signaled to initiate shutdown. For a Web API this doesn't really matter as the new application version won't receive traffic until the deploy swaps the active application and then all traffic goes to the new version. But background services will be running on both at the same time. This isn't a problem in general but something you need to be aware of. Nexus will ensure that there's only one instance of a job running at the same time, but if you have dependencies between jobs it can affect you. During the time of a deploy there can be two different versions of the same [job](../jobs/index.md) or [function](../nexus-functions/index.md) running. Nexus will ensure that a single job never runs at the same time in different instances, but if you've scheduled a job to run every second then the old and new version of the job will compete to start running the job during the deployment. Let's say that you have a Nexus function that looks like this: ```cs public class ExampleService { public void DoSomethingInteresting() { _logger.LogInformation("This is interesting"); } } ``` And you have calls to that service frequently scheduled using `await _nexusFunction.RunInBackgroundAsync(x => x.DoSomethingInteresting());`. Then you change the log to `"This is very interesting!"`. During the deploy of that new implementation your log can look like this: ``` [20XX-XX-XX XX:XX:01] This is interesting [20XX-XX-XX XX:XX:02] This is interesting [20XX-XX-XX XX:XX:03] This is very interesting! [20XX-XX-XX XX:XX:04] This is interesting [20XX-XX-XX XX:XX:05] This is very interesting! [20XX-XX-XX XX:XX:06] This is interesting [20XX-XX-XX XX:XX:07] This is very interesting! [20XX-XX-XX XX:XX:08] This is very interesting! ``` It's up to you to consider if this timing is an issue for you, or if it's fine that different versions of the function, job or enqueuing will run at the same time. :::note[Azure Container Apps or Google Gloud Run] Since Azure App Service/Web Apps isn't optimized for a worker/background service workload you should consider using Azure Container Apps or Google Gloud Run instead which both has great support for both worker and web API workloads. ::: --- # Instance events Nexus is built to work on multiple instances/servers at the same time in a scaled out environment. You can have as many servers running jobs as you like and the system still guarantees that a single job will only run on a single instance at any given time and that only a single queue message is processed by a single server. In some cases the system needs to communicate with all running instances which is called instance events. At the heart of it is [`IInstanceEvent`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Core/InstanceEvents/IInstanceEvent.cs) and [`IInstanceEventBroker`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Core/InstanceEvents/IInstanceEventBroker.cs). `IInstanceEvent` represents an event that should be sent to either all or a specific instance. Specific implementations of the interface exists for different cases such as broadcasting a message to cancel a job or a message to start queue processing for a specific queue. `IInstanceEventBroker` is the interface to use when you want to listen to or broadcast these events and it's up to the implementation of it to make sure that the events are sent to the correct instance(s). ## Single instance mode There's two implementations of the broker interface built in. One that you can use if you only run a single instance at any given time. It uses a memory based solution to register listeners and broadcast events. You register it by calling this in your DI setup code: ```cs builder.Services.AddNexus().AddSingleServerInstanceEvents(); ``` ## Multi instance mode If you have multiple instances running like in an auto scaled environment where there might be multiple servers jobs running you can't use the single instance mode. Your options are to either use database polling, use Azure Service Bus, or write your own broker. Both the database polling and Azure Service Bus implementation takes an option func that lets you specify `BrokerType` which is either `InstanceEventBrokerType.ProduceAndConsumeEvents` or `InstanceEventBrokerType.ProduceEvents`. For an application that only adds messages to Nexus queues using the `CommerceMind.Nexus.Queueing` package you can set the broker type to `InstanceEventBrokerType.ProduceEvents`. This means that Nexus will skip setting upp polling/subscriptions for that application. ```cs builder.Services.AddNexus().AddDatabasePollingInstanceEvents(options => { options.BrokerType = InstanceEventBrokerType.ProduceEvents; }); // Or: builder.Services.AddNexus().AddAzureServiceBusInstanceEvents(options => { options.BrokerType = InstanceEventBrokerType.ProduceEvents; }); ``` :::note[CommerceMind.Nexus.ApiClient] If you're using the `CommerceMind.Nexus.ApiClient` package that [enqueues messages over the HTTP API](/queues/enqueueing#enqueueing-using-the-apiclient-package) you don't need to configure instance events at all for that application. ::: ### Database polling If you want to limit external dependencies you can use the database to store instance events and have your servers poll the database for new events. You configure database polling events like this: ```cs builder.Services.AddNexus().AddDatabasePollingInstanceEvents(options => { options.PollDelay = TimeSpan.FromSeconds(1); // Default polling is every second }); ``` This implementation will check if there's multiple running instances and then save a copy of the event for each running instance except for itself where it broadcasts the event directly. It knows which instances are running by using the [`INexusInstanceHeartbeatService`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Jobs/Heartbeats/INexusInstanceHeartbeatService.cs). Read more about [heartbeats here](./monitoring.md). ### RabbitMQ If you're using RabbitMQ you can use that as your instance event broken. Install the package `CommerceMind.Nexus.RabbitMQ` and configure the instance event broker: ```cs builder.Services .AddNexus() .AddRabbitMq(options => { options.UserName = "..."; options.Password = "..."; options.HostNames = ["localhost"]; // Or: options.ConnectionString = new Uri("amqp://..."); }) .AddRabbitMqInstanceEvents(options => { // This is the default value, you only need to change this if you're using // multiple Nexus applications with the same RabbitMQ instance. options.ExchangeName = "nexus_instance_events"; }); ``` ### Azure Service Bus If you're using Azure there's a NuGet package called `CommerceMind.Nexus.Azure` which uses a service bus topic to broadcast instance events. After you've installed it you configure it like this in your `Program.cs`: ```cs builder.Services.AddNexus().AddAzureServiceBusInstanceEvents(); ``` The default name of the topic is `nexus-instance-events` but that can be configured by passing an options func like this: ```cs builder.Services.AddNexus().AddAzureServiceBusInstanceEvents(options => { options.TopicName = "my-topic-name"; }); ``` By default Nexus will look for the connection string using `IConfiguration.GetConnectionString("azureServiceBus")` but you can add a loader func on the options object if you need to fetch it from somewhere else: ```cs builder.Services.AddNexus().AddAzureServiceBusInstanceEvents(options => { options.ConnectionStringLoader = (serviceProvider) => "Endpoint=sb://xxx.servicebus.windows.net/;..."; }); ``` Nexus will automatically create topic subscriptions for each instance which works best in an auto scaled environment where you don't know how many instances you will have. In order to be able to create subscriptions on-the-fly you also need to configure an administration connection string. You do that either by setting the connection string `azureServiceBusAdmin` with `IConfiguration` or set the `AdminConnectionStringLoader` property on the options object. If you're not using an auto scaled environment and you instead want to create the subscriptions yourself you can skip configuring the admin connection string and instead add `SubscriptionNameLoader`: ```cs builder.Services.AddNexus().AddAzureServiceBusInstanceEvents(options => { options.SubscriptionNameLoader = (serviceProvider) => "my-subscription"; }); ``` :::note[One subscription per server] Make sure that all instances use a different subscription when you're manually creating them. This setup only works if you have a number of dedicated machines such as VMs that are always running. ::: #### Batching events By default instance events are batched so that events that happen within 0.5 seconds are sent as a single service bus message. This is done to limit the number of messages since Azure Service Bus has a cost per message. If a 0.5 second delay doesn't work for you it's possible to change the delay like this: ```cs builder.Services.AddNexus().AddAzureServiceBusInstanceEvents(options => { options.DebounceDelay = TimeSpan.FromSeconds(5); options.MaxMessageBatchSize = 10; // The default is 10 }); ``` The `MaxMessageBatchSize` is to guard against getting a message size that is bigger than what's allowed. So if there's more than 10 events within 0.5 seconds they will get sent as mutiple messages. If you're using the standard tier service bus the limit is 256 kB but if you're using the premium tier the limit is 100 MB. ### Redis If you're using Redis you can use that as your instance event broker. Install the package `CommerceMind.Nexus.Redis` and configure the instance event broker: ```cs builder.Services .AddNexus() .AddRedisInstanceEvents(options => { options.ConnectionString = "localhost:6379"; }); ``` If you already have an `IConnectionMultiplexer` registered in DI you can reuse it instead of providing a connection string: ```cs builder.Services .AddNexus() .AddRedisInstanceEvents(options => { options.ConnectionMultiplexerFactory = sp => sp.GetRequiredService(); }); ``` The default pub/sub channel name is `nexus_instance_events` but that can be changed if you're using the same Redis instance for multiple Nexus applications: ```cs builder.Services .AddNexus() .AddRedisInstanceEvents(options => { options.ConnectionString = "localhost:6379"; options.ChannelName = "my-app-instance-events"; }); ``` Redis uses pub/sub for broadcasting, so all subscribers on the channel receive every message — no per-instance group or subscription configuration is needed. ### Kafka If you're using Kafka you can use that as your instance event broker. Install the package `CommerceMind.Nexus.Kafka` and configure both the Kafka connection and the instance event broker: ```cs builder.Services .AddNexus() .AddKafka(options => { options.ConsumerConfig.BootstrapServers = "localhost:9092"; options.ConsumerConfig.GroupId = "my-nexus-app"; options.ProducerConfig.BootstrapServers = "localhost:9092"; }) .AddKafkaInstanceEvents(); ``` The default name of the topic is `nexus_instance_events` but that can be configured: ```cs builder.Services .AddNexus() .AddKafka(options => { ... }) .AddKafkaInstanceEvents(options => { // Change this if you're using the same Kafka instance for multiple Nexus applications. options.TopicName = "my-topic-name"; }); ``` If you need separate producer/consumer configuration specifically for instance events you can override them via the options: ```cs builder.Services .AddNexus() .AddKafka(options => { ... }) .AddKafkaInstanceEvents(options => { options.ProducerConfig = new ProducerConfig { BootstrapServers = "localhost:9092" }; options.ConsumerConfig = new ConsumerConfig { BootstrapServers = "localhost:9092", GroupId = "my-nexus-app" }; }); ``` Because Kafka delivers messages to only one consumer per consumer group, every instance must use a **unique** `GroupId` in order to receive all broadcast events. A common pattern is to include the hostname or a random ID in the group ID: ```cs builder.Services .AddNexus() .AddKafka(options => { options.ConsumerConfig.BootstrapServers = "localhost:9092"; options.ConsumerConfig.GroupId = $"nexus-instance-events-{Environment.MachineName}"; options.ProducerConfig.BootstrapServers = "localhost:9092"; }) .AddKafkaInstanceEvents(); ``` ## Custom application instance events You can use the Nexus instance event infrastructure to broadcast events within your application. To do this, define your own event class that implements the `IApplicationInstanceEvent` interface and define handlers that implement the `IApplicationInstanceEventHandler` interface. You can then use the `CommerceMind.Nexus.Abstractions.Core.InstanceEvents.IApplicationInstanceEventPublisher` to publish events that will be broadcast to all running instances. Eg: ```cs public class MyEvent : IApplicationInstanceEvent { public string? MyProperty { get; set; } } public class MyEventHandler : IApplicationInstanceEventHandler { public async Task HandleAsync(MyEvent myEvent) { // Do something interesting with the event } } builder.Services.AddTransient>(); var app = builder.Build(); var publisher = app.Services.GetRequiredService(); await publisher.PublishAsync(new MyEvent()); ``` ## Writing your own broker If you don't feel like any of the above options works for you then you can always create your own implementation of [`IInstanceEventBroker`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Core/InstanceEvents/IInstanceEventBroker.cs). An important thing to think about is that `IInstanceEvent` has an `InstanceId` property that can be `null` or have a value. If it's `null` then the event should be sent to all instances. But if it's not null it should only be published as an event on the instance where [`INexusInstance.InstanceId`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Jobs/INexusInstance.cs) is the same as `IInstanceEvent.InstanceId`. You can of course decide to send the event to all instances and then do the check in your implementation before calling `OnInstanceEventRecieved?.Invoke()` if the instance id matches. When you're done with the implementation all you need to do is register it like this: ```cs builder.Services.AddSingleton(); ``` --- # Nexus and JSON Nexus uses JSON heavily for serialization of different data structures. In v1 of Nexus the default serialization used `Newtonsoft.Json` but since v2 the default is `System.Text.Json` instead. If you're using `Newtonsoft.Json` you can install the package `CommerceMind.Nexus.NewtonsoftJson` and call `builder.Services.AddNexus().AddNewtonsoftJson()` to use `Newtonsoft.Json` for serialization instead of `System.Text.Json`. ## Customizing the serialization Both `System.Text.Json` and `Newtonsoft.Json` allows you to add custom converters, but `System.Text.Json` doesn't allow you to modify the global serialization options. In Nexus all of the serialization is controlled by the different serialization interfaces, except for the response data in the API. Nexus passes object instances to ASP.NET which then handles the serialization. Nexus will however still control the serialization of queue messages inside the API models. Which mean that you might get a response looking like this: ```json { "id": 4823771, "status": "Processed", "retryCount": 0, "message": { "Id": null, "SomeProperty": null, "ExampleEnum": "Value1" } } ``` Where the outer object has gotten kebabCased by ASP.NETs JSON formatting rules but the `message` object is PascalCased because that's the default formatting in Nexus. Similar to ASP.NET there's also a hook to modify the Nexus serialization: ```cs builder.Services.AddNexus().ConfigureJsonOptions(options => { options.SerializerOptions.Converters.Add(new MyCustomConverter()); }); ``` ## Dealing with interfaces and abstract types in serialization Sometimes Nexus needs to deal with serializing to and from abstract types such as interfaces or abstract classes. In those cases Nexus will serialize a .NET type name into the JSON structure so the deserialization later on will know which type to deserialize to. Note that Nexus never does this on input from data coming in through the API, only on data that Nexus itself has serialized and stored in the database. This works just like Newtonsoft.Jsons `TypeNameHandling`. Since there are security implications in this Nexus also includes a hash of the type name in the serialized JSON which is then verified before loading the type during deserialization. You can control this with the following configuration: ```cs builder.Services.AddNexus().ConfigureJsonOptions(options => { // The salt used when hashing types options.TypeHashSalt = "mysalt"; // If a $typehash property should be included or not in the JSON. Including a type hash // makes the serialized JSON incompatible with Newtonsoft.Json which you might want to keep // compatibility with. options.WriteTypeHash = false; // If Nexus should reject any JSON with a $type but without a $typehash options.RequireTypeHash = true; }); ``` You're able to even further control this with the interfaces `ISerializationBinder` and `ISerializationTypeHashCalculator`. The `ISerializationBinder` interface is what reads and writes a .NET type to it's string representation and works just like the standard .NET [ISerializationBinder`](https://learn.microsoft.com/en-us/dotnet/api/system.runtime.serialization.serializationbinder?view=net-8.0). `ISerializationTypeHashCalculator` is responsible for creating a hash for a type name. If you ever want to change the type hash salt you need to implement your own `ISerializationTypeHashCalculator` to handle both the new and old salt until no data is using the old salt anymore. --- # Metrics with Open Telemetry Nexus exposes a number of metrics using Open Telemetry. Open Telemetry is an open standard for exposing metrics and telemetry that can be plugged into almost any monitoring service such as Azure Application Insights, Splunk, Elastic APM etc. If you're using Azure Application Insights/Azure Monitor for example you can install the Open Telemetry exporter for Azure Monitor: [https://learn.microsoft.com/en-us/azure/azure-monitor/app/opentelemetry-configuration?tabs=aspnetcore](https://learn.microsoft.com/en-us/azure/azure-monitor/app/opentelemetry-configuration?tabs=aspnetcore) Install the NuGet package `Azure.Monitor.OpenTelemetry.AspNetCore` and then call: ```cs builder.Services .AddOpenTelemetry() .UseAzureMonitor(options => { options.ConnectionString = ""; }); ``` This will export the metrics and telemetry data from Nexus into Azure Monitor. You'll be able to see counters such as how many queue messages are processed and how many fail. Each job run is exported as a trace with a start and end in the same way as Application Insights represents an HTTP request. If you also use the Application Insights SDK you'll see all the database queries or HTTP requests that happens inside that job run as well. This is not specific to Application Insights though as Nexus has no knowledge of that. It only exposes the telemetry in the Open Telemetry standard format and then its up to you to send that data to whichever APM you're using. --- # Monitoring The system integrates with and extends [Health checks in ASP.NET Core](https://docs.microsoft.com/en-us/aspnet/core/host-and-deploy/health-checks?view=aspnetcore-6.0). All functions, jobs, and queues automatically get health checks that signals their health, the only thing you need to do in an ASP.NET Core application is call this in your `Program.cs`: ```cs app.MapHealthChecks("/health"); ``` Now that endpoint will fail if any functions, jobs, or queues are failing/has errors. The health checks are also included in the admin UI to get an overview of the system status. If you're not using ASP.NET you can still run the health checks by calling this: ```cs builder.Service.AddNexus().AddHealthChecks(options => { options.UseHealthCheckBackgroundService = true; }); ``` When not using ASP.NET and enabling `UseHealthCheckBackgroundService` you can still register an implementation of [`IHealthCheckPublisher`](https://docs.microsoft.com/en-us/dotnet/api/microsoft.extensions.diagnostics.healthchecks.ihealthcheckpublisher?view=dotnet-plat-ext-6.0) in `IServiceCollection` like this: ```cs builder.Service.AddSingleton(); ``` The Nexus background service will then periodically call your publisher twice per minute. Since it's a singleton instance you should store the last status and only ping someone if the status goes from healthy to unhealthy. :::note[Only for non ASP.NET] If you're using ASP.NET the `IHealthCheckPublisher` is still called even if nobody is pinging the `/health` endpoint. ::: You can see the health checks included in the demo environment here:\ [https://commerce-mind-nexus-a9bsbwgth3cgfnde.swedencentral-01.azurewebsites.net/admin/healthchecks](https://commerce-mind-nexus-a9bsbwgth3cgfnde.swedencentral-01.azurewebsites.net/admin/healthchecks) ### Adding more health checks There's two health check interfaces included. One that runs on demand, and one that is scheduled. #### `INexusHealthCheck` Use this implementation for checks that you always want to run on demand. Their results are never cached but should then respond quickly and not be very costly. #### `INexusScheduledHealthCheck` In some cases you want a health check to run on a schedule rather than firing every time someone pings the health status endpoint. For this you can use the `INexusScheduledHealthCheck` interface which much like jobs has a cron schedule. When the service executes the health checks it will only execute a scheduled check if enough time has passed since the last time it was called. Otherwise the system will return the previous result for that check. Another benefit of `INexusScheduledHealthCheck` is that the `CheckHealthAsync()` method is passed the previous result of that check. This means that you can determine health based on a previous run. An example: ```cs public class ExampleScheduledHealthCheck : INexusScheduledHealthCheck { public string Schedule => "@every_minute"; public string Name => "MyHealthCheck"; // This needs to be unique public string DisplayName => "Health check running every minute"; public string? Description => "A longer description of what the health check does"; // You can return null if you don't need a description public string? Category => "My health checks category"; // Used to group checks in the Admin UI public IEnumerable Tags => new string[] { }; public string? AdminUILinkUrl => null; public async Task CheckHealthAsync(NexusHealthCheckResult? previousHealthCheckResult, ScheduledNexusHealthCheckInitiator initiator, HealthCheckContext context, CancellationToken cancellationToken) { var previousCount = (int?)previousHealthCheckResult?.Result.Data?["count"]; var data = new Dictionary(); var currentCount = GetCurrentCount(); data["count"] = currentCount; if (currentCount - previousCount < 10) { return HealthCheckResult.Unhealthy($"Everything is NOT fine!", null, data); } return HealthCheckResult.Healthy($"Everything is fine!"); } private int GetCurrentCount() { // Look up value } } ``` Here we're using the `Data` property on `HealthCheckResult` to store any properties we need in the next run to determine the health. This can be used to see if counters such as the amount of `Pending` messages in a certain queue increases too much. #### Schedule The `Schedule` property is a cron expression just like [`DefaultSchedule`](../jobs/index.md#ischeduledjobdefaultschedule) on jobs, and it can be any expression supported by [Cronos](https://github.com/HangfireIO/Cronos). The above example uses one of the supported [macros](https://github.com/HangfireIO/Cronos#macro) in Cronos. Note that there's a difference between the `DefaultSchedule` property on `IJob` and the `Schedule` property on `INexusScheduledHealthCheck`. The default schedule on a job is only read by Nexus the first time the job is deployed and registered in the database. Using eg `CronSchedule.EveryMinute()` for the job will generate a random second (eg `16`) for the second during in a minute that the job will start on. This means that you can have many jobs using `CronSchedule.EveryMinute()` but they won't start at exactly the same time. Instead they will start on random seconds to spread out the load they generate. The health checks `Schedule` property on the other hand are read every time Nexus checks if it's time to invoke them. So if you use `CronSchedule.EveryMinute()` for a health check it'll sometimes execute the check multiple times in a minute and sometimes less frequent than a minute. Eg if the clock is `12:00:10` and the schedule says `9 *, *, *, *, *` the check is executed. And if the next time the schedule is evaluated it becomes `19 *, *, *, *, *` it means that it'll only be 10 seconds between the times the check is called. If you still want to generate a random second for a health check you can generate it when the application starts and then keep returning that cron schedule. Like this: ```cs public class ExampleScheduledHealthCheck : INexusScheduledHealthCheck { private static string _schedule = CronSchedule.EveryMinute(); public string Schedule => _schedule; } ``` This will make sure that the schedule is stable during the time that the application lives and only generate a new schedule when the application is restarted or a new version is deployed. ## Registering custom health checks If you created an implementation of `INexusHealthCheck` or `INexusScheduledHealthCheck` you register it by calling: ```cs builder.Services.AddNexus().AddHealthCheck(); ``` --- # 2.58.0 - In the [new Admin UI](../admin-ui/index.md#experimental-new-version) it's now possible to [upload files as parameters to jobs](../jobs/parameters.md) - Various improvements to the [new Admin UI](../admin-ui/index.md#experimental-new-version) - Since the webhook integration with Microsoft Teams has been deprecated by Microsoft there's now a new [Teams integration](../healthchecks/teams.md) # 2.57.2 _2026-08-06_ - Performance improvement in job-start loop when setting job loop delay to less than 1s - Fix potential memory leak in in-memory queues # 2.57.1 _2026-05-20_ - Minor fixes and tweaks to the new admin UI # 2.57.0 _2026-05-19_ - Add a **new admin UI** which can be used by [setting an option](../admin-ui/index.md#experimental-new-version) on the Admin UI options during configuration # 2.56.0 _2026-05-13_ - Support idempotency and message ids in the [Redis](../queues/redis.md) external pending storage - Add `[IdempotencyKey]` [as a way to control how idempotency is calculated](../queues/index.md#idempotent-messages) # 2.55.0 _2026-05-12_ - Make it possible for Nexus to process messages directly from RabbitMQ, Azure Service Bus, Redis, and Kafka using [external pending storage](../queues/external-pending.md) - Add Redis package for instance events and external pending storage - Improve performance of memory queues # 2.54.1 _2026-05-08_ - Fix admin UI taking over all paths that starts with the admin UI path, eg taking over `/nexus-path` when the admin UI is at `/nexus`. # 2.54.0 _2026-05-08_ - Add `VirtualHost` to RabbitMQ options. # 2.53.0 _2026-05-08_ - Add new package `Nexus.Mcp` to expose an MCP endpoint to let AI agents work with Nexus. See [docs about MCP](../api/mcp.md) # 2.52.0 _2026-05-05_ - Prevent jobs scheduled at every second from being able to run more frequent than once per second when the job loop is configured to run more often than once per second. - Add Entity Framework integration for IEnqueuer where it waits for EFs transaction to commit queue messages together with EF - Add more logging in Azure Service bus queue ingestion # 2.51.0 _2026-03-10_ - Add option `MaxStartupParallelization` which will attempt to create queue tables in parallel. Use this if faster startup time is important and you have a lot of queues. - Fix issue causing job mutexes to be released when they shouldn't. # 2.50.3 _2026-01-30_ - Don't wait for Rabbit connection before starting Nexus when using RabbitMQ # 2.50.2 _2025-12-16_ - Fix issue with property default values (eg false and 0) in partial messages # 2.50.1 _2025-12-10_ - Use incremental backoff delay when job loop fails due to database errors instead of a fixed 30s delay # 2.50.0 _2025-12-10_ - Allow [configuring the job loop delay](../jobs/index.md#job-loop-delay) - Avoid null referecnce exception in health check that can happen during high parallelism # 2.49.3 _2025-11-28_ - Remove dependency on System.Linq.Async - Show logs with level Critical as red in job log viewer - Handle health check result description being null # 2.49.2 _2025-11-20_ - Force a new async context when starting a job from an instance event to prevent enrolling the job run in any transactions - Don't compile lambda when expression is a constant expression since compiling is expensive # 2.49.1 _2025-11-18_ - Don't block thread by event handler when receiving instance event # 2.49.0 _2025-11-11_ - Make default parameters and next run parameters editable in admin UI # 2.48.6 _2025-11-05_ - Make all IHostedServices use try-catch in StopAsync() - Mark a queue job run as NothingToDo when all messages where skipped/NotProcessed # 2.48.5 _2025-10-14_ - Wait for 5m instead of 1m before warning about a job that hasn't started yet since some jobs might not start right away when many jobs are running # 2.48.4 _2025-10-03_ - Fix issue causing some historical job data not to be deleted correctly # 2.48.3 _2025-09-24_ - Don't retry SQL Server deadlock exceptions when an external transaction is set - Don't scan the same assembly twice for the same queue messages - Add more/longer retries when RabbitMQ connection fails # 2.48.2 _2025-09-09_ - Fix issue with dropping unused queue indexes in Postgres and Sqlite # 2.48.1 _2025-09-05_ - Make it possible to skip sending Degraded health checks to Teams # 2.48.0 _2025-08-22_ - Make it possible to change the health status level a job or queue has when the job is failing or when the queue has error messages in it - Add health check publisher to [Microsoft Teams](../healthchecks/teams.md) # 2.47.5 _2025-06-26_ - Fix issue causing `[Queue(AutomaticRetryAfter = "xxx")]` not to be correctly used # 2.47.4 _2025-06-24_ - Fix support for scoped services in Nexus Functions # 2.47.3 _2025-06-09_ - Add locks when acknowledging messages with RabbitMQ - Don't load the same log from Elastic multiple times when looking at logs for a current run # 2.47.2 _2025-05-04_ - Fix issue causing the Elastic job log reader not to return all logs # 2.47.1 _2025-05-04_ - Downgrade Elastic from 9 to 8 in `CommerceMind.Nexus.Elastic` # 2.47.0 _2025-05-04_ - Add [queue message validation](../queues/validation.md) # 2.46.0 _2025-04-26_ - Add `CommerceMind.Nexus.Elastic` NuGet package to [load job logs from Elasticsearch](../jobs/logs.md#elasticsearch) instead of the database - Use explicit lock in SQL Server when leasing functions for execution # 2.45.0 _2025-04-26_ - Prevent admin UI job list from flickering when scrolling - Allow updating processed message retention in the admin UI - Make it possible to add a delay on a queue before a job is allowed to process a message - Handle exceptions when service provider fails to create a job instance - Increase query timeout for operations that update all or most items in a queue - Fix url to queue webhook endpoint in admin UI - Handle errors that might occur when organizing queue messages - Handle ProcessedMessageRetention being TimeSpan.MinValue in memory queues # 2.44.4 _2025-04-07_ - Include cookies for different sub domains in requests made by the Admin UI to make auth easier to implement # 2.44.3 _2025-04-05_ - Fix issue causing some memory queue changes not to get persisted to fallback storage properly - Ignore any DbConnectionScope when explicitly asking for a new db connection # 2.44.2 _2025-04-03_ - Only register ApplicationInstanceEventBroker if instance events have been registered # 2.44.1 _2025-04-02_ - Fix issue during startup when instance events are not configured, eg when only using Nexus API enqueuers # 2.44.0 _2025-04-02_ - Make it possible to use Nexus instance events infrastructure for [application specific events](./instance-events.md#custom-application-instance-events) - Use retries when cleaning up historical job runs # 2.43.0 _2025-04-02_ - Add queue export to Excel in Admin UI - Make it possible to control how ServiceBusClient is created in CommerceMind.Nexus.Azure - Throw better error when there are multiple queues with the same name # 2.42.0 _2025-03-22_ - Make it possible for virtual queue transformers to [return an `EnqueueContext`](../queues/virtual.md#enqueuecontext) to control things like processing date and priority - Add `EnqueueContextFactory` option when enqueueing from [Azure Service Bus](../queues/azure.md), [Rabbit MQ](../queues/rabbitmq.md) and [Kafka](../queues/kafka.md) - Fix issue with job progress message disappearing # 2.41.1 _2025-03-18_ - Support JSON arrays in Azure Service Bus messages - Prevent unnecessary delay when requesting a job to start - Fix issue with job progress total disappearing # 2.41.0 _2025-03-13_ - Allow specifying `MaxConcurrentCalls` for Azure Service Bus # 2.40.1 _2025-03-13_ - Run Kafka message processing in dedicated tasks # 2.40.0 _2025-03-12_ - Handle changes in casing of job names - Allow setting new priority and update message data when returning NotProcessed() for a message - Change default Admin UI path from /admin to /nexus and redirect /admin to /nexus - Skip setting up Azure Service Bus listener and log if not properly configured instead of throwing - Don't log TaskCanceledExceptions caused by service shutdown # 2.39.0 _2025-03-10_ - Use Ignore instead of Null as key for Kafka consumer - Allow passing a message deserializer to EnqueueFrom() when declaring a RabbitMQ queue # 2.38.0 _2025-03-10_ - Generate explicit OpenApi definitions for queue webhook endpoints # 2.37.0 _2025-03-08_ - Add `ProcessResults.NotProcessed()` for situations where you can't process and message and you want to return it to the queue untouched - Ensure that job progress table can't have multiple rows for a single job - Return data from update queries and fill job meta data cache to avoid having to fetch the data again # 2.36.0 _2025-03-07_ - Make it possible to force update/delete messages in Admin UI - Fix deserializing nullable number type - Support passive queues in RabbitMQ # 2.35.4 _2025-03-07_ - Include job meta data version in instance event to correctly purge the cache # 2.35.3 _2025-03-07_ - Fix handling of multiple progress entries for the same job # 2.35.2 _2025-03-07_ - Improve caching of job meta data - Add option to turn off v1 compatbility checks - Avoid unnecessary message parsing for memory queues # 2.35.1 _2025-03-06_ - Avoid creating a job instance to check `StopProcessingOnError` if that property is not defined on the type - Show message deserialization errors in Admin UI instead of crashing # 2.35.0 _2025-03-06_ - Make it possible to dynamically [generate queues and queue jobs](../queues/generated.md) - Show filtered queue counts for jobs with message filter in Admin UI - Make PauseOnError work with filtered queue jobs - Make it possible to choose how many queue messages to see per page in Admin UI - Health checks for jobs now becomes unhealthy if it hasn't started on schedule - Fix queue counts disappearing in job list when scrolling - Refresh queue item list when navigating - Refresh queue status counts when updating queue item(s) in Admin UI - Allow queue processing retries to be scheduled in 10s or less from now - Increase query timeout when enqueueing a lot of messages # 2.34.1 _2025-02-25_ - Avoid making HTTP request in Admin UI that doesn't fetch any data - Respond with 400 Bad Request instead of 500 in queue webhook when the incoming message is malformed - Don't create unused RabbitMQ channel when using RabbitMQ for instance events # 2.34.0 _2025-02-19_ - Make it possible to have dynamic (untyped) content in queue messages [using `IDynamicQueueMessage`](../queues/index.md#dynamic-message-content) # 2.33.0 _2025-02-09_ - Include queue message type in Open Telemetry metrics - Extract UI into separate NuGet package so that you can host the UI [in a separate application](../admin-ui/index.md#hosting-the-ui-on-its-own) # 2.32.1 _2025-02-06_ - Fix Postgres returning null when any property in the message is null # 2.32.0 _2025-02-01_ - Fix alignment issues for buttons on queue page - Fix issue with hidden deleted jobs and queues in the admin UI - Make it possible to hide job and queue "More" settings with the escape button - Make it possible to start the queue job from the queue page - Fix filtered "Message id" text field on refreshing queue page # 2.31.2 _2025-01-24_ - Start queue jobs where job start has been requested before other jobs - Handle multiple manual job run requests executing at the same time # 2.31.1 _2025-01-24_ - Fix issue with EnqueueAndStartProcessingAsync() causing intermittent errors - Fix SQL Server issue with upgrading from an old version # 2.31.0 _2025-01-22_ - Add [CommerceMind.Nexus.Kafka](../queues/kafka.md) which enables instance events over Kafka as well as automatically enqueueing to Nexus queues from Kafka topics - Don't wait for the next job loop tick when enqueueing a message and requesting processing to start # 2.30.0 _2025-01-21_ - Expose more options to RabbitMQ binding object # 2.29.0 _2025-01-18_ - Fix issue causing reloads in Admin UI when scrolling long jobs or queues lists - Make it possible to update a message as part of processing it - Make it possible to hide jobs and queues in the Admin UI - Show notification that data could not be refreshed instead of showing error screen - Cache frequently executed job meta data query # 2.28.0 _2024-12-20_ - Generate XML comment files for NuGet packages - Retry failed requests in the Admin UI - Only load queue status counts for visible queues in the Admin UI - Base Admin UI polling on when the last request ended instead of started to prevent overloading the server when it responds slowly # 2.27.2 _2024-12-17_ - Fix issue with invalid existence check on SQL Server queue column # 2.27.1 _2024-12-15_ - Fix issue with Azure Postgres not always returning all columns in information_schema.columns # 2.27.0 _2024-12-15_ - Add [CommerceMind.Nexus.RabbitMQ](../queues/rabbitmq.md) and make it possible to use [RabbitMQ for instance events](./instance-events.md) - Make it possible to use [dynamic job mutexes](../jobs/index.md#job-parallelism-and-mutexes) - Make it possible to read queue messages by a list of message ids - Add search/filter to header bar in the Admin UI - Make it possible to retry messages instead of marking them as Abandoned - Make it possible to limit which instance a job should run on through code and in the Admin UI - Fix issue that caused enable/disable buttons for jobs not to be responsive in the Admin UI - Support [top level dictionaries in partial messages](../queues/index.md#partial-messages) - Allow queue jobs to control [the job result message](../queues/jobs.md#controlling-the-job-result-message) - Make it possible for a [queue job to filter which messages it wants to process](../queues/jobs.md#filtering-messages) - Make it possible to [filter on queue properties](../queues/filterable.md) # 2.26.0 _2024-12-05_ - Show dates as labels instead of time when showing more than latest 24h in statistics - Hide sorting controls on queues list until hovered - Add automatic webhook endpoint for all queues, [see docs](../queues/webhooks.md) - Wrap ActivitySource usage in order to trap exceptions during dispose, fixes issue where Elastic.APM causes jobs to be reported as failed # 2.25.2 _2024-12-02_ - Fix exception caused by queues that no longer exist in code - Correctly handle status change statistics for idempotent queues # 2.25.1 _2024-12-02_ - Retry TimeoutExceptions that occur in the database layer # 2.25.0 _2024-12-02_ - Make it possible to sort jobs and job categories in the Admin UI - Don't show processed count in queue stats if processed messages are deleted since it's confusing to see "Processed: 0" when the graph says lots of processed messages - Only show status counts below graphs of the statuses shown in the graph - Optimize performance of updating statistics graphs in Admin UI # 2.24.1 _2024-11-30_ - Generate random color for custom statuses in queue statistics graphs - Fix issue causing queue meta data to be reset when updating meta data # 2.24.0 _2024-11-30_ - Make it possible to control order and items in the Admin UI main menu. [See docs](../admin-ui/index.md). - Add statistics/graphs over queue message status changes. [See docs](../queues/statistics.md). - Allow setting which status a message should get when reaching max retries. [See docs](../queues/retry.md#status-when-max-retries). - Allow updating queue display name, category, max retries, statistics retention, and history retention in the Admin UI/API. - Make it possible to change the sort order of queues and queue categories in the Admin UI. # 2.23.0 _2024-11-24_ - Make it possible to pause jobs after they fail until the error has been resolved. See more [in the docs](../jobs/index.md#pausing-on-error). - Reset retry count when updating a single message in the Admin UI. - Enable editing job category, historical run retention and if a job should be considered degraded when disabled in the Admin UI/API. # 2.22.0 _2024-11-22_ - Make it possible to delete by status in the Admin UI/API - Reset retry count when changing status of a queue item - Respect low MaxNumberOfMessagesPerRun values in non-batched queue jobs - Make it possible to auto cancel jobs after max duration - Allow setting max duration and auto cancel for a job in the AdminUI/API # 2.21.0 _2024-11-18_ - Only filter on status being Pending when leasing messages to make it slightly faster - Remove queue message id database index for queues that doesn't have id - Remove queue database index no longer used - Fix issue with unclosed db connections in memory queue fallback storage - Support authorization in the API and Admin UI, see more [in the docs](../api/index.md#authorization) - Add error handling for unexpected errors in admin UI # 2.20.1 _2024-11-16_ - Only wrap JSON string in quotes if not already wrapped # 2.20.0 _2024-11-16_ - Make it possible to invoke arbitrary Nexus Functions in the Admin UI (see [docs](../nexus-functions/index.md#invoking-functions-in-the-admin-ui) # 2.19.2 _2024-11-11_ - Fix issue where CommerceMind.Nexus.Azure package initialization didn't return the fluent Nexus initialization # 2.19.1 _2024-11-11_ - Pass topic/queue name to the connection string and subscription name loaders in Azure Service Bus config # 2.19.0 _2024-11-11_ - Make it possible to automatically retry errors in queues through Admin UI and code - Fixes issue that reset retry count to 0 when reaching max retry count - Support fetching from Azure Service Bus Topics as well as Azure Service Bus Queues # 2.18.0 _2024-11-09_ - Make it possible for a queue to be [in-memory](../queues/in-memory.md) - Extract SQL `COUNT(*)` into separate query since it's faster for large data sets # 2.17.0 _2024-10-26_ - Optimize handling of queue messages without id # 2.16.0 _2024-10-21_ - Set the same data type in queue message import temp table as the source table to avoid bad query plan - Make it possible to filter to only see jobs with issues in the admin UI - Group jobs list by category first instead of job type first and category second in admin UI # 2.15.0 _2024-10-20_ - Add more retries to SQL queries for functions - Don't load job progress when it's not needed - Handle SQL Server concurrent update for memory optimized tables error - Delay creating job progress until the first progress is reported - Make enqueueing of messages without id faster - Enqueue messages one by one when the batch is small since it's faster # 2.14.0 _2024-10-12_ - Store latest job run details in main jobs table to avoid joins in hot query - Disable lock escalation in SQL Server to avoid use of rowlock hint - Add retries when fetching historical job runs # 2.13.0 _2024-10-02_ - Make unbatched queue processing as fast as batched - Don't use NOLOCK hint for SQL Server when using read committed snapshot # 2.12.0 - Support calling private methods in functions - Prevent long job result text to break the job details UI - Add support for middlewares for functions - Wrap creation and population of temp tables during enqueueing in an OTel transaction so the performance of it can be measured # 2.11.0 - Add retries to query that runs as part of message enqueueing - Release batch messages that haven't been processed or failed - Show in admin UI if any functions failed to load instead of just hiding them # 2.10.5 - Rewrite sql query so that an index can be correctly used # 2.10.4 - Fix regression that didn't respect the global max retry count for functions # 2.10.3 - Fix issue with using messages as records with batch jobs # 2.10.2 - Await function batches to avoid getting into a state where new functions are started before existing ones finish # 2.10.1 - Add missing database index for deleting processed queue messages - Ensure that next runs in UI and API are always in the future - Ensure that parameters passed to JobResults is of the correct type # 2.10.0 - Improve the layout of job list when you have many jobs split into many categories # 2.9.0 - Fix styling for lists and paragraphs in job markdown description - Fix health check migrations for SQLite - Make it possible to mass update queue item priority in admin UI # 2.8.6 - Add retries when reading job meta data - Keep retrying migrations on transient errors to avoid crashing the application - Lower log levels on warnings that are meant for debugging # 2.8.5 - Fix failing migration for health checks # 2.8.4 - Changes the category of Nexus log messages for queue jobs to make it easier to filter out # 2.8.3 - Fix for supporting queue messages to be records rather than classes # 2.8.2 - Separate short description and description of a job and show short description in listings and description on the details page in the admin UI # 2.8.1 - Fix issue with Nexus functions with arguments not getting called correctly # 2.8.0 - Adds description text for jobs in the admin UI which can be edited in the UI, markdown supported - Adds cancellation token support for Nexus functions - Fixes admin UI to not say "Starts in in X seconds" with a duplicate "in" - Fixes admin UI job start dates to not show dates in past time # 2.7.2 - Fix incorrect use of service scope # 2.7.1 - Use service provider scope in all places where a scheduled job instance is used # 2.7.0 - Changes jobs and functions to use a service provider [scope](https://learn.microsoft.com/en-us/dotnet/core/extensions/dependency-injection#scoped) when creating service instances in order to support services registered with a scoped lifetime. # 2.6.2 - Fix issue with using enums as arguments to Nexus functions # 2.6.1 - Correctly report outcome on Open Telemetry activities/traces # 2.6.0 - Expose metrics and telemetry through [Open Telemetry](https://opentelemetry.io/docs/languages/net/) # 2.5.1 - Fix how casts of Nexus function arguments are handled # 2.5.0 - Adds more filtering and sorting to job runs in admin UI - Fixes bug with serializing GUIDs with `System.Text.Json` # 2.4.0 - Adds filtering of job runs in admin UI - Adds missing database index for job runs # 2.3.0 - Support List.Contains() in IQueueItemUpdater.UpdateStatusWhereAsync() - The health checks system now has a watch dog that will reset any check that is stuck in a running state. Eg if the process crashed after the check finished but before Nexus was able to update the status. - Fix issue where enabling queue history for all queues included queues where messages don't have id - Ensure that the ILoggerProvider for job logs aren't registered multiple times # 2.2.4 - Improve error handling with instance events # 2.2.3 - Add transient database error retries on all methods on `IQueueReader` - No longer assumes that a temp table still exists since it might get deleted when a connection is reopened # 2.2.2 - Fix issue with Postgres and SQL Server that caused an error if you called `IEnqueuer.EnqueueAsync()` with an empty list of messages - Reset migrations if they're stuck in a running state because of a server crash # 2.2.1 - Explicitly drop temp tables instead of relying on connection close to delete them # 2.2.0 - Make it possible to change priority on a message when scheduling a future processing - Improve handling of merging partial messages in SQL Server - Add retries on transient database errors in more situations # 2.1.1 - Fix issue with queue messages that has propertiy types without parameterless constructors # 2.1.0 - Fix issue using `MaxRetries` in `[NexusFunction]` attributes - Fix issue causing JSON serialization errors in the Slack healty reporter - Support enqueueing partial messages, [read more here](../queues/index.md#partial-messages) - Add retries on SQL queries with SQLite that caused "database is locked" errors - Run retries on transient database errors more granularly and in more places # 2.0.0 This release contains a few breaking changes that are described in [the upgrade guide](./upgrade-guide.md). The breaking changes are mostly a code and project reorganization that should be easy to upgrade to. Besides the reorganization some notable changes are: - The scheduled jobs system now has a watch dog that will reset any job that is stuck in a running state. Eg if the process crashed after the job finished but before Nexus was able to update the job status. - All NuGet dependencies have been updated to the latest version as well as .NET. - Serilog is no longer a dependency to Nexus. - Newtonsoft.Json is no longer a direct dependency to Nexus, the support for Newtonsoft.Json has been extracted to a separate package. - It's now possible to register [middlewares around job execution](../jobs/index.md#job-middlewares). - It's now possible to set job and log history [per job](../jobs/index.md#log-retention). # 1.23.0 - Allows queue jobs to set custom statuses for messages. [Read more here](../queues/jobs.md#returning-a-custom-status). # 1.22.0 - Adds message priority for queues. [Read more here](../queues/priority.md). # 1.21.1 - Adds feature detection for if queue history is enabled to avoid SQL queries running against tables that doesn't exist # 1.21.0 - Adds opt-in message history for queues with identity. [Read more here](../queues/history.md). # 1.20.3 - Adds retries to the database maintenance job when using Postgres. # 1.20.2 - Fixes risks for deadlocks when job progress hasn't finished writing before the job is done. When a job finishes and cleanup attempts to delete job progress before the job progress has finished writing a deadlock could occur because the update and cleanup/delete tries to write the same row. # 1.20.1 - Removes a query hint when using SQL Server that required READ COMMITTED isolation level. The recommended isolation level is READ COMMITTED SNAPSHOT, but is not a hard requirement. # 1.20.0 - Makes it possible to initiate a soft graceful shutdown which means that all jobs, functions, etc that has already started will run until finished but no new work will be started. See more [here](/general/hosting). # 1.19.0 - Fixes regression around running internal Nexus migrations. # 1.19.0 - Makes it possible to pass a transaction to `DbConnectionScope`. # 1.18.0 - Makes it possible to stop queue processing when an error occur, and pause it until no messages in the queue has an `Error` or `AwaitingRetry` status any more. See more [here](/queues/jobs#pausing-on-error). # 1.17.0 - Adds `IDbConnectionScope` which lets you pass an existing db connection object to Nexus. See more [here](/general/database#controlling-the-connection). # 1.16.1 - Fixes an issue where extra job runs (such as a retry) disappeared if they needed to aquire a mutex and that mutex was already taken # 1.16.0 - `IQueueItemUpdater.DeleteAndReturnAllAsync()` added as a safe way to read and delete all messages in a queue in a single and safe operation - `IScheduledJobMetaDataRepository.GetJobMetaDataAsync()` added # 1.15.6 - Let outcome descriptions line break in admin UI # 1.15.4 - Expire the jobs meta data cache after five seconds to prevent stale cache issues # 1.15.3 - Add debug logging to queue processing jobs # 1.15.2 - Use `long` for queue message ids everywhere - Ensure that changing a job schedule doesn't immediately trigger a new run if the schedule is moved forward in time # 1.15.1 - Fix issue causing `MaxNumberOfMessagesPerRun` not to work on queue jobs # 1.15.0 - Make it possible to control how many messages a single queue job run is allowed to process - Show the Nexus version in the admin UI - Don't show a Disable button for a health check for a disabled job # 1.14.4 - Fix issuing causing some health checks to not show up in admin UI # 1.14.3 - Fix issuing making it impossible to resolve a failed job run # 1.14.2 - Fix links to job details in health check details - Add caching for health check statistics to avoid polling the database on the health overview - Health periods are now not filled out to include the space in time between two periods # 1.14.1 - Change job health check to not generate a unique result description every time it's called # 1.14.0 - Stores health check results so it's possible to see the history of a health check in the Admin UI # 1.13.1 - Fix issue which caused migrations to be run out-of-order # 1.13.0 - Make it possible to use Nexus job logs in the admin UI without Serilog. Previously you needed to use Serilog in order for your job logs to get the correct context/scope values attached to know which job they originated from. Nexus now uses standard .NET logging for this which means that you no longer need to use Serilog for it. See more in the documentation under Jobs / Logs. # 1.12.1 - Reset healthchecks when an instance is starting. If an instance crashed or was deployed in the middle of when a healthcheck was running it can still be reported as running after the deploy or restart finishes. This fixes that all healthchecks are reset that claims to be running on the instance that is just starting up. # 1.12.0 - Added `JobResults.CompletedWithWarnings()` which lets a job return a successful status but have the run flagged with a warning sign in the Admin UI and the job health check status become `Degraded` # 1.11.0 - Messages that are `Leased` will now be transitioned into a new status called `Abandoned` if the lease expires due to a server crash. See [docs about abandoned messages](../queues/index.md#abandoned-messages). - Messages that are scheduled for a retry now has a new status called `AwaitingRetry` instead of being in the `Error` status. - A new max retry limit has been introduced which defaults to 10. If the same message has been retried more times than that the message is transitioned to the `Error` status and won't be automatically retried again. - Instance id and correlation id is now set on queue messages when they are leased to be able to find logs for the job that processed a message. # 1.10.1 - Minor styling issue fixed in Functions list in Admin UI # 1.10.0 - Make it possible to group queues in the Admin UI by category just like what's possible for jobs # 1.9.0 - Fix issue where the admin UI could display "0" instead of "No logs found" - Make it possible to have a new job be initially disabled # 1.8.2 - Fix issue which could cause an infinite loop in `ScheduledJobExecutorService` when using multiple servers and jobs that executes frequently - Lowered log level from information to debug on some logs that occur frequently # 1.8.1 - Includes debug symbols in new `snupkg` packages, published to the Github NuGet feed # 1.8.0 - Removed unnecessary database calls that were made to release job mutexes, even for jobs without any mutexes - Nexus will now react slightly quicker to the cancellation token that signals that the app is shutting down - A new way of dealing with exceptions in background services ensures that Nexus is more resilient against intermittent database failures # 1.7.0 - Made it possible to customize the example message shown in the admin UI when enqueueing a new message manually. - `IScheduledBatchQueueJob.GetBatchSize()` has been deprecated in favor of `IScheduledBatchQueueJob.BatchSize`. The `GetBatchSize()` method will be removed in the next major version. # 1.6.8 - Add missing indexes to mutex table to avoid deadlocks when attempting to release job mutexes # 1.6.7 - Apply less locking when fetching job data from SQL Server to avoid deadlocks # 1.6.6 - Fix issue in API usage of Nexus Functions with SQL Server # 1.6.5 - Fix concurrency issue with leasing queue messages and marking functions as running when using Postgres. # 1.6.4 - Fix issue where instance heartbeat healthchecks reported the time in 12 hour - Fix text overflow in health check admin UI for long descriptions # 1.6.3 - Dispose `CancellationTokenSource` in `ScheduledJobExecutorService` after use # 1.6.2 - Add horizontal scrollbar for long queue processing log in admin UI # v1.6.1 - Remove testing code in the admin UI that slipped through in 1.6.0 # v1.6.0 - Make it possible for jobs to report their progress and display it in the admin UI - Use Tailwind instead of CSS Modules in admin UI # v1.5.4 - Use higher query timeout for all queries in IDatabaseMaintenanceService - Show resolved/dismissed state for latest run in job list # v1.5.3 - Fix issue with database polling instance events # v1.5.2 - Fix issue with admin UI not rendering correctly when not having a trailing slash in the url (ie just /admin and not /admin/) # v1.5.1 - Fix issue with duration incorrectly being stored as decimal in SQLite when integer is expected # v1.5.0 - Make it possible to mark a failed job run as resolved or dismissed with an optional markdown comment - Make it possible to add a comment to queue messages - Store both latest job run and latest meaningful job run to since some jobs run very frequently but doesn't perform work all the time - Fix when cancelling a queue job the historical run status becomes should be Cancelled rather than Completed # v1.4.0 - Add RetryLater as a successful job outcome # v1.3.2 - Fix issue with database maintenance on SQL Express # v1.3.1 - Reset any health checks marked as running when an instance is starting # v1.3.0 - Ask for a connection string every time the database is opened to make it possible to use connection strings with updated credentials # v1.2.1 - Add query timeout of 5 minutes for database maintenance queries - Make it possible to update queue message status by filtering on message JSON properties - Increase timeout when deleting old job runs and logs # v1.2.0 - Show a health check for a disabled job as disabled - Implement graceful shutdown which attempts to stop Nexus background processes before the .NET application begins to terminate to ensure that all jobs etc aren't running when the application terminates - Generate a unique Nexus instance id for Azure App Service where a single VM can run multiple applications # v1.1.0 - Increase query timeout for updating queue messages to correct status - Make it possible to update all queue items to a certain status - Make it possible to mark a queue item as processed and schedule another processing of the message in the future - Only show main menu items in the admin UI if they have been configured - Add WITH (HOLDLOCK) to all SQL Server MERGE statements to avoid insert conflicts # v1.0.0 Initial stable release. --- # From v1 to v2 Version 2 contains a number of breaking changes from version 1. The core logic and concepts all remain the same in Nexus and this release is mostly about code and project reorganization, removing unnecessary dependencies and updating to .NET 8. ### New NuGet package structure Instead of having the different sub systems such as `CommerceMind.Nexus.Jobs` in their own NuGet packages they have been consolidated into the new `CommerceMind.Nexus` package and the abstraction packages have been consolidated into `CommerceMind.Nexus.Abstractions`. This reorganization means that many interfaces have been moved to new namespaces. Eg `IScheduledJob` which now lives in `CommerceMind.Nexus.Abstractions.Jobs` instead of `CommerceMind.Nexus.Jobs.Abstractions`. The interface names are still the same, so it should be easy to either search-and-replace the namespaces or let Visual Studio/Rider/etc guide you to the new names. Another reorganization change in this is that the implementations for the different databases are no longer included in the main packages. Instead they have been extracted into `CommerceMind.Nexus.Postgres`, `CommerceMind.Nexus.SqlServer`, and `CommerceMind.Nexus.Sqlite`. Allowing you to only include the database(s) you want. ### New fluent service registration In v1 there was a number of extension methods on `IServiceCollection` that you had to call in a specific order. In v2 this has changed to a fluent registration on `IServiceCollection` where the order of which method you call is no longer important. This is how the v2 registration looks like: ```cs builder.Services. .AddNexus() .AddSingleServerInstanceEvents() .AddFunctions() .AddScheduledJobs() .AddQueues() .AddSqliteConnection() // Don't forget this one! .Build() ; ``` Note that you can still split up your service registration as long as the call to `.Build()` is done last, as that will register additional services based on your configuration. You can for example do this: ```cs builder.Services.AddNexus().AddSingleServerInstanceEvents(); builder.Services.AddNexus().AddFunctions(); builder.Services.AddNexus().AddScheduledJobs(); builder.Services.AddNexus().AddQueues(); builder.Services.AddNexus().AddSqliteConnection(); builder.Services.AddNexus().Build(); ``` Each of these lines could be in different projects as well, as long as the call to `Build()` happens last. All calls to `builder.Services.AddNexus()` will return the same Nexus registration object for that specific service collection. It's also safe to call any method on the Nexus registration object multiple times. ### Nexus no longer uses Newtonsoft.Json by default Instead `System.Text.Json` is used by default and the Newtonsoft.Json serializers have been extracted into a separate package called `CommerceMind.NewtonsoftJson` which you can use by calling `builder.Services.AddNexus().AddNewtonsoftJson()`. You probably don't need Newtonsoft.Json, this is only if you've modified how the JSON in Nexus is serialized with custom converters. A notable change here is that the JSON formatting rules for ASP.NET will no longer be applied for queue messages in the API. Instead the JSON formatting rules for Nexus is applied, which means that you might get a response looking like this: ```json { "id": 4823771, "status": "Processed", "retryCount": 0, "message": { "Id": null, "SomeProperty": null, "ExampleEnum": "Value1" } } ``` Where the outer object have gotten kebabCase properties but the message object has PascalCase properties. You can control both of these ([see docs about JSON](./json.md)) formatting options. ### Nexus now requires >= .NET 8 v1 supported .NET 6 but v2 now reqires at least .NET 8. The only thing you need to do is to upgrade your .NET version in your application if you haven't already done so. ### Deprecated methods and properties have been removed Methods and properties that have been deprecated in v1 have been removed in v2. If you upgrade to v1.23.0 and fix all build warnings about obsolete methods and properties you'll automatically be prepared for the removals of them in v2. ### Default job history retention is now 30 days Previously the default was to keep job history in the database forever. To avoid having Nexus growing your database indefinitely the default has been changed to 30 days. You can still configure this just like before, but since you can now have different retention per job the property is now called `DefaultHistoricalRunsRetention`: ```cs builder.Services.AddNexus().AddScheduledJobs(options => { options.DefaultHistoricalRunsRetention = TimeSpan.FromDays(10); }); ``` You set it per job with the `[ScheduledJob]` attribute like this: ```cs [ScheduledJob("MyJob", HistoricalRunsRetention = "30.00:00")] public class MyJob : IScheduledJob { ... } ``` ### The IEnqueuer and IEnqueuer interfaces have been simplified Previously `IEnqueuer` inherited from `IEnqueuer` which it no longer does. In v1 every message type was registered as a `IEnqueuer` which means that you could get all enqueuers by requesting `IEnumerable` like this: ```cs public class DynamicEnqueuer(IEnumerable enqueuers) { public Task EnqueueAsync(IQueueMessage message) { var enqueuer = enqueuers.Single(e => e.MessageType == message.GetType()); return enqueuer.EnqueueAsync(message); } } ``` In v2 there is only a single non-generic `IEnqueuer` registered which does the above for you. Meaning that you can pass a `IQueueMessage` message to it and it will figure out which concrete `IEnqueuer` to pass it on to. Both the `IEnqueuer` and `IEnqueuer` interfaces have been simplified to take a message and an `EnqueueContext` instead of having multiple different overloads for when you want to specify a custom status, a date for when to process it, etc. So when you did this in v1: ```cs await enqueuer.EnqueueAsync(message, DateTime.UtcNow.AddHours(1)); ``` To enqueue a message to be processed in one hour you now do this: ```cs await enqueuer.EnqueueAsync(message, new EnqueueContext { ProcessAfter = TimeSpan.FromHours(1) }); ``` ### No longer using Serilog Nexus previously used Serilog to capture logs happening inside job runs. With v2 it instead has a [`ILoggerProvider`](https://learn.microsoft.com/en-us/dotnet/api/microsoft.extensions.logging.iloggerprovider?view=net-8.0) which is registered by the scheduled job system. In v2 you can use any logging framework you want as long as it writes to external logging providers. Note that this is turned off by default in Serilog and that you need to enable it, eg like this: ```cs builder.Host.UseSerilog((context, services, configuration) => { configuration.ReadFrom.Configuration(context.Configuration); // This one is important }, writeToProviders: true); ``` If you've used the jobs logging in v2 you also need to remove the Nexus Serilog sink from your configuration: ```json "Serilog": { "Using": ["CommerceMind.Nexus.Jobs"], "WriteTo": [{ "Name": "NexusJobLogs" }] } ``` ### Changes to the `IJobLogProvider` interface For consistencys sake the `IJobLogProvider` interface has been renamed to `IJobLogReader` since there's also a `IJobLogWriter` interface. And instead of just getting a date to fetch logs newer than Nexus now passes in a date range. You also get the log levels passed as `LogLevel` enums rather than strings. Also note that `JobLogEvent.Level` has changed data type from `string` to the `LogLevel` enum. --- # Healthchecks Nexus integrates with and extends [Health checks in ASP.NET Core](https://docs.microsoft.com/en-us/aspnet/core/host-and-deploy/health-checks?view=aspnetcore-6.0). All functions, jobs, and queues automatically get health checks that signals their health. The only thing you need to do in an ASP.NET Core application is call this in your `Program.cs`: ```cs app.MapHealthChecks("/health"); ``` The jobs, queues and functions subsystems in Nexus initializes the health checks subsystem as part of their initialization so you don't have to do that explicitly. But if you want to change any of the default options you can initialize it directly like this: ```cs builder.Service.AddNexus().AddHealthChecks(options => { // Set this to true if you're not using ASP.NET. Then Nexus will periodically call the health checks for you // and publish a health report just like in ASP.NET. options.UseHealthCheckBackgroundService = false; // If you're not using ASP.NET you can configure how often the background service calls the health checks. options.HealthCheckBackgroundServicePollDelay = TimeSpan.FromSeconds(30); // How long Nexus should keep results of health checks to display historical status in the admin UI options.StatisticsRetention = TimeSpan.FromDays(30); // If you have health checks definied in other assemblies you can add them here to get Nexus to find them options.AssembliesToScanForHealthChecks.Add(typeof(MyType).Assembly); // This lets you dynamically exclude any health check types that you don't want Nexus to register options.HealthCheckFilter = healthCheckType => !healthCheckType.Name.Contains("HealthCheckIDontWant"); }); ``` You can see the health checks included in the demo environment here:\ [https://commerce-mind-nexus-a9bsbwgth3cgfnde.swedencentral-01.azurewebsites.net/admin/healthchecks](https://commerce-mind-nexus-a9bsbwgth3cgfnde.swedencentral-01.azurewebsites.net/admin/healthchecks) ## Teams and Slack integrations If you wish to get Teams or Slack notifications when health checks are failing you can use the [Teams integration](./teams.md) or [Slack integration](./slack.md) for Nexus. ## When are health checks invoked? If you're using ASP.NET the health checks will be called when anyone is pinging the `/health` endpoint and when you access the health checks in the admin UI. If you've registered a `IHealthCheckPublisher` that will also cause the health checks to be called an additional two times per minute by ASP.NET. When you create a custom health check as described below you can choose to have a scheduled or unscheduled health check. If you create a scheduled health check Nexus will ensure that your health check is never executed more often than the schedule allows. Nexus will return the previous result for such a health check until it's time to call it again. ## When not using ASP.NET If you're not using ASP.NET you can still use the Nexus health checks. You should initialize the health check system like this: ```cs builder.Service.AddNexus().AddHealthChecks(options => { options.UseHealthCheckBackgroundService = true; options.HealthCheckBackgroundServicePollDelay = TimeSpan.FromSeconds(30); }); ``` You should also create your own [`IHealthCheckPublisher`](https://docs.microsoft.com/en-us/dotnet/api/microsoft.extensions.diagnostics.healthchecks.ihealthcheckpublisher?view=dotnet-plat-ext-6.0) and reigster it with the `IServiceCollection` like this: ```cs builder.Service.AddSingleton(); ``` The Nexus background service will then periodically call your publisher twice per minute. Since it's a singleton instance you should store the last status and only ping someone if the status goes from healthy to unhealthy. ## Adding more health checks There's two health check interfaces included. One that runs on demand, and one that is scheduled. ### `INexusHealthCheck` Use this implementation for checks that you always want to run on demand. Their results are never cached but should then respond quickly and not be very costly. For example: ```cs public class ExampleHealthCheck : INexusHealthCheck { public string Name => "MyHealthCheck"; // This needs to be unique public string DisplayName => "Health check running a lot"; public string? Description => "A longer description of what the health check does"; // You can return null if you don't need a description public string? Category => "My health checks category"; // Used to group checks in the Admin UI public IEnumerable Tags => new string[] { }; public string? AdminUILinkUrl => null; public async Task CheckHealthAsync(HealthCheckContext context, CancellationToken cancellationToken = default) { return HealthCheckResult.Healthy($"Everything is fine!"); } } ``` ### `INexusScheduledHealthCheck` In some cases you want a health check to run on a schedule rather than firing every time someone pings the health status endpoint. For this you can use the `INexusScheduledHealthCheck` interface which much like jobs has a cron schedule. When the service executes the health checks it will only execute a scheduled check if enough time has passed since the last time it was called. Otherwise the system will return the previous result for that check. Another benefit of `INexusScheduledHealthCheck` is that the `CheckHealthAsync()` method is passed the previous result of that check. This means that you can determine health based on a previous run. An example: ```cs public class ExampleScheduledHealthCheck : INexusScheduledHealthCheck { public string Schedule => "@every_minute"; // See: https://github.com/HangfireIO/Cronos#macro public string Name => "MyHealthCheck"; // This needs to be unique public string DisplayName => "Health check running every minute"; public string? Description => "A longer description of what the health check does"; // You can return null if you don't need a description public string? Category => "My health checks category"; // Used to group checks in the Admin UI public IEnumerable Tags => new string[] { }; public string? AdminUILinkUrl => null; public async Task CheckHealthAsync(NexusHealthCheckResult? previousHealthCheckResult, ScheduledNexusHealthCheckInitiator initiator, HealthCheckContext context, CancellationToken cancellationToken) { var previousCount = (int?)previousHealthCheckResult?.Result.Data?["count"]; var data = new Dictionary(); var currentCount = GetCurrentCount(); data["count"] = currentCount; if (currentCount - previousCount < 10) { return HealthCheckResult.Unhealthy($"Everything is NOT fine!", null, data); } return HealthCheckResult.Healthy($"Everything is fine!"); } private int GetCurrentCount() { // Look up value } } ``` Here we're using the `Data` property on `HealthCheckResult` to store any properties we need in the next run to determine the health. This can be used to see if counters such as the amount of `Pending` messages in a certain queue increases too much. #### Schedule The `Schedule` property is a cron expression just like [`DefaultSchedule`](../jobs/index.md#ischeduledjobdefaultschedule) on jobs, and it can be any expression supported by [Cronos](https://github.com/HangfireIO/Cronos). The above example uses one of the supported [macros](https://github.com/HangfireIO/Cronos#macro) in Cronos. Note that there's a difference between the `DefaultSchedule` property on `IJob` and the `Schedule` property on `INexusScheduledHealthCheck`. The default schedule on a job is only read by Nexus the first time the job is deployed and registered in the database. Using eg `CronSchedule.EveryMinute()` for the job will generate a random second (eg `16`) for the second during in a minute that the job will start on. This means that you can have many jobs using `CronSchedule.EveryMinute()` but they won't start at exactly the same time. Instead they will start on random seconds to spread out the load they generate. The health checks `Schedule` property on the other hand are read every time Nexus checks if it's time to invoke them. So if you use `CronSchedule.EveryMinute()` for a health check it'll sometimes execute the check multiple times in a minute and sometimes less frequent than a minute. Eg if the clock is `12:00:10` and the schedule says `9 *, *, *, *, *` the check is executed. And if the next time the schedule is evaluated it becomes `19 *, *, *, *, *` it means that it'll only be 10 seconds between the times the check is called. If you still want to generate a random second for a health check you can generate it when the application starts and then keep returning that cron schedule. Like this: ```cs public class ExampleScheduledHealthCheck : INexusScheduledHealthCheck { private static string _schedule = CronSchedule.EveryMinute(); public string Schedule => _schedule; } ``` This will make sure that the schedule is stable during the time that the application lives and only generate a new schedule when the application is restarted or a new version is deployed. ## Registering custom health checks If you created an implementation of `INexusHealthCheck` or `INexusScheduledHealthCheck` it becomes automatically registered if the type exists in your entry/application assembly. If your checks exists in a different assembly you can instruct Nexus which assemblies to scan: ```cs builder.Services.AddNexus().AddHealthChecks(options => { options.AssembliesToScanForHealthChecks.Add(typeof(MyHealthCheck).Assembly); }); ``` --- # Slack integration If you're using Slack you can use the `CommerceMind.Nexus.Slack` NuGet package to get realtime updates when jobs and queues are failing. The package contains a [`IHealthCheckPublisher`](https://docs.microsoft.com/en-us/dotnet/api/microsoft.extensions.diagnostics.healthchecks.ihealthcheckpublisher?view=dotnet-plat-ext-6.0) implementation that publishes [health check failures](./index.md) to Slack. ## Notifications Queue errors:  "Show errors" button clicked:  Errors resolved:  Job errors:  "Show error logs" button clicked:  ## Installation The first thing you need to do is to create a Slack app for your organization. Go to [https://api.slack.com/apps](https://api.slack.com/apps) and click `Create New App`. You can use the following manifest to create the app from: ```yml display_information: name: Nexus description: Info notices from Nexus background_color: "#18113f" features: bot_user: display_name: Nexus always_online: false oauth_config: scopes: bot: - chat:write.public - chat:write settings: interactivity: is_enabled: true org_deploy_enabled: false socket_mode_enabled: true token_rotation_enabled: false ``` If you want to you can use our logo for the app icon: [nexus.png](./img/nexus.png) After you've created the app you need to install the `CommerceMind.Nexus.Slack` NuGet package and configure it in your `Program.cs` like this in the application that runs background jobs: ```cs builder.Services.AddNexus().AddSlackPublisher(options => { options.AccessToken = "xoxb-XXX"; options.AppToken = "xapp-XXX"; options.ChannelId = "CXXX"; options.AdminUIUrl = "https://url-to-nexus-adminui.com"; }); ``` You should only enable the Slack publisher in the application that runs background functions and/or jobs. The Slack publisher can handle multiple instances/servers and will correctly handle instances starting and stopping, but if you run the publisher in an application that doesn't run jobs you won't get notified when jobs are failing. You can find your Slack app access token in the left menu item OAuth & Permissions on the page for your app ([https://api.slack.com/apps/](https://api.slack.com/apps/)). The value starts with `xoxb`. The app token can be found under Settings / Basic Information -> App-Level Tokens. The value starts with `xapp`. The channel id can be found by opening the channel/DM/group chat you want Nexus to post messages to and click on the name at the top. In the bottom of the About tab you find the channel id. Note that the channel id is not visible in the Slack mobile app. The admin UI url doesn't have to be publically available. The Nexus Slack app use it to generate links that a Slack user can click on. All communication between Slack and Nexus is done over web sockets so your Nexus application doesn't have to be publically accessible. --- # Health check statistics The result of all Nexus health checks are stored when they go from one health status to another or when the description returned from the check changes. This means that if you click on a health check in the Admin UI you get statistics about when the health check status changed and for how long the period was when it was unhealthy or degraded. The built-in checks for Nexus queues and jobs for example will show a history of when jobs fail or when queues have messages with error status and for how long. The health check dashboard also shows indicators for each check of which status the previous results for the check was. This makes it easier to spot that a check that is currently healthy recently had an unhealthy period. ## Importance of the result description Nexus uses the description text of a health check result to determine that something changed about the check. For the health check statistics every time the result description changes while the status is unhealthy or degraded the result is stored. And if the status is healthy and the description changes a counter is incremented to signal how many different healthy checks there's been in a period. This means that you should consider what you include in the description. You should not include values such as the current date/time because that will change every time the check is called. Avoid doing this: ```cs public async Task CheckHealthAsync(NexusHealthCheckResult? previousHealthCheckResult, ScheduledNexusHealthCheckInitiator initiator, HealthCheckContext context, CancellationToken cancellationToken) { return HealthCheckResult.Healthy($"Everything was fine on {DateTime.UtcNow}"); } ``` But you should be doing things like this: ```cs public async Task CheckHealthAsync(NexusHealthCheckResult? previousHealthCheckResult, ScheduledNexusHealthCheckInitiator initiator, HealthCheckContext context, CancellationToken cancellationToken) { var errorCount = GetErrorCount(); if (errorCount === 0) { return HealthCheckResult.Healthy($"No errors found"); } else { return HealthCheckResult.Unhealthy($"{errorCount} errors found"); } } ``` By including the error count in the description Nexus will store the result if the error count changes and you'll get more details when you look at the unhealthy period. If you want to include even more details you should use the data dictionary on `HealthCheckResult` like this: ```cs var data = new Dictionary(); data.Add("Something", "Interesting"); return HealthCheckResult.Unhealthy($"{errorCount} errors found", null, data); ``` The data dictionary will be serialized and stored and displayed in the health check details. ## Statistics retention By default Nexus will store the health check statistics for 30 days before it's deleted. You can set a different value on the options object when initializing Nexus health checks: ```cs builder.Services.AddNexus().AddHealthChecks(options => { options.StatisticsRetention = TimeSpan.FromDays(7); }); ``` If you set this to `TimeSpan.Zero` then Nexus will never store any statistics for health checks. --- # Microsoft Teams integration If you're using Teams you can use the `CommerceMind.Nexus.Teams` NuGet package to get realtime updates when jobs, queues, and health checks are failing. The package contains an [`IHealthCheckPublisher`](https://docs.microsoft.com/en-us/dotnet/api/microsoft.extensions.diagnostics.healthchecks.ihealthcheckpublisher?view=dotnet-plat-ext-8.0) implementation that publishes [health check failures](./index.md) to Teams through a Teams bot. Teams webhook connectors have been retired by Microsoft. Nexus still contains the old `AddTeamsPublisher` webhook API for a migration window, but new installations should use direct Teams bot mode or outgoing-only bridge mode. ## Register a Teams bot If you haven't already you need to register a Teams bot. This guide helps you with the steps: https://learn.microsoft.com/en-us/microsoftteams/platform/teams-sdk/get-started/quickstart-register?pivots=csharp Note that when registering your bot you will need a public url into Nexus that Teams can communicate with. So read the rest of the docs first before registering your bot if you're unsure about the url. The guide helps you register a Teams bot in your Teams workspace and will create a `Teams` section in appSettings.json with `ClientId`, `ClientSecret`, `TenantId` which you'll need regardless if you use the bridge mode or direct mode. If you use the direct mode you'll add these to the appSettings.json of your Nexus application. If you use the bridge mode you'll add them to the appSettings.json file of your bridge app instance. Note that regardless if you use direct mode or bridge mode Teams needs a public url that it can ping you on. In direct mode that's directly to your Nexus instance and in bridge mode it's a separate, public facing web application that acts as a bridge between your Nexus application and Teams. ## Notifications The Teams bot publisher stores the conversation where the bot is installed and sends proactive Adaptive Card messages to that chat or channel. It can update the original card when a failing check changes, when someone clicks a bot-side button, and when the check becomes healthy again. Supported actions include: - Failing scheduled job: show or reload error logs, and open the job in the Nexus UI. - Long-running scheduled job: show or reload logs, and open the job in the Nexus UI. - Queue errors: show or reload the latest 5 errors, and open the queue in the Nexus UI. - Generic health check: show the health check status and open the health checks page in the Nexus UI. Nexus keeps the active Teams message IDs in its key-value store, so cards can still be updated after the application restarts. In multi-instance Nexus deployments, only one Nexus instance publishes Teams health messages at a time. ## Direct mode Use direct mode when the Nexus application that publishes health checks is reachable by Teams over public HTTPS. Install the NuGet package `CommerceMind.Nexus.Teams` and configure the Teams bot publisher in `Program.cs`: Configure the Teams bot publisher in `Program.cs`: ```cs builder.Services .AddNexus() .AddTeamsBotPublisher(options => { // A required secret that protects linking a Teams conversation to notifications. Use a long random value. options.LinkCode = builder.Configuration["NexusTeamsBot:LinkCode"]!; // This is optional and used to add link buttons to the messages to open the Nexus UI. options.AdminUIUrl = "https://url-to-nexus-adminui.com"; // The default is Degraded. Set to Unhealthy to skip degraded health checks. options.MinimumHealthStatus = HealthStatus.Degraded; }); var app = builder.Build(); app.UseNexusTeamsBot(); ``` Teams commonly uses `/api/messages` in Bot Framework examples, but the path itself is not a Teams requirement. `UseNexusTeamsBot()` registers the Teams bot endpoint at `/nexus-teams/bot/messages` by default but if you need another route, pass it explicitly: ```cs app.UseNexusTeamsBot("api/messages"); ``` Create and configure a Teams app/bot using the standard Microsoft Teams SDK configuration. The application that runs Nexus must be reachable by Teams, and the bot messaging endpoint should point to your Nexus application, for example: ```text https://your-nexus-host.example.com/nexus-teams/bot/messages ``` If `AdminUIUrl` is configured, add that host to the Teams app manifest `validDomains` list. Teams requires this for `Action.OpenUrl` buttons in Adaptive Cards. Linking a conversation requires the configured `LinkCode`, so that being able to install or message the bot is not by itself enough to subscribe a conversation to health alerts. ### Link conversation Installing the bot does not auto-link the conversation; send `@name-of-bot link {linkCode}` to link the current Teams conversation, and send `@name-of-bot unlink` from the same conversation to remove it. Multiple chats or channels can be linked, and health notifications are sent to every linked conversation. Keep the link code out of shared channels where possible, and remove the linking message after the link has been created. ## Outgoing-only bridge mode Use bridge mode when Nexus is hosted as an internal application that Teams cannot reach over the internet. The public bridge receives Teams traffic, while the internal Nexus application only makes outbound HTTPS requests to the bridge. Install `CommerceMind.Nexus.Teams` in the internal Nexus application and configure bridge publishing: ```cs builder.Services .AddNexus() .AddTeamsBridgePublisher(options => { options.BridgeUrl = new Uri("https://teams-bridge.example.com"); options.NexusAppId = "erp-prod"; options.ApiKey = builder.Configuration["NexusTeamsBridge:ApiKey"]!; }); ``` The internal Nexus application does not need Teams bot credentials and does not need to call `UseNexusTeamsBot()`. Host a public bridge application with `CommerceMind.Nexus.Teams.Bridge`: ```cs // Program.cs builder.Services.AddNexusTeamsBridge(); builder.Services.AddNexusTeamsBridgeSqlServer(builder.Configuration.GetConnectionString("NexusTeamsBridge")!); // or AddNexusTeamsBridgePostgres(...) // or AddNexusTeamsBridgeSqlite(...) var app = builder.Build(); app.UseNexusTeamsBridge(); ``` The bridge is just a simple ASP.NET Core web application with at least the `CommerceMind.Nexus.Teams.Bridge` package installed that you need to deploy to somewhere that Teams can access. It needs a stable IP or domain that you can use when you register your teams bot. The bridge needs a database to maintain state between restarts of the bridge application. It uses Entity Framework internally. You can point it to an existing database, but you should not point it to a database already owned by EF. The bridge registers the Teams bot endpoint at `/nexus-teams/bot/messages`. Configure the Teams app/bot messaging endpoint to point to the public bridge: ```text https://teams-bridge.example.com/nexus-teams/bot/messages ``` A single bridge can serve many Nexus applications. The bridge application is designed to run as one active server, with support for short blue-green deployment overlap. ## Bridge database schema The bridge uses EF Core migrations for its own tables. `AddNexusTeamsBridge()` runs `Database.MigrateAsync()` at startup by default, so a new bridge database is created automatically when the configured database user has schema permissions. Set `AutoMigrateDatabase = false` when migrations are applied by a deployment pipeline, migration bundle, or another operational process. ## Bridge app registration Nexus app registrations are managed directly in the bridge database. There is intentionally no bridge admin HTTP API. Insert one row per Nexus application: ```sql INSERT INTO NexusTeamsBridgeApps ( NexusAppId, DisplayName, AdminUIUrl, ApiKey, LinkCode, IsEnabled, CreatedAtUtc, UpdatedAtUtc ) VALUES ( 'erp-prod', 'ERP Production', 'https://internal-erp-nexus/admin', 'generated-long-random-secret', 'generated-long-random-link-code', 1, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP ); ``` API keys and link codes are stored in plain text in the bridge database. Protect the bridge database accordingly, use long random values, and rotate them by directly updating the `ApiKey` and `LinkCode` columns. The `ApiKey` authenticates the internal Nexus application to the bridge (server-to-server). The `LinkCode` is a separate secret that protects who may subscribe a Teams conversation to an app's notifications: because the `NexusAppId` is not secret and is easy to guess, linking additionally requires the link code so that being able to message the bot is not enough to receive an app's health alerts. Keep the link code out of shared channels where possible; because it is typed into a Teams message it is visible in that conversation's history, so rotate it if it is exposed. Delete the link message after the link is created. The default Nexus-to-bridge authentication uses the `X-Nexus-Teams-Bridge-Api-Key` HTTP header. This keeps bridge API credentials separate from the Teams SDK's own `Authorization` handling. Internal Nexus applications can replace request signing by registering `INexusTeamsBridgeRequestSigner` before calling `AddTeamsBridgePublisher()`. Bridge hosts can replace request authentication by registering `INexusTeamsBridgeRequestAuthenticator` before calling `AddNexusTeamsBridge()`. ### Link conversation Link a Teams conversation to a registered Nexus app by sending one of these messages to the bridge bot, using the app's configured link code: ```text @name-of-bot link erp-prod {linkCode} ``` The link code must match the `LinkCode` stored for that app or the link is refused. Running the same command from another chat or channel adds that conversation to the app's notification destinations; the bridge sends each health notification to every linked conversation for that Nexus app. Remove the current Teams conversation from a Nexus app with: ```text @name-of-bot unlink erp-prod ``` --- # Introduction Commerce Mind Nexus is a .NET (6+) based library and solution from [Commerce Mind](https://www.commercemind.se) for running scheduled background functions and jobs, and populating and processing queues of messages. The solution is built primarily with e-commerce data flows in mind but it's a general solution that works outside of e-commerce as well. ## Key features - Distributed as NuGet packages, [Apache 2.0 licensed](tou.md) - Over 600 integration and unit tests to ensure stability - Execute a class method with retries in the background using [Nexus functions](./nexus-functions/index.md) - Schedule [jobs](./jobs/index.md) to run every X seconds/minutes/hours/etc - Out-of-schedule [job runs](./jobs/index.md#extra-job-runs) - Scale out and run jobs and functions on multiple servers - SQL based queues with [repository](https://martinfowler.com/eaaCatalog/repository.html) like access - Extreme throughput [when pure SQL based queues aren't enough](./queues/external-pending.md) - [In-memory](./queues/in-memory.md) queues with fallback storage when performance requires it - Versioned queue messages with optional identity to ensure safety - Access to [previous version of messages](./queues/index.md#previous-message-version) to enable sophisticated change tracking - [Virtual queues](./queues/virtual.md) with one or more backing queues - Processing queue messages in [batch or one by one](./queues/jobs.md) - Full [HTTP API](./api/index.md) and [Admin UI](./admin-ui/index.md) - Enqueue over [HTTP API](./queues/enqueueing.md) to let other services create messages - Great introspection, debuggability, and visibility into the contents of the queues - Easily re-enqueue failed messages - [Monitoring](./general/monitoring.md) of failed jobs, queue messages, and functions - Custom statuses for queue messages to allow grouping and saving for later - Enqueueing messages for [processing in the future](./queues/index.md#scheduling-messages-for-future-processing) - Optionally [keep processed messages](./queues/retention.md) - Retry processing a queue message in X minutes - Scheduled [health checks](./general/monitoring.md) - Can easily be embedded into an existing application - No external dependencies required, everything can be run locally on a dev machine - Metrics exposed with Open Telemetry - Out-of-the-box integrations with [Azure](./queues/azure.md), [Teams](./healthchecks/teams.md), [Slack](./healthchecks/slack.md), and [RabbitMQ](./queues/rabbitmq.md) See the section about [Nexus vs FaaS](./faas.md) if you read this and thought "Well I already have Azure Functions for this". If you've used [Hangfire](https://www.hangfire.io/) before you might want to read the [Hangfire vs Nexus](./general/hangfire-vs-nexus.md) comparison. ## Demo environment There's a demo environment that's deployed to a free tier Azure Web App here: **Admin UI**\ [https://commerce-mind-nexus-a9bsbwgth3cgfnde.swedencentral-01.azurewebsites.net/admin/](https://commerce-mind-nexus-a9bsbwgth3cgfnde.swedencentral-01.azurewebsites.net/admin/) **API**\ [https://commerce-mind-nexus-a9bsbwgth3cgfnde.swedencentral-01.azurewebsites.net/swagger/](https://commerce-mind-nexus-a9bsbwgth3cgfnde.swedencentral-01.azurewebsites.net/swagger/) You can click on any buttons you like, start any job or function or do anything with the queues to get a feel for what the system does. :::note[Demo environment] Since the demo environment is using a free tier Azure Web App the scheduler is paused when nobody is using the API or admin UI which means that the jobs won't run exactly according to schedule. ::: ## Getting started The first thing you need to do is to get access to the NuGet packages which are hosted on a Github packages. Just talk to us and we'll add you to the repo and give you access to the NuGet feed! Once you've gotten a Personal Access Token for the feed you need to configure NuGet to look for packages in this feed:\ [https://nuget.pkg.github.com/Commerce-Mind/index.json](https://nuget.pkg.github.com/Commerce-Mind/index.json) You use your Github username as username and the Personal Access Token as password. Read more about the Github NuGet feed here:\ [https://docs.github.com/en/packages/working-with-a-github-packages-registry/working-with-the-nuget-registry](https://docs.github.com/en/packages/working-with-a-github-packages-registry/working-with-the-nuget-registry) When you have access you need to install these NuGet packages in a .NET Core (6 or later) project: ``` CommerceMind.Nexus # One or more of: CommerceMind.Nexus.Postgres CommerceMind.Nexus.SqlServer CommerceMind.Nexus.Sqlite ``` This package is for the API and admin UI which is only relevant if you're using ASP.NET. ``` CommerceMind.Nexus.Api ``` When you've installed the packages you open your `Program.cs` and add this initialization code: ```cs var builder = WebApplication.CreateBuilder(args); builder.Services.AddSingleton(); builder.Services.AddControllers(); builder.Services .AddNexus() .AddSingleServerInstanceEvents() .AddFunctions() .AddScheduledJobs() .AddQueues() .AddApi() .AddSqliteConnection() // Or: // .AddPostgresConnection() // .AddSqlServerConnection() // Don't forget to call this, as this is what finializes the Nexus registration and validates it! .Build() ; var app = builder.Build(); // If you want the admin UI app.UseNexusAdminUI(); app.UseHttpsRedirection(); app.UseAuthorization(); app.MapControllers(); app.Run(); ``` Now let's create the most simple functions, jobs and queues possible, make sure to place it in the same .NET project as your `Program.cs` file. ```cs public class ExampleService { public void ExampleMethod(int arg) { Console.WriteLine("ExampleMethod was called with arg: " + arg); } } public class ExampleJob : IScheduledJob { private readonly IEnqueuer _enqueuer; private readonly INexusFunction _nexusFunction; public ExampleJob(IEnqueuer enqueuer, INexusFunction nexusFunction) { _enqueuer = enqueuer; _nexusFunction = nexusFunction; } public string DefaultSchedule => CronSchedule.TimesPerMinute(10); public async Task ExecuteAsync(CancellationToken cancellationToken) { await _enqueuer.EnqueueAsync(new ExampleQueueMessage()); await _nexusFunction.RunInBackgroundAsync(x => x.ExampleMethod(123)); return JobResults.Completed("This went great!"); } } public class ExampleQueueMessage : IQueueMessage { } public class ExampleQueueProcessingJob : IScheduledQueueJob { public string DefaultSchedule => CronSchedule.TimesPerMinute(10); public async Task ProcessMessageAsync(ExampleQueueMessage message, CancellationToken cancellationToken) { return ProcessResults.Completed("The message was processed!"); } } ``` Place a breakpoint in the `ExampleMethod()`, `ExecuteAsync()` and `ProcessMessageAsync()` methods and start the application. The breakpoints should get hit within a couple of seconds. Now we have a very basic example of one job scheduling a function as well as producing messages and placing them in a queue, and another job that process the messages. If using a web application you'll also be able to access the API on eg `/api/jobs` and `/api/jobs/ExampleJob/historical-runs`. And the admin UI is available on `/admin`. With the above setup you're using SQLite as the database which is great for quickly getting started, but it's not really fit for a heavy production load. Read more about database setup in the [database section](./general/database.md). When you've got the above working you should dig deeper into the docs here to understand everything you can do with the system! ## Trusted by Nexus is used in production by some of the Nordic region's most well-known e-commerce companies. ### Lyko Lyko is a leading Nordic beauty and hair care retailer operating across Sweden, Norway, Finland, and several other European markets. With a complex, constantly evolving IT environment spanning multiple group companies, Lyko manages B2C flows through third-party logistics providers using a custom-built Order Management System. Nexus is integrated directly into that OMS, giving developers immediate visibility into logs and making it easy to pinpoint and resolve problems across a distributed infrastructure. The flexible adapter concept makes it straightforward to add new integrations rapidly as Lyko expands into new markets and channels. > "The adapter concept provides flexible support and enables quick implementation of new integrations with high visibility out of the box." > > — Peter Gunnarsson, CTO Lyko ### Cervera Cervera is one of Sweden's most established kitchenware and home decor retailers, currently undergoing a major IT transformation — relocating their IT environment, replacing their ERP, and reworking the integrations that connect it all. Through Nexus, Cervera has unified their order flows across ERP, billing, purchasing, and analytical systems into a single, well-monitored integration layer. A key outcome has been shifting day-to-day monitoring from the technical team to business personnel, with live health check dashboards displayed on office screens. This gives the whole organisation fast, clear visibility into whether integrations are running as expected — and makes it easy to respond when they are not. > "Commerce Mind Nexus is a stable product that works well handling our integrations. What stands out is the visual overview and the ability to easily receive alerts about problems and rerun jobs with a single click." > > — Rasmus Andersson, CTO Cervera ### Kjell & Company Kjell & Company is Sweden's leading specialist retailer in consumer electronics, cables, and accessories, with more than 80 stores across the Nordic region alongside a growing e-commerce presence. Running an omnichannel operation means keeping a large number of systems — storefronts, warehouse management, ERP, and supplier feeds — continuously in sync. Nexus handles the background job orchestration and queue processing that keeps order and inventory data flowing reliably between those systems, both in real time and on schedule. The monitoring capabilities help the development team surface and fix integration issues quickly, before they reach customers. ### Thule Thule is a globally recognised Swedish brand specialising in outdoor lifestyle products — roof racks, bike carriers, bags, and luggage — sold in more than 100 countries. Running e-commerce operations at global scale demands tight, dependable integration between online storefronts, ERP systems, and an international network of logistics partners. Nexus manages the scheduled background jobs and queue processing that keep orders, product data, and fulfilment information flowing accurately across Thule's digital infrastructure. The solution's stability and built-in observability are a natural fit for a company where reliability is non-negotiable. ### Lekia Lekia is one of Sweden's most beloved toy retail chains, with stores across the country offering a wide range of toys, games, and children's products. Behind the scenes, keeping stock, orders, and supplier data in sync across both physical and digital channels requires a robust integration layer. Nexus manages the background job processing and queue handling that ensures Lekia's order flows run reliably, whatever the volume. The health check and monitoring features give the team confidence that everything is working as it should — and a clear path to action when something needs attention. ### Nordic Nest Nordic Nest is one of Europe's leading online destinations for Scandinavian design, founded in 2002 and offering more than 250 curated brands — from iconic names like Georg Jensen and Marimekko to contemporary favourites like Ferm Living and Normann Copenhagen. Operating across multiple European markets with worldwide shipping, Nordic Nest has built a reputation not only for their exceptional product selection but for staying at the forefront of e-commerce technology. Their engineering team embraces modern tooling and infrastructure to deliver a seamless shopping experience at scale, making them one of the most technically ambitious retailers in the Nordics. Nexus fits naturally into that culture — providing reliable, observable background processing that keeps their order and data flows running without friction. ### Ark Ark is Norway's largest bookstore chain, with more than 150 stores across the country alongside a thriving e-commerce platform at ark.no and a dedicated reading app for digital books. Owned by Gyldendal ASA, Ark has built one of the most ambitious omnichannel retail operations in Scandinavia, seamlessly connecting physical stores, online orders, and click-and-collect services for millions of customers. The scale of their operations — handling millions of book titles and a high volume of daily orders across every channel — demands an integration layer that is both robust and highly observable. Nexus provides the background job orchestration and queue processing that keeps Ark's order flows, inventory data, and system integrations running reliably across their entire operation. ### Hatstore Hatstore is a Swedish e-commerce company founded in 2011 and today one of Europe's leading online retailers for headwear, with over 20,000 caps, hats, and beanies from more than 100 brands including New Era, '47, and Mitchell & Ness. Operating regional webshops across more than ten European markets — including Sweden, Germany, the UK, France, Spain, and the Netherlands — Hatstore manages a complex multi-market setup with localised storefronts and a shared logistics infrastructure. Beyond off-the-shelf products, Hatstore also offers a custom design service with embroidery and printing, adding another layer of operational complexity to their order flows. Nexus helps keep those flows — across markets, product types, and fulfilment paths — running smoothly with full visibility into what is happening at every step. ### Science Fiction Bokhandeln Science Fiction Bokhandeln has been Sweden's home for science fiction, fantasy, and horror since 1984 — selling books, comics, manga, board games, role-playing games, and collectibles across stores in Stockholm, Gothenburg, Malmö, and Linköping, as well as online at sfbok.se. With a deeply passionate customer base and a catalogue spanning tens of thousands of titles and products, keeping inventory, orders, and supplier data in sync across both physical and digital channels is no small task. Nexus handles the background job processing and queue management that keeps their order flows running reliably, giving the team clear visibility into their integrations and the ability to act fast when something needs attention. --- # Jobs A job is a C# class that implements the interface [`IScheduledJob`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Jobs.Abstractions/IScheduledJob.cs). The simplest possible implementation of a job would look like this: ```cs public class MyJob : IScheduledJob { public string DefaultSchedule => CronSchedule.TimesPerMinute(10); public async Task ExecuteAsync(CancellationToken cancellationToken) { Console.WriteLine("The job ran"); return JobResults.Completed(); } } ``` By default the name used a key for the job in the database is its type/class name. So in the example above it would be `MyJob`. This means that if you rename the job class name the meta data will be lost and the system will think it's a new job. The class name might also not be a great candidate for the display name of the job and to solve that you can use the `[ScheduledJob]` attribute like this: ```cs [ScheduledJob("myjob", DisplayName = "My really great job")] public class MyJob : IScheduledJob { } ``` Now we'll use `myjob` as the key in the database and the admin UI will display it as `My really great job`. Now you can also rename the job class without loosing the meta data. ## IScheduledJob.DefaultSchedule This property tells the job engine what the default schedule of the job is, and in the above example we're scheduling it to run ten times per minute. But `DefaultSchedule` is just a string that should contain a valid cron schedule (including seconds) so you can return any valid cron expression. `CronSchedule` here is just a helper class for generating such schedules. The engine is using [`Cronos`](https://github.com/HangfireIO/Cronos) to parse cron expressions, so any expression that is valid in `Cronos` will work. The reason for the property being called `DefaultSchedule` rather than `Schedule` is because it will only be read once by the system and that is the first time the engine registers the job. After that the schedule is saved to the database and can be changed through the UI and the API. If you set `DefaultSchedule` to an empty string (or use `CronSchedule.NotScheduled()`) the job will get registered but the engine will never execute it unless you start it manually through the UI or the API. ## IScheduledJob.ExecuteAsync() This method is where you'll place the code that should be performed by the job. The example just contains a simple `Console.WriteLine()` which won't be very useful in a real scenario. The job instance is created using the .NET `IServiceProvider` so you can request any services you need through constructor injection. :::note[Instantiation] Note that the job might be instantiated for other reasons than to call `ExecuteAsync()` so don't do any work in the constructor, you should only set instance fields. ::: ## CancellationToken The `ExecuteAsync()` method is passed a [`CancellationToken`](https://docs.microsoft.com/en-us/dotnet/api/system.threading.cancellationtoken?view=net-6.0) by the engine which the job should respect. If you've never worked with cancellation tokens before it's a way for the outside to signal into the job that someone wants to cancel the job execution. It's up to every job to respect the cancellation token passed in and if you don't do that the job can't be cancelled. It'll either always run to completion or be forcefully killed if the job service is killed. The cancellation token passed in is a combination of the general service cancellation token and a cancellation token for that specific job run. The service cancellation token can be triggered by a deploy or a machine restart whereas the cancellation token for a specific job is only triggered if someone manually cancels the job through the UI or the API. The difference here is not something a job needs to think about, it only have to concern itself with the cancellation token passed to `ExecuteAsync()`. Here's an example of how you would use the cancellation token: ```cs public async Task ExecuteAsync(CancellationToken cancellationToken) { foreach (var item in await _service.GetBigListOfThingsAsync()) { cancellationToken.ThrowIfCancellationRequested(); // Do stuff with the item } return JobResults.Completed(); } ``` You add calls to `cancellationToken.ThrowIfCancellationRequested();` inside loops or between chunks of code that are time consuming. A general rule is that the job should never execute code for more than one second before calling `cancellationToken.ThrowIfCancellationRequested();` again. `ThrowIfCancellationRequested()` will throw an `OperationCanceledException` which the job engine will catch and report that job run as cancelled. ## JobResults The return type of a job is [`JobResults`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Jobs/JobResults.cs) which is a class that represents the outcome of a job run. `JobResults` contains static methods such as `Completed()` that tells the engine if the job ran successfully. The `Completed()` method takes an optional string describing the outcome of the job which is visible under the historical runs of the job in the UI. Something like `"Exported 5 orders, 2 of which failed"`. ### JobResults.NothingToDo() If your job starts up and you see that there's no work to do, you should return `JobResults.NothingToDo()` rather than `JobResults.Completed()`. Returning `NothingToDo()` signals to the engine that this run isn't interesting, so it won't get saved to the historical runs. This helps to not clutter up the job history in the UI with runs that didn't perform any interesting work. ### JobResults.RetryLater() Sometimes your job detects that it can't do what it's supposed to do right now. It might be that an external system is down or that the data it needs isn't available right now. For jobs that run less frequently this can be an annoying thing. Lets say that you schedule a job to run every night but external systems might have downtime during the night so now you have to wait a whole day for it to run again. Instead of waiting that whole day your job can return eg `JobResults.RetryLater(TimeSpan.FromMinutes(10))` to tell the engine to schedule an extra run of that job in ten minutes. More on extra runs down below. ### JobResults.Failed() `JobResults` contains a method called `Failed()` which you don't need to call manually. Nexus will catch any exceptions that occur and mark the run as failed. This method is useful if you want to return parameters to the next run of the job. Read more about [jobs with parameters here](parameters.md). ### JobResults.CompletedWithWarnings() Sometimes your job does what it's supposed to do but detects that some things aren't really what they should. Something might be taking a longer time than expected, or some import or exports succeeds but generates warnings. In these cases you can return `JobResults.CompletedWithWarnings()` and pass a descriptive text of the warnings. The jobs health checks state will now be `Degraded` (rather than `Healthy` or `Unhealthy`) and will show with an orange warning sign in the Admin UI. You should use standard logging inside your job with additional details about the warnings to be able to resolve it by logging at the job logs. The recommendation is to use this for things that needs to be looked into but isn't urgent or critical enough to use `JobResults.Failed()`. Don't use it for things that aren't actionable since it'll only create noise. ## Max job duration Most jobs have an average expected duration and can sometimes take a much longer time to complete than expected. For some jobs this doesn't matter and the duration can vary a lot. Such as a queue job that sometimes needs to process a lot of messages. But if you have jobs where you want to be notified if the job runs for too long you can set the max duration either in the [Admin UI](../admin-ui/index.md) under the More button in the job details, or by setting `MaxDuration` on the `[ScheduledJob]` attribute like this: ```cs [ScheduledJob("myjob", MaxDuration = "00:10")] public class MyJob : IScheduledJob { ... } ``` The value should be any string that can be passed to [`TimeSpan.Parse()`](https://docs.microsoft.com/en-us/dotnet/api/system.timespan.parse?view=net-6.0). In the above example the max duration is set to 10 minutes. When a job has been running for a longer time than `MaxDuration` the [Admin UI](../admin-ui/index.md) will show a warning message about the jobs duration, and the [health check](../general/monitoring.md) for the job will report a `Degraded` health status. Or be cancelled if automatic cancellation is enabled. Read more about automatic cancellation below. ## Automatic cancellation Sometimes you might have jobs that aren't allowed to run for a longer period of time than the defined max duration. In such cases you can enable automatic cancellation either in the [Admin UI](../admin-ui/index.md) under the More button in the job details, or through the job attribute like this: ```cs [ScheduledJob("myjob", MaxDuration = "00:10", AutoCancel = true)] public class MyJob : IScheduledJob { ... } ``` In the above case the jobs cancellation token will be triggered after ten minutes. ## Disabling a job A job can be disabled using the [Admin UI](../admin-ui/index.md) or the API. This means that the job won't start on it's schedule, or when queue processing is requested. It will only start if you manually start it through the UI or API. This is very useful for when external systems that the job depends on are having issues, or when the job isn't behaving as it should. In order to not forget about enabling important jobs again you can say that the health check for the job should get a `Degraded` status when it's disabled: ```cs [ScheduledJob("myjob", DegradedWhenDisabled = true)] public class MyJob : IScheduledJob { ... } ``` You can also create a new job directly in a disabled state. This can be useful when the job does senstive work where you need to be in control of when it starts. Any new job will have its first run directly when its deployed for the first time which may not be ideal in all cases. You can set the job to be initially disabled like this: ```cs [ScheduledJob("myjob", InitiallyDisabled = true)] public class MyJob : IScheduledJob { ... } ``` ## Pausing on error Sometimes a job can be very sensitive to errors and in those cases you might want to prevent a job from running again until the error has been resolved. Nexus lets you handle this by setting `PauseOnError` in the `[ScheduledJob]` attribute like this: ```cs [ScheduledJob("myjob", PauseOnError = true)] public class MyJob : IScheduledJob { ... } ``` Previously this could be done by implemeting a `bool StopProcessingOnError { get; }` on queue jobs but that has been deprecated and will be removed in the next major version. Note that this can be overriden in the Admin UI/API under the More button in the job details page. ## How to register jobs The job engine will use reflection to look for classes implementing the [`IScheduledJob`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Jobs.Abstractions/IScheduledJob.cs) or [`IScheduledJob`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Jobs.Abstractions/IScheduledJob.cs) interfaces and automatically register them. This is only done automatically for the entry assembly so if you have multiple .NET projects that contains job classes you need to register the assemblies with the service. There's an extension method on [`IServiceCollection`](https://docs.microsoft.com/en-us/aspnet/core/fundamentals/dependency-injection?view=aspnetcore-6.0) called `AddNexus().AddScheduledJobs()` which you can call to scan that assembly for jobs like this: ```cs builder.Services.AddNexus().AddScheduledJobs(options => { options.AssembliesToScanForJobs.Add(typeof(MyJob).Assembly); }); ``` If you have assemblies that contains some jobs you don't want to register for some reason you can use the options property `ScheduledJobFilter` like this: ```cs builder.Services.AddNexus().AddScheduledJobs(options => { options.ScheduledJobFilter = jobType => jobType != typeof(JobThatIDontWant); }); ``` ## Log retention By default the job history is stored for 30 days but you can configure for how long job history and [logs](./logs.md) should be stored like this: ```cs builder.Services.AddNexus().AddScheduledJobs(options => { options.DefaultHistoricalRunsRetention = TimeSpan.FromDays(10); }); ``` You can also specify this per job with the `[ScheduledJob]` attribute like this: ```cs [ScheduledJob("MyJob", HistoricalRunsRetention = "30.00:00")] public class MyJob : IScheduledJob { ... } ``` The string you set is any string that can be passed to `TimeSpan.Parse()` or the special string `forever` to indicate that Nexus should never delete the job history. Note that this value can be updated in the Admin UI under the More button on the job details page. ## Extra job runs A job typically starts either by its schedule saying that it should start or by someone manually starting the job. But a job can also have extra runs outside of it's normal schedule. An extra run can be scheduled either using the API endpoint `POST /jobs/{jobName}/start?startAt=2022-08-19T12:00:00` (date should be in UTC) or using the service [`IScheduledJobMetaDataRepository.ScheduleExtraRunAsync()`](https://github.com/Commerce-Mind/Nexus/blob/aeccf863f9d3980a556c45cd95bcb2819557b40b/src/job-engine/Nexus.Jobs/IScheduledJobMetaDataRepository.cs). ## Starting a job programatically To explicitly start a job through code you have two options. You can either use the API endpoint `POST /jobs/{jobName}/start` or use the service [`IJobStartRequester`](https://github.com/Commerce-Mind/Nexus/blob/aeccf863f9d3980a556c45cd95bcb2819557b40b/src/job-engine/Nexus.Jobs/IJobStartRequester.cs). When you do this an extra run of the job is scheduled to start as soon as possible. If the job is currently running another run will start immediately after. ## Job parallelism and mutexes The engine guarantees that a single job never executes multiple times in parallel. Even if it's scheduled to run for eg ten times per minute it will only run once per minute if it takes 60 seconds to complete. Only one instance is allowed to have the job running at any given time. This is done because many jobs communicate with external data sources and having the same job doing that in parallel can often cause bugs. If you wish to speed things up you should start multiple threads inside your job to parallelize the work. If you have two or more jobs that you want to ensure never run at the same time you can use the [`[RequireJobMutex]`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Jobs/Attributes/RequireJobMutexAttribute.cs) attribute on your job. Here's an example: ```cs [RequireJobMutex("SomeMutex")] [RequireJobMutex("SomeOtherMutex")] public class Job1 : IScheduledJob { // Code omitted for brevity } [RequireJobMutex("SomeMutex")] public class Job2 : IScheduledJob { // Code omitted for brevity } [RequireJobMutex("SomeOtherMutex")] public class Job3 : IScheduledJob { // Code omitted for brevity } ``` In this example `Job1` can never run at the same time as `Job2` and `Job3`. `Job2` can run at the same time as `Job3` but not as `Job1` since `Job2` and `Job3` needs different mutexes. In more advanced scenarios where you want multiple jobs to process the same queue with different [filters](../queues/jobs.md#filtering-messages) you can acquire mutexes dynamically as part of the job execution instead: ```cs public class MyJob(IScheduledJobMetaDataRepository jobMetaDataRepository) { public async Task ExecuteAsync(CancellationToken cancellationToken) { await using var _ = await jobMetaDataRepository.AcquireMutexesAsync(["some-mutex-name"], cancellationToken: cancellationToken); // The mutex has been acquired return JobResults.Completed(); } } ``` The mutexes are decentralized and safe to use with multiple Nexus servers/instances running your jobs. If you're only using a single server you're probably better of using `SemaphoreSlim` to handle locking. ## Job middlewares In some cases you want to wrap the execution of a job. You might want to add additional log context properties or instrument the job execution in some way. To achieve this you can register one or more `IScheduledJobMiddleware` instances in the service collection. Eg: ```cs public class MyJobMiddleware(ILogger logger) : IScheduledJobMiddleware { public async Task ExecuteAsync(ScheduledJobDescriptor scheduledJobDescriptor, Func> executeJob) { using logger.BeingScope(new Dictionary {{ "JobName", scheduledJobDescriptor.Name }}); return await executeJob(); } } // Register the middleware builder.Services.AddSingleton(); ``` ## Organizing the Admin UI If you have a lot of jobs you can group jobs in the Admin UI by a category, just like you can with queues. Set a category in the `ScheduledJob` attribute like this: ```cs [ScheduledJob("myjob", Category = "My category")] public class MyJob : IScheduledJob { } ``` Or set the category through the Admin UI under the More button on the job details page. ## Job loop delay By default Nexus will check once per second if there are any new jobs to start since it's not possible to schedule jobs to run more often than that. But in some situations you might want to lower that delay, eg if you use schedule extra runs frequently or if your queue jobs use `MaxMessagesPerRun` and you want the job to restart as fast as possible. In such cases you can configure the job loop delay like this: ```cs builder.Services.AddNexus().AddScheduledJobs(options => { options.JobLoopDelay = TimeSpan.FromSeconds(0.5); }); ``` --- # Job logs Nexus has a [`ILoggerProvider`](https://learn.microsoft.com/en-us/dotnet/api/microsoft.extensions.logging.iloggerprovider?view=net-8.0) which will collect logs by job run and store in the database. Nexus is compatible with any logging framework such as Serilog or NLog as long as it supports external providers. With Serilog writing to providers is disabled by default so you need to enable it to get Nexus logs to work: ```cs builder.Host.UseSerilog((context, services, configuration) => { configuration.ReadFrom.Configuration(context.Configuration); // This one is important }, writeToProviders: true); ``` ## `IJobLogWriter` Collected job logs are continuously flushed to the [`IJobLogWriter`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Jobs/IJobLogWriter.cs) by the job executor as they occur. There's three implementations of this interface built in. One for Postgres, one for SQL Server and one for SQLite and whichever database you've chosen to store job meta data is also used for logs. Logging to the database is enabled by default, but you can turn it off like this: ```cs builder.Services.AddNexus().AddScheduledJobs(options => { options.SaveLogsToDatabase = false; }); ``` Job logs are then stored in the database for as long as you'e set the retention for historical runs. When a historical run is deleted the logs for that run is also deleted. The default is to store historical runs and logs for 30 days, but you can configure it like this: ```cs builder.Services.AddNexus().AddScheduledJobs(options => { options.DefaultHistoricalRunsRetention = TimeSpan.FromDays(10); }); ``` You can also specify this per job with the `[ScheduledJob]` attribute like this: ```cs [ScheduledJob("MyJob", HistoricalRunsRetention = "30.00:00")] public class MyJob : IScheduledJob { ... } ``` The string you set is any string that can be passed to `TimeSpan.Parse()` or the special string `forever` to indicate that Nexus should never delete the job history. ## `IJobLogReader` When requesting job logs for a specific run the [`IJobLogReader`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Abstractions/Jobs/IJobLogReader.cs) service is called. The built in implementations that stores logs in the database also implement this interface. When calling `AddNexus().AddScheduledJobs()` like the example above both the writer and the reader gets registered. If you send your logs to an external source like Elastic Search or Application Insights you can provide your own implementation of this interface. :::note You can view the job logs in the [admin UI](/admin-ui) where all logs for a single job run are displayed to make troubleshooting a job easier ::: ## Elasticsearch If you're sending logs to Elasticsearch using [`Elastic.Serilog.Sinks`](https://www.elastic.co/docs/reference/ecs/logging/dotnet/serilog-data-shipper) you can install the package `CommerceMind.Nexus.Elastic` and read job logs from Elastic instead. After you've installed if you add a call to `AddElasticJobLogs()` like this: ```cs builder.Services .AddNexus(...) // ... other Nexus initialization calls .AddScheduledJobs(options => { var isLocalDev = ...; options.SaveLogsToDatabase = isLocalDev; }) .AddElasticJobLogs(options => { options.ElasticUrl = "https://elastic:mypassword@localhost:9200"; options.LogIndexName = (serviceProvider) => "myindex"; // Or if you want complete control of the client settings: options.ClientSettings = (serviceProvider) => new ElasticsearchClientSettings(new Uri("https://elastic:mypassword@localhost:9200")) .DefaultIndex("myindex"); }) .Build(); ``` If you're sending logs to Elastic using some other transport than [`Elastic.Serilog.Sinks`](https://www.elastic.co/docs/reference/ecs/logging/dotnet/serilog-data-shipper) it might not work as expected because your transport might not be creating Elastic documents in the same way as the Serilog sink. If it doesn't work for you then check the property names that the job name, correlation id and log level is written to and set them in the configuration function like this: ```cs .AddElasticJobLogs(options => { options.CorrelationIdPropertyName = "..."; options.JobNamePropertyName = "..."; options.LogLevelPropertyName = "..."; }) ``` Please reach out to us and we'll try to modify the log reader to accommodate your transport as well. :::note CommerceMind.Nexus.Elastic currently only works with Elastic 8. If you're running Elastic 9 on the server you'll need to copy the implementation of the log reader using the Elastic 9 SDK as it's not possible for Nexus to support multiple major versions. ::: --- # Jobs with parameters In some situations it is useful for a job to take parameters: a data bag supplied from outside the job run. Some example use cases are: - A job that fetches data from an external system based on a change date can include a `DateTime` in its parameters and pass a date to the next execution. The future run can then fetch only changes that happened after the previous run. - Parameters can be supplied explicitly when starting a job from the Admin UI, allowing a manual run to affect how the job is executed. - An occasionally useful side effect, such as writing detailed troubleshooting output, can be controlled for an individual run instead of through application configuration. - A user can upload a file that the job reads during its execution. ## Defining a parameterized job Define a scheduled job that takes parameters by implementing [`IScheduledJob`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Abstractions/Jobs/IScheduledJob.cs). For example: ```cs public class MyJobParameters { public DateTime LastRunDate { get; set; } } public class MyJobWithParameters : IScheduledJob { public string DefaultSchedule => CronSchedule.TimesPerMinute(10); public MyJobParameters DefaultParameters => new(); public async Task ExecuteAsync(MyJobParameters parameters, CancellationToken cancellationToken) { var now = DateTime.UtcNow; var changes = await _service.GetChangesAsync(from: parameters.LastRunDate, to: now); // Do something interesting with the changes return JobResults.Completed($"Processed {changes.Count} changes", new MyJobParameters { LastRunDate = now, }); } } ``` The parameters class can be any class that can be serialized to and from JSON. Nexus stores default, next-run, retry, and manually supplied parameters as JSON. No separate database schema is created for a parameters class. To customize parameter serialization, register an implementation of [`IJobParametersSerializer`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Abstractions/Jobs/IJobParametersSerializer.cs) in the dependency injection container. ## Editing parameters in the Admin UI When Nexus can describe a job's parameter properties, the Admin UI offers two modes: - **Fields** is selected initially and renders controls appropriate for each property. - **JSON** provides direct access to the complete parameters object for technical users. The same editor is used when starting a job and when editing its default or next-run parameters on the job details page. The start-job editor is initialized from the saved default parameters, falling back to the job's parameter example when no defaults are available. Changes are preserved when switching modes. Switching from JSON to Fields requires a valid JSON object. Fields are validated before switching to JSON or submitting the form. File selection is available only in Fields mode. ### Supported fields Nexus describes public properties that have both a public getter and public setter. Read-only, indexed, and JSON-ignored properties are omitted. Objects, collections, and scalar types without a dedicated control are represented by an individual JSON editor rather than being recursively expanded. The JSON tab always shows the serialized representation and does not perform timezone conversion or provide file pickers. ### Labels, descriptions, and serialized names By default Nexus creates a label from the property name. Attributes can provide a better label, help text, or serialized property name: ```cs public class ImportParameters { [DisplayName("Changes from")] [Description("Only import changes made on or after this date.")] [JsonPropertyName("fromDate")] public DateOnly From { get; set; } } ``` `DisplayNameAttribute` and `DisplayAttribute` control the label, while `DescriptionAttribute` adds help text. `JsonPropertyNameAttribute`, Newtonsoft.Json's `JsonPropertyAttribute`, and the configured System.Text.Json property naming policy are respected when available. Both System.Text.Json and Newtonsoft.Json ignore attributes are respected. The API exposes this metadata in `JobSummary.ParameterFields`. `JobSummary.FileUploadsEnabled` tells clients whether file controls can currently accept uploads. ## Default parameters `DefaultParameters` should return an instance of the parameters class. It is passed to `ExecuteAsync()` when Nexus has no other parameters to use, such as for the first run or when the previous run did not provide parameters for the next run. Default parameters can be edited in either Fields or JSON mode on the job details page. Values saved in the Admin UI override the `DefaultParameters` property on the job class. To remove the saved override and use the value from code again, use the Reset action on the job details page. ## Next-run parameters A job can pass parameters to its next execution by returning them in its `JobResults`, as shown in the first example. This is optional. The next-run parameters can also be edited in Fields or JSON mode on the job details page. If neither a previous run nor a manual start supplies parameters, Nexus uses the default parameters. ## Parameters and job failure If a non-manual job run with parameters fails, its next run receives the same parameters. In the first example, `LastRunDate` therefore contains the date set by the last successful run. To use different parameters after a failure, catch the error and pass explicit parameters to `JobResults.Failed()`: ```cs public async Task ExecuteAsync(MyJobParameters parameters, CancellationToken cancellationToken) { var now = DateTime.UtcNow; try { // Implementation excluded for brevity } catch (Exception e) { return JobResults.Failed(e, new MyJobParameters { LastRunDate = now, }); } } ``` The run is still marked as failed, but the next run receives the parameters supplied to `JobResults.Failed()`. ## Uploaded files Use [`JobFile`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Abstractions/Jobs/JobFile.cs) for a file parameter: ```cs public class ImportParameters { [Description("The CSV file to import.")] public JobFile InputFile { get; set; } = null!; } public class ImportJob : IScheduledJob { public string DefaultSchedule => CronSchedule.NotScheduled(); public ImportParameters DefaultParameters => new(); public async Task ExecuteAsync(ImportParameters parameters, CancellationToken cancellationToken) { var contents = parameters.InputFile.ReadAsText(); // Import the contents return JobResults.Completed(); } } ``` ### Configuring filesystem storage File uploads are disabled by default. Set `UploadedFilesPath` when registering scheduled jobs to enable the built-in filesystem storage: ```cs builder.Services .AddNexus() .AddScheduledJobs(options => { options.UploadedFilesPath = @"\\fileserver\shared\nexus-job-files"; options.UploadedFileRetention = TimeSpan.FromDays(14); options.UploadedFileCleanupInterval = TimeSpan.FromHours(1); }); ``` Use a shared folder when a job can be uploaded on one application instance and executed on another. The built-in storage uses generated opaque filenames and never uses the uploaded filename as a filesystem path. Storage keys are validated before paths are resolved. `UploadedFileRetention` defaults to seven days, and `UploadedFileCleanupInterval` defaults to one hour. A hosted cleanup loop sweeps once at startup and then at the configured interval. It also removes uploads abandoned when a user selects a file but never successfully starts or saves the job parameters. Uploaded references can be reused by retries and later runs only until cleanup removes the underlying file. Configure retention to be longer than the maximum expected delay before a run, including retries. Be especially careful when placing uploaded references in default or next-run parameters, because those references do not make the file permanent. When uploads are disabled, the Admin UI keeps file controls disabled and explains that `UploadedFilesPath` or a custom storage implementation is required. ### Using custom file storage Implement [`IJobFileStorage`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Abstractions/Jobs/IJobFileStorage.cs) to store files somewhere else, such as Azure Blob Storage: - `IsEnabled` reports whether the storage is available for uploads. - `UploadAsync()` stores the supplied stream and returns a `JobFile` containing a persistable opaque reference. - `OpenReadAsync()` reconstructs a readable stream from that reference before job execution. - `DeleteOlderThanAsync()` deletes storage objects older than the supplied UTC cutoff. Register the custom implementation before adding scheduled jobs so it replaces the default registration: ```cs builder.Services.AddSingleton(); builder.Services .AddNexus() .AddScheduledJobs(options => { options.UploadedFileRetention = TimeSpan.FromDays(30); }); ``` `UploadedFilesPath` is not required for custom storage. The same retention and cleanup interval options control calls to the custom implementation's `DeleteOlderThanAsync()` method. --- # Job progress Some jobs are expected to take a long time to complete and for such jobs it's nice to be able to see the progress for the job. Nexus has built-in support for jobs to report their progress through the `IJobProgress` interface like this: ```cs public class ExampleJob : IScheduledJob { private readonly IJobProgress _jobProgress; public ExampleJob(IJobProgress jobProgress) { _jobProgress = jobProgress; } public async Task ExecuteAsync(CancellationToken cancellationToken) { var thingsToProcess = await GetThingsToProcessAsync(); _jobProgress.SetTotal(thingsToProcess.Count); foreach (var thingToProcess in thingsToProcess) { cancellationToken.ThrowIfCancellationRequested(); await ProcessAsync(thingToProcess); _jobProgress.Increment(1); } return JobResults.Completed(); } } ``` The important lines here are `IJobProgress.SetTotal(thingsToProcess.Count);` and `IJobProgress.Increment(1);`. The call to `IJobProgress.SetTotal()` tells Nexus how much work the job expects to do and `IJobProgress.Increment(1);` tells Nexus that we've done some work. The numbers to use here are up to you. If your job does multiple things and some operations take a longer time to complete than others you can increment the progress with a higher number than `1`. You can call the `SetTotal()` method multiple times during a job. Eg if the job sees that there's more work to do than first expected. If you increment the progress more than what the total has been set to the total will automatically increase. ## Progress message The progress can report an optional message as well. Something like this: ```cs public class ExampleJob : IScheduledJob { private readonly IJobProgress _jobProgress; public ExampleJob(IJobProgress jobProgress) { _jobProgress = jobProgress; } public async Task ExecuteAsync(CancellationToken cancellationToken) { _jobProgress.SetMessage("Calculating work to be done..."); var thingsToProcess = await GetThingsToProcessAsync(); var otherThingsToProcess = await GetOtherThingsToProcessAsync(); _jobProgress.SetTotal(thingsToProcess.Count + (otherThingsToProcess.Count * 2)); _jobProgress.SetMessage("Processing things..."); foreach (var thingToProcess in thingsToProcess) { cancellationToken.ThrowIfCancellationRequested(); await ProcessAsync(thingToProcess); _jobProgress.Increment(1); } _jobProgress.SetMessage("Processing other things..."); foreach (var otherThingToProcess in otherThingsToProcess) { cancellationToken.ThrowIfCancellationRequested(); await ProcessAsync(otherThingToProcess); _jobProgress.Increment(2); } return JobResults.Completed(); } } ``` Before we know how much work we have to do, we set a progress message which at least feedbacks that's something happening and where we are in the job execution. In the above example we also give `otherThingsToProcess` a weight of `2` instead of `1` because we anticipate that they will take a longer time to process. Which gives a better indication as to when the job will be finished. For some jobs you might not have numbers to increment but you still want to give feedback to what's happening. In such cases you can use just the progress message to indicate what the job is doing. The admin UI will display the message but won't show a progress bar. Note that the message isn't cleared if you increment progress or change the total. If you wish to clear the current message you'll need to call `IJobProgress.SetMessage(null)`. ## Calling `IJobProgress` outside of a job Most of the time you will call `IJobProgress` directly from inside the job implementation but there's nothing stopping you from calling it inside other services. If you call `IJobProgress` from a service when the service isn't invoked in a job context it will simply not doing anything. `IJobProgress` stores a [`AsyncLocal`](https://learn.microsoft.com/en-us/dotnet/api/system.threading.asynclocal-1?view=net-7.0) instance connected to it's current running job and if there's no context the call to `IJobProgress` will simply not do anything. If you try to get a hold of `IJobProgress` in an assembly or application which hasn't called registered the jobs system with `IServiceCollection.AddNexus().AddScheduledJobs()` you'll need to inject your own mock implementation of the interface as otherwise you'll get an exception that there's no registered implementation of the interface. --- # Functions Nexus functions are a lightweight way of executing a method on a service in the background. Sometimes you want to execute some method but you want it to run in the background. Either because the method takes a long time or because you want to make sure that it's retried until it completes successfully. Let's take a simplified example: ```cs public class OrderController : Controller { private readonly IOrderConfirmationEmailService _orderConfirmationEmailService; public OrderController(IOrderConfirmationEmailService orderConfirmationEmailService) { _orderConfirmationEmailService = orderConfirmationEmailService; } public async Task OrderPlaced(string orderId) { await _orderConfirmationEmailService.SendEmailForOrderAsync(orderId); return Json(new { Success = true, }); } } ``` We need to send the order confirmation email, but there's a couple of issues with this approach. What happens if `SendEmailForOrderAsync()` throws an error? And what if it takes a couple of seconds to complete? If it throws an exception we'll miss sending the email for that order. And if it takes a couple of seconds we'll slow down the experience for the end customer. We can fix that by using a Nexus function instead, like this: ```cs public class OrderController : Controller { private readonly INexusFunction _nexusFunction; public OrderController(INexusFunction nexusFunction) { _nexusFunction = nexusFunction; } public async Task OrderPlaced(string orderId) { await _nexusFunction.RunInBackgroundAsync(x => x.SendEmailForOrderAsync(orderId)); return Json(new { Success = true, }); } } ``` With this change we've moved the actual execution of the `SendEmailForOrderAsync()` method into the background (maybe even to a different server). When Nexus executes the method it will catch any exceptions that happen and retry the call later. You can control the retry behavior using the `[NexusFunction]` attribute which you can read about below. ## Registering functions The only things you need to do is to call `AddFunctions()` like this: ```cs builder.Services.AddNexus().AddFunctions(); ``` And then install either `CommerceMind.Nexus.Sqlite`, `CommerceMind.Nexus.Postgres`, or `CommerceMind.Nexus.SqlServer` and initialize it: ```cs builder.Services .AddNexus() .AddFunctions() // Either: .AddSqliteConnection() // Or: .AddPostgresConnection() // Or: .AddSqlServerConnection() .Build(); ``` After that you can use `INexusFunction` and start running methods in the background. Note that by default the application will also start to run functions in the background. If you want an application to only add functions but not run them you can set `UseBackgroundService` to false like this: ```cs builder.Services.AddNexus().AddFunctions(options => { options.UseBackgroundService = false; }); ``` You must have at least one application that runs functions so this is only if you have multiple applications where one or more is dedicated for background processing. If you're not using a continously running server and instead using something like Azure Functions or Azure Container Instances that start on a schedule you can set `UseBackgroundService` to `false` and instead call `INexusFunctionExecutorService.RunAllFunctionsPendingSinceLastCallAsync()` from your entrypoint. That method will run all functions that have been scheduled to run since the last time that method was called. See the example [Azure function trigger here](../faas.md). :::note[Keeping processed functions] By default Nexus will store meta data about processed functions for one day in the database before deleting them. You can control this by setting `AddNexus().AddFunctions(options => options.KeepProcessedRunsFor = TimeSpan.Zero);`. You can also set the retention per function using the `[NexusFunction]` attribute which you can read about below. ::: ## Service registration You can use any service you want when calling `RunInBackgroundAsync()`. The only requirement is that the service is registered in the `IServiceCollection`. Which means that you can use interfaces as well as long as you register an implementation for it. So for the `IOrderConfirmationEmailService` example above you'd do something like this: ```cs serviceCollection.AddSingleton(); ``` :::note[Multi threading] If the same method on the same service is called multiple times Nexus can sometimes call the same service in multiple threads at the same time. Make sure that all services that you use with Nexus functions has thread safe implementations. ::: ## Invoking functions in the Admin UI If you want to be able to invoke functions yourself through the Admin UI you need to make Nexus aware of which methods you intend to use as a function. Nexus allows invoking any method using `INexusFunction` in code, but requires your methods to be decorated with the `[NexusFunction]` attribute in order to list them as a function in the admin UI. When invoking a function in the admin UI you input the arguments to the method as JSON. ## Method arguments Nexus will automatically capture the values of arguments passed to the method like in the example above with the `orderId` argument. The only requirement is that it can only be values that can be serialized to JSON. Which means that you can pass strings, numbers, dates, and data objects to the service method. But you can't pass another service for example. If you have a method that requires another service as an argument you need to create a separate Nexus friendly method. Something like this: ```cs public class ExampleService { private readonly OtherService _service; public ExampleService(OtherService service) { _service = service; } public void ExampleMethod(OtherService service, string someString) { service.SomeMetod(someString); } public void ExampleMethod(string someString) { ExampleMethod(_service, someString); } } ``` So instead of having Nexus call the `ExampleMethod(OtherService service, string someString)` method you'd let it call `ExampleMethod(string someString)`. You can of course create a separate service that does this instead of having the method in the same service. If you need to change the way arguments are serialized you can register a custom implementation of `INexusFunctionArgumentsSerializer` with the service provider. ## Cancellation token The methods on `INexusFunction` has overloads that passes a cancellation token to the provided function. This lets you pass the cancellation token on to your service method like this: ```cs await _nexusFunction.RunInBackgroundAsync((s, ct) => s.LongOperation(ct)); ``` The cancellation token passed to the service is the [graceful shutdown](../general/hosting.md#graceful-nexus-shutdown) token that will be triggered either if a graceful shutdown has been initiated or if the application is shutting down because of a deploy or server restart. ## Running a method after a delay If you don't want to run the method as fast as possible you can call `RunInBackgroundAfterAsync()` instead of `RunInBackgroundAsync()`. Like this: ```cs await _nexusFunction.RunInBackgroundAfterAsync(TimeSpan.FromMinutes(10), x => x.SendEmailForOrderAsync(orderId)); ``` ## Running a method until a condition is met Sometimes you want to keep running a method until some condition is true such as polling an external system. Let's say that you don't want to send the order confirmation email until the customer on the order has been exported to the CRM. You can use recursion to let a method called by Nexus schedule a new run, something like this: ```cs public MyOrderConfirmationEmailService { private readonly ICustomerService _customerService; private readonly IOrderService _orderService; private readonly INexusFunction _nexusFunction; public MyOrderConfirmationEmailService(ICustomerService customerService, IOrderService orderService, INexusFunction nexusFunction) { _customerService = customerService; _orderService = orderService; _nexusFunction = nexusFunction; } public async Task SendEmailForOrderAsync(string orderId) { var order = await _orderService.GetAsync(orderId); if (!await _customerService.ExistsAsync(order.CustomerNumber)) { // Wait 5 minutes to see if the customer has been exported await _nexusFunction.RunInBackgroundAfterAsync(TimeSpan.FromMinutes(5), x => x.SendEmailForOrderAsync(orderId)); } else { // Customer exists, let's break the recursion and send the email await SendEmailAsync(order); } } private async SendEmailAsync(Order order) { // Send the email } } ``` In this example Nexus will continuously call `SendEmailForOrderAsync(orderId)` every five minutes until the customer has been exported. ## [NexusFunction] attribute The `[NexusFunction]` attribute lets you control some aspects of how Nexus will run a method. You specify the attribute either on the service class/interface or on individual methods. You can use `[NexusFunction]` to control for how long meta data about successful runs of a method is stored like this: ```cs [NexusFunction(KeepProcessedRunFor = "10")] public void ExampleMethod() { } ``` This will store all calls to this method for 10 days as the value for `KeepProcessedRunFor` must be a string that can be parsed to a `TimeSpan` using `TimeSpan.Parse()`. ### Retries By default Nexus will retry a function if an exception is thrown. The default strategy is to use linear back-off where the first retry is after 5 seconds, the second retry is after 10 seconds, the third is fter 15 seconds, etc. You can control the delay and the max number of retries globally by setting the `RetryAfter` and `MaxRetries` options like this: ```cs builder.Services.AddNexus().AddFunctions(options => { options.RetryAfter = TimeSpan.FromSeconds(10); options.MaxRetries = 5; }); ``` But you can also set the max retry count on a function level like this: ```cs [NexusFunction(MaxRetries = 5)] public void ExampleMethod() { } ``` By default Nexus will keep calling the method forever until it completes successfully without any max retry number. You can change the retry strategy from linear back-off to either `NexusFunctionRetries.RetryUntilSuccess` or `NexusFunctionRetries.ManualRetry`. `NexusFunctionRetries.RetryUntilSuccess` will keep calling the method without any delay whereas `NexusFunctionRetries.ManualRetry` won't retry at all. You can trigger a retry manually through the Admin UI or the API. ```cs [NexusFunction(Retries = NexusFunctionRetries.RetryUntilSuccess)] public void ExampleMethod() { } ``` ## Function middlewares In some cases you want to wrap the execution of a function. You might want to add additional log context properties or instrument the function execution in some way. To achieve this you can register one or more `INexusFunctionMiddleware` instances in the service collection. Eg: ```cs public class MyFunctionMiddleware(ILogger logger) : INexusFunctionMiddleware { public async Task ExecuteAsync(NexusFunctionDescriptor nexusFunctionDescriptor, Func> executeFunction) { using logger.BeingScope(new Dictionary {{ "FunctionName", nexusFunctionDescriptor.DisplayName }}); return await executeFunction(); } } // Register the middleware builder.Services.AddSingleton(); ``` ## How Nexus Functions works When you call `RunInBackgroundAsync(x => x.SomeMethod(someArgument))` Nexus will use .NET reflection to gather information about what it is you want to run. The full name of the service type, the name of the method to use and the arguments passed. This is then stored in the database in a row that looks something like this: | status | service_type | method | parameter_types_json | arguments_json | run_at_utc | | --------- | --------------------------------------------------------- | ------------------------ | ------------------------- | -------------- | --------------------- | | "Pending" | "My.Namespace.IOrderConfirmationEmailService, MyAssembly" | "SendEmailForOrderAsync" | ["System.String, System"] | ["123"] | "2022-01-01T00:00:00" | Nexus will then read this data from the database and again use reflection to locate the .NET types needed and deserialize the arguments from JSON. It will then locate the service through the `IServiceProvider` and execute the method in a [background service](https://learn.microsoft.com/en-us/aspnet/core/fundamentals/host/hosted-services?view=aspnetcore-6.0&tabs=visual-studio#backgroundservice-base-class). The type signature for `RunInBackgroundAsync()` is a lambda that gets the `TService` passed in and either returns a `Task` or nothing. So the type system allows you to write something like this: ```cs await _nexusFunction.RunInBackgroundAsync(x => orderId == "123" ? x.SendEmailForOrderAsync(orderId) : x.DoSomethingElseAsync(orderId)); ``` Nexus will however throw an exception if you do this as the only allowed lambda to pass in is a method call on the service passed into the lambda. You can think of it as a more readable and type safe way of writing this: ```cs await _nexusFunction.RunInBackgroundAsync(typeof(IOrderConfirmationEmailService), nameof(IOrderConfirmationEmailService.SendEmailForOrderAsync), new object?[] { orderId }); ``` ### Return value The method you use in your lambda is allowed to return a value for synchronous methods and `Task` or `Task` for asynchronous methods but Nexus won't do anything with the return value. If you need to do something with the return value you should wrap the method in another method that does something with it. If you want to call a method that returns `false` if it didn't go as expected you should wrap it in a method that throws an exception instead. Something like this: ```cs public class ExampleService { public bool TryDoSomething() { // Do something interesting here and return true/false } [NexusFunction] public void DoSomething() { if (!TryDoSomething()) { throw new Exception("It didn't go as planned"); } } } ``` ## Renaming classes or methods Since Nexus will store full type names such as `My.Namespace.MyService, MyAssembly` in the database it's important not to rename classes that are used as Nexus functions. If you rename such a class or method existing functions will fail and you need to update the database with the new type names. --- # Functions vs Jobs & Queues If you've already read about [jobs](../jobs/index.md) and [queues](../queues/index.md) you might ask yourself "when should I use functions and when should I use jobs and queues?". There's a lot of similarities between them and both can often be used for the same use case. You can think of functions as a lightweight version of jobs & queues that are well suited for use cases where creating a job and a queue feels like overkill. A key difference is that functions can only be created by .NET code and you can't have an external system adding a function call through the API like you can with [queues](../queues/enqueueing.md). Another big difference is that queues allows you to separate concerns in a better way. The part of the application that creates a queue message has no idea what will happen when that message is processed, but when you use a Nexus function the caller has to specify what service should get called. There's also no way of doing batching with functions like you can with queue jobs. That is, if 100 calls to the same method gets scheduled there's no way of grouping them into a single call to the service. A rule of thumb is to start with a Nexus function if that works for your use case, and later on move it to a queue job if you need to. :::note[Tip: Use only functions] The functions, jobs and queues sub systems are independant from each other and can be used in isolation. So if you don't have any need for jobs & queues you can skip registering those sub systems and only call `builder.Services.AddNexus().AddFunctions()`. ::: --- # Processing directly from Azure Service Bus queues Nexus allows you to use Azure Service Bus as an external pending storage, which means that Nexus reads and processes messages directly from RabbitMQ without first storing them in an SQL backed Nexus queue. Use it by first installing `CommerceMind.Nexus.Azure` and then configuring the external storage: ```cs builder .AddNexus() .AddAzureServiceBusExternalPendingStorage(); ``` And then on your queue: ```cs [Queue("export_order", StorageMode = QueueStorageMode.ExternalPending)] public class ExportOrderQueueMessage : IQueueMessageWithId { } ``` Read more about [external pending storage here](./external-pending.md). # Enqueueing from Azure Service Bus queues A common scenario is to have queues in Azure Service Bus that other systems sends messages to. Since many systems already have integrations to Azure Service Bus it makes it easy to consume events from such systems. An option is of course to use the `Azure.Messaging.ServiceBus` NuGet package and process the messages using a `ServiceBusProcessor` and add any logic you need into the `ProcessMessageAsync` event handler to handle the message. Another option is to use the `CommerceMind.Nexus.Azure` package to take the messages from the service bus queue and place them in a Nexus queue. Doing so lets you use [virtual queues](./virtual.md), [idempotent messages](./index.md#idempotent-messages) and [to store processed messages](./retention.md) which can be very useful. But even if you don't need any of those features you'll benefit from the visibility, introspection and error handling of Nexus queues. ## Reading from an Azure queue or topic into a Nexus queue Start by installing the `CommerceMind.Nexus.Azure` NuGet package. After that you configure it like this in your `Program.cs`: ```cs builder.Services .AddNexus() .AddQueues() .AddNexusAzureServiceBusQueues() .EnqueueFromAzureQueue("azure-queue") .Build() ; ``` By default Nexus will look for the connection string using `IConfiguration.GetConnectionString("azureServiceBus")` but you can add a loader func on the options object if you need to fetch it from somewhere else: ```cs builder.Services .AddNexus() .AddQueues() .AddNexusAzureServiceBusQueues(options => { options.ConnectionStringLoader = (serviceProvider, queueOrTopicName) => "Endpoint=sb://xxx.servicebus.windows.net/;..."; // Or: options.ServiceBusClientLoader = (serviceProvider, queueOrTopicName) => new ServiceBusClient(...); }) .EnqueueFromAzureQueue("azure-queue") .Build() ; ``` If you want to read from a Service Bus Topic you also need to add a subscription name loader like this: ```cs builder.Services .AddNexus() .AddQueues() .AddNexusAzureServiceBusQueues(options => { options.SubscriptionNameLoader = (serviceProvider, topicName) => "my-subscription"; }) .EnqueueFromAzureTopic("azure-topic") .Build() ; ``` With the above example you also need a `ExampleQueueMessage` defined that implements `IQueueMessage` or `IQueueMessageWithId`. Read more about that in [the Nexus queues intro](./index.md). The assumption is that the JSON on the Azure queue can be deserialized directly to the `ExampleQueueMessage` class. If that doesn't work for you it's possible to specify a separate type for the Azure message and also pass a function that maps the Azure message to the Nexus message like this: ```cs public class ExampleQueueMessage : IQueueMessage { public string SomeProperty { get; set;} } public class AzureQueueMessage { public string SomeOtherProperty { get; set;} } builder.Services .AddNexus() .AddQueues() .AddNexusAzureServiceBusQueues(options => { // This allows you to control things like which status or when messages are processed options.EnqueueContextFactory = message => new EnqueueContext { Priority = message is MyImportantQueueMessage ? 10 : 0 } }) .EnqueueFromAzureQueue( "azure-queue", m => new ExampleQueueMessage { SomeProperty = m.SomeOtherProperty } ) .Build() ; ``` The `CommerceMind.Nexus.Azure` package only expects there to be registered `IEnqueuer`s for the message types used which means that it can enqueue to any of the underlying enqueuers; [`IVirtualEnqueuer`](./virtual.md), [`IDatabaseEnqueuer`](./enqueueing.md#enqueueing-using-the-queueing-package), [a memory enqueuer](./in-memory.md) or [`IApiEnqueuer`](./enqueueing.md#enqueueing-using-the-apiclient-package). --- # Enqueueing from outside the job service The typical scenario is to deploy the code that runs background jobs as a separate service from the rest of your application(s). Eg a public website that takes traffic from public users is deployed separately from the service running background job. But you still want to enqueue messages from the public website that the background jobs will process. To split this up you have three options. ## Enqueueing using the Queueing package The NuGet package `CommerceMind.Nexus.Queueing` can be included in other .NET projects than the one running background jobs. This means that the .NET project for your public website or other non-jobs service can still enqueue messages. In order for this to work you need to keep your `IQueueMessage` implementation classes in a separate .NET project that is used by both your background job service and your website. A simplified example: Core project: ```cs namespace MyNamespace.Core { public class MyQueueMessage : IQueueMessage { } } ``` Jobs project: ```cs namespace MyNamespace.Jobs { public class MyMessageJob : IScheduledQueueJob { // Implementation excluded for brevity } public static void Main(string[] args) { var builder = WebApplication.CreateBuilder(args); builder .AddNexus() .AddDatabasePollingInstanceEvents() // Or: .AddAzureServiceBusInstanceEvents() .AddScheduledJobs() .AddQueues(options => { options.AssembliesToScanForQueueMessages.Add(typeof(MyQueueMessage).Assembly); }) .AddPostgresConnection() .Build(); var app = builder.Build(); app.Run(); } } ``` Website project: ```cs namespace MyNamespace.Website { public class MyMessageController : Controller { private IEnqueuer _enqueuer; public MyMessageController(IEnqueuer enqueuer) { _enqueuer = enqueuer; } // Implementation excluded for brevity } public static void Main(string[] args) { var builder = WebApplication.CreateBuilder(args); builder.Services .AddNexus() .AddDatabasePollingInstanceEvents(options => { options.BrokerType = InstanceEventBrokerType.ProduceEvents; }) // Or: .AddAzureServiceBusInstanceEvents(options => ...) .AddQueues(options => { options.AssembliesToScanForQueueMessages.Add(typeof(MyQueueMessage).Assembly); }) .AddPostgresConnection() .Build(); var app = builder.Build(); app.Run(); } } ``` Note that the website project still needs to configure [instance events](../general/instance-events.md) when only running the queue system, but we can signal that we're only producing instance events and are not interested in listening to them. This skips setting up any polling for messages on the website application. Now both the jobs and website project uses the queue system and has a reference to the core project to use the `MyQueueMessage` class. The website project can use `IEnqueuer` to enqueue messages that the jobs project will pick up and process. This is the simplest approach but has the downside that your website project needs access to the database containing the queues which may or may not be a problem for you. If it is a problem you probably want to enqueue through the API instead. ## Enqueueing using the ApiClient package There's a separate NuGet package called `CommerceMind.Nexus.ApiClient` that contains an `IEnqueuer` implementation that enqueues messages over the API. You install the package and configure it like this in `Program.cs`: ```cs builder.Services.AddHttpClient(ApiEnqueuerOptions.DefaultHttpClientName).ConfigureHttpClient(client => { client.BaseAddress = new Uri("https://the-url-to-the-api.com/"); }); builder.Services.AddNexus().AddApiEnqueuers(options => { // Important to register assemblies which contains IQueueMessage classes options.AssembliesToScanForQueueMessages.Add(typeof(MyMessage).Assembly); }); ``` This will find all `IQueueMessage` classes and register `IEnqueuer`s for them that can be used to enqueue messages through the API. This means that you don't have to reference any of the other NuGet packages, and your deployed application won't need access to the database. Registering an HTTP client is important since it lets you set the url to the API as well as setting additional headers such as authentication. If you don't want the default HTTP client name you can change it by doing: ```cs builder.Services.AddHttpClient("my-http-client").ConfigureHttpClient(...); builder.Services.AddNexus().AddApiEnqueuers(options => { options.HttpClientName = "my-http-client"; }); ``` If you need to differentiate between an API enqueuer, a database, or virtual enqueuer there's these additional interfaces that you can use to check against: `IApiEnqueuer`, `IVirtualEnqueuer`, and `IDatabaseEnqueuer`. ## Enqueueing over HTTP using the API If you want to enueue over the API yourself without using the `CommerceMind.Nexus.ApiClient` that's of course also possible, but should only be done if you really can't use the API client package. Such as if you want to enqueue from a non-.NET solution. The included API has endpoints to enqueue messages over HTTP, more specifically the endpoints `POST /api/queues/{queueName}` and `POST /api/queues/{queueName}/batch` that you can find in the [API reference](/api/reference/nexus-api/). --- # Entity Framework with Nexus queues If you're using Entity Framework you might want to sync your `SaveChangesAsync()` calls with calls to `EnqueueAsync()` on Nexus queues to ensure that they are committed together. To achieve this you can install the `CommerceMind.Nexus.EntityFramework` package and initialize it like this in Program.cs: ```cs builder.Services .AddNexus() .AddEntityFramework(); builder.Services.AddDbContext((serviceProvider, options) => { options.UseNexus(serviceProvider); }); ``` This will install interceptors with Entity Framework that instead of sending a Nexus message directly to the database will keep it in the background until you call `context.SaveChangesAsync()` and will then use the same database connection and transaction that EF uses for your other data. Note that if there's any messages to save Nexus will create an explicit transaction if there isn't already one. Which means that your EF data will be saved in the same transaction as your Nexus messages. ## Only works with EnqueueAsync() Note that this sync with `SaveChangesAsync()` only works with `IEnqueuer.EnqueueAsync()` and not other Nexus services like `IQueueItemUpdater`. If you need that you can always create an explicit transaction yourself and use `DbConnectionScope` to make sure that any Nexus operations that happen in that scope use your explicit transaction. --- # External pending queue storage The great thing about Nexus is the control and insight you get into queue messages from using an SQL based queue storage. But for some cases the message throughput is just too high for it to make sense to use SQL based queues for everything. This is where external pending queue storage in Nexus comes in, specifically the `IExternalPendingStorage` interface. It lets you use Nexus without forcing you to store pending messages in an SQL backed queue. One of the `IExternalPendingStorage` implementations in Nexus is Redis and using that means that message producers can enqueue directly to Redis or through the Nexus enqueue API and the message won't be stored in SQL at all. When the Nexus queue job starts it will fetch messages from Redis, process them and then delete them from Redis if nothing failed. If a message fails the message is inserted into the SQL version of the queue which gives you the ability to troubleshoot like you're used to with Nexus and typically don't get with a Redis backed queue. If you decide to retry the message it will get deleted from the SQL based queue and added to Redis again. This is very useful for cases like event ingestion where you get tons of messages/events but don't really care about individual messages when they're in the happy path of Pending -> getting processed -> Processed/deleted. But you do care about failures and want the insights you get from using Nexus. ## Idempotency, message id, etc There are some Nexus features that you may not be able to use with external pending queue storage such as idempotency, queue message history, messages with id getting deduplicated etc. How well these features can be supported depends on the feature set of the external queue storage. The Redis package supports idempotency, message ids and messages scheduled for future processing, so the Redis package is the best option if you're just looking to maximize message throughput and don't have any requirements on which underlying queue storage to use. ## Supported packages At the time of writing the Nexus.Redis, Nexus.RabbitMQ, Nexus.Kafka and Nexus.Azure packages offer external pending storage out-of-the-box. The core Nexus package also ships with an in-memory provider suited for local development and testing. Use it by calling the extension method from the different packages, eg: ```cs builder .AddNexus() .AddRedisExternalPendingStorage(); ``` And in your queue message set it as external like this: ```cs [Queue("export_order", StorageMode = QueueStorageMode.ExternalPending)] public class ExportOrderQueueMessage : IQueueMessageWithId { } ``` ### In-memory The core `Nexus` package ships with an in-memory provider that requires no external infrastructure. It is a good fit for single-server deployments where you can guarantee that the shutdown token is properly triggered on process exit. On startup any messages persisted by the previous run are restored from a temp file and the file is deleted. When shutdown is triggered the in-memory state is flushed to that file. After that point all further enqueue calls write directly to the file, while lease, count and peek calls return empty/zero, so that no new work is picked up during shutdown. Messages are not preserved across crashes or forced kills. The temp file is named after the provider ID and the Nexus instance ID (normally the hostname) and is placed in the OS temp directory. ```cs builder .AddNexus() .AddInMemoryExternalPendingStorage(); ``` An optional provider ID can be supplied when running multiple providers side-by-side: ```cs builder .AddNexus() .AddInMemoryExternalPendingStorage("my-in-memory"); ``` ### Using multiple different external storages Note that you can use both eg Redis and RabbitMQ at the same time on different queues like this: ```cs builder .AddNexus() .AddRedisExternalPendingStorage() .AddRabbitMQExternalPendingStorage(); ``` ```cs [Queue("export_order", StorageMode = QueueStorageMode.ExternalPending, ExternalProvider = "redis")] public class ExportOrderQueueMessage : IQueueMessageWithId { } [Queue("import_order", StorageMode = QueueStorageMode.ExternalPending, ExternalProvider = "rabbitmq")] public class ImportOrderQueueMessage : IQueueMessageWithId { } ``` --- # Filterable message properties In some cases you might want to find messages where one or more property contains certain values. Like if you have a queue of inventory messages and you don't want to process it until the article data has come in. If your inventory message class looks like this: ```cs public class InventoryChangeQueueMessage : IQueueMessage { public required int ChangedQuantity { get; set; } public required string Sku { get; set; } } ``` When the message is processed we check if an article exists with that SKU and if it doesn't we want to skip processing it, but we don't want to delete it. We'll schedule it to be retried later, but we want to minimize the wait time after the article has been created and when we process its inventory messages. So when the article has been created we can update the status of messages by passing in a lambda like this: ```cs class ProductController(IQueueItemUpdater updater) { public async Task OnArticleCreatedAsync(string sku) { await updater.UpdateStatusWhereAsync(QueueItemStatus.Pending, m => m.Sku == sku); } } ``` The lambda is translated into SQL and by default it will use the JSON functionality in the database to filter by message property. If you have a lot of messages in the queue that can become a performance issue as it won't be able to use any indexes for the query. If you're concerned about performance you should use the `[Filterable]` attribute as described below. # Reading queue items by property If you have more advanced use cases than just updating status by message property you can use the `IQueueReader` to read queue items that matches a lambda filter. Eg: ```cs class ProductController(IQueueReader reader) { public async Task OnArticleCreatedAsync(string sku) { var items = await reader.ReadAsync(m => m.Sku == sku); foreach (var item in items) { // Do something interesting } } } ``` # [Filterable] attribute If you want to filter on message properties and you're concerned about performance you should add the `[Filterable]` attribute on your property like this: ```cs public class InventoryChangeQueueMessage : IQueueMessage { public required int ChangedQuantity { get; set; } [Filterable] public required string Sku { get; set; } } ``` This tells Nexus that you want enhanced filtering on the `Sku` property which means that we will extract the contents of that property into a separate database column that gets a database index to improve performance of filtering. The `[Filterable]` attribute is allowed on properties of primitive type (including `DateTime` and `Guid`) as well as lists of primitive types. Eg: ```cs public class InventoryChangeQueueMessage : IQueueMessage { public required int ChangedQuantity { get; set; } [Filterable] public required List Skus { get; set; } } ``` Nexus will create a separate table per list property in order to efficiently filter on values for it. Nexus will also automatically populate new filterable columns and tables from the messages in the queue when you've added `[Filterable]` attributes. This is done when the application starts up. Nexus automatically creates database indexes for these columns and tables, but won't combine multiple columns in these indexes. Which means that if you want to filter on multiple properties in a single query you might want to create a combined index on those columns. # Lambda limitations The lambda passed to `IQueueItemUpdater` and `IQueueReader` can combine and/or comparisons as well as grouping by parentheses such as `m => m.Sku == sku && (m => m.Quantity == 1 || m => m.Quantity == 2)`. The lambda can filter on deep properties such as `m => m.Some.Deep.Object == value` but note that the `[Filterable]` attribute only work on top-level properties on a message class. So deep properties like this will always use the less performant JSON operators in the database. The lambda can check if a list of primitive values contains a value or not, such as `m => m.ListOfSkus.Contains(sku)` or `m => !m.ListOfSkus.Contains(sku)` but no other comparisons on lists are allowed. Dictionaries are not supported for filtering. If you have a dictionary on your message and you can't convert that to a list you can create a separate filterable property that's derived from the dictionary instead: ```cs public class ArticleChangeQueueMessage : IQueueMessage { public required Dictionary Categories { get; set; } [Filterable] public required List CategoryIds => Categories.Keys.ToList(); } ``` # Filtering in the Admin UI By default all properties with the `[Filterable]` attribute will be extra visible and filterable in the Admin UI. If you don't want a filterable property to show up in the Admin UI like that you can hide it by doing `[Filterable(VisibleInAdminUI = false)]`. # Filtering guarantees In order to guarantee consistency Nexus stores primitive filterable properties as columns in the queue table. Which means that you won't risk getting out-of-sync between the message itself and the filterable property column since it's committed to the database in the same query. In the case of list properties Nexus will populate the separate tables after the messages are inserted/updated in a queue table, but it's done inside a transaction to prevent inconsistency. If you're explicitly using an isolation level in the database that allows dirty reads (such as `read uncommitted` in SQL Server) you risk reading or updating the wrong messages. --- # Auto generated queues Sometimes you have messages with the same structure but one or more discriminator properties makes them significantly different than others. Different enough that you want such message split up into multiple different queues. An example might be a message that has a `Language` property and you want to have different queues per language so each language can be processed individually. Or if you want to enable `PauseOnError` so that any error for that language pauses processing for any other message in that language. It's possible to manually define a bunch of queues and jobs using inheritance to handle this, but it comes quite inelegant quickly because of the boilerplate needed. To handle this use-case Nexus lets you auto-generate queues and jobs for those queues by giving Nexus a base class. For example: ```cs // Program.cs [Queue("language")] public class LanguageQueueMessage : IQueueMessage { } public abstract class LanguageQueueMessageJob : IScheduledQueueJob where TMessage : LanguageQueueMessage { public async Task ProcessMessageAsync(TMessage message, CancellationToken cancellationToken) { return ProcessResults.Processed(); } } var generatedQueues = new List(); var generatedJobs = new List(); foreach (var language in GetLanguages()) { var queueName = "language_" + language; generatedQueues.Add(new GeneratedQueue { Name = queueName, DisplayName = "Language queue " + language, Category = "Language", ShouldEnqueue = message => message.Language == language, }); generatedJobs.Add(new GeneratedScheduledQueueJob { ScheduledJobBaseType = typeof(LanguageQueueMessageJob<>), Name = "language_job_" + language, DisplayName = "Language job " + language, Category = "Language", QueueName = queueName, }); } builder.Services .AddNexus() .AddQueues(options => { options.GeneratedQueues = generatedQueues; }) .AddScheduledJobs(options => { options.GeneratedScheduledJobs = generatedJobs; }); ``` With this code you you'll get one concrete queue and job per language returned from the fictional `GetLanguages()` method. The `ShouldEnqueue` func you set on your `GeneratedQueue` class determines if a message is relevant for your generated queue or not. Because when you want to enqueue to any of the generated queues you ask for an `IEnqueuer` and that will iterate over all the generated queues and see which of them are interested in the message. You can also use the API to enqueue to the base `LanguageQueueMessage` queue which will have the same effect. The generated job needs to be an abstract class that is generic on the queue message like in the example above. The generated job will be generic on the generated queue message. The job can be a batch job and doesn't have to be a single message job like in the example above. ## Enqueueing messages Since the `IQueueMessage` types for each generated queue is generated at runtime you can't ask the service provider for a `IEnqueuer` for a specific queue. Instead you should request `IEnqueuer` where `BaseMessage` is the type you used to define the generated queue in `GeneratedQueue`. Nexus will register a specific enqueuer for that type which then iterates over all the generated queues of that type, calls the `ShouldEnqueue` func and enqueues it into the concrete queue if it returns `true`. If you explicitly want to get the enqueuer for a generated queue you'll need to request `IEnumerable` from the service provider and then filter out based `IObjectEnqueuer.QueueName`. ## Other queue operations If you need any of the other services for a generated queue you'll need to request a list of the non-generic version and filter out based on queue name. So let's say that we want the Swedish queue reader from the language example at the top on this page. We'd do something like this: ```cs var queueReaders = serviceProvider.GetServices(); // Or IEnumerable in your constructor var language = "sv"; var swedishQueueReader = queueReaders.Single(q => q.QueueName == "language_" + language); ``` --- # Queue message history Nexus can keep track of the history of messages with [identify](./identity.md). This means that you get a full audit trail of the content of the messages, when they change and what changed on them. Message history is disabled by default in Nexus since the history can grow a lot which can become costly. You can enable history on all queues where the queue message implements `IQueueMessageWithId` by configuring it when initializing the queue system: ```cs builder.Services.AddNexus().AddQueues(options => { // History will be stored forever on all queues with ids options.DefaultHistoryRetention = TimeSpan.MaxValue; }); ``` This will set the default retention which can then be overriden per queue using the `[Queue]` attribute: ```cs [Queue("myqueue", HistoryRetention = "30.00:00:00")] ``` The above will store the history for that queue for 30 days. The string is a `TimeSpan` in string format but also supports the special `forever` value to represent `TimeSpan.MaxValue`. ## Accessing the history You can access the history in the [admin UI](../admin-ui/index.md) when going into a queue and selecting a queue message. The button to show message history will only be visible for queues with identity and where the history retention is not set to `TimeSpan.MinValue`. You can also access the message history using the API endpoint `GET /api/queues/:queueName/:messageId/history`. --- # Message identity and IQueueMessageWithId The `IQueueMessage` interface doesn't contain any properties so you can define your message in any way you want. But using `IQueueMessage` means that the system won't have any knowledge of message identity. The system won't compare two messages to see if they are identical to prevent duplicates in a queue. If you need to ensure that a queue never contains duplicates of the same message you should instead use the `IQueueMessageWithId` interface. The only property that you're forced to have on your message class is the `Id` property. What the id should be depends on your queue. If you want you can use a combination of values in the implementation of the property. The property is allowed to return `null` so you can have messages in the same queue where some have identity and others don't. If we take the `SendOrderConfirmationQueueMessage` from the example in the [Overview](./index.md) we can define it like this instead: ```cs [Queue("send_order_confirmation")] public class SendOrderConfirmationQueueMessage : IQueueMessageWithId { public string? Id => OrderId; public string OrderId { get; set; } public string Email { get; set; } } ``` By doing this the system will ensure that if we already have a `SendOrderConfirmationQueueMessage` message for order with id `123` we will never allow duplicates to be enqueued. If someone enqueues it again we'll instead update the existing message rather than adding another one. This is true regardless of which status the message has. Other queue systems has a separate deadletter queue where failed messages are placed which can allow duplicates to still happen. But if we have a message for order `123` in the `Error` status and someone enqueues a message with that id the existing message gets updated and the status is set to `Pending` again. Read more about message statuses [here](./index.md#message-status). ## Idempotent messages If you use message identity you might also want to use [message idempotency](./index.md#idempotent-messages). It tells Nexus to discard incoming messages where the message property valies are exactly the same as a message with the same id in the queue. Which is handy if you only want to process changes but the queue is populated with data from a system that only allows full exports and can't tell you which messages/entities have changed since the last sync. --- # In-memory queues Queues in Nexus are backed by a database but in some cases you want to squeeze the last bit of performance out of your queues and then the overhead of a database can become too much for certain use cases, or if you need to relieve your database of the work needed to handle some very busy queues. For those scenarios Nexus lets you have in-memory queues without the overhead of reading and writing to a database first. Note however that this _only_ works if you only deploy Nexus to a single server. See the section below about single-server use. You can however allow other servers to enqueue but only if you use `IApiEnqueuer` to enqueue through the API to the single server hosting the API and the memory queue(s). ## Creating a memory queue You create a memory queue just like any other queue by creating a class that implements `IQueueMessage` or `IQueueMessageWithId`: ```cs [Queue("example_memory", InMemory = true)] public class ExampleMemoryQueueMessage : IQueueMessage { // Properties here } ``` When you set `InMemory = true` in the attribute it gets registered as a memory queue. Note that you can switch back and forth between having a queue as an in-memory queue or not because an in-memory queue will use its database queue as fallback storage to ensure that messages aren't lost because of deploy or application restarts. Messages and message changes are scheduled to be persisted to the database as soon as possible, but it's done on a background thread. The in-memory queue registers a listener for application shutdown to prevent the application from shutting down before all changes are persisted. ## Durability Even if the queue is in-memory Nexus still persists changes to it in the background in order for queue items to survive a deploy and any application crash. The queue item change are persisted as fast as possible on a background thread so the durability guarantees are very close to the guarantees of database queues. All changes are synchronously flushed to the database when the application starts to shutdown. The only time you would risk data loss if you have a large amount of messages coming in at the same time as a power outage or if your host doesn't give the application enough time to shut down gracefully. ## Fallback storage The in-memory queue will fallback to it's own regular database table by default which makes the transition from a non-memory queue to an in-memory queue smooth. But if you want to use an in-memory queue to lessen the load on your database it doesn't have the intended effect. In that case you can explicitly use the Sqlite driver and use Sqlite tables as fallback storage by installing the `CommerceMind.Nexus.Sqlite` NuGet package and adding this in your `Program.cs`: ```cs builder.Services .AddNexus() .AddSqliteMemoryQueueFallbackStorage(); ``` If using disk as fallback storage isn't an option for you it's possible to create your own implementation of the interface `IMemoryQueueFallbackStorage` and register it in the service collection. ## Single server use _only_ In-memory queues can generally only be used if you only have a single server/instance running Nexus. Even if changes to an in-memory queue is persisted to storage Nexus won't read from the storage except during initialization. You can use in-memory queues in a multi-instance application if the queues are local to that instance and messages in the same queue have no effect on messages in other instances in terms of sort order or processing side effects. In this case however you must implement your own `IMemoryQueueFallbackStorage` as the built in implementation uses a memory counter to increment assign queue item ids. If you're using a hosting service such as Azure App Service you probably can't use in-memory queues unless you make sure to stop the application before starting a deploy as the default in Azure App Service is to bring up a parallel deployment slot and swap with the live slot. Which means that during deploy you will have multiple instances running. If this is an issue for your application will depend on how you use the queues. ## Application shutdown In-memory queues registers a callback on the applications stopping token (`CancellationToken`) and performs a synchronous flush of all pending changes to the fallback storage. Nexus will also attempt to stop all jobs when shutdown starts and won't start any jobs after that. But it's up to you not to enqueue to memory queues after shutdown has started. If you do you will risk enqueueing a message that doesn't have time to be persisted before the process exits. ## In-memory queues and transactions In-memory queues does not offer a way to integrate with the transaction support in `DbConnectionScope`. All changes to an in-memory queue are visible immediately to the whole application, and changes are sent to the fallback storage explicitly without transaction support to not accidentally enlist those queries in unrelated transactions. If you need transaction support you should enqueue your messages to a memory list first and then send them to Nexus after your database transaction has successfully completed. --- # Queues Unlike popular message buses and event queues this system uses an SQL database to store messages and also offers [in-memory storage](./in-memory.md). Message bus systems are typically built to be able to handle tons of messages and have the assumption that you're not really interested in looking at a single message and doesn't offer great support for finding and manipulating messages. Nexus on the other hand assumes that you won't have millions of messages per second and that you are interested in looking at individual messages. And for that an SQL database is a perfect fit. At the time of writing we support Postgres, SQL Server and SQLite. Feel free to reach out to us if you need support for another database engine like MySQL or MariaDB. Each registered queue is its own database table which is automatically created by the system when you've registered your queue. And a message is a row in that database table which contains the message in JSON format together with meta data such as the status and when it was created. ## When SQL based queues aren't enough If you get more incoming messages than what a SQL based queue can handle in a performant way you should look into [external pending storage](./external-pending.md) for Nexus queues. It gives you the best of both worlds - the insights and troubleshooting capabilities from Nexus with the raw throughput of queue systems like RabbitMQ, Redis, etc. ## Defining and registering a queue The first thing to do is to create a C# class representing the message type of the queue. It's just a .NET class implementing the interface [`IQueueMessage`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Abstractions/Queueing/IQueueMessage.cs). Here's an example: ```cs [Queue("send_order_confirmation")] public class SendOrderConfirmationQueueMessage : IQueueMessage { public string OrderId { get; set; } public string Email { get; set; } } ``` You can add any properties you want on the class, the only important thing to remember is that the message will get serialized to and from JSON. There's also the `IQueueMessageWithId` interface which you can read about in the section about [message identity](./identity.md). ### [Queue] attribute The queue attribute isn't strictly required but it's recommended as it sets meta data for the queue such as the queue name. That name is used for naming the database table and to access the queue in the API and admin UI. If the queue attribute isn't set the name is taken from the class name. If you don't use the `[Queue]` attribute it's easy to accidentally rename the class which will create a completely new and empty queue. The `[Queue]` attribute also lets you set other types of meta data such as which database engine to use and for how long we should keep processed messages. ### Registering the queue Just like scheduled jobs the queues are automatically registered by finding all types that implements [`IQueueMessage`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Abstractions/Queueing/IQueueMessage.cs) (`IQueueMessageWithId` inherits from `IQueueMessage`). And just like for jobs this is only done automatically for messages defined in the entry assembly. If you have multiple .NET projects that contains queue message classes you need to register the assemblies with the service. There's a property on the options object passed into `AddNexus().AddQueues()` called `AssembliesToScanForQueueMessages` where you can add additional assemblies to scan for queue messages like this: ```cs builder.Services.AddNexus().AddQueues(options => { options.AssembliesToScanForQueueMessages.Add(typeof(MyQueueMessageInAnotherProject).Assembly); }); ``` ## Adding a message to a queue There's two ways to add queue messages. The first is through the C# SDK using the [`IEnqueuer`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Queueing/IEnqueuer.cs) service. That service is injectable through the `IServiceProvider` so you can take it as a constructor dependency. The generic argument `TMessage` is the message type that you want to add a message for. An example: ```cs public class OrderController { private readonly IEnqueuer _enqueuer; public OrderController(IEnqueuer enqueuer) { _enqueuer = enqueuer; } [HttpPost] public async Task OrderPlacedAsync(string orderId, string email) { await _enqueuer.EnqueueAsync(new SendOrderConfirmationQueueMessage { OrderId = orderId, Email = email, }); return NoContent(); } } ``` The message is now placed in the queue with the status `Pending` which means that it'll get processed by the queue processing job as soon as it starts. There might be a delay between the time the message is enqueued and when the queue processing job is scheduled to run. If no delay is acceptable you can use the `EnqueueAndStartProcessingAsync()` method rather than the `EnqueueAsync()` method on [`IEnqueuer`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Queueing/IEnqueuer.cs). That will request the queue processing job to start as quickly as possible regardless of its schedule unless the job has been manually disabled. There's also the non-generic version of `IEnqueuer` which lets you enqueue any object of type `IQueueMessage`. The non-generic `IEnqueuer` will find the correct enqueuer for the messages you pass in and send it to the correct queue. ### Adding a message through the API The other way of adding a message to a queue is by using the API. The API is just an way to access the same [`IEnqueuer`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Queueing/IEnqueuer.cs) over HTTP at the endpoint `POST /queues/{queueName}`. ### Adding multiple messages at once Both the C# SDK and the API allows you to enqueue multiple messages at the same time and will use batch operations (except SQLite which doesn't have batch imports) to make enqueuing a lot messages as fast as possible. ## Message status A message has a status and the default status is `Pending`. It means that the message is up-for-grabs for a queue processing job. When a queue processing job starts it will see if there's any `Pending` messages and then try to lease them. This is done by using the [`IQueueReader.LeaseAsync()`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Queueing/IQueueReader.cs) method which will update the status of any `Pending` messages to `Leased` together with setting the date for how long they've been leased for. Leasing a message ensures that multiple queue processors can never lease the same message at the same time. If the lease expires - that is, if the processing job hasn't finished in the time it set for the lease - the message status till be set to `Abandoned`. When this happens it most likely means that the job that leased the message crashed and won't recover. When a message is set as `Abandoned` you need to manually look into what happened and decide if you want to retry the message by changing its status back to `Pending` or if you want to delete the message and handle the work in some other way. In the Admin UI it's possible to configure a queue (under the More button on the queue details page) to retry messages when the lease expires instead of setting them to `Abandoned`. This should only be enabled for messages where it's safe to process the same message twice. The message might have been processed once, but the server crashed before Nexus was able to update the status to `Processed`. When a queue processing job is done with the message the status will either become `Error` or `Processed`. And if the queue is not configured to keep processed messages the message will be deleted rather than set to `Processed`. The system also allows you to set custom statuses on messages through the UI and the API which allows you to categorize messages for troubleshooting as well as prevent them from being processed by the system. Messages with custom statuses are invisible to the processing system and are only visible in the UI and API. ## Updating status or reading by property in message Sometimes you might want to update the status of messages where a certain property inside messages contains a specific value. An example would be that you have a queue of inventory messages and you don't want to process it until the product data has come in. So your queue processing job for the inventory messages keeps rescheduling such messages. When the product data then comes in you can update the inventory message(s) like this: ```cs public async Task OnProductCreated(string sku) { await _inventoryQueueItemUpdater.UpdateStatusWhereAsync(QueueItemStatus.Pending, m => m.Product.Sku == sku); } ``` See more [in the docs about filterable properties](./filterable.md). ## Default enqueue status As mentioned above the default enqueue status is `Pending` which makes the message immediately available for processing jobs. In some cases you might want to change this to a different status so that new messages that comes in are kept in a different status and becomes invisible to processing jobs. You can do this in the Admin UI for the queue or through the API and you can use any text value you want such as `Paused` or `OnHold`. This can be useful if a queue processing job is experiencing problems and you only want to run it on selected messages. Then you can change the default enqueue status and manually change only the messages you want to process to `Pending`. After the issues have been resolved you can filter out all the messages with status `Paused` or `OnHold` and update to `Pending` again. ## Processing delay Sometimes you want to wait a bit before processing a message. Maybe you need to make sure that you only process the last message of it's kind and for that you need to wait a while to see if a new one comes in before you process it. Or maybe you need a delay to give some other system time to prepare first before you send them data. You can control a delay on message processing either on the `[Queue]` attribute using eg `[Queue("myqueue", ProcessingDelay = "30m")]` or you can set it on-the-fly in the Admin UI. ## Previous message version For messages that has [identity](identity.md) the previous version of a message is stored when enqueueing a new version. This is useful for doing change based processing since it allows you to compare the current message with the previous message and add additional processing logic when certain properties of the message changes. In order to access the previous message from inside a [queue processing job](./jobs.md) you use the `IPreviousMessageProvider` service. Note that it's only usable inside queue processing jobs and will always return `null` if you try to use it outside of a job. If you need access to previous messages outside of a job you can read `QueueItem`s using `IQueueReader`. `QueueItem` contains everything about the queue item including the current and previous message. :::note[Previous version] The previous version is only updated when calling `IQueueItemUpdater.ProcessedAsync()` which is done for you when using the built in queue processing jobs. It's then updated to be the message passed to that method. This ensures that the previous version of a message will only ever be a version that was successfully processed. If the message processing fails the previous version stays untouched to let you retry the message. ::: ## Idempotent messages If messages are fully self-contained, meaning that a processing job will only use data on the message itself to perform it's processing and not look at any external data your messages are idempotent. Processing an idempotent message will always result in the same effect. For the queues you have which has this characteristic you can tell the `[Queue]` attribute that it's idempotent like this: ```cs [Queue("product_updates", IdempotentMessages = true)] public class ProductUpdatedQueueMessage : IQueueMessageWithId { ... } ``` When `IdempotentMessages` is set to `true` it means that if a message is enqueued Nexus will check if we already have that message with the same message id and JSON and then discard the message. Nexus will compare the full message JSON by default, but if your message contains timestamps or other things that doesn't matter from an idempotency perspective you can have a specific property be the idempotency key. If you set the `[IdempotencyKey]` attribute on the property then Nexus will use that for deduplication instead of the whole message. This is useful for big data syncs such as a product catalog. You can have a job that reads the entire product catalog from another system and enqueues a message per product. If nothing has changed on the product since the last time it was enqueued the message won't get processed again. The comparison is done at a JSON string level so a message will only be discarded if the JSON is exactly the same. Which means that the order of properties in the serialized message is important. This is only something you need to consider if your message contains dynamic collections such as dictionaries where the order can change. :::note[Idempotency and processed messages] When you set `IdempotentMessages` to `true` the `KeepProcessedMessagesFor` property is automatically set to never delete processed messages unless you explicitly specify a timespan to keep processed messages for in the queue. ::: ## Partial messages Sometimes you might have multiple data sources that are each responsible for their own section of a message. Such as in an e-commerce context where the ERP might be responsible for the price and the PIM is responsible for the product data, but you want to process the data from these systems in a single message. In such a case you can use partial messages in Nexus, which is message classes where the top level properties are set by different enqueuers. Take this message for example: ```cs public class ProductUpdatedMessage : IQueueMessageWithId { public string? Id { get; set; } public ProductInformation? ProductInformation { get; set; } public ProductPrice? Price { get; set; } } ``` When listening to events from the PIM you do this: ```cs public ProductInformationController(IEnqueuer enqueuer) : Controller { [HttpPost("{productNumber}")] public async Task ProductUpdatedAsync(string productNumber, [FromBody] ProductInformation productInformation) { await enqueuer.EnqueueAsync(new ProductUpdatedMessage { Id = productNumber, ProductInformation = productInformation }, new EnqueueContext { PartialMessage = true }); return Ok(); } } ``` And when listening for ERP price updates you do: ```cs public PriceController(IEnqueuer enqueuer) : Controller { [HttpPost("{productNumber}/price")] public async Task PriceUpdatedAsync(string productNumber, [FromBody] ProductPrice price) { await enqueuer.EnqueueAsync(new ProductUpdatedMessage { Id = productNumber, ProductPrice = price }, new EnqueueContext { PartialMessage = true }); return Ok(); } } ``` What will happen in the background here is that Nexus will merge the incoming message with an existing message with the same id. Meaning that when the PIM data comes in it won't remove the price from the message and vice versa. If there's no existing message with the same id a new message with be enqueued with only that data. If the processing job needs both in order to correctly process the product it can just return a custom status or just return that it processed the message. The message will get the `Pending` status again as soon as any of the systems involved have sent their data. A word of caution about the values `false`, `0`, `null` together with partial messages is that Nexus will ignore any properties that have default values. This because Nexus can't know if you explicitly set a property to a default value or if the property hasn't gotten a value and should be ignored. So you can't use partial messages to update eg a boolean property to `false` because Nexus won't know if you explicitly set the value to false or if that was just the default value. The recommendation is to use nullable types or enums instead. :::note[Only top level properties are merged] When using partial messages Nexus won't do a deep, recursive merge. Which means that if you have a message that looks like this: `{"Key1": {"DeepKey1": "deepvalue1"}, "Key2": "value2"}` and enqueue a partial message that looks like this: `{"Key1": {"DeepKey2": "deepvalue2"}, "Key3": "value3"}` the end result will be: `{"Key1": {"DeepKey2": "deepvalue2"}, "Key2": "value2", "Key3": "value3"}`. That is, `Key1` is replaced and `Key1.DeepKey1` was removed. Note however that top level dictionaries are supported, so if `Key1` on your message is a `Dictionary` it would instead be merged into: `{"Key1": {"DeepKey1": "deepvalue1", "DeepKey2": "deepvalue2"}, "Key2": "value2", "Key3": "value3"}`. Top level dictionaries can have any type and not just strings as in this example. ::: ## Dynamic message content Sometimes you get messages from external sources that have varying structure depending on what type of message it is but you don't want to create one Nexus queue per message type. And you might not want to create a ton of different properties on your message class where only some of them are relevant together. In such cases you can let your message implement `IDynamicQueueMessage`. This adds a property called `MessageData` to your message which represents the JSON that will be serialized. Which means that the `MessageData` property holds the raw JSON data and you can have other properties read from that. Something like this: ```cs class MyDynamicQueueMessage : IDynamicQueueMessage { JsonElement MessageData { get; set; } public string? Type => MessageData.GetProperty("type").GetString(); public IMessageContent Content => Type == "FirstType" ? MessageData.Deserialize() : MessageData.Deserialize(); } public interface IMessageContent { ... } public class FirstTypeContent : IMessageContent { ... } public class SecondTypeContent : IMessageContent { ... } ``` This lets you use the `MyDynamicQueueMessage` class in a more convenient way as you don't have to include all the properties from all the different types. This also allows you to store data in the messages that aren't defined in the message class which can come in handy if an external system defines the message structure and you want to make sure that you always store exactly what they send you. When using `IDynamicQueueMessage` Nexus will only serialize and deserialize the `MessageData` property, all other properties on your message is ignored. If you want to use [messages with identity](./identity.md) with dynamic messages you simply let your message class implement both `IDynamicQueueMessage` and `IQueueMessageWithId`. Note that if you're using [Json.NET](https://www.newtonsoft.com/json) it still works, but Json.NET won't be serializing or deserializing a `IDynamicQueueMessage` even when you're using the `CommerceMind.Nexus.NewtonsoftJson` NuGet package. If you need Json.NET specific converters you'll need to first convert to/from JSON with Json.NET before passing it to `IDynamicQueueMessage`. ## Scheduling messages for future processing There's an additional status called `Sleeping` that messages that should be processed in the future will have. When you enqueue a message you can specify a `DateTime` or a `TimeSpan` (relative to right now) for when a message should be processed. This is useful for scenarios where you know that something will happen at a certain time in the future and you want to run some processing at that time. A typical e-commerce scenario is that a product can have prices with a start and stop date. You want to export your prices to another system but that system might not support date based validity for prices so you need to export the current price at any given time. To deal with this you can enqueue messages that should get processed at the time that one price stops being active and another starts. And your queue processing job can send the new price to the external system exactly when it starts being valid. ### Processing the same message again in the future Sometimes when you're using [idempotent messages](./index.md#idempotent-messages) to update external systems you might have dates in your message that affects the outcome of the processing. Such as a product message which contains a list of prices that have a valid from and to date. At the time of processing you pick the price that is valid right now, but you want to run the processing again when the next price becomes valid. In your queue processing job you can return an optional `DateTime` for when to process the message again. See more details in the section about [queue jobs](jobs.md#scheduling-a-processed-message-to-be-processed-again). ## Example queue messages The admin UI needs to create an example message to prefill the text box where you can manually enqueue a new message. By default Nexus will try to create an instance of your queue message class with default values in properties. If you want to customize that message, you can add a static method on your message class called `GetExampleMessage` like this: ```cs private class MessageWithExampleMethod : IQueueMessage { public string? SomeProperty { get; set; } public static IQueueMessage? GetExampleMessage() { return new MessageWithExampleMethod { SomeProperty = "From example" }; } } ``` If you want even more control over this you can register your own implementation of the interface `IExampleQueueMessageFactory` which is the service that will be called to create these example messages. ## Leasing messages The way that Nexus ensures that multiple queue processors won't process the same message is by requiring them to first lease messages before processing them. And Nexus ensures that if two processes try to lease the same message only one of them will win. The default lease time is 30 minutes. If you have a batch job with a very large batch size and the job takes a long time to complete you might want to increase the lease time for that job. You do that by implementing the `MessageLeaseTime` property from `IScheduledQueueJobBase`, like this: ```cs public class ExampleQueueJob : IScheduledQueueJob { public TimeSpan MessageLeaseTime => TimeSpan.FromMinutes(60); } ``` ### Abandoned messages If a server crashes in the middle of processing messages the lease will eventually expire. Note that this only happens in extraordinary cases. Exceptions that are thrown by the processing job are caught by Nexus and correctly transitions the message status from `Leased` to `Error`. But there can be a power outage or the database server might become unavailable which are things that Nexus can't automatically handle. In such cases the lease will eventually expire and at that point Nexus will transition the message status from `Leased` to `Abandoned`. If you see that any of your queues contains abandoned messages you need to investigate what happened first before resetting the status to run them again. It could be that the messages were fully processed but the server or database crashed right before Nexus was able to update the status to `Processed`. The best way to start investigating is by looking at the correlation id set in the messages. That is the correlation id of the job that leased it which means that you can use the Admin UI or your log UI of choice to try to find out how far the job with that correlation id got. If the correlation id is "123" your logs related to that job run would have a log property called `Nexus.CorrelationId` with the value `123`. ## Health status All queues automatically get a health check and the default behavior is to mark the queue as unhealthy when there's a message with the `Error` status in it. However that's not always what you want because you can have qeueues for systems that you know have intermittent failures and you don't want the health check monitoring to keep bugging you about it. In that case you can either disable the health check for the queue or you can change which health status the queue has when it has messages of `Error` status in it, like this: ```cs [Queue("myqueue", HealthStatusWhenHasErrors = HealthStatus.Healthy)] private class MyQueueMessage : IQueueMessage { ... } ``` ## Organizing the Admin UI If you have a lot of queues you can group queues in the Admin UI by a category, just like you can with jobs. Set a category in the `Queue` attribute like this: ```cs [Queue("myqueue", Category = "My category")] private class MyQueueMessage : IQueueMessage { ... } ``` --- # Queue processing jobs There's two interfaces for queue processing jobs included in the SDK. One implementation that process messages one by one and one that process messages in batch. ## Accessing the previous message For some types of jobs and queues you want to get both the current and the previous version of a message. Such as a message for product information updates where you want to compare the previous and the current message to know what changed. You access it using the `IPreviousMessageProvider` service from inside your job. Read more [here](./index.md#previous-message-version). ## Processing messages one by one This is the easiest and safest way to process a message since processing one by one never risks missing a message, it will either complete or fail. You implement such a job by creating a class that implements [`IScheduledQueueJob`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Jobs.Abstractions/Queueing/IScheduledQueueJob.cs). The job executor takes care of leasing messages and all you have to do is to implement the `ProcessMessageAsync()` method like this: ```cs public class ExampleQueueJob : IScheduledQueueJob { public string DefaultSchedule => CronSchedule.TimesPerMinute(10); public async Task ProcessMessageAsync(ExampleQueueMessage message, CancellationToken cancellationToken) { await DoSomethingInteresting(message); return ProcessResults.Completed(); } } ``` The job will run until there are no more `Pending` messages in the queue and will then return a short summary of how many messages that got processed and how many of them that failed as the job results. Note that even though the `ProcessResults` class has a `Failed()` method you don't have to catch exceptions yourself. The executor will catch any exceptions that happen in `ProcessMessageAsync()` and mark that message with the `Error` status together with the exception details in its processing log. ## Updating message during processing Sometimes you want to store meta data about the processing of a message and use that meta data the next time you process it. An example being syncing entities to another system and then storing the id the entity got in the other system so the next time we process the message we can use that id when communicating with the external system. Or you want to store error details that the next run can use when retrying a message. This can be achieved by returning an updated message as part of the returned `ProcessResult`. Eg: ```cs public class ExampleQueueJob(ILogger logger) : IScheduledQueueJob { public async Task ProcessMessageAsync(ExampleQueueMessage message, CancellationToken cancellationToken) { var id = await SendToExternalSystemAsync(message); message.IdInExternalSystem = id; return ProcessResults.Processed(updatedMessage: message); } } ``` Note that if a new version of the message is enqueued the default behavior is to replace the message, which means that your updated message with the id is cleared. You can work around that by using [partial messages](./index.md#partial-messages) since that won't replace the message, only replace the properties that the new message contains. ## Pausing on error If you have a queue and a job where the processing of a message is dependent on that previous messages have been successfully processed you can tell Nexus to pause processing when an error occurs. Nexus will by default process messages in the order they are created, but will continue to process the next message if a previous message fails. To pause processing when an error happen you set `PauseOnError` to `true` in the `[ScheduledJob]` attribute on your job. This means that as soon as any message has either `Error` or `AwaitingRetry` status in the queue the processing will be paused and won't continue until those messages gets another status. Use this with caution as you risk pausing processing indefinitely unless you monitor the queue for errors. ```cs [ScheduledJob(PauseOnError = true)] public class ExampleQueueJob(ILogger logger) : IScheduledQueueJob { public async Task ProcessMessageAsync(ExampleQueueMessage message, CancellationToken cancellationToken) { // Important to wrap the entire job in a try-catch with a retry. Otherwise you risk exceptions // stopping the queue processing until someone notices it. try { await DoSomethingInteresting(message); return ProcessResults.Processed(); } catch (Exception e) { logger.LogError(e, "An error occured"); return ProcessResults.RetryLater(TimeSpan.FromMinutes(10)); } } } ``` Note that this can be overriden in the Admin UI/API under the More button in the job details page. ## Processing messages in batches The other interface is [`IScheduledBatchQueueJob`](https://github.com/Commerce-Mind/Nexus/blob/main/src/job-engine/Nexus.Jobs.Abstractions/Queueing/IScheduledBatchQueueJob.cs) where you're passed a batch of messages instead of getting them one by one. When you're creating a queue processing job that will read data from an external system it's often more efficient to batch load such data rather than requesting it one by one. Such as having a queue for when product data is updated and your job needs to read the full product object. Processing messages in a batch is a bit harder to do correctly than processing one by one, but the executor will take care of most of the complexity for you. Here's an example implementation: ```cs public class ExampleQueueJob : IScheduledBatchQueueJob { public string DefaultSchedule => CronSchedule.TimesPerMinute(1); public async Task ProcessMessagesAsync(QueueMessageBatch batch, CancellationToken cancellationToken) { foreach (var message in batch.Messages) { cancellationToken.ThrowIfCancellationRequested(); try { await DoSomethingInteresting(message); batch.Processed(message); } catch (Exception e) { batch.Failed(message, e); } } } } ``` You are responsible for letting the system know which messages you've completed and which failed. If you didn't have the `try-catch` and there was an exception when processing the fifth message in a batch of ten the system will mark all messages as failed that hasn't explicitly been passed to `batch.Processed(message)`. :::note[Queue job schedule] How often you schedule your job depends on the type of queue you have and if you process it in batches or not. For some queues it's more efficient to run it less frequently and in larger batches but for other jobs you might want to run the job very frequently. There's also the option of requesting processing to start directly when enqueueing to avoid any type of delay. ::: ## Filtering messages In some cases you might want multiple queue jobs to process messages in a single queue. Eg if your queue contains messages for all languages but you want to be able to process messages for different languages individually. This functionality builds on [filterable message properties](./filterable.md) so make sure to read that section. To implement filtering on your job you implement one of these interfaces: ``` IScheduledFilteredQueueJob IScheduledFilteredQueueJob IScheduledFilteredBatchQueueJob IScheduledFilteredBatchQueueJob ``` Read more about [jobs with parameters here](../jobs/parameters.md) if you're wondering what `TParameters` mean. These interfaces extend the base queue job interfaces with a way for you to return a lambda to filter which messages you want to process. Eg: ```cs class MyMessage : IQueueMessage { [Filterable] public required string Language { get; set; } public required SomeDataType Data { get; set; } } class EnglishQueueJob : IScheduledFilteredQueueJob { public Expression> MessageFilter => m => m.Language == "en"; public async Task ProcessMessageAsync(MyMessage message, CancellationToken cancellationToken) { // Only english messages will be passed to this job return ProcessResults.Completed(); } } ``` The interface with parameters lets you create a dynamic lambda based on the passed in parameters which can be useful if you want to process all languages by default, but want to be able to limit to one language in a parameter: ```cs class MyParameters { public required string? Language { get; set; } } class MyQueueJob : IScheduledFilteredQueueJob { public MyParameters DefaultParameters => new(); public Expression> GetMessageFilter(MyParameters parameters) => parameters.Language != null ? m => m.Language = parameters.Language : null; public async Task ProcessMessageAsync(MyMessage message, CancellationToken cancellationToken) { return ProcessResults.Completed(); } } ``` In case you want even more control you might want to take a look at [generated](./generated.md) queues and jobs which lets you generate dynamic jobs and queues at startup. ## Returning a custom status Sometimes it's not specific enough to say that a queue message was either processed or failed. You might have different actions you need to take for different outcomes that you want to specifically monitor. Nexus allows you to process a message and set it directly in a custom status rather than one of the built in statuses. In a batched queue job you call the `CustomStatus` method on the batch: ```cs var processingLog = "... something descriptive of what happened ..."; batch.CustomStatus(message, "MyCustomStatus", processingLog, isError: true); ``` And in a regular queue job you return `ProcessResults.CustomStatus()`: ```cs var processingLog = "... something descriptive of what happened ..."; return ProcessResults.CustomStatus("MyCustomStatus", processingLog, isError: true); ``` The `isError` argument lets you determine if the job should be marked as failed or not because of this custom status. Note that it's up to you to monitor queues for these custom statuses and take action. If you want to be notified about it you can implement a [custom healthcheck](../healthchecks/index.md) that checks if any queue has such a status. ## Batch size By default the batch size of a job is 100. You can control this by implementing the `BatchSize` property from `IScheduledBatchQueueJob`. Like this: ```cs public class MyBatchJob : IScheduledBatchQueueJob { public int BatchSize => 200; } ``` ## Number of messages per job run By default a Nexus queue job will run until the queue is empty. Sometimes that can be too aggressive. It might put too much pressure on other systems that the job is communicating with, and digging into the logs of that single run can be hard since it'll contain tons of logs. Nexus lets you set a max number of messages that a single job run can process by implementing the property `MaxNumberOfMessagesPerRun` in your queue job. Like this: ```cs public class MyQueueJob : IScheduledQueueJob // Or IScheduledBatchQueueJob { public int MaxNumberOfMessagesPerRun => 1000; } ``` ## Multithreading The interfaces `IScheduledBatchQueueJob` and `IScheduledQueueJob` have a property called `NumberOfProcessingThreads` which by default returns `1`. If you want your job to process messages in multiple threads in parallel you can implement the method in your job to increase parallelism like this: ```cs public class ExampleQueueJob : IScheduledQueueJob { public int NumberOfProcessingThreads => 10; // Other methods and properties excluded for brevity } ``` Make sure not to store any instance state that isn't thread safe when you turn on multi threading. The same instance of your job will be used by all threads so if you need to store instance state make sure to use concurrent collections and/or locks. ## Controlling the job result message By default Nexus will create a job result message based on the number of messages processed by the job run. In case you want to create your own job result message you can implement the method `IScheduledQueueJobBase.GetResultsText()` and return your own job result message: ```cs public class ExampleQueueJob : IScheduledQueueJob { public string DefaultSchedule => CronSchedule.TimesPerMinute(10); public async Task ProcessMessageAsync(ExampleQueueMessage message, CancellationToken cancellationToken) => ProcessResults.Completed(); public string GetResultsText(QueueJobProcessResult result) { if (result.FailedCount == 0) { return null; // returning null means we'll get the default result message } return $"ERROR! {result.FailedCount} messages failed"; } } ``` ## Writing your own queue processing You don't have to use the built in interfaces for processing queues in a `IScheduledJob`. You can implement it yourself using `IQueueReader.LeaseAsync()`, `IQueueItemUpdater.ProcessedAsync()`, and `IQueueItemUpdater.FailedAsync()`. Should however let your job implement the marker interface `IScheduledQueueJobBase` as that is what the UI and API uses to tell which job processes which queue. ## Scheduling a processed message to be processed again Sometimes when you're using [idempotent messages](./index.md#idempotent-messages) to update external systems you might have dates in your message that affects the outcome of the processing. Such as a product message which contains a list of prices that have a valid from and to date. At the time of processing you pick the price that is valid right now, but you want to run the processing again when the next price becomes valid. For a batch processing job you can call: ```cs batch.Processed(message, processAgainAtUtc: DateTime.UtcNow.AddHours(1)); ``` And for a job that process messages one by one you do: ```cs return ProcessResults.Completed(processAgainAtUtc: DateTime.UtcNow.AddHours(1)); ``` This will mark the queue message as `Processed` together with the date for when it should be processed again. Note that it's not marked as `Sleeping` like unprocessed, future messages. This to be able to differentiate between messages that have been processed from ones that are just scheduled for processing. The message status is then changed to `Pending` again at the date you set, and the queue processing job is called again with the same message. If a change is enqueued for the same product/message before the date you set the status for the message becomes `Pending` directly and it's up to the job to recalculate when/if the message should get processed again. --- # Processing directly from Kafka topics Nexus allows you to use Kafka as an external pending storage, which means that Nexus reads and processes messages directly from Kafka without first storing them in an SQL backed Nexus queue. Use it by first installing `CommerceMind.Nexus.Kafka` and then configuring the external storage: ```cs builder .AddNexus() .AddKafkaExternalPendingStorage(); ``` And then on your queue: ```cs [Queue("export_order", StorageMode = QueueStorageMode.ExternalPending)] public class ExportOrderQueueMessage : IQueueMessageWithId { } ``` Read more about [external pending storage here](./external-pending.md). # Enqueueing from Kafka topics Nexus has built-in support to use the very popular Apache Kafka platform for both [instance events](../general/instance-events.md) as well as taking messages from a Kafka topic and enqueueing into a Nexus queue. In order to use this you need to install the NuGet package for RabbitMQ called `CommerceMind.Nexus.Kafka`. You set it up like this: ```cs builder.Services .AddNexus() .AddQueues() .AddKafka(options => { options.ConsumerConfig.BootstrapServers = "..."; options.ConsumerConfig.GroupId = "..."; }) .AddNexusKafkaQueues(options => { // This allows you to control things like which status or when messages are processed options.EnqueueContextFactory = message => new EnqueueContext { Priority = message.Id != null ? 10 : 0 } }) .EnqueueFrom("kafka-topic"); ``` With this examples all messages posted to the `kafka-topic` will get placed in the Nexus queue for `KafkaQueueMessage`. ## Instance events You can also use Kafka as the way to [send signals](../general/instance-events.md) between Nexus instances in a scaled out environment. ```cs builder.Services .AddNexus() .AddQueues() .AddKafka(options => { options.ConsumerConfig.BootstrapServers = "..."; options.ConsumerConfig.GroupId = "..."; options.ProducerConfig.BootstrapServers = "..."; }) .AddKafkaInstanceEvents(options => { // Change this to something unique if you're using the same Kafka instance for multiple // different Nexus applications options.TopicName = "nexus_instance_events"; }); ``` --- # Message priority The default behavior in Nexus is to process messages in the order that they come in; oldest first. In some cases this is not granular enough and you have some messages that needs to be prioritized over others. Nexus allows you to control this by passing a number as priority when you enqueue the message. Eg: ```cs public class MyController(IEnqueuer enqueuer) { public async Task Enqueue() { await enqueuer.EnqueueAsync(new MyMessage { ... }, new EnqueueContext { Priority = 10 }); return Ok(); } } ``` With the above your message is guaranteed to be processed before any message with a lower priority than 10. For other messages with the same priority the order will be first in - first out. ## Default priority The default priority in Nexus is `0`. When you enqueue messages you're encouraged to create an enum or a static class with your named priority levels rather than setting numbers like in the example above. Note that you can have low prio messages that have a negative priority. Which means that they will be processed after any message with the default (`0`) priority. ## Programmatically determining the priority In some cases messages comes in through the API and you don't want to let the external party dictate the priority, but if they use the Nexus API directly to enqueue you don't get to set the priority. For this scenario and others you can implement `IMessagePriorityCalculator` and register with `IServiceCollection`. ```cs public static class MessagePriority { public const int High = 100; public const int Low = 0; } public class MyMessagePriorityCalculator : IMessagePriorityCalculator { public async Task CalculatePriorityAsync(MyMessage message) { if (message.SomeProperty == "some value") { return MessagePriority.High; } return MessagePriority.Low; } } ``` Note that this class is only called for messages that doesn't have an explicit priority set already. --- # Processing directly from RabbitMQ queues Nexus allows you to use RabbitMQ as an external pending storage, which means that Nexus reads and processes messages directly from RabbitMQ without first storing them in an SQL backed Nexus queue. Use it by first installing `CommerceMind.Nexus.RabbitMq` and then configuring the external storage: ```cs builder .AddNexus() .AddRabbitMqExternalPendingStorage(); ``` And then on your queue: ```cs [Queue("export_order", StorageMode = QueueStorageMode.ExternalPending)] public class ExportOrderQueueMessage : IQueueMessageWithId { } ``` Read more about [external pending storage here](./external-pending.md). # Enqueueing from RabbitMQ queues RabbitMQ is a great way to communicate between different applications. But sometimes you want to have more visibility into queue messages, when they're processed and if they fail. And to be able to retry in a controlled fashion when they fail. Nexus lets you get the best of both worlds by automatically enqueueing from a RabbitMQ queue to a Nexus queue like the example below. In order to use this you need to install the NuGet package for RabbitMQ called `CommerceMind.Nexus.RabbitMQ`. ```cs builder.Services .AddNexus() .AddQueues() .AddRabbitMq(options => { options.UserName = "..."; options.Password = "..."; options.HostNames = ["localhost"]; // Or: options.ConnectionString = new Uri("amqp://..."); }) .AddNexusRabbitMqQueues(options => { // This allows you to control things like which status or when messages are processed options.EnqueueContextFactory = message => new EnqueueContext { Priority = message is MyImportantQueueMessage ? 10 : 0 } }) .EnqueueFrom(new NexusRabbitMqQueue { Name = "rabbit_queue", Bindings = [new NexusRabbitMqQueueBinding { ExchangeName = "my_exchange", RoutingKey = "" }] }); ``` ## Instance events You can also use Kafka as the way to [send signals](../general/instance-events.md) between Nexus instances in a scaled out environment. ```cs builder.Services .AddNexus() .AddQueues() .AddRabbitMq(options => { options.UserName = "..."; options.Password = "..."; options.HostNames = ["localhost"]; // Or: options.ConnectionString = new Uri("amqp://..."); }) .AddRabbitMqInstanceEvents(options => { // Change this to something unique if you're using the same RabbitMQ instance for multiple // different Nexus applications options.ExchangeName = "nexus_instance_events"; }); ``` --- # Processing directly from Redis Nexus allows you to use Redis as an external pending storage, which means that Nexus reads and processes messages directly from Redis without first storing them in an SQL backed Nexus queue. Use it by first installing `CommerceMind.Nexus.Redis` and then configuring the external storage: ```cs builder .AddNexus() .AddRedisExternalPendingStorage(); ``` And then on your queue: ```cs [Queue("export_order", StorageMode = QueueStorageMode.ExternalPending)] public class ExportOrderQueueMessage : IQueueMessageWithId { } ``` Nexus expects Redis to contain a sorted set with the same name as the Nexus queue and will pop messages from that for processing. Read more about [external pending storage here](./external-pending.md). # Enqueueing from Kafka topics Nexus has built-in support to use the very popular Apache Kafka platform for both [instance events](../general/instance-events.md) as well as taking messages from a Kafka topic and enqueueing into a Nexus queue. In order to use this you need to install the NuGet package for RabbitMQ called `CommerceMind.Nexus.Kafka`. You set it up like this: ```cs builder.Services .AddNexus() .AddQueues() .AddKafka(options => { options.ConsumerConfig.BootstrapServers = "..."; options.ConsumerConfig.GroupId = "..."; }) .AddNexusKafkaQueues(options => { // This allows you to control things like which status or when messages are processed options.EnqueueContextFactory = message => new EnqueueContext { Priority = message.Id != null ? 10 : 0 } }) .EnqueueFrom("kafka-topic"); ``` With this examples all messages posted to the `kafka-topic` will get placed in the Nexus queue for `KafkaQueueMessage`. ## Instance events You can also use Kafka as the way to [send signals](../general/instance-events.md) between Nexus instances in a scaled out environment. ```cs builder.Services .AddNexus() .AddQueues() .AddKafka(options => { options.ConsumerConfig.BootstrapServers = "..."; options.ConsumerConfig.GroupId = "..."; options.ProducerConfig.BootstrapServers = "..."; }) .AddKafkaInstanceEvents(options => { // Change this to something unique if you're using the same Kafka instance for multiple // different Nexus applications options.TopicName = "nexus_instance_events"; }); ``` --- # Storing processed messages The default behavior of a Nexus queue is to delete messages when they have been processed. But for some use cases it's interesting to keep the processed messages. You can control how for long processed messages are stored on the `[Queue]` attribute like this: ```cs [Queue("example", DisplayName = "Example queue", KeepProcessedMessagesFor = "10")] public class ExampleQueueMessage : IQueueMessage { } ``` In this example the messages are stored for 10 days. The property `KeepProcessedMessagesFor` should be either a string that can be parsed into a `TimeSpan` by [`TimeSpan.Parse()`](https://docs.microsoft.com/en-us/dotnet/api/system.timespan.parse?view=net-6.0) or the special string `"forever"` which tells Nexus to never delete processed messages. ## Use cases One use case for storing processed messages is to enable troubleshooting of important messages. Let's say that you have a queue of orders that should get exported to another system. The export from your side might have worked perfectly but the receiving system might reach out to you to say "hey could you send the message for order 123 again?" or "hmm what exactly did you send to us for order 123?". If you delete processed messages you'd have to dig through logs to maybe be able to reconstruct the message. If you instead keep processed messages you can easily answer both questions. Another use case is when an external system is only able to send full exports of data rather than incremental changes. The system might be scheduled to export everything daily into a Nexus queue. If you store processed messages and turn on [idempotent messages](./index.md#idempotent-messages) on your queue Nexus will only set messages to `Pending` where the data differs from yesterday. A third use case is to do change tracking inside your queue job by reading the [previous version](./index.md#previous-message-version) of the message and compare the current to the previous to see what has changed. If you don't store processed messages you won't be able to load the previous message since it's been deleted. --- # Retrying queue messages When you implement a queue processing job you might end up in a situation where you can't process the message right now but you know that you'll soon want to try it again. Such as if you detect that a system that you want to send the message to is down at the moment so you want to try again in X minutes. Nexus lets you handle that by returning `ProcessResults.RetryLater()` from a queue job or call `batch.RetryLater()` in a batch queue job. You specify the amount of time to wait and then Nexus will handle the rest. Nexus will place the message in a new status called `AwaitingRetry` and it will sit there until it's time to try agian. At that point the message status is set to `Pending` again and the processing job will pick it up again. ## Automatic retries You can also instruct Nexus to automatically retry messages in a queue. In the Admin UI you can click on the More button on the queue page and set the field "Automatic error retry" to eg `24h` (or `1d` or `1h 30m` etc) which will automatically retry any message that gets the Error status after that time period. Note that the time period is per message, and does not run periodically to batch update all Error messages to Pending. You can also control this through code by either setting it in the `[Queue]` attribute: ```cs [Queue("my-queue", AutomaticRetryAfter = "00:15:00")] public class MyQueueMessage : IQueueMessage { } ``` `AutomaticRetryAfter` represents a string that can be passed to `TimeSpan.Parse()` so in the example above it'll be retried 15 minutes after it first failed. Another way of setting it through code is to use `IQueueMetaDataRepository.SetRetryErrorsIntervalAsync()` which lets you set a runtime value for it. This is the same method that gets called when updating it in the Admin UI/API. If no value has been set through `IQueueMetaDataRepository` Nexus falls back to what's set in the attribute, and if no value is set in the attribute no automatic retries are performed. Automatic retries respects the max retries for a queue which is described below. ## Max retries To avoid having messages that end up in an inifinite loop of retries Nexus has a default limit on how many times a message can be retried. The default limit is 10 and once a message has reached that number of retries it will get the status `Error` and a comment saying `Message was retried too many times (10) and has been placed in the Error status`. At that point it will not be automatically retried again but you can always manually set it back to `Pending` which gives the message 10 more tries. If you want to change the limit of max number of retries you can either set it on all queues like this in your `Program.cs`: ```cs builder.Services.AddNexus().AddQueues(options => { options.DefaultMaxRetries = 5; }); ``` You can also set this per queue using the `[Queue]` attribute: ```cs [Queue("example", MaxRetries = 5)] public class ExampleQueueMessage : IQueueMessage { } ``` Note that you can allow an infinite number of retries by setting `-1` as the max. ## Status when max retries By default the status becomes `Error` when a message has been retried too many times. But you can control which status it gets by passing the status as part of the `ProcessResults.RetryLater()` call, like this: ```cs return ProcessResults.RetryLater(TimeSpan.FromMinutes(5), statusWhenMaxRetried: "MyErrorStatus"); ``` --- # Queue statistics Nexus can store statitics for your queues to help visualize the activity in a queue. If you've enabled statistics you can find them in the Admin UI under Statistics and in the queue list and queue details. The statistics shown in the queues list is the total number of status changes/total activity in a queue. The statitics consists of counters for each status. The statistics represents the changes in status rather than the current count of a status at any given time. That is, if the statistics reports zero Error messages it means that no messages got the status Error at that point in time. By default statistics are turned off but you can opt-in to them by either: **Turning it on for all queues by doing:** ```cs builder.Services.AddNexus().AddQueues(options => { options.StatisticsRetention = TimeSpan.FromDays(30); // TimeSpan.MaxValue to store forever }); ``` **Turning it on on a specific queue**: ```cs [Queue(StatisticsRetention = "30d")] public class MyQueueMessage : IQueueMessage { } ``` The string set for `StatisticsRetention` must either by a parsable `TimeSpan` or in the human readable format, eg `6h` or `30d`. **Updating through the Admin UI/API**: In the Admin UI you can go to the queue details and click on the More button and edit the Statistics retention per queue. The Admin UI uses the API to do this, so it's also possible to control through the API directly. See the Swagger docs for endpoint details. ## Disabling statistics By default statistics tables are created per queue to make it possible to enable it dynamically through the Admin UI. If you really don't want to see the statistics feature you can disable it completely by doing: ```cs builder.Services.AddNexus().AddQueues(options => { options.DisableStatistics = true; }); ``` This deletes all statistics tables and prevents statistics features from being displayed in the UI. ## `IQueueStatisticsReader` / `IQueueStatisticsWriter` The reader and writer interfaces are interfaces you can override to either implement your own statistics storage or decorate to send the statistics to analytics. The writer interface gets called directly when any message status changes (or new messages are created), so you should store the changes in memory and persist the changes in batches in a background thread. --- # Queue message validation If you need to validate messages before they're added to Nexus there's two approaches. Either you can let your message class implement `IQueueMessageWithValidation` like this: ```cs public class MyMessage : IQueueMessageWithValidation { public required int MyProperty { get; set; } public ValidationResults Validate() { return MyProperty < 10 ? ValidationResults.Valid() : ValidationResults.Invalid("Value too high"); } } ``` Or you can implement a separate validator class like this: ```cs public class MyMessage : IQueueMessageWithValidation { public required int MyProperty { get; set; } } public class MyMessageValidator : IQueueMessageValidator { public async Task ValidateAsync(MyMessage message) { return message.MyProperty < 10 ? ValidationResults.Valid() : ValidationResults.Invalid("Value too high"); } } ``` If you have a validator class you don't need to register it, Nexus will find it as long as it's placed in an assembly that's scanned for queue messages. If you need to place the validator in an assembly that's not scanned for messages you can explicitly add the validator by calling `AddQueueValidator()` on the Nexus registration. The effect of returning `ValidationResults.Invalid()` will be that a `QueueValidationException` will be thrown and in the case of the API the request will fail with a `400 Bad Request` and with the validation message as the response. ## Skipping validation In some cases you might want to explicitly skip validation, like if the message comes from an external source and it's more important to store the message than validating it. In such cases you can use `SkipValidation` on the `EnqueueContext` like this: ```cs await enqueuer.EnqueueAsync(message, new EnqueueContext { SkipValidation = true }); ``` ## API validation Note that some validation can happen even before your validators are called. Like if you have `required` properties on your message class and incoming API requests are missing such values. In that case the deserialization will fail and the API will respond with `400 Bad Request` before your validator is called. If you want to handle these cases yourself you either need to skip the `required` part of your properties or prevent the serialization from throwing by implementing your own `IQueueMessageSerializer`. --- # Virtual queues Depending on your application you might not want the message producer to be in complete control of the message that is added to a queue. And in some cases a single event should cause multiple messages to be added to different queues. But you don't want to force the message producer to have to think about that. To solve this the system has a concept of virtual queues. A virtual queue is an `IQueueMessage` that doesn't have it's own backing storage, it just acts as a proxy in front of one or more concrete queues. You can then register one or more message transformers to place messages in concrete queues. Hopefully this all makes more sense with a concrete example. Let's say we're in an e-commerce context and we receive an event that an order has been placed. When an order is placed we want to do at least two things; send an order confirmation email and export the order to some other system such as the ERP. The naive implementation of this would something like this: ```cs [Queue("send_order_confirmation")] public class SendOrderConfirmationQueueMessage : IQueueMessageWithId { public string? Id { get; set; } } [Queue("export_order")] public class ExportOrderQueueMessage : IQueueMessageWithId { public string? Id { get; set; } } public class OrderController { private readonly IEnqueuer _sendOrderConfirmationEnqueuer; private readonly IEnqueuer _exportOrderQueueMessageEnqueuer; public OrderController(IEnqueuer sendOrderConfirmationEnqueuer, IEnqueuer exportOrderQueueMessageEnqueuer) { _sendOrderConfirmationenqueuer = sendOrderConfirmationEnqueuer; _exportOrderQueueMessageEnqueuer = exportOrderQueueMessageEnqueuer; } [HttpPost] public async Task OrderPlacedAsync(string orderId, string email) { await _sendOrderConfirmationenqueuer.EnqueueAsync(new SendOrderConfirmationQueueMessage { Id = orderId, }); await _exportOrderQueueMessageEnqueuer.EnqueueAsync(new ExportOrderQueueMessage { Id = orderId, }); return NoContent(); } } ``` There's a couple of problems with this approach. The first one is that the controller needs to know which queues are interested in getting notified when orders are placed. This means that if we add an additional queue for order processing we'd need to update the controller to enqueue that message as well. The second problem is that the controller here is in charge of creating and setting properties to the message that is actually placed in the queue. In the above example it's really simple but constructing a message can be more complex than that. In simple applications this isn't really an issue but in larger application it becomes a problem. So let's update the example to use virtual queues instead! ```cs [VirtualQueue("order_placed")] public class OrderPlacedQueueMessage : IQueueMessageWithId { public string? Id { get; set;} } [Queue("send_order_confirmation")] public class SendOrderConfirmationQueueMessage : IQueueMessageWithId { public string? Id { get; set;} } public class SendOrderConfirmationQueueMessageSyncTransformer : IVirtualQueueMessageSyncTransformer { public TransformResult Transform(OrderPlacedQueueMessage message, EnqueueContext? context) { return new(new SendOrderConfirmationQueueMessage { Id = message.Id }); } } [Queue("export_order")] public class ExportOrderQueueMessage : IQueueMessageWithId { public string? Id { get; set;} } public class ExportOrderQueueMessageSyncTransformer : IVirtualQueueMessageSyncTransformer { public TransformResult Transform(OrderPlacedQueueMessage message, EnqueueContext? context) { return new(new ExportOrderQueueMessage { Id = message.Id }); } } public class OrderController { private readonly IEnqueuer _orderPlacedEnqueuer; public OrderController(IEnqueuer orderPlacedEnqueuer) { _orderPlacedEnqueuer = orderPlacedEnqueuer; } [HttpPost] public async Task OrderPlacedAsync(string orderId, string email) { await _orderPlacedEnqueuer.EnqueueAsync(new OrderPlacedQueueMessage { Id = orderId, }); return NoContent(); } } ``` So there's a lot of stuff to unpack here. First we have a new attribute called `[VirtualQueue]`. This is what tells the system that this message should not get a concrete queue storage setup, it's only used for enqueuing messages to other queues. The second thing is that we have two message transformers. These are just classes that implements either `IVirtualQueueMessageSyncTransformer` or it's async counterpart `IVirtualQueueMessageTransformer`. They're responsible for translating a virtual message into a concrete message. The message transformers are automatically registered with the service collection as long as they exists in assemblies which are scanned for `IQueueMessage` implementations. In the controller we now take in an `IEnqueuer` for our virtual queue. Regular queues gets more services registered such as `IQueueReader` and `IQueueItemUpdater` but the only thing you can do with a virtual queue is to add messages to it. What happens behind the scenes when you call the enqueuer is that the message transformers we created are called and the enqueuers for `SendOrderConfirmationQueueMessage` and `ExportOrderQueueMessage` are called behind the scenes. As you might have noticed the transformer can return null in its `Transform()` method which is a way to say "this message isn't interesting for this queue". If all transformers return null then nothing is enqueued. ### EnqueueContext If the call site that wants to enqueue messages includes an `EnqueueContext` it gets passed into your transformer as in the example above. But you can override the context by passing a new `EnqueueContext` to the `TransformResult` as the second argument, eg: ```cs public class ExportOrderQueueMessageSyncTransformer : IVirtualQueueMessageTransformer { public async Task> TransformAsync(OrderPlacedQueueMessage message, EnqueueContext? context) { return new( new ExportOrderQueueMessage { Id = message.Id }, new EnqueueContext { ProcessAfter = TimeSpan.FromMinutes(15) } ); } } ``` ## API Virtual queues are extra interesting if you let other systems POST messages into your queues through the API since the virtual queues are also exposed there. This gives you a way to intercept the data sent from the other system and run logic before the message is stored. ## Benefits So when should you use virtual queues? If you have a not-so-large application with just a few developers it might not worth it. But if you have a larger application and maybe even multiple teams it will be worth it. In such a case you should probably create a virtual queue for each queue, and keep the concrete queue message as internal and the virtual queue message as public. That prevents anyone outside to directly manipulate your underlying queue. Something like this: ```cs [VirtualQueue("public_queue")] public class PublicQueueMessage : IQueueMessageWithId { public string? Id { get; set; } } [Queue("internal_queue")] internal class InternalQueueMessage : IQueueMessageWithId { public string? Id { get; set; } } internal class InternalQueueMessageSyncTransformer : IVirtualQueueMessageSyncTransformer { public InternalQueueMessage? Transform(PublicQueueMessage message) { return new InternalQueueMessage { Id = message.Id }; } } ``` Now other .NET projects will only be able to use `IEnqeuer` which gives you full control over your queue. Another option is to start with a public `IQueueMessage` and at a later point convert it from a concrete queue to a virtual queue. Something like this in step 1: ```cs [Queue("internal_queue")] public class PublicQueueMessage : IQueueMessageWithId { public string? Id { get; set; } } ``` And then when you need it change to: ```cs [VirtualQueue("public_queue")] public class PublicQueueMessage : IQueueMessageWithId { public string? Id { get; set; } } [Queue("internal_queue")] internal class InternalQueueMessage : IQueueMessageWithId { public string? Id { get; set; } } ``` There's a couple of downsides with this that you should be aware of: 1. Anyone using the API to enqueue messages needs to update their code to use `public_queue` instead of `internal_queue`. 1. Since we moved the `[Queue("internal_queue")]` to a different type there might still be old JSON messages in that queue that are serialized `PublicQueueMessage`s. --- # Webhooks for Nexus queues [Webhooks](https://en.wikipedia.org/wiki/Webhook) is a popular way of letting another web application notify your web application that something has happened by posting data to an endpoint in your application. Nexus supports this out-of-the-box by automatically creating an endpoint for each queue in the system that can be used given to another application to send notifications to. The webhook url is: ``` https://url-to-your-nexus-app.com/api/queues/[queue name]/webhook ``` You can also click on the More button in the admin url of your queue to get the exact url. ## Request details Nexus requires the incoming HTTP request to be a POST request and to have a request body in JSON. The body can either be a single message or an array of messages: Either: ```json { "someMessageProperty": "some value" } ``` Or: ```json [ { "someMessageProperty": "message1" }, { "someMessageProperty": "message1" }, { "someMessageProperty": "message1" } ] ``` The JSON body is deserialized into the structure of your queue message and sent to the queue. Which means that any properties in the JSON body that doesn't exist in your message type is discarded, so make sure to include all the properties you need. You can customize the message deserialization by implementing your own `IQueueMessageSerializer` in which you can take the entire original JSON body and store in a separate property on your message. It's also a good idea to use [virtual queues](./virtual.md) for webhooks since it lets you register a single webhook and then pass the webhook notitication to multiple internal queues. --- # Terms of use Nexus is free-of-charge and licensed under the [Apache 2.0](https://www.apache.org/licenses/LICENSE-2.0) license. It is however not publically available as the repository is private and the NuGet packages are hosted on a private feed. It may seem counter-intuitive to have an open source licensed code base in a private repository but the reason for it is that we want to know who is using it and be able to help out. It's possible that we at some point will release Nexus completely public. If you want access just hit us an email at [movefaster@commercemind.se](mailto:movefaster@commercemind.se)! The important thing for you to know is that once you've gotten access to the repository and NuGet packages your access to it is protected by the [Apache 2.0](https://www.apache.org/licenses/LICENSE-2.0) license. Which means that you will forever have the right to the code and do whatever you want with it as long as you follow the license terms.