Listen to Database Changes Through the Postgres WAL

Debezium published this doc on Exactly-once delivery with their most recent 3.3.0 version. They dont support it natively, but say it can be achieved via Kafka Connect

https://debezium.io/documentation/reference/stable/configura...

You could probably achieve something similar with the NATS Jetstream sink as well, which has similar capabilities - though I think it doesnt have quite the same guarantees.

I switched to using Debezium a few months ago, after a Golang alternative to Debezium + Kafka Connect - ConduitIO - went defunct. I should have been using Debezium all along, as it is clearly the most mature, most stable option in the space, with the best prospects for long-term survival. Highly recommended, even if it is JVM (though they're currently doing some interesting stuff with Quarkus and GraalVM that might lead to a jvm-free binary at some point)

I saw the post on the clojure subreddit! I’m stoked to try it out.

Ever since I saw Martin Kleppman’s “Turning the database inside out” talk I’ve wanted an easy way to hook into a transaction log. Apache samza/kafka is very cool but I’m not going to set it up for personal projects. It’d be VERY cool to make materialized views straight from the log!!

The problem is that unclean connector shutdowns are a thing that can happen in real life.

I have listen/notify on most changes in my database. Not sure I've experienced any performance issue though but I can't say I've been putting things through their paces. IMHO listen/notify's simplicity outweighed the perf gains by WAL.

I'm only sharing this should it be helpful:

  def up do
    whitelist = Enum.join(@user_columns ++ ["tick"], "', '")

    execute """
    CREATE OR REPLACE FUNCTION notify_phrasing() RETURNS trigger AS $$
    DECLARE
      notif jsonb;
      col_name text;
      col_value text;
      uuids jsonb := '{}'::jsonb;
      user_columns text[] := ARRAY['#{whitelist}'];
    BEGIN
      -- First, add all UUID columns
      FOR col_name IN
        SELECT column_name
        FROM information_schema.columns
        WHERE table_name = TG_TABLE_NAME AND data_type = 'uuid'
      LOOP
        EXECUTE format('SELECT ($1).%I::text', col_name)
        INTO col_value
        USING CASE WHEN TG_OP = 'DELETE' THEN OLD ELSE NEW END;

        IF col_value IS NOT NULL THEN
          uuids := uuids || jsonb_build_object(col_name, col_value);
        END IF;
      END LOOP;

      -- Then, add user columns if they exist in the table
      FOREACH col_name IN ARRAY user_columns
      LOOP
        IF EXISTS (
          SELECT 1
          FROM information_schema.columns
          WHERE table_name = TG_TABLE_NAME AND column_name = col_name
        ) THEN
          EXECUTE format('SELECT ($1).%I::text', col_name)
          INTO col_value
          USING CASE WHEN TG_OP = 'DELETE' THEN OLD ELSE NEW END;

          IF col_value IS NOT NULL THEN
            uuids := uuids || jsonb_build_object(col_name, col_value);
          END IF;
        END IF;
      END LOOP;

      notif = jsonb_build_object(
        'table', TG_TABLE_NAME,
        'event', TG_OP,
        'uuids', uuids
      );

      PERFORM pg_notify('phrasing', notif::text);
      RETURN NULL;
    END;
    $$ LANGUAGE plpgsql;
    """

    # Create trigger for each table
    Enum.each(@tables, fn table ->
      execute """
      CREATE TRIGGER notify_phrasing__#{table}
      AFTER INSERT OR UPDATE OR DELETE ON #{table}
      FOR EACH ROW EXECUTE FUNCTION notify_phrasing();
      """
    end)
  end

I do react to most (70%) of my database changes in some way shape or form, and post them to a PubSub topic with the uuids. All of my dispatching can be done off of uuids.

I wrote a Haskell library for this as well: https://hackage.haskell.org/package/postgresql-replicant