Added output handler to handle external processing in workflows

[oskardudycz]

This wraps up the last missing piece: how to call external services based on the workflow output (e.g. external Web API, queuing system, etc).

In the article @yreynhout showed the handle method that was just calling external APIs without returning anything. I think that's fine, but I wanted to make it a bit more accessible for people coming from non-EDA background.

I had two options:

• make workflow Async (so decide could be returning Promise),
• make output handler return message.

For now, I went with the former, and added possibly to pass outputHandler settings containing handle method that can return: nothing, single input, array of input.

I chose the second option instead of async workflow as the problem with async decide would create confusion. Currently, decide in a conceptually clear space. It takes state + input, returns outputs. The moment it can call an external API, it becomes an infrastructure function disguised as business logic. You lose the ability to unit-test it without mocking, replay it safely, or reason about it deterministically. Having separated output handler makes it clear that:

• decide is about business logic, can be unit tested,
• The output handler can be tested in integration separately.

The fact that the output handler handle method can return a message is helpful for observability (we see the result of the external call and have some sort of double-entry bookkeeping), and it also streamlines workflow processing. We don't need to provide an additional reactor to just handle external API routing. We can set it up here and return the input that will trigger the next step eventually.

If the output handler is set up in processor settings, then its handle method will be called separately from the regular decision logic, it'll be appended, and then processed once the consumer polls it.

The workflow

sequenceDiagram
    participant S as Event Store Stream
    participant P as Processor
    participant W as Workflow (decide)
    participant R as Output Handler (handle)
    participant E as External API

    S->>P: GuestCheckedOut<br/>(input type)
    P->>W: dispatch — it's an input
    W-->>S: append ReleaseRoomInPms<br/>(output, no input flag)

    S->>P: ReleaseRoomInPms<br/>(output type)
    P->>R: dispatch — not an input
    R->>E: POST /rooms/release
    E-->>R: 200 OK

    R-->>S: append PmsRoomReleased<br/>(input flag = true)

    S->>P: PmsRoomReleased<br/>(input flag = true)
    P->>W: dispatch — input flag set
    W-->>S: append nothing<br/>(workflow done)

Loading

Example:

const outputHandler = workflowOutputHandler<CheckoutInput, CheckoutOutput, ReleaseRoomInPms>({
  canHandle: ['ReleaseRoomInPms'],
  handle: (msg) => { 
    const { reservationId, roomId } = output.data;

      try {
        await pmsApi.releaseRoom({ reservationId, roomId });

        return {
          type: 'PmsRoomReleased',
          data: { reservationId, roomId, releasedAt: new Date() },
        };

      } catch (err) {
        return {
          type: 'PmsRoomReleaseFailed',
          data: {
            reservationId,
            roomId,
            reason: err instanceof Error ? err.message : 'Unknown error',
            failedAt: new Date(),
          },
        };
      }
    }
});

const processor = workflowProcessor({
  workflow: checkoutWorkflow,

  getWorkflowId: (event) =>
    event.type === 'GuestCheckedOut' ? event.data.reservationId : null,

  inputs: {
    commands: [],
    events: ['GuestCheckedOut', 'PmsRoomReleased', 'PmsRoomReleaseFailed'],
  },

  outputs: {
    commands: ['ReleaseRoomInPms'],
    events: [],
  },

  outputHandler,
});