Love and hate the pm

[codeliner]

"Love and hate the pm" This comment by @lunetics caused a discussion about process managers and their role in an event sourced system in the prooph/improoph chat.

@bweston92 has provided his implementation of a process manager being an event sourced aggregate.

This issue tracks the discussion so that we can point to it later.

[codeliner]

@bweston92 's implementation

along with explanation:

It is a PM that's root AR is a "lead" creates various parts of an application and tracks the opening of the application at the end
And on failures sends an e-mail

So, I will accept via HTTP a message RecordBusinessLoanLead, this will include all the information about the applicant, legal entity and business loan requirements along with the introducers id. This will then open up a new AR BusinessLoanLead and emit an event BusinessLoanLeadRecorded which will be emitted over AMQP and return 202 accepted.

Now the AMQP consumer for the leads service will consume BusinessLoanLeadRecorded in the process manager, it will locate the AR and call createApplicantForBusinessLoanLead($stateResolver, $commandBus) this will get the information about the lead from the state resolver and using the command bus dispatch a message CreateApplicant this will then because that it a HTTP call underneath will succeed or throw an exception, in which case I can try again depending on the error and eventually emit either a command RecordLeadApplicantCreated or RecordLeadApplicantCreationFailed.

Now the AMQP consumer for the leads service will consume RecordLeadApplicantCreated in the process manager, it will locate the AR and call createLegalEntityForBusinessLoanLead($stateResolver, $commandBus) this will get the information about the lead from the state resolver and using the command bus dispatch a message CreateLegalEntity this will then because that it a HTTP call underneath will succeed or throw an exception, in which case I can try again depending on the error and eventually emit either a command RecordLeadLegalEntityCreated or RecordLeadLegalEntityCreationFailed.

This will then happen for both opening the business loan application and tracking the source of the application.

However explaining this just made me realise my diagram is a bit wrong but hey that is the detail

[codeliner]

@bweston92

The command bus will be injected into the AR

Can you explain me why you need to inject the command bus into an AR?

[codeliner]

@bweston92 uses prooph/micro which is good to take a look on how you would work with a process manager in a functional world.

@bweston92

Injecting the command bus is the easy part, validating my logic on doing this is what I need to do

Sascha-Oliver Prolic @prolic 12:42

however I asked about where you inject 😀

Bradley Weston @bweston92 12:42

So the following is called by the main PM after receiving BusinessLoanRecorded is will dispatch
attemptToCreateApplicantForLead which calls a command with ApplicantCreator interface.

function createApplicant(callable $stateResolver, Command\AttemptToCreateApplicantForLead $command, ApplicantCreator $applicantCreator): array {
        $state = $stateResolver();
        $attempts = 0;

        try {
            do {
                $response = $applicantCreator->create($state['applicant']);

                if ($response->isSuccessful()) {
                    break;
                }

                $retry = $response->canRetry() and $attempts++ < 5;
            } while($retry);

            if ($response->isSuccessful()) {
                return [
                    Event\LeadApplicantWasCreated::withData(
                        LeadId::fromString($state['lead_id']),
                        nextVersion($state)
                    ),
                ];
            } else {
                return [
                    Event\FailedToCreateApplicant::withData(
                        LeadId::fromString($state['lead_id']),
                        $response->errorMessage(),
                        $attempts,
                        nextVersion($state)
                    ),
                ];
            }
        } catch (\Throwable $unknownError) {
            return [
                Event\FailedToCreateApplicant::withData(
                    LeadId::fromString($state['lead_id']),
                    $unknownError->getMessage(),
                    $attempts,
                    nextVersion($state)
                ),
            ];
        }
    }

Sascha-Oliver Prolic @prolic 12:44
this is for micro right?
this looks great

Bradley Weston @bweston92 12:44
Yes
Then I have this in my dispatcher

    'Leads.AttemptToCreateApplicantForLead' => [
        'handler' => function (callable $stateResolver, Message $message) use ($factories): array {
            return \Cosmic\Lead\Model\createApplicant($stateResolver, $message, $factories['applicant_creator']);
        },
        'definition' => LeadAggregateDefinition::class,
    ],

[bweston92]

@codeliner so basically I was using the CommandBus as a "service" for dispatching my commands as I'm using an AR to be a PM.

[bweston92]

However I removed the command bus idea and injected an ApplicantCreator as you can see in the createApplicantForLead AR method. (that you've got as createApplicant)

[prolic]

For prooph/micro I think the solution is very very easy, as you only got to do this:
f ( f (history) -> state, event ) -> event[]

For a process manager that is supposed to send new commands, you can do something like this:

function (callable $stateResolver, Message $event) use ($commandBus) {
    $state = $stateResolver();
    if ($state['something'] === true) {
        $command = new SomeCommand($event->someThing());
        $commandBus->dispatch($command);
    }

    return [];
}

Now when using prooph/event-sourcing things get a little more complicated:

I need to have access to the command bus to send new commands (this could be done with the aggregate repository, which would need a custom repository factory).

Maybe prooph could add some infrastructure to support event sourced process managers with the event-sourcing component.

When using prooph/event-store projections, things are a little easier again, but still not as easy as with micro:

example:

$projection = $projectionManager->createModelProjection('some name');

$projection
    ->fromStream('event_stream')
    ->when([
        'someEvent' => function ($state, Message $event) use ($commandBus): array {
            if ($state['something'] === true) {
                $command = new SomeCommand($event->someThing());
                $commandBus->dispatch($command);
            }

            return $state;
        },
        'someOtherEvent' => function ($state, Message $event): array {
            $state['someting'] = true;

            return $state;
        }
    ])
    ->run();

Last but not least, there is the option to use custom persistence process managers, where the state is persisted somewhere else (f.e. mongodb): There is no infrastructure at all to support this atm.

Summary:
There are at least 4 options on how to implement sages (aka stateful process managers) with prooph:
a) using micro
b) using event-sourcing
c) using event-store projections
d) using custom state persistence

In micro things are so easy, as you only need to write your function, that some example code is enough. For all the others, we might need to have a discussion, on how the best way would be to implement those and maybe add some infrastructure supporting this in the prooph ecosystem.

[prolic]

Another note for event-sourcing:

We need to have a mechanism, that fetches the correct aggregate and passes it to an event handler, so something like this:

$handler = function (SomeThingHappened $event) use ($aggregateRepository): void {
    $ar = $aggregateRepository->get($event->someId());
    $ar->onEvent($event);
}

$eventRouter->route(SomeThingHappened::class)->to($handler);

Maybe it's not easy to come up with a general abstraction here, but at least some example code would be nice, so we can point to something when the question comes up again next time.

[codeliner]

The question is: Why do we need the aggregate state in the process manager?

The aggregate handles invariantes and therefor uses its state to make decisions. Like shown by @prolic an aggregate function type looks like this:

f ( f (history) -> state, event ) -> event[]

Following the discussion here an "aggregate function" should not have side effects. The function type above underlines this. No dependencies, just current state (left fold of past events) as input and new events (effects) as output. This can be tested easily in isolation and will always produce the same output when given the same input. Great!

Now let's look at a possible function type of a stateless process manager:

f (event, sideEffect) -> Maybe success

So the process manager handles or performs side effects (I/O, send message to messaging system, send new command to aggregate, ...) but it is itself stateless. All it needs is an event to react on. The event is the effect and the process manager executes that effect with an output of Maybe success or Result = Error | Success.

This means that aggregate and process manager form a team. The aggregate tells the outside world (where side effects are performed): SomethingHappened
The outside world (where process managers live): react on SomethingHappened by changing the outside world. Once done they "send back" a new message to the aggregate (a new command) telling the aggregate that the outside world has changed so the aggregate should change, too.

Let's take @bweston92 's implementation.

//domain model
function openBusinessLoanLead(callable $resloveState, RecordBusinessLoanLead $command): array
{
  //...
  return [BusinessLoanLeadRecorded::with(/* ... */)];
}

//process manager
function createApplicantForBusinessLoanLead(BusinessLoanLeadRecorded $event, AmqpChannel $channel, int $retries = 0): array
{
  try {
    $channel->connect();
    $channel->publish(CreateApplicant::fromEvent($event));
  } catch (\Throwable $error) {
    if(canRetry($error, $retries)) {
      sleep(5);      
      return createApplicantForBusinessLoanLead($event, $channel, ++$retries);
    } else {
      return ["Failed to send command ...", false];
    }
  }
    
  return [null, true];
}

function canRetry(\Throwable $error, int $retries): bool {
  return $error instanceof AmqpChannelNotAvailable && $retries < 5;
}

Let's assume creating the Applicant is handled in another bounded context and once it is done we get an event back from that system.

We have again a simple process manager to react on that event:

//process manager
function trackRecordLeadApplicantCreated(RecordLeadApplicantCreated $event, BusinessLoan\CommandBus $commandBus, int $retries = 0): array
{
  try {
    $commandBus->dispatch(TrackRecordLeadApplicantCreated::fromEvent($event));
  } catch (\Throwable $error) {
    if(canRetry($error, $retries)) {
      sleep(5);
      return createApplicantForBusinessLoanLead($event, $commandBus, ++$retries);
    } else {
      return ["Failed to send command ...", false];
    }
  }
    
  return [null, true];
}

function canRetry(\Throwable $error, int $retries): bool {
  return $error->code >= 500 && $retries < 5;
}

In the second part of the example you can see that I'm sending a new command to BusinessLoanLead aggregate. I'm telling the aggregate that it should "track" progress of the process, but I still use the process mangers to perform side effects.

The point here is that you can use an aggregate to "track" progress of a business process but you should use process managers to perform side effects.

You can now set up a monitoring process manager that observes the event stream of BusinessLoanLead to make sure that the business process is finished within a defined time frame to make sure that the final goal is reached:

It is a PM that's root AR is a "lead" creates various parts of an application and tracks the opening of the application at the end

The monitoring process manager is a bonus. Maybe it is enough to just show progress in the UI (a projection is constantly updating the read model) or send the admin team emails in case of errors. Whatever works for your system.

The point here is, that it maybe looks like overhead but it is a simple set of rules, no magic or any other crazy stuff involved.

[codeliner]

Addition: I've used functions in the example to avoid boilerplate, but the OO version would pretty much look the same just with classes. It is the same simple concept.

[lunetics]

What i found very clear for understanding was that:
https://github.com/justeat/ProcessManager

So you have 1 ProcessRouter which actually is responsible for retrieving the state of the process manager instance and decoupled the "logic" of the Process manager.

Also looking of having such stuff decoupled from the event-store (lot of people don't use ES)