Create more robust "Event Bus"

[brent-hoover]

Currently we have a well-thought out and generally universally implemented "appEvent" system that allows for decoupling of components via event emitters and event listeners. This system has a few limitations that this work seeks to overcome.

1. Currently arguments passed via emitters are not validated and have no standard format. It is easy to forget an argument and pass the wrong data to a listener without even knowing it.
2. Currently appEvents just go into internal memory. If a large number of appEvents are emitted and a pod were to go down those events would be lost. Also this means that a single pod would need to handle all the work for all the listeners rather than being able to distribute this work across the cluster.
3. These events are only available to processes within Reaction itself. If an external process were to want to listen to events it would need to write a custom plugin to listen and act upon these events.

So the solution would be twofold:

1. Allow you to optionally attach a schema to an event type that all event emitters would need to conform to. The obvious validation tool for this would be simple-schema but open to other ideas that aren't TypeScript.
2. Events would be emitted along some sort of pub/sub system. The obvious choice now that Redis is in the stack would be Redis, but it should be written with adapters for Redis and Kafka and ability to write an adapter for any other type of event bus. This should be written in a pub/sub sort of way so that external process could subscribe to an event and get those message while not removing them for other listeners.
3. Optionally we might want to have a way to register a webhook, so that external systems could subscribe to an event with a URL and have it hit with information from that event.

One of the big challenges is going to be security. How does one authorize external systems to access some events and not others? Would this be a an API key? If so how to we create and manage those.

I think this is a good first step in our plan for improve the way we move data in and out of our system while also making internal processes more robust.

Note that the first step of this work would be creating a proposal for how this would all work that would be codified as an ADR

cc/: @tedraykov @aldeed

[tedraykov]

The first and non-breaking change would be to allow to define the schema of app events during startup. The registerPlugin function already has a property called appEvents which allows configuring the app events. We can extend its API by adding a schemas property:

export default async function register(app) {
  await app.registerPlugin({
    label: "Orders",
    name: "orders",
    version: pkg.version,
    functionsByType: {
      startup: [startup]
      ...
    },
    appEvents: {
      schemas: eventSchemas  // <-----
    },
    simpleSchemas: {
      Order,
      ...
    }
  });
}

The easiest route is to continue using simpl-schema as we can use the same simpl-schema definitions we already have for the domain objects.
For example for the afterOrderCreate, the schema for the event payload might look like this:

import { Order } from "./simpleSchemas.js";

const AfterOrderCreateSchema = new SimpleSchema({
  createdBy: {
    type: String
  },
  order: {
    type: Order
  }
});

export default async function register(app) {
  await app.registerPlugin({
    name: "orders",
    ...
    appEvents: {
      schemas: {
        // Here the key is the app event name and the value is the schema
        // We can provide a more detailed interface if we need to pass some extra information if the future like so:
        // {
        //   name: "afterOrderCreate",
        //   schema: AfterOrderCreateSchema
        // }
        afterOrderCreate: AfterOrderCreateSchema
      }
    },
    simpleSchemas: {
      Order,
      ...
    }
  });
}

This way we will use the same instance of the Order SimpleSchema that will be shared across the app and also be extended by other plugins. For example, when the promotions plugin extends the Order schema, all app event schemas that contain the Order schema would also be extended as well because they share the same instance.

[tedraykov]

Regarding point 1 of the solution proposal:
I would even suggest making the event schemas required rather than optional or at least highly suggested to have schemas. My reasoning about that comes from the work we did on the webhooks plugin that exposes the app event we did for the Merchstack integration. A problem we faced there was we didn't know what events to listen to.

There is no way to list supported events at startup so we had to go through all plugins and find all the appEvents.emit(*eventName*) occurrences and created a static list of supported events. This wasn't enough because later we realized there are also custom events emitted by custom plugins so we had to go back through all custom plugins and extract all events from there as well.
If each app event has a schema, we can build the list of all events at startup and then the events would be discoverable by the webhooks plugin, rather than hardcoding them. If an app event doesn't have a schema attached to it, it won't be discovered by the webhooks plugin.

Regarding point 2 of the solution proposal:
It is a great idea to support high availability of the emitted events and distribute the load in a multi-instance scenario by switching from in-memory to Redis but I have a concern with exposing the Pub/Sub system to external consumers.

Let's say we have a custom plugin that imports pricing for products and emits an event when each product is updated. We might want to automatically publish the product with the updated prices in the catalog. This is possible by listening to the afterProductUpdated event and triggering the Publish Product workflow.

An important note here is that we want only one of the reaction instances in a multi-instance scenario to catch this event because if the event is distributed to all pods by first sending it to Redis, this will cause all pods to catch the event and trigger the publishing. From this scenario, we can conclude that when an event is emitted, only the first consumer should be able to process the event. In the case of Redis, we should configure an event to be consumed only by the first consumer.

If we do that, we won't be able to consume the events from the external system by connecting to Redis directly because it might "steal" an event that is intended to be consumed by Reaction itself.

That's why I think even if we switch to Redis, we should only consume the Redis events in reaction and use the webhooks plugin proposed in point 3 to allow external systems to listen to events. Unless there is a way to configure Redis to send events to only one of the reaction pods and also to any other system.

[brent-hoover]

Great comments @tedraykov
As discussed at the arch meeting I think we should emit two streams: An internal Queue so that only one pod would consume the event, and then a PubSub for external consumption as we don't want/need to control what people do. Also there would need a layer in there to provide security and another layer of curation (Maybe we would want to remove certain fields?)

Still open on what that final API subscription layer would look like. Makes sense to not make the raw PubSub available (which we can't for security reason) but I know I don't want to do something clumsy like Webhooks