Some more questions

[cloudcompute]

Hi @hnaderi

a. You have used the outbox pattern, which allows us to write domain events in one table and integration events in another (Outbox) table in a single transaction. Now I want an async process that monitors the outbox table for new entries and writes them to Kafka. As result, we have publishing guarantees. Our internal data storage structure (domain events) is not exposed since the integration events have their own schemas.

The first transactional process is called Edomata and the second async process is called Stomata.. am I right?

b. How does Edomata fit in the context of microservices where each microservice should have its own database.. in this case, each microservice will write its event of streams in its database. Right? Suppose we have an aggregate root Product which has children like Category, Tags. Pricing. So the entire data would be written in a single event stream?

c. There are two different table schemas.

Is this an event journal?

CREATE TABLE IF NOT EXISTS $table (
id uuid NOT NULL, "time" timestamptz NOT NULL, seqnr bigserial NOT NULL, "version" int8 NOT NULL, stream text NOT NULL, payload #${codec.oid.name} NOT NULL,
);

Outbox table? What are causation and correlation fields for?

CREATE TABLE IF NOT EXISTS $table(
seqnr bigserial NOT NULL, stream text NOT NULL, correlation text NULL, causation text NULL, payload #${codec.oid.name} NOT NULL, CONSTRAINT outbox_pk PRIMARY KEY (seqnr)
);

d. Does Edomata support Ad-hoc polymorphism (monads / type classes) out-of-the-box?

[hnaderi]

Hi @cloudcompute ,
A. Yes, there are two completely different set of messages, one is the events (as in event sourcing, which are internal events), the other one is notifications, which are external events that can be used as integration events. the latter uses outbox.
The rationale behind this separation is to separate the concerns, and guide the design; as these messages have different use cases and properties.
In your case, integration events that are stored in Kafka, are from notifications. So that, your application changes state by emitting events, and may or may not publish some notifications (which are for external uses only and doesn't have any effect on internal state and behavior). These actions are all done in a single transaction, so it's either done or not, and is also retried by default based on the back-end configuration.
One different process is reading the outbox to receive notifications, and writes what's necessary on Kafka.

B. Each Edomaton in event sourcing, or Stomaton in CQRS style state machines, models a single aggregate root in DDD terms. It encapsulates a domain and is also the transaction boundary. The entities and value objects belonging to the aggregate are just the state part of the state machine.
There is no limitation on how to use it in a single database or a separate database. Default storage drivers create separate postgres schemas for each domain model.
Each aggregate root is completely independent of other aggregate roots (whether its own type or other types).

Product which has children like Category, Tags. Pricing

In your example, if category, tag and pricing are entities or value objects in product, so it's just like fields of the case class that serves as the state, and all the events for that aggregate root are together in one journal.

C. Journal is where the events are stored, and Outbox is the place that stores the notifications.

What are causation and correlation fields for?

These fields are used to track the history of system changes and are commonly used in event sourced systems.
The use case is as follows:

• You send a command with id a
• It emits 2 events with different unique ids, but both have a as their causation and correlation.
• One of those events, causes a process manager to send a new command, each command has also a correlation and causation id you can set.
• If you set these metadata in the command you are sending, you have the complete causal chain for your system, so that you know which events caused which events, how and when.

D. I'm not sure what do you mean by that, can you elaborate more?
Basically, all the state machines in edomata are implemented as algebraic effects, and have all the instances that can be defined.
For example, all of the state machines implemented currently, have a monad instance and also an applicative instance.
Also, you can add your own storage codecs by providing the required instances yourself.

[cloudcompute]

@hnaderi, Sounds great !

A.

Default storage drivers create separate postgres schemas for each domain model.

So this means Skunk would read the Edomata's domain models (internal events) and create the tables accordingly? How does the schema look like, simple relational tables like we have in relational databases, for example separate tables for Product, Tags, Category, etc.

OR

"all the events for that aggregate root are together in one journal"

Do you mean to say that a single table would be there for the entire aggregate root, Product, in our case where one row of this single table stores the data for a single event, probably in JSON format.. simple hierarchical nesting which includes Tags, etc. data.

B.
Similarly, does the Outbox table has a schema (of external events) like this, as shown in the Outbox table below:

C.

I'm not sure what do you mean by that, can you elaborate more?

Being naive to functional programming, I am not sure whether I used the correct terms. But let me try to explain this by an example: We have a Pricing microservice, it can encapsulate a number of pricing calculation algorithms. I want that we should be able to extend Pricing by adding new behaviors at any point in time easily.

The service's pure-functions can take arbitrary number of inputs, for example discount could be there, a time range like Holidaying time is also optional may or may not be be there. And it returns a tuple or probably a simple numeric that holds the final price. Probably, the core Edomata has nothing to do with this requirement. But you know better, how this use case can be implemented.

D.
I hope that I'll be able use Zio Interop Cats for interacting with the outside world like Kafka.

E. Great if you can explain causation and correlation with a real-world use case.

Thank you very much.

[hnaderi]

A. The tables are created when the back-end is setting up.
The table structure differs for event-sourcing style and the CQRS one. In event-sourcing, the journal table contains all the events, each row is a unique event.
In CQRS style, you can either have a single table with jsonified states per row, or you can implement your own repository, which can be a normalized database schema. See this discussion for more info.

B. Outbox table in edomata is very similar to the example diagram in your reply.

C. These are logics implemented in the application layer when handling a command, there are no limitations in implementing logic when you use edomata.
Given that the command contains the information required for the user intent, you can leverage any technique that's applicable in your situation to handle the command and return the result type, and it doesn't need to be functional to work either. For instance you can implement it as a strategy pattern or command pattern as well.
Edomata only requires you to return its state machine type, so it can handle the rest.

D. Yes, you absolutely can. Using ZIO interop, you can read the outbox using ZIO streams, and handle the effects completely in ZIO. The project example contains a ZIO implementation that might be of your interest.

E. Sure!

1. Command received: id=1,payload=exchange
2. Event published id=e1, payload=deposit, causation=1, correlation=1
3. Event published id=e2, payload=withdrawn, causation=1, correlation=1
4. Some PM is listening to all withdrawn messages, and it receives e2
5. It then sends a command to another service
6. Command id=2, payload=update report, causation=1, correlation=e2

If we collect all the events, and look at their correlation and causation, we can build the causal chain, so for example we can understand that the reason for update report command in 6 was the event in 3 which was published due to the command 1 and so on.

[cloudcompute]

Wonderful replies.

I am sorry for asking so many questions, your guidance would help me put on track.

A. So for my microservices, I can use ZIO-gRPC (in Protobuf format) to let the data in and feed it to Edomata state machine. Right?

B. Edomata's Outbox table has 3 columns: seqnr, payload, and stream. Which of these columns correspond to which columns of the Outbox table listed in the image above?

C. I hope it won't be difficult to alter the implementation of Outbox and Journal tables.

For example, I want that minimal data should be entered in the payload column of an event row i.e. on the change that occurred, Edomata won't interfere whatever payload we decide to enter in the Journal table.

Or, in the outbox table want to add a column named global_sequence_number that will help Kafka (or its consumer) determine the correct order of events published by a producer.

D. Is PostgreSQL a good choice for event sourcing where they could be tens of millions of records in a single journal table when running in production? While Skunk is completely non-blocking (async), Doobie uses JDBC which is sync/blocking, correct? Skunk would be an obvious choice if performance is vital.

Alternatively, I can use ZIO, it has a support for Cassandra. I hope there would be a very loose coupling between the core edomata and its backend which will allow me introduce ZIO-Cassandra (or some other) easily.

E. You have implemented Snapshotting too which is quite tough. If using event-based microservices and Kafka as an event broker, snapshots are used by consumers so that they can quickly update their projections. In the microservices setup, a microservice cannot access the databases of others' microservices. So how can we pass these snapshots to other microservices, are you storing it in a separate Postgres table?

F. Does event sourcing support Optimistic Concurrency control, I think it'd be relying on the Postgres built-in OCC.

Regards

[hnaderi]

A. Yes, edomata exposes a single function that receives a command and handles the rest and gives back the response. You can receive the command from grpc, kafka, http or in any other way that suits your situations.

B. Seqnr has no alternative in that diagram, stream is AggregateId and payload is Payload.

C. Payloads are completely under the control of user. Edomata storage handles them as a blob of data. They are the serialized form of your events.
About global_sequence_number, that's exactly what seqnr is for, no need to add anything.

D. That completely depends on your specifics of the problem at hand. For event-sourced DDD applications it almost always is a good fit, as the number of events for each aggregate is bounded by some properties of the domain itself (most aggregate roots have limited lifecycle in real world), and while the total number of events in a journal can be tremendously high, indices, caches and snapshots make it unnoticeable under heavy use in production.

E. Snapshots are not meant to be used like that. they are a mere caching strategy and may or may not be persisted, let alone the fact that snapshots are aggregate state that should be completely encapsulated in each domain.

On the other hand, IMHO I think what you are talking about is a completely different kind of using events and snapshots, that is more of an event streaming approach rather than event-sourcing. I suggest you to see this talk to notice the difference. Some people use the term event-sourcing when using kafka and solutions in that ecosystem, but that's a completely misuse of the term that creates a lot of confusion, as event-sourcing is another old technique, used for at least 40 years in software design and tries to solve an unrelated problem to what kafka stream tries to solve.
You need to disambiguate which one of these approaches are right or better suited to your situation, as they are completely different approaches for two different problems.

F. Yes, it does. It implements a convergent update mechanism, protected by a version constraint on the database, it doesn't use any specific PG capabilities and is totally portable to other databases as well.

[cloudcompute]

Hi

I know that the event streaming and event sourcing are different things. Event sourcing has nothing to do with Kafka. CQRS can be achieved in either of the two ways:

a. with event streaming: in this case, producers just write events to Kafka and projections are typically handled by the consumers.

b. without event streaming: reads (projections) can be both synchronous and asynchronous. Snapshot is nothing but a special type of projection. I think the projections that you are doing are synchronous.

I will watch the video that you had shared.. Martin Fowler is a genius.

Thank you for your detailed response. I will get back to let you know about my proceedings relating to Event sourcing.. may take some time.