Merge pull request #241 from commanded/bug/postgres-trigger

Fix bug with subscriptions trigger in older Postgres versions

.dialyzer_ignore.exs

@@ -0,0 +1,4 @@
+[
+  {"deps/postgrex/lib/postgrex/type_module.ex", :improper_list_constr},
+  {"lib/postgrex/type_module.ex", :improper_list_constr}
+]

.github/workflows/test.yml

@@ -9,8 +9,8 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        otp: ['23.3', '24.0']
-        elixir: ['1.11.4', '1.12.2']
+        otp: ['23.3']
+        elixir: ['1.12.2']
 
     services:
       postgres:

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Next release
+
+### Bug fixes
+
+- Fix bug with subscriptions trigger in older Postgres versions ([#241](https://github.com/commanded/eventstore/pull/241)).
+
 ## v1.3.1
 
 ### Bug fixes

lib/event_store/sql/init.ex

@@ -186,6 +186,7 @@ defmodule EventStore.Sql.Init do
     CREATE OR REPLACE FUNCTION notify_events()
       RETURNS trigger AS $$
     DECLARE
+      old_stream_version bigint;
       channel text;
       payload text;
     BEGIN
@@ -196,8 +197,14 @@ defmodule EventStore.Sql.Init do
         --  * last `stream_version`
         -- Each separated by a comma (e.g. 'stream-12345,1,1,5')
 
+        IF TG_OP = 'UPDATE' THEN
+          old_stream_version := OLD.stream_version + 1;
+        ELSE
+          old_stream_version := 1;
+        END IF;
+
         channel := TG_TABLE_SCHEMA || '.events';
-        payload := NEW.stream_uuid || ',' || NEW.stream_id || ',' || COALESCE(OLD.stream_version, 0) + 1 || ',' || NEW.stream_version;
+        payload := NEW.stream_uuid || ',' || NEW.stream_id || ',' || old_stream_version || ',' || NEW.stream_version;
 
         -- Notify events to listeners
         PERFORM pg_notify(channel, payload);
@@ -267,7 +274,7 @@ defmodule EventStore.Sql.Init do
   defp record_event_store_schema_version do
     """
     INSERT INTO schema_migrations (major_version, minor_version, patch_version)
-    VALUES (1, 3, 0);
+    VALUES (1, 3, 2);
     """
   end
 end

lib/event_store/tasks/migrations.ex

@@ -18,7 +18,8 @@ defmodule EventStore.Tasks.Migrations do
     "0.17.0",
     "1.1.0",
     "1.2.0",
-    "1.3.0"
+    "1.3.0",
+    "1.3.2"
   ]
 
   @dialyzer {:no_return, exec: 2, handle_response: 1}

mix.exs

@@ -155,6 +155,7 @@ defmodule EventStore.Mixfile do
 
   defp dialyzer do
     [
+      ignore_warnings: ".dialyzer_ignore.exs",
       plt_add_apps: [:ex_unit, :jason, :mix],
       plt_add_deps: :app_tree,
       plt_file: {:no_warn, "priv/plts/eventstore.plt"}

priv/event_store/migrations/v1.3.2.sql

@@ -0,0 +1,36 @@
+DO $migration$
+BEGIN
+  CREATE OR REPLACE FUNCTION notify_events()
+    RETURNS trigger AS $$
+  DECLARE
+    old_stream_version bigint;
+    channel text;
+    payload text;
+  BEGIN
+      -- Payload text contains:
+      --  * `stream_uuid`
+      --  * `stream_id`
+      --  * first `stream_version`
+      --  * last `stream_version`
+      -- Each separated by a comma (e.g. 'stream-12345,1,1,5')
+
+      IF TG_OP = 'UPDATE' THEN
+        old_stream_version := OLD.stream_version + 1;
+      ELSE
+        old_stream_version := 1;
+      END IF;
+
+      channel := TG_TABLE_SCHEMA || '.events';
+      payload := NEW.stream_uuid || ',' || NEW.stream_id || ',' || old_stream_version || ',' || NEW.stream_version;
+
+      -- Notify events to listeners
+      PERFORM pg_notify(channel, payload);
+
+      RETURN NULL;
+  END;
+  $$ LANGUAGE plpgsql;
+
+  -- Record schema migration
+  INSERT INTO schema_migrations (major_version, minor_version, patch_version) VALUES (1, 3, 2);
+END;
+$migration$ LANGUAGE plpgsql;

test/list_event_store_migrations_test.exs

@@ -25,6 +25,7 @@ defmodule EventStore.ListEventStoreMigrationsTest do
         1.1.0     \tpending
         1.2.0     \tpending
         1.3.0     \tpending
+        1.3.2     \tpending
 
       """
 

test/migrate_event_store_test.exs

@@ -85,6 +85,7 @@ defmodule EventStore.MigrateEventStoreTest do
       "0\t17\t0\tNOW" -> true
       "1\t1\t0\tNOW" -> true
       "1\t2\t0\tNOW" -> true
+      "1\t3\t0\tNOW" -> true
       _line -> false
     end)
   end