keinsell / escentia

📚 Escentia is software design library that delivers ready-to-go set of data structures and abstractions to be used in software systems.

-----BEGIN PGP PUBLIC KEY BLOCK-----

xjMEYgokaRYJKwYBBAHaRw8BAQdAWjhCNiBbCIcOXErTneEdVDlfExR+Pz9R
kfWx+w3s2IDNMWtlaW5zZWxsQHByb3Rvbm1haWwuY29tIDxrZWluc2VsbEBw
cm90b25tYWlsLmNvbT7CjwQQFgoAIAUCYgokaQYLCQcIAwIEFQgKAgQWAgEA
AhkBAhsDAh4BACEJEEAtI2TafKU2FiEE+y7KjXkUNdxoPYDRQC0jZNp8pTbG
pgEA6b90m1KWMaDQUWCFlaAcLlMWmhaZvJnvq+Ib1Fgm3ncA/i5mTpBDmyW/
veULeGufAr5gkQ4guJ/GmjH9S41+KNYCzjgEYgokaRIKKwYBBAGXVQEFAQEH
QGK6WibCD9WiaLeb5S/JbKqvYrP+8VGEKoWNMf/pkjQCAwEIB8J4BBgWCAAJ
BQJiCiRpAhsMACEJEEAtI2TafKU2FiEE+y7KjXkUNdxoPYDRQC0jZNp8pTbx
jAD/XWkgaKDmUqu9xLdy49IXwO4GYAfyapu9TDGHx5a1Nv0A/1VRAY5Iwvra
hyjUQ7Zrs6uxKvOz47A7u+71Gu4+AZAE
=aRAo
-----END PGP PUBLIC KEY BLOCK-----

escentia's Issues

Managing aggregate state with Finite State Machine (FSM)

Aggregate can increase version automatically

[escentia] src/domain-modeling/aggregate-root.ts (Lines 9-33)

export abstract class AggregateRoot<ENTITY extends Entity<UniqueIdentifier>> {
  protected readonly _id: ENTITY["_id"]
  private _version: SequentialId
  private _events: DomainEvent<AggregateRoot<ENTITY>>[] = []
  protected root: ENTITY

  protected constructor(root: ENTITY, version?: SequentialId) {
    this._id = root._id
    this._version = version || sequentialId(0)
    this.root = root
  }

  protected incrementVersion(): void {
    this._version++
  }

  addEvent<T extends DomainEvent<AggregateRoot<ENTITY>>>(event: T): void {
    this._events.push(event)
  }

  get events(): DomainEvent<AggregateRoot<ENTITY>>[] {
    return this._events
  }
}

Open in IDE · Open on GitHub

^{Created from JetBrains using CodeStream}

Add `AsyncSpecification`

I'm not kinda sure this is a viable design pattern or if it exists under different name, but I find this useful to specify ex. IsUserInDatabase to have specific check that can be used in multiple places.

https://www.npmjs.com/package/bspec

Channel should have registry of subscriptions

In case where we have Queue which should most likely one subscriber there we should not allow for adding more than one subscriber or actually take it in constructor. Not sure there Subscriber or Handler should be used.

[escentia] src/messaging/channel.ts (Lines 17-64)

export interface ChannelConfiguration {
	serialization?: {
		serializer: any
		deserializer: any
	}
	schedulingMethod?: SchedulingAlgorithm
	scheduling?: MessageScheduling
}

/** Channels, also known as topics, queues, or exchanges */
export abstract class Channel<M extends Message> {
	public readonly _name: string = kebabSpace(this.constructor.name)
	public readonly _type: ChannelType = ChannelType.DATATYPE
	public readonly _schedulingAlgorithm: SchedulingAlgorithm | undefined

	constructor(
		protected readonly broker: Broker<any>,
		protected readonly configuration?: ChannelConfiguration
	) {
		this._schedulingAlgorithm = configuration?.schedulingMethod

		if (configuration?.scheduling) {
			this._schedulingAlgorithm = configuration.scheduling.type
		}
	}

	async publish<T extends M>(message: T): Promise<void> {
		this.broker.publish(this._name, this.serialize(message))
	}

	async subscribe(subscriber: Subscriber): Promise<void> {
		this.broker.subscribe(this._name, subscriber)
	}

	async unsubscribe(subscriber: Subscriber): Promise<void> {
		this.broker.unsubscribe(this._name, subscriber)
	}

	protected serialize(message: unknown) {
		let formattedMessage: unknown = message

		if (this.configuration?.serialization) {
			formattedMessage = this.configuration.serialization.serializer(message)
		}

		return formattedMessage
	}
}

Open in IDE · Open on GitHub

^{Created from JetBrains using CodeStream}

How to implement serialization based on `protobuf`?

Package for standalone serivice

It would be good move to deliver a standalone package that would be aggregate all commands, events, views and notifications from specific software domain/infrastructure and will expose them through easy to use HTTP, GraphQL and/or gRPC interfaces.

Add `Router`

A component that routes messages based on predefined rules or conditions. (Content-Based Routing)

`Agggregate` Reqiurement Specification

• Aggregate is and uses Entity class
• Entity should track it's version by events happened
• Aggregate should annotate methods run on commands
• Aggregate should annotate methods run when replying events
• Aggregate must be identified by Contextual ID
• Aggregate must have unique identifier
• Aggregate must have methods to commit events
• Aggregate must have methods to revert events

Add support for Workflows and Sagas

How to perform Aggregate snapshots?

In Event Sourced approach there may be need for snapshots of aggregates, I wonder if there should be any half-automated solution for such or just developers should attach their own method for building snapshot.

Add `Queue` class

Add support for building Queue which will allow injection of different scheduling algorithms.

Do channel should be registred in broker or broker should be injected in channel?

I wonder about one thing, when there's a Channel and there's Broker - of course Broker is responsible for sending out messages into another component, however we're publishing messages to Channel not a broker.

new Channel(broker)
Channel.constructor { broker.registerChannel(this) }

new Channel(broker)
Channel.publish { broker.publish }

Add Event Store

Add additional abstraction that will allow to store event chain on specific aggregate and then fetch same chain from store.

Potential Resources

• https://www.npmjs.com/package/@avanzu/eventstore
• https://www.npmjs.com/package/eventstore

Library should be toolchain-agnostic

• For persistence we should allow users to use typeorm as well prisma, these options seems to most popular, however Model class may seems to be good if user of library will model by himself.

Do there should be `MessageRegistry` for all kinds of messages or separated ones per message's type?

Add Event Replay / Projection

Literally everybody in this ecosystem did it wrong, it can be done easily on decorators to project all of the events on aggregate with version tracking - probably will be first one to do it easy in typescript.

Add `Signal` abstraction

Choreography & Orchestration

Library should have support as well for choreography and orchestraction

Delete `@Serializable()` decorator and related functionality

Depending on the approach, I think extending main Message class seems to be more natural and flexible than actually overinjecting properties.

Add `Actor` abstraction

Actor class typically refers to a concept from the actor model of concurrent computation. The actor model is a theoretical framework for designing and implementing concurrent and distributed systems, where actors are the fundamental units of computation.

• https://github.com/cqrsfk/core/blob/master/src/Actor.ts

Merge `MessageSerializer` and `MessageDeserializer` into class called `Serialization`

Do `MessageDeserializer` should take addional metadata beside serialized bytes?

I wonder about one thing, if we have serialized message we're not really supposed to serialize everything if message broker have support for message headers - and if there are data passed to headers we need to attach them back to message.

Alternatively, we can put metadata in payload and headers 🤷🏻

Add `Endpoint`

A component that represents the source or destination of a message.

Add `Producer`

Define Producer and Publisher

Add MD5 Hash generation from `Query`'s payload

MD5 Hash by query's payload will allow to identify same queries and hit their result from cache for example.

Add `Aggregator`

A component that combines multiple messages into a single message.

Do Query should be a Message?

// TODO: Do query should be a message?


export abstract class Query<RESPONSE> {
	public readonly _response: RESPONSE = undefined as unknown as RESPONSE
}

^{Created from JetBrains using CodeStream}

How do Point-to-Point works with Queue?

Add `Transformer`

A component that transforms messages from one format to another.

Add Lifecycle Hooks

Optimally every class should receive hooks that will allow to better logic separation, for example beforeHandle in case of Handler class which may be useful for logging on some shit.

Bus Hooks

• beforeSend
• beforePublish
• afterReceive
• beforeDispatch
• afterDispatch
• onError

Add repository dedicated for file management

Repository that supports buffers or streams

Add `ulid` unique identifier

https://github.com/keinsell/neuronek/tree/trunk/apps/server/src/shared/%40foundry/indexing/ulid

Automatically generate `x-host` message header

This header should clearly tell from which IP message was sent

Add `cuid` unique identifier

https://github.com/keinsell/neuronek/blob/trunk/apps/server/src/shared/%40foundry/indexing/cuid.ts

Add `Splitter`

A component that splits a message into multiple messages.

Add `nanoid` unique identifier

https://github.com/keinsell/neuronek/tree/trunk/apps/server/src/shared/%40foundry/indexing/nanoid

Research `MediatR`

https://github.com/jbogard/MediatR

How to process queue in other components?

[escentia] src/scheduling/queue.ts (Lines 17-44)

export abstract class Queue<T = unknown> {
	protected readonly capacity: number
	protected readonly store: Store<QueuePosition<T>>
	protected readonly scheduling: MessageScheduling

	constructor(configuration: {
		capacity?: number
		store: Store<QueuePosition<T>>
		scheduling: MessageScheduling
	}) {
		this.capacity = configuration.capacity || 0
		this.store = configuration.store
		this.scheduling = configuration.scheduling
	}

	abstract enqueue<T>(message: T, details?: PositionMetadata): Promise<void>

	abstract dequeue<T>(batchSize?: number): Promise<T[]>

	abstract peek<T>(): Promise<QueuePosition<T> | undefined>

	abstract isEmpty(): Promise<boolean>

	abstract clear(): Promise<void>

	abstract size(): Promise<number>
}

Open in IDE · Open on GitHub

^{Created from JetBrains using CodeStream}

Integration with RabbitMQ

Messaging should have support for RabbitMQ as solution for message transport in message bus, this should also have implementation as Message Broker.

The so called „Transport” is underlying Message Broker in terms of Enterprise Integration Patterns.

Prepare `example` directory with two contexts communicating with each other

Add `Filter`

A component that selectively filters messages based on specified criteria.

Research `Seedwork`

https://learn.microsoft.com/pl-pl/dotnet/architecture/microservices/microservice-ddd-cqrs-patterns/seedwork-domain-model-base-classes-interfaces

Add `Processor`

There's a Queue however, Queue itself is just data structure and we need Processor class that will poll for specific queue and will take actions such as publishing in terms of PointToPointChannel.

Add `Consumer`

Add Consumer