tomferon / eventsourcing Goto Github PK

To make the type signatures terser.

At least from Database.CQRS.PostgreSQL.Projection.

Helpers to create a global registry (possibly sharded) of public identifiers such that we can go from a pair of a public identifier and a stream family to a fully qualified stream identifier (FQSI) containing a database pool, a shard ID, the stream family and a stream identifier.

From a FQSI, we have all the information required to actually find the corresponding stream.

Public identifiers could have a form such as "[shard]:[UUIDv4]" while FQSI could be "[pool name]:[shard number]:[stream family name]:[integer ID]"..

If a projection produces actions of type Action that, for example, get turned into SQL statements that are sent to a database.
If we have a read model that is essentially equivalent to streamId -> m ProjectedState.
If we have a function of type ProjectedState -> Action -> ProjectedState.

Then we can have a helper that:

• gets the projected state from the read model and the identifier of the last event and the internal state of the projection from the corresponding tracking table,
• fetches any new event for that stream,
• transforms the new events into actions with the projection and the internal state (which can be discarded after that),
• applies the actions to the projected state and
• return the final projected state.

This can be used when we want to get the latest version of something without having to wait for the projection to catch up in the background and when using AggregateStore is not appropriate.

The user would have to be careful that applying the action to the projected state in memory yields the same result than applying the transformed action to a database. Maybe some helper functions could help write tests for this.

Support for projecting streams to key-value pairs in Redis.

Add new library eventsourcing-cli to create command-line tools that can be used to replay aggregators/projections on event streams interactively (thing gdb-style).

Some feedback I received:

[...] possibly some type aliases / newtype wrappers would make the code more readable I think, take

data StreamFamily streamId eventId metadata event = StreamFamily
{ connectionPool :: forall a. (PG.Connection -> IO a) -> IO a
, relation :: PG.Query
, notificationChannel :: PG.Query
, parseNotification :: BS.ByteString -> Either String (streamId, eventId)
, streamIdentifierColumn :: PG.Query
, eventIdentifierColumn :: PG.Query
, metadataColumns :: [PG.Query]
, eventColumn :: PG.Query
}

all those PG.Query don't give much info, you need to look at the selector to understand their use (disclaimer: I like to look at type name more than anything else so I'm biased here). eventColumn maybe can be a bit clearer e.g. eventDataColumn / dataColumn or the likes?

There could be a library eventsourcing-debugging or something that provides helpers to serve the internal state of the program on some port with HTTP, e.g. the aggregates contained in an aggregate store.

Add a 'ShardedStreamFamily' which groups different stream families into one and dispatches to one of the underlying stream family based on the stream identifier.

Given an aggregator of type Aggregator event st ~ event -> State st (), a projection can be derived of type Projection event st Void ~ event -> State st [Void] since () is equivalent to [Void].

Together with TopUpReadModel, it enables the aggregated state to be pre-computed by the derived projection.