proophsoftware/event-machine

The world's only CQRS / ES framework that lets you pick your Flavour

event-machine's Issues

Use ImmutableRecord getters in toArray

For computed properties it is important that the ImmutableRecordLogic uses the getter methods instead of the property values in the toArray method.

Test if json schema for a message can be passed as string (path to json file)

Shorthand for message creation

EventMachine::dispatch should allow to pass message name as first argument and message payload as second argument instead of the message itself

Generate messagebox api docs from message json schemas

swagger would be preferred if compatible with json schema draft 4

Build messagebox explorer

The messagebox provides similar functionality to GraphQL in that it allows us to send both commands and queries to a single endpoint. Unlike GraphQL and GraphiQL, we don't have a good way to document and explore the capabilities of the messagebox. If we can build something similar to GraphiQL, though, we can have the benefits of an explorable API and use the full capabilities of JSON schema instead of losing some to support GraphQL.

I believe a first step that needs to be completed for this is to support functionality from JSON schema draft 7 (hyper schema) and to include all definitions in the messagebox-schema. This would allow for the schema to define the response types on the query objects.

Once the responses can be fully documented, we can build out a web interface to support parsing the schema and displaying it for the end user. Something like https://github.com/mozilla-services/react-jsonschema-form could provide a good basis for the interface.

Use PSR-Logger to collect metrics

A PSR-Logger should be used within different parts of Event Machine to log metrics

• Projector should log execution time of each projection
• Projector should log current stream position
• Logger should be attached to buses and log handled messages
• MessageBox should log received messages

It should be possible to filter out payload that must not be part of the log message e.g. user data or other critical information.

Remove query support

See #2 for reasons

Behat extension

Provide a feature context for behat (in a separate repo) that offers useful default definitions.

Convert JsonSchema to type language

Belongs to: #28

Type Language: http://graphql.org/learn/schema/#type-language

Test commands

Add a second arg to EventMachine::initialize() to turn on test mode. In test mode EventMachine replaces the event store with an InMemoryEventStore and does not publish events but instead return them to the caller:

EventMachine::testCommand(GenericJsonSchemaCommand $command): GenericJsonSchemaEvent[]

If test mode is enabled EventMachine will throw an exception if the normal dispatch method is used.

EventMachine CLI

• Run upcast
• Reset projections
• Consume filesystem queue

Document v1.0 Features

See milestone: https://github.com/proophsoftware/event-machine/milestone/1

• Swagger Integration
  • Retry of events using swagger (testing process managers)

Handle arbitrary read models

It should be possible to use arbitrary read models if default read models (aggregate, document, stream-to-stream) are not suitable for a feature. Example could be an elastic search read model.

Use zend-problem-details for error responses

See zendframework/zend-problem-details

ImmutableRecord VOs inside arrays are not converted

When ImmutableRecordLogic::toArray() is called it converts the entire state to an array which in this case will be persisted using the aggregate projection. There is however a problem when the ImmutableRecord contains an array of ValueObjects because they aren't converted but instead end up in the result as objects. This will lead to the read model containing something like this {"someData": [{},{},{}]} which is rather unexpected. My suggestion therefore is that event-machine should either loop through the array recursively and convert all VOs to arrays or forbid this and throw an exception. The second option would force the developer to create List VOs or explicitly convert the VOs to a native type inside the state.

Store definitions of all possible JsonSchema types in event machine

• pass them as definitions to each schema assertion: JsonSchemaAssertion->setDefinitions($eventMachine->jsonSchemaDefinitions);
• enable registering schema types: $eventMachine->registerType('MyType', JsonSchema::object(...));
• convert JsonSchema::defintion('MyType') to an inline referenz pointing to type defined with registerType and injected by event machine
• use storeDocumentsOfType() method as proxy to $eventMachine->registerType()

API

$eventMachine->watch(Stream::ofWriteModel())
 	->with('my_type_projection', MyTypeProjector::class)
	->storeDocumentsOfType('MyType', JsonSchema::object(
		[
			'requiredField' => [
				'type' => 'string', 
			], 
			'nullableRequiredFieled' => ['type' => ['string', 'null']
		], 
		[
			'optionalField' => ['type' => 'string']
		]));

Add more tests

• closure assertion for all cached config properties
• EventMachine::fromCachedConfig works with full featured set up
• messagebox works

Rm context from cmd and queries in GraphQL context

Using a dot or backslash as namespace separator for messages does not work well together with GraphQL and adding the context to the message name is not needed because the Context is defined by the GraphQL endpoint. In other cases like sending cmds/queries using a message broker it makes sense that the message name contains the context.

So ideally the context is removed from message name within the GraphQL schema but added again when the GraphQL query is translated to the real message.

Register query

$eventMachine->registerQuery('GetListOfMyType', JsonSchema::object([
    'limit' => ['type' => 'integer', 'default' => 10] //args as JsonSchema, Validation using JsonSchemaAssertion
])) 
	->resolveWith(MyListOfTypeResolver::class)
	->queryComplexity(10) //Optional for GraphQL
	->return(JsonSchema::array(JsonSchmea::definition('MyType')));

EventStreamWatcher as C Extension

Advanced projection is a C Extension (w/ Zephir) that runs event machine projections in parallel

See original issue: #22

JSON schema validation optimization

In the JustinRainbowJsonSchemaAssertion::assert implementation, we are currently forced to do something like the below to ensure empty objects and empty arrays are differentiated.

$enforcedObjectData = json_decode(json_encode($data));
$jsonSchema = json_decode(json_encode($jsonSchema));

With a large enough schema, this decode/encode pattern could become problematic. This isn't really a big issue right now, I think, but I wanted to get it documented.

80% Test Coverage

• set up travis
• set up coverals
• reach at least 80% test coverage

Add duplicate return type check to ReflectionBasedContainer

Register MessageWrapper

It should be possible to register a message wrapper.

This object would allow users to use their own message implementations.
Whenever event machine passes a message to userland it checks if a message wrapper was registered and calls $messageWrapper->wrap($emMessage)

If event machine receives a message from userlande code it checks if a message wrapper was registered and calls $messageWrapper->unwrap($customMessage)

Add method to load aggregate state

It would be useful for process managers to use event machine to load state of an aggregate so a process manager need not to rely on the read model.

EventMachine::loadAggregateState(string $aggregateType, string $aggregateId)

EventMachine can use the aggregate repository to get the aggregate and then call GenericAggregateRoot::currentState()

Add a republish method

Add a method to EventMachine that can republish an event.

EventMachine should reset the handled_async flag in event metadata and dispatch the event again on the internal event bus.

Use aggregate projections for snapshots

Implement handling of queries

Document limitations of JsonSchema integration

It is not possible to use advanced json schema for message payload validation like anyOf, oneOf, ...
because those validation patterns cannot be translated to GraphQL easily.

One example for the need of oneOf is that IF property X has the value Y then property Z must not be NULL.
A simple workaround is to allow NULL as an input arg and move the complex validation to a value object. This should be documented to avoid questions or frustration.

Configurable event stream

It should be possible to use an alternative event_stream name.

Allow setting an alternative name using an env var.

Generate GraphQL Schema

Event Machine should expose a method to generate GraphQL schema for:

• Document projections: #27
• Aggregate state: #21

yield a Tuple of [event name, payload] from aggregate

At the moment an aggregate method can only yield a predefined (set of) event(s) but different conditions can cause different events so the aggregate should be able to tell event machine what event should be recorded.

So all possible events and corresponding apply methods should be described upfront, but the aggregate can decide at runtime which events are recorded based on current aggregate state, command and defined conditions.

Consider making this the standard way to yield events. It increases readability because you don't have to check description to know which event is yielded by an aggregate method. (release a new dev version before breaking BC here!)

HTTP endpoints should use PSR-15

Wait until expressive 3.0 is available

Replace json schema validator with newer package

Allow user to use another event_stream name per aggregate (type)

Find a simple way to override default setting of event_stream name.

Set up aggregate repository in one place (currently we have two: CommandProcess, EventMachine::loadAggregateState())

A repository provider could do the job initialized by event machine and passed down to CommandProcessor.

Remove possibility to pass JsonSchema for a message as string

Schema files are not needed, because defining a schema with PHP arrays works like a charm and is the most performant way in PHP.

So the option to pass a filename to schema definition of message should be removed.

ImmutableRecordLogic calls fromInt instead of fromFloat

When the fraction points of the float inside a float VO happen to be zero it gets converted to an integer inside the JSON representation eg. ["prop1" => 3.2, "prop2" => 4.0] will result in {"prop1":3.2,"prop2":4}. When event machine now tries to replay the events to recreate the aggregate state it will encounter an error because it is going to call ::fromInt() for prop2 which will obviously fail since ::fromFloat() is the method that should be called instead. A simple solution might be to add a check like this here https://github.com/proophsoftware/event-machine/blob/master/src/Data/ImmutableRecordLogic.php#L295

return method_exists($type, 'fromInt')
    ? $type::fromInt($value)
    : $type::fromFloat($value);

What do you think @codeliner ? Is there a better solution?

EventMachine Aggregate Schema

• Optionally pass JsonSchema of aggregate state when aggregate is described for first time
• OR aggregate state implements JsonSchemaProvider (ImmutableRecord implements it by default, can be overridden)
  • Consider nullable properties
• Index definition can also be passed as optional parameter or DocumentIndexProvider is implemented
• JsonSchema is translated into GraphQL Schema to access aggregate state using GraphQL
• Auto Schema Versioning by creating a diff of old and new schema OR using a commit tag?

StreamToStream projections

Sub task of #22

• StreamReadModel is a wrapper around event store

Allow aggregate methods to yield NULL

if no new event should be recorded an aggregate method should be able to yield NULL (or similar) to tell event machine that no event should be recorded.

This is useful in case an aggregate detects a duplicate command and decides to ignore it instead of raising an error or recording an error event.

Support Upcasting

Define a good upcasting strategy.
On-the-fly upcasting is one option
but there should also be a way to easily project and upcast into a v2 stream:
It should be documented and validated if this scenario is supported by Event Machine PaaS.

Document projections

Sub task of: #22

• DocumentReadModel is a wrapper around simple document store

Message::get

• Add GenericJsonSchemaMessage::get(key): mixed
• Add GenericJsonSchemaMessage::getOrDefault(key, default): mixed

C-Extension for InMemoryDocumentStore

to keep documents in memory across requests

EventMachine Watch

$eventMachine->watch(Stream::ofWriteModel() || Stream::ofService('service_name', 'stream-projection-name')) //Resolved to service event stream, if no stream-projection-name given
    ->with('projection-name', Projector::class) 
    //DocumentProjector::__invoke($prjState, DocumentReadModel $readModel, Message $event)
    //StreamProjector::__invoke($prjState, StreamReadModel $readModel, Message $event)
    ->filterAggregateType('Aggregate\Type') //Optional
    ->filterEvents(array $listOfEventNames) //Optional
    ->storeDocumentsWithSchema(JsonSchema::object([...])); //JsonSchema of the stored read model documents
    //->storeAsStream(); //Alternative to ->storeDocumentsWithSchema()

• Simple implementation is a single prooph projection: EventStreamWatcher
• $prjState is the one provided by prooph
• DocumentReadModel is a wrapper around simple document store (see #27)
  • JsonSchema is translated into GraphQL Schema and projection read model can be queried by GraphQL
• StreamReadModel is a wrapper around event store (see #29)

Automatically add causation id

Detect dispatch sessions automatically by using causation id and nesting level

Each call to $eventMachine->on() should add a nesting level.
Each call to $eventMachine->dispatch() should remove a nesting level.

First call to $eventMachine->dispatch() should use message.uuid as causation id.

Nesting level + causation id should be set in message metadata.

Register type interface

$eventMachine->registerTypeInterface('InterfaceName', JsonSchema::object(...));

In json schema this becomes a definition like a normal type, but in GraphQL this becomes an interface that types can use.

To describe the dependency using JsonSchema one can do:

$eventMachine->registerType('TypeOfInterface', JsonSchema::allOf(
    JsonSchema::typeRef('Interfacename'),
    JsonSchema::object(....)
))

Belongs to: #28

Aggregate projection

Sub task of #22

Depends on #27

• Use DocumentStore abstraction from #27 to store aggregate state
• Move AggregateReadModel from skeleton to event machine. Enable projection by default.

EM watch http stream

Follow up of #22

$eventMachine->watch(Stream::ofService('service_name', 'stream-projection-name'))

Event Machine should use a ServiceResolver to resolve a service_name to a service stream base url
In DEV mode a simple ServiceResolver uses the service name directly and assumes that the service is available within the docker network. In production the service resolver should use Consul/etcd or a registry provided by cloud infra.

Roadmap Schema UI

Tasks are listed here: prooph/event-store-mgmt-ui#1
fyi @camuthig