Refactor the notifiers logic and plan for connectors

[supabase]

Refactor the notifiers logic and plan for connectors

[fracek]

It's one I wanted to discuss more too. At the moment we are only filtering on equality. It would be good to throw around some ideas how we can do more/better. Let's discuss on slack or in a new issue

We can discuss it in a separate issue. For now I will stub the transaction filter module with just the basics.

[fracek]

I think this one was fixed by #36. You can add connectors by adding a new genserver and notify it from the replication genserver.

[kiwicopple]

Great - remaining questions:

• let's say I create a genserver called "connectors" which stores all the different locations that a customer wants to send the events too
• I create a genserver called "webhook" which will POST the event to a URL

As a customer, I want to start the server and have it send the events to 2 endpoints:

• https://webhook.site/44e4457a-77ae-4758-8d52-17efdf484853
• https://webhook.site/c451c09e-6363-45fe-aeff-9cf577681058

How would I pass those to the server? Env vars don't seem like a good way, as it needs to be an array (and other connectors might need additional config).

Ideas:

• Maybe some sort of JSON file that the "connectors" genserver can read from? To update the sources/connectors we can then just restart the "connectors" genserver which will read the file again?

[fracek]

One option is to use config providers in releases.exs and add a provider for the configuration.

Another (and IMHO better) option is to use an agent to hold the configuration. At the beginnig you can keep it simple (simply read from a json configuration file), but can be easily improved by adding configuration through HTTP endpoints/consul etc. The important thing is that all processes read the configuration from the agent and subscribe to be notified when the configuration changes so that the data they need.

[kiwicopple]

I like your option 2 - the use of an agent. Especially as we plan for customers to keep dynamically adding connectors via our dashboard.

I agree with the simple start, and probably always having an option for something as simple as a JSON file to make the install more approachable for devs.

I've never used Consul, but I get the idea with the agent/subscriber model. Perhaps something we are going to need to do on a few of our repos actually.

I'll keep this task open and let's scope it done:

• create a "connectors" genserver which reads from a JSON file
• create a "webhooks" genserver that POSTs events
• when an event is received, the "connectors" genserver sends to all relevant subscribers

I imagine the JSON will be something like:

[
  {
    type: "webhook",
    event: "*", // all events
    relation: "*", // all changes
    config: {
      endpoint: "https://webhook.site/44e4457a-77ae-4758-8d52-17efdf484853",
    },
  },
  {
    type: "webhook",
    event: "INSERT",
    relation: "public", // only the public schema
    config: {
      endpoint: "https://webhook.site/44e4457a-77ae-4758-8d52-17efdf484853",
    },
  },
  {
    type: "webhook",
    event: "UPDATE",
    relation: "public:users.id=eq.2", // specific row
    config: {
      endpoint: "https://webhook.site/44e4457a-77ae-4758-8d52-17efdf484853",
    },
  },
  {
    type: "kafka", // idea for the future
    event: "DELETE",
    relation: "public:users.id=eq.2", // specific row
    config: {
      /* some relevant conf */
    },
  },
]

[fracek]

Can you expand more on how you plan to allow customers to configure the service from the dashboard? How will this configuration be accessible by realtime? (http, direct db access, etc.) I'm trying to think of a design that's easy to setup but that can also be expanded in the future to let customers load the configuration from multiple sources.

[kiwicopple]

I'll put this one on Slack with a video

[fracek]

So I tought more about this and if the configuration resides outside the process (e.g. json file, supabase server) there is no need for an agent . We would have a ConfigurationManager whose repsonsibility is to fetch the configuration and parse it into elixir types that other processes can query.

Then I add a Connectors genserver that simply forwards notifications to other connectors (for now only webhooks). The Webhooks genserver looks up endpoints by the notification payload and class them in a separate Task.

I sketched the genservers and messages below, rectangles are genservers and arrows are messages. In green what we have, in black what needs to be added for this issue, in grey what can be added in the future.

In the future you can swap the configuration manager with something more complex (and configurable), for now it will load the configuration from json.

[kiwicopple]

This looks perfect with perhaps one missing piece: the SubscribersNotification genserver that you just split out. At the moment that sits between Replication and Channels.

The Connectors genserver needs to also "pull apart" the events into many different events (eg: *, INSERT, columns.eq...). We have 2 options:

1. the SubscribersNotification genserver could also send everything to Connectors which would then match against all events in Config Manager
2. the Connectors directly receives the changes, fetches the config from Config Manager and then see if the Replication event matches any of the config

Preference is 1 (less to maintain) unless you have any opinions on how you think it might fit together

[fracek]

Agree with option 1, so the final design for now is like this.

Connectors job is to dispatch messages to the right connector based on the event type and relation.

[kiwicopple]

Looks perfect 👍

Love the diagrams

[fracek]

I made progress on this. A couple of questions before I can open a PR:

• I added a filter by event type and schema/relation, but not on column value (I think it needs a separate discussion for it).
• In the configuration, I added what we need for the connectors. Basically it loads the json configuration and parse it into a nice datastructure we can work with. This is kinda related to #31, but I think it's important to distinguish between static configuration properties (port, db connection) and dynamic ones (webhooks, connectors).
• I'm using HTTPoison to make the call to the webhooks, let me know if you want me to use a different library.
• What should happen if a webhook returns a status code that's not 2xx? At the moment I log a warning.

[kiwicopple]

Nice Francesco

I'm using HTTPoison

👍

What should happen if a webhook returns a status code that's not 2xx? At the moment I log a warning.

I think that's good - we can send all the logs to BigQuery and set up alerts for the user.

I think it's important to distinguish between static configuration properties

Agreed

I added a filter by event type and schema/relation, but not on column value (I think it needs a separate discussion for it).

It's one I wanted to discuss more too. At the moment we are only filtering on equality. It would be good to throw around some ideas how we can do more/better. Let's discuss on slack or in a new issue

[kiwicopple]

This change is amazing, thanks @fracek. It opens up a number of new possibilities.

It looks like we have 2 remaining points to discuss

• the transaction filter module
• a way to dynamically update/create the connectors config

Should we spin off separate issues for those?

[fracek]

I think they're both issues important enough to require their own issue.

[kiwicopple]

Great - I imagine you already have some ideas for them both? Do you want to start the issues and add your thoughts?

I'll close the one