osstotalsoft/nbb

.Net Building Blocks

NBB

.Net Building Blocks

Our philosophy

Build cutting-edge, cloud-ready, scalable, maintainable and fun LOB services with .Net Building Blocks

Combining domain-driven design tactical patterns with clean architecture by decoupling the business model and use-cases of the application with the rest of the technology and infrastructure, you get a technology-independent, hand-crafted, stable, encapsulated business model that will evolve over time, regardless the UI, Database, Messaging or other infrastructure or technology.

Applying concepts from EDA, CQRS or ES we decouple furthermore the business domain from the read-side so that the domain would not change when the UI views needs to change.

Applying concepts from the Microservices architectural style, you get a new beginning with every new bounded context (module).

Architectural considerations

The goal of software architecture is to minimize the human resources required to build and maintain the required system.

-- Robert C. Martin

With NBB you can power a great diversity of architectures from a Monolithic one to a Multi-Container Microservices based one. It is important to mention that NBB does not impose any kind of architecture.

This repo contains a sample Microservices application decomposed around three bounded contexts: Contracts, Invoices and Payments. They are autonomous and the integration is based on events delivered with NATS. The sample application contains scripts for building CI / CD pipelines for docker-compose or kubernetes.

The blocks

• NBB.Core - core abstractions and functionality that other packages rely upon
• NBB.Application - application layer specific functionality
• NBB.Domain - building blocks for domain modelling in DDD
• NBB.Data - data access abstractions and implementations
• NBB.Messaging - distributed application infrastructure that enables loosely-coupled, message-based asynchronous communication
• NBB.EventStore - event store functionality
• NBB.Correlation - facilitates the grouping of all requests, messages, logs, and traces belonging to a business flow
• NBB.ProcessManager - a way of orchestrating your events
• NBB.MultiTenancy - building blocks for multi-tenant applications
• NBB.ProjectR - functional style read-model projections inspired by Elm

The samples

Please see our samples folder to get you started with various architectures built with NBB.

The templates

NBB templates is a collection of custom templates to be used by the dotnet cli (dotnet new command).

License

NBB is licensed under the MIT license.

nbb's Issues

Add support for parallel message handling in topic subscribers

In the current solutions messages received on a topic are processed serially one at a time.

Add support for processing of multiple messages of the same type (topic) in parallel. This behavior should be configurable per subscription.

StopWatch from Diagnostic assembly is used in too many places

We need a Profilling decorator where to keep up reference to Diagnostics because the change of a StopWatch doesn’t ripple through the rest of the system.

Process manager - state should not be contrained to beeing a struct

Currently, the state in process manager is a struct. It should be a class.

Using a struct is a way to insure the state does not have behaviour.

However, the "framework" should not be THAT opinionated and should allow the programmer to have the responsibility of its own code.

ExcelDataExport.Export, GetCommandHandler, CsvDataExport.Export, ParseInternal has too many local variables

To refactor such method and increase code quality and maintainability, certainly you'll have to split the method into several smaller methods or even create one or several classes to implement the logic.

Some duplicate on DefaultTopicRegistry class and multiple if clauses should be rewritten with a chain of responsability or strategy patterns

   public string GetTopicForMessageType(Type messageType, bool includePrefix = true)
        {
            var topic = GetTopicNameFromAttribute(messageType);
            if (topic == null)
            {
                if (typeof(ICommand).IsAssignableFrom(messageType))
                {
                    topic = $"ch.commands.{messageType.GetLongPrettyName()}";
                }
                else if (typeof(IEvent).IsAssignableFrom(messageType))
                {
                    topic = $"ch.events.{messageType.GetLongPrettyName()}";
                }
                else if (typeof(IQuery).IsAssignableFrom(messageType))
                {
                    topic = $"ch.queries.{messageType.GetLongPrettyName()}";
                }
                else
                {
                    topic = $"ch.messages.{messageType.GetLongPrettyName()}";
                }
            }

            return topic;
        }

This should be in a private method and if-else branches should be rewritten with chain of responsibility like so:

        public static string GetTopic(Type messageType, IConfiguration configuration) =>
            GetTopicNameFromAttribute(messageType, configuration) ?? new CommandTypeValidatorHandler()
                .Then(new EventTypeValidatorHandler())
                .Then(new QueryTypeValidatorHandler())
                .Then(new DefaultTypeHandler()).Handle(messageType);

where handlers are:

  public class CommandTypeValidatorHandler : MessageTypeHandler<Type>
    {
        public override string Handle(Type request)
        {
            if (typeof(ICommand).IsAssignableFrom(request))
            {
                return $"ch.commands.{request.GetLongPrettyName()}";
            }
            return base.Handle(request);
        }
    }

    public class EventTypeValidatorHandler : MessageTypeHandler<Type>
    {
        public override string Handle(Type request)
        {
            if (typeof(IEvent).IsAssignableFrom(request))
            {
                return $"ch.events.{request.GetLongPrettyName()}";
            }

            return base.Handle(request);
        }
    }

  public class QueryTypeValidatorHandler : MessageTypeHandler<Type>
    {
        public override string Handle(Type request)
        {
            if (typeof(IQuery).IsAssignableFrom(request))
            {
                return $"ch.queries.{request.GetLongPrettyName()}";
            }
            return base.Handle(request);
        }
    }

public class DefaultTypeHandler : MessageTypeHandler<Type>
    {
        public override string Handle(Type request)
            => $"ch.messages.{request.GetLongPrettyName()}";
    }


 public interface IMessageTypeHandler<T> where T : class
    {
        IMessageTypeHandler<T> Then(IMessageTypeHandler<T> next);
        string Handle(T request);
    }

    public abstract class MessageTypeHandler<T> : IMessageTypeHandler<T> where T : class
    {
        private IMessageTypeHandler<T> Next { get; set; }

        public virtual string Handle(T request)
        {
            return Next?.Handle(request);
        }

        public IMessageTypeHandler<T> Then(IMessageTypeHandler<T> next)
        {
            Next = next;
            return Next;
        }
    }

IMessageSerDes has two duties: Deserialization and Serialization. Serialization and Deserialization are cohesive?

   public interface IMessageSerDes
    {
        string SerializeMessageEnvelope(MessagingEnvelope envelope, MessageSerDesOptions options = null);
        MessagingEnvelope<TMessage> DeserializeMessageEnvelope<TMessage>(string envelopeString, MessageSerDesOptions options = null);
        MessagingEnvelope DeserializeMessageEnvelope(string envelopeString, MessageSerDesOptions options = null);
    }

We could split up this interface in two interfaces (Interface Segregation Principle):

1. IMessageSerializer
2. IMessageDeserializer

And, besides, why MessageSerDesOptions optional parameter is spreading in all methods?
We could take if off by extracting an abstract class for a builder. I will come back with this implementation in a future issue.

Model NBB.Effect types with structs instead of classes and benchmark the results

The NBB.Effect family of types (Effect, Pure, Impure, EffectFnSeq, etc) is modeled via reference types which puts a lot of presure on the GC.
We should be able to cut off a lot of managed memory allocations using value types.
Be carefull with value types, as they get boxed when casted to interface types, we should avoid polymorhism in those cases.

Process Manager - Concurrency exception for anonymous object correlation key

Event<chEv.InsurancePolicyValidated>(builder => builder.CorrelateBy(z => new { z.ChDocumentId, z.ChSiteId }));

Received OutOfMemoryException after many retries:

Process manager - Add command support

Current behaviour: process manager only listens and allow creating definitions for events.
Desired behaviour: process manager should also listen to commands and allow definitions for commands

Reference the templates repo in main Readme

We should reference the templates repo and maybe write a short paragraph in the main Readme.md

Some classes and its contracts(interfaces) are in the same assembly.

MessageBusPublisher and IMessageBusPublisher are in NBB.Messaging.Abstractions assembly for instance. These are supposed to be in different assemblies.

F# mediator boilerplate

Currently the NBB.Application.Mediator,FSharp does not formalize a mediator abstraction, instead it defines an algebra for dealing with application requests and evenets pipelines. Those pipelines cannot be directly referenced in other request/event handlers due to the cyclic dependency between the specific handler and the whole pipeline. Therefore we need a mediator abstraction in order to invert dependencies betwwen the handler and the pipeline. The pipeline will depend on the specific handlers, the handlers will depend on the mediator abstraction and the mediator implementation will depend on the whole pipeline.
See https://github.com/osstotalsoft/functional-guy/tree/master/Chapter7.%20FSharp%20microservices/Sample/NBB.Invoices.FSharp/NBB.Invoices.FSharp/Application

One should be able to register some mediator by just providing the command-pipeline, query-pipeline and event-pipeline, see examples in the link above.

Event store - multiple calls

one can call this code multiple times:
services.AddEventStore() .WithNewtownsoftJsonEventStoreSeserializer() .WithAdoNetEventRepository();

Result:
System.Data.SqlClient.SqlException (0x80131904): WrongExpectedVersion at System.Data.SqlClient.SqlConnection.OnError(SqlException exception, Boolean breakConnection, Action1 wrapCloseInAction) at System.Data.SqlClient.SqlInternalConnection.OnError(SqlException exception, Boolean breakConnection, Action1 wrapCloseInAction) at System.Data.SqlClient.TdsParser.ThrowExceptionAndWarning(TdsParserStateObject stateObj, Boolean callerHasConnectionLock, Boolean asyncClose) at System.Data.SqlClient.TdsParser.TryRun(RunBehavior runBehavior, SqlCommand cmdHandler, SqlDataReader dataStream, BulkCopySimpleResultSet bulkCopyHandler, TdsParserStateObject stateObj, Boolean& dataReady) at System.Data.SqlClient.SqlCommand.FinishExecuteReader(SqlDataReader ds, RunBehavior runBehavior, String resetOptionsString) at System.Data.SqlClient.SqlCommand.CompleteAsyncExecuteReader() at System.Data.SqlClient.SqlCommand.EndExecuteNonQueryInternal(IAsyncResult asyncResult) at System.Data.SqlClient.SqlCommand.EndExecuteNonQuery(IAsyncResult asyncResult) at System.Threading.Tasks.TaskFactory1.FromAsyncCoreLogic(IAsyncResult iar, Func2 endFunction, Action1 endAction, Task1 promise, Boolean requiresSynchronization) --- End of stack trace from previous location where exception was thrown --- at NBB.EventStore.AdoNet.AdoNetEventRepository.AppendEventsToStreamExpectedVersionAsync(String streamId, IList1 eventDescriptors, Int32 expectedVersion, CancellationToken cancellationToken) ClientConnectionId:e86fba0e-c407-4228-96fb-65e85c8cb171 Error Number:50000,State:1,Class:16 --- End of inner exception stack trace --- at NBB.EventStore.AdoNet.AdoNetEventRepository.AppendEventsToStreamExpectedVersionAsync(String streamId, IList1 eventDescriptors, Int32 expectedVersion, CancellationToken cancellationToken) at NBB.EventStore.AdoNet.AdoNetEventRepository.AppendEventsToStreamAsync(String streamId, IList1 eventDescriptors, Nullable1 expectedVersion, CancellationToken cancellationToken) at NBB.EventStore.EventStore.AppendEventsToStreamAsync(String stream, IEnumerable1 events, Nullable1 expectedVersion, CancellationToken cancellationToken) at NBB.ProcessManager.Runtime.Persistence.InstanceDataRepository.Save[TData](Instance1 instance, CancellationToken cancellationToken) in ...`

Desired behaviour: some validation on this topic.

Async extensions over IQueriable

One should be able to work directly with IQueryable in Query handlers without direct referencing EntityFramework. e.g. .ToListAsync, .FirstOrDefaultAsync.

Iqueryable can be injected in DI like this:
services.AddEfQuery<SomeProjection, SomeDbContext>();

Process killed when a deserialization error occurred in messaging host

Describe the bug
Process killed when a deserialization error occurred in messaging host

To Reproduce
Steps to reproduce the behavior:

1. Send an invalid message in the bus
2. Subscribe to the message in the messaging host
3. Register publish decorator that throws an error
4. Publishing the deserialization error to the dead letter queue throws an exception that kills the process.

Expected behavior
The error must be logged but the process should not be killed.

Screenshots

Microservices orchestration Sample

In Microservices sample, use Orchestration via NBB.ProcessManager instead of Choreography.
If necessary move Integration event handlers logic in Command/CommandHandlers.
Add additional fields in commands/events required by process correlation

NBB.ProcessManager - ES events should be resilient to publish language updates

If services publish language updates, but the process does not update it's behaviour, the event sourcing mechanism should be able to hydrate history events.
See events:

• public record EventReceived(TEvent ReceivedEvent);
• public record ProcessCompleted(TEvent ReceivedEvent);

Maybe we should have events that encapsulate state updates.

We should introduce an abstract class Specification that allows us to encapsulate some piece of domain knowledge into a single unit - specification - and reuse it in different parts of the code base

public abstract class Specification<T>
{
    public abstract Expression<Func<T , bool>> ToExpression();
 
    public bool IsSatisfiedBy(T entity)
    {
        Func<T , bool> predicate = ToExpression().Compile();
        return predicate(entity);
    }
}

public class AndSpecification<T> : Specification<T>
{
    private readonly Specification<T> _left;
    private readonly Specification<T> _right;
 
    public AndSpecification(Specification<T> left, Specification<T> right)
    {
        _right = right;
        _left = left;
    }
 
    public override Expression<Func<T , bool>> ToExpression()
    {
        Expression<Func<T , bool>> leftExpression = _left.ToExpression();
        Expression<Func<T , bool>> rightExpression = _right.ToExpression();
 
        BinaryExpression andExpression = Expression.AndAlso(
            leftExpression.Body, rightExpression.Body);
 
        return Expression.Lambda<Func<T , bool>>(
            andExpression, leftExpression.Parameters.Single());
    }
}

And an example related to tasks:

  public class CloseEventIsComingSpecification : Specification<Event>
    {
        private TaskDefinition TaskDefinition { get; }

        public CloseEventIsComingSpecification(TaskDefinition taskDefinition)
        {
            TaskDefinition = taskDefinition;
        }

        public override Expression<Func<Event, bool>> ToExpression()
        {
            return @event => @event.EventDefinitionName == TaskDefinition.CloseEvent;
        }
    }

 public class CancelEventIsComingSpecification : Specification<Event>
    {
        private TaskDefinition TaskDefinition { get; }

        public CancelEventIsComingSpecification(TaskDefinition taskDefinition)
        {
            TaskDefinition = taskDefinition;
        }

        public override Expression<Func<Event, bool>> ToExpression()
        {
            return @event => @event.EventDefinitionName == TaskDefinition.CancelEvent;
        }
    }

 public class EvaluatorSpecification : Specification<Event>
    {
        private IExpressionEvaluationService ExpressionEvaluationService { get; }
        private TaskDefinition TaskDefinition { get; }

        public EvaluatorSpecification(IExpressionEvaluationService expressionEvaluationService, TaskDefinition taskDefinition)
        {
            ExpressionEvaluationService = expressionEvaluationService;
            TaskDefinition = taskDefinition;
        }

        public override Expression<Func<Event, bool>> ToExpression()
        {
            return @event => ExpressionEvaluationService.EvaluateEventExpression(@event, TaskDefinition.CloseExpression);
        }
    }

Some extensions over IQueryable are not used anywhere from QueryableAsyncExtensions anymore

Some extensions over IQueryable are not used anywhere

Event store SQL scripts should be re-runnable + Create a method named CreateDatabaseObjects in AdoNetEventStoreDatabaseMigrator for creating at the first time all tables required for Event Sourcing.

Event store SQL scripts should be re-runnable + Create a method named CreateDatabaseObjects in AdoNetEventStoreDatabaseMigrator for creating at the first time all tables required for Event Sourcing.
This way there would be no need to execute all scrips manually.

ICommand, IMessage are empty. Are these marker interfaces???

Interfaces define members that provide a behavior or usage contract. The functionality that is described by the interface can be adopted by any type, regardless of where the type appears in the inheritance hierarchy. A type implements an interface by providing implementations for the members of the interface. An empty interface does not define any members. Therefore, it does not define a contract that can be implemented.
If this identification will occur at run time, the correct way to accomplish this is to use a custom attribute. Use the presence or absence of the attribute, or the properties of the attribute, to identify the target types. If the identification must occur at compile time, then it is acceptable to use an empty interface.

IInvoiceRepository not used

https://github.com/osstotalsoft/nbb/blob/master/samples/MicroServices/NBB.Invoices/NBB.Invoices.Domain/InvoiceAggregate/IInvoiceRepository.cs

The generic interfere and crud repo is used :)

NBB.ProcessManager - Db persisted TimeoutRepository

Add an implementation for a db persisted TimeoutRepository - maybe event sourced.
Add a configuration for TimeoutsManager PollingInterval.
Discuss the posibility of replacing the polling mechanism with an evented one.

dotenv configuration provider

We need to plugin external configuration .env files
We need the provider and a builder extension.

ProcessManager - AppendEventsToStreamAsync - SQL call incorrect exception handler

If SQL insert statement fails, the same exception code is raised - WrongExpectedVersion - and the initial error is not returned to the application.

declare @p5 Web.NewEventStoreEvents
insert into @p5 values('D318360C-CBF7-42DB-843E-730D2D85668A',N'{"ReceivedEvent":{"PartnerId":33608627,"Params":[{"Item1":"TriggeredByDocumentPartnerId","Item2":"54102"}]},"ReceivedEventType":"Charisma.Leasing.PublishedLanguage.Events.Partner.MasterPartnerConsentUpdated"}','NBB.ProcessManager.Runtime.Events.EventReceived, NBB.ProcessManager.Runtime','FBF236D4-1A6D-430C-915F-756F4D341B7C')

exec sp_executesql N'declare @NewEventsCount int
select @NewEventsCount = count(*) from @NewEvents
if @NewEventsCount = 0
begin
	return;
end

declare @ActualVersion int
select @ActualVersion = count(*) from EventStoreEvents where StreamId = @StreamId

if @ActualVersion <> @ExpectedVersion
BEGIN
	RAISERROR(''WrongExpectedVersion'', 16, 1);
	RETURN;
END



BEGIN TRY
	insert into EventStoreEvents(EventId, EventData, EventType, CorrelationId, StreamId, StreamVersion)
	select EventId, EventData, EventType, CorrelationId, @StreamId, @ExpectedVersion + OrderNo
	from @NewEvents
END TRY
BEGIN CATCH
	RAISERROR(''WrongExpectedVersion'', 16, 1);
END CATCH

',N'@StreamId varchar(200),@ExpectedVersion int,@NewEvents [NewEventStoreEvents] READONLY',@StreamId='Deimos.Worker.Processes.PartnerSyncProcess:NBB.Core.Effects.Unit:33608627',@ExpectedVersion=95,@NewEvents=@p5

Hello from an OpenFaaS core contributor!

This is not really an issue, I just wasn't able to find any other way to contact anyone on this project.

I discovered this while searching through NuGet.org for existing OpenFaaS packages. I am one of the core contrtibutors to the OpenFaaS project and would like to hear more about how you're using the project.

Please feel free to contact me directly. My email is on my profile. Alternatively, please consider joining the OpenFaaS community on Slack

Thank you

ITenantService is not a proper name for an interface and TenantService could be refactored in a cleaner way.

ITenantService should be named more abstract than any concrete implementation implementing it. ITenantService is a too concrete name for an interface. ITenantIdentifierProcessor could be a well-chosen name. But, nevertheless, ITenantService name is too tied to TenantService implementation so If I want to create another implementation this name wouldn't help me.

This:

  public interface ITenantService
    {
        /// <summary>
        /// Gets current tenant id
        /// </summary>
        /// <returns>Tenant Id</returns>
        /// <exception cref="TenantNotFoundException"></exception>
        Task<Guid> GetTenantIdAsync();

        /// <summary>
        /// Tries to get current tenant id
        /// </summary>
        /// <returns>Tenant Id or null</returns>
        Task<Guid?> TryGetTenantIdAsync();
    }

can be rewritten as follows:

  public abstract class TenantIdentifierProcessor
    {
        public async Task<Guid> GetTenantIdAsync()  // this is a default method. This is a template method because it will remain the same for other implementation. This way we can avoid the duplicate behavior of this method for other implementations.
        {
            return await TryGetTenantIdAsync() ?? throw new TenantNotFoundException();
        }

        protected abstract Task<Guid?> TryGetTenantIdAsync(); //this is the method with the logic inside of if. We can change this logic in other implementation.
    }

 public class TenantService : TenantIdentifierProcessor
    {
        private readonly IEnumerable<TenantIdentificationStrategy> _tenantIdentificationStrategies;

        public TenantService(IEnumerable<TenantIdentificationStrategy> tenantIdentificationStrategies)
        {
            _tenantIdentificationStrategies = tenantIdentificationStrategies;
        }

        protected override async Task<Guid?> TryGetTenantIdAsync()
        {
            foreach (var tenantIdentificationStrategy in _tenantIdentificationStrategies)
            {
                var tenantId = await tenantIdentificationStrategy.TryGetTenantIdAsync();

                if (tenantId.HasValue)
                {
                    return tenantId;
                }
            }

            return null;
        }
    }

And the interface will contain just GetTenantIdAsync method so for the other implementation you can change the behavior of this method.

 public interface ITenantIdentifier
    {
        /// <summary>
        /// Gets current tenant id
        /// </summary>
        /// <returns>Tenant Id</returns>
        /// <exception cref="TenantNotFoundException"></exception>
        Task<Guid> GetTenantIdAsync();
    }

Example of another implementation:

public class TenantService2 : TenantIdentifierProcessor, ITenantIdentifierProcesor
    {
        private readonly IEnumerable<TenantIdentificationStrategy> _tenantIdentificationStrategies;

        public TenantService2(IEnumerable<TenantIdentificationStrategy> tenantIdentificationStrategies)
        {
            _tenantIdentificationStrategies = tenantIdentificationStrategies;
        }

        protected override async Task<Guid?> TryGetTenantIdAsync()
        {
            foreach (var tenantIdentificationStrategy in _tenantIdentificationStrategies)
            {
                var tenantId = await tenantIdentificationStrategy.TryGetTenantIdAsync();

                if (tenantId.HasValue)
                {
                    return tenantId;
                }
            }

            return null;
        }

        async Task<Guid> ITenantIdentifierProcesor.GetTenantIdAsync() //explicit implementation
        {
            return await TryGetTenantIdAsync() ?? throw new TenantNotFoundException();
        }
    }

NBB.ProcessManager RequestTimeout method

RequestTimeout should be renamed as Schedule, change it's signature to:
public EventActivitySetBuilder<TEvent, TData> RequestTimeout(Func<TEvent, InstanceData, T> messageFactory,
TimeSpan timeSpan,
EventPredicate<TEvent, TData> predicate = null)

TenantConfiguration and ConfigurationTenantRepository - switch from tenant list to tenant map

Is your feature request related to a problem? Please describe.
When working with multiple configuration sources, merging configuration lists is based on index, witch is hard and error prone.

Describe the solution you'd like
We need to configure and bind tenant configuration and tenant repository from a tenant map with the tenant code as key.

HttpFormatter, BodyParser, HeaderParser might be turned into a static type

A stateless class or structure might be turned into a static type.
Such class or structure is a stateless collection of pure functions, that doesn't act on any this object data. Such collection of pure functions is better hosted in a static class. Doing so simplifies the client code that doesn't have to create an object anymore to invoke the pure functions.

isNew flag argument from ApplyChanges method is a strong sign the function is doing two or more things

Flag arguments are strong signs the functions have more responsibilities. We can break them into smaller functions.

GetTopicForName has some rules inside of it that could be rewritten

public string GetTopicForName(string topicName, bool includePrefix = true)
        {
            if (topicName == null)
            {
                return null;
            }

            var topic = (includePrefix ? GetTopicPrefix() : string.Empty) + topicName;
            topic = topic.Replace("+", ".");  //this rules are ugly here
            topic = topic.Replace("<", "_");
            topic = topic.Replace(">", "_");

            return topic;
        }

could be rewritten such as:

private IEnumerable<RuleValue> Rules { get; } = new[] // could be moved away from this class to container
        {
            new RuleValue("+", "."),
            new RuleValue("<", "_"),
            new RuleValue(">", "_"),
        };

public string GetTopicForName(string topicName, bool includePrefix = true)
        {
            if (topicName == null)
            {
                return null;
            }

            var topic = (includePrefix ? GetTopicPrefix() : string.Empty) + topicName;

            return Rules.Aggregate(topic, (current, rule) => current.Replace(rule.OldValue, rule.NewValue));
        }

Where RuleValue is:

 public class RuleValue
    {
        public string OldValue { get; }
        public string NewValue { get; }
        public RuleValue(string oldValue, string newValue)
        {
            OldValue = oldValue;
            NewValue = newValue;
        }
    }

CloudEvents format for messages

Adopt the CNCF Cloud events standard for messaging

https://cloudevents.io/
https://github.com/cloudevents/sdk-csharp

Proposed solution

FeatureFlag: CloudEvents compatible
Keep MessagingEnvelope; Add ToCloudEvent FromCloudEvent functions
Modify MessagingTransport to receive/return MessagingEnvelope instead of ReceiveContext/ SendContext
Add custom ProtocolBinding and Formatter for legacy message format or just use exiting code.
Use existing or develop Formatter/ ProtocolBinding for Stan.
For Rusi Transport, just add enough information related to Cloud Events and let Rusi do the envelope.

Introduce Projections in ES

Projections are a primary pattern you’ll likely use with Event Sourcing. It’s the answer to the most common question I get about event sourcing which is: How do you build out UI? If you have to replay all the events in an event stream to get to the current state, isn’t that very inefficient when it comes to displaying in a UI or reporting?

The answer, depending on your situation, could be yes. It could be very inefficient to have to replay an entire event stream to get to the current state.

This is where projections come in.

As a starter you can take a look at https://github.com/bolicd/practicalcqrs - it's simple and on point. It can be improved more, though.
Yet, there are a few open-source projects that use projections such as EventStore, Axon or EventFlow.

F# effect computation expressions

Move Effect builders in different modules and name both instances effect.
The default one (in namespace NBB.Core.Effects.FSharp) should be lazy.

Also, implement TryFinally and TryWith methods, see https://docs.microsoft.com/en-us/dotnet/fsharp/language-reference/computation-expressions#creating-a-new-type-of-computation-expression

NBB.ProcessManager documentation

Move documentation from NBB.docs to package readme.md, if any.
The documentation should also include samples in order to help anyone in the getting started process.

Process manager - introduce generic empty state

Not all processes require state. One can simply use the event it receives and create a new event based on it, without the need of a state.
Put as part of the code an empty state in order not to require the programmer to create an empty one.

Example:
public class EmptyState
{

}

And yes, make the state a class not a struct.

Refactor TenantDatabaseConfig Service

Refactor ITenantDatabaseConfigService.GetConnectionString to a more general ITenantConfiguration.GetValue which can provide any type of configuration

Scope validation error when registering NBB.Core.Effects

The problem is in the Effects composition root: https://github.com/osstotalsoft/nbb/blob/master/src/Core/NBB.Core.Effects/DependencyInjectionExtensions.cs

The singleton Sequenced.Handler injects the scoped IInterpretor.
Please check for other validation problems.

To enable scope validation:

.UseDefaultServiceProvider(
      fun context options ->
          options.ValidateScopes <- false
          options.ValidateOnBuild <- false
  )

Property Name from Enumeration class should be renamed to Value

throw new DomainException(ErrorCode.TaskFinnishedOrCancelled.Name);

I think the Value is more suitable for the client of this code. The client wants value from this enumeration, not a name.

throw new DomainException(ErrorCode.TaskFinnishedOrCancelled.Value);

Upgrade to Jetstream

JetStream was developed with the following goals in mind:

The system must be easy to configure and operate and be observable.

The system must be secure and operate well with NATS 2.0 security models.

The system must scale horizontally and be applicable to a high ingestion rate.

The system must support multiple use cases.

The system must self-heal and always be available.

The system must have an API that is closer to core NATS.

The system must allow NATS messages to be part of a stream as desired.

The system must display payload agnostic behavior.

The system must not have third party dependencies

You can view an implementation on this branch
https://github.com/osstotalsoft/nbb/tree/feature/Jetstream

Integrate with Rusi sidecar

Implement a Rusi publisher / subscriber

Provide a sample (inside of samples/Microservices folder) of how tu set up messaging with rusi

To be more specific, please provide the yaml files and dapr components I need to have and how to bootstrap them. (Pubsub: nats streaming).

event store - WithAdoNetEventRepository

please write some comments in the class so programmers know where to get the connection string from (appsettings..etc).

Inconsistent api for IMessageBusSubscriber.SubscribeAsync

Current contract:

Task SubscribeAsync(Func<MessagingEnvelope, Task> handler, CancellationToken token, string topicName = null, MessagingSubscriberOptions options = null);

Should be:
Task SubscribeAsync(Func<MessagingEnvelope, Task> handler, MessagingSubscriberOptions options = null, CancellationToken token =default(CancellationToken));

topicName should be included in MessagingSubscriberOptions

For usability reasons there should be an extension over the interface providing the method contract:
Task SubscribeAsync(Func<MessagingEnvelope, Task> handler, CancellationToken token =default(CancellationToken));

Fluent api Worker Builder

Currently it is very hard to manage the boilerplate for creating a messaging host worker project.
We should provide a builder that guides the user through a fluent api.
Check if we cand provide this functionality via extension methods over the HostBuilder allready provided by Microsoft.Extensions.Hosting

Also, check if we can to change the namespaces of different DependencyInjectionExtensions classes to Microsoft.Extensions.DependencyInjection so people don't have to open a lor of namespaces. Check wether this is a practice for different libraries

process manager - duplicate definitions

One can call services.AddProcessManagerDefinition multiple times or can have multiple definitions for the same type. Now, the runtime dies with
"Message of type XXX could not be processed due to the following exception System.InvalidOperationException: Sequence contains more than one element at System.Linq.ThrowHelper.ThrowMoreThanOneElementException() at System.Linq.Enumerable.SingleOrDefault[TSource](IEnumerable1 source) at NBB.ProcessManager.Runtime.ProcessExecutionCoordinator.Invoke"

Desired behaviour: similar with a DI container's Build process, the program should check for duplicate definitions after startup is complete (maybe also introduce a build function) and throw a more programmer friendly exception like "multiple definitions have been loaded for the same.. ".

However, it is to be discussed if multiple definitions for the same event should be supported, eg maybe I want to have a fork coming out of the same event. Example: an user is created, I want to start an email sending process and also start a flow where I check the person against an fraud internal database.

NBB.ProcessManager - dynamic dispatch for Emit/Apply

Leplace the dynamic dispatch of ES events in the emit - apply mechanism with a more resilient one.
Decide what to do with history events that do not match any Apply method.

WithAdoNetEventRepository - override

support custom connection string override.
example:

        services.AddEventStore()
            .WithNewtownsoftJsonEventStoreSeserializer()
            .WithAdoNetEventRepository(o=> {o.ConnectionString= ""; o.TopicPrefix = "";} );

ParseInternal - the God Method phenomenon

A "God Method" is a method that does way too many processes in the system and has grown beyond all logic to become The Method That Does Everything. When need for new processes increases suddenly some programmers realize: why should I create a new method for each processe if I can only add an if.
A large and complex method should be split in smaller methods, or even one or several classes can be created for that.
Large switch…case structures might be refactored through the help of a set of types that implement a common interface, the interface polymorphism playing the role of the switch cases tests.
Unit Tests can help: write tests for each method before extracting it to ensure you don't break functionality.