fix sql and using pgcrypto extension in tests

src/Simplex/Messaging/Agent/Store/Postgres/Util.hs

@@ -21,30 +21,32 @@ import Database.PostgreSQL.Simple.SqlQQ (sql)
 createDBAndUserIfNotExists :: ConnectInfo -> IO ()
 createDBAndUserIfNotExists ConnectInfo {connectUser = user, connectDatabase = dbName} = do
   -- connect to the default "postgres" maintenance database
-  bracket (PSQL.connect defaultConnectInfo {connectUser = "postgres", connectDatabase = "postgres"}) PSQL.close $
-    \postgresDB -> do
-      void $ PSQL.execute_ postgresDB "SET client_min_messages TO WARNING"
-      -- check if the user exists, create if not
-      [Only userExists] <-
-        PSQL.query
-          postgresDB
-          [sql|
-            SELECT EXISTS (
-              SELECT 1 FROM pg_catalog.pg_roles
-              WHERE rolname = ?
-            )
-          |]
-          (Only user)
-      unless userExists $ void $ PSQL.execute_ postgresDB (fromString $ "CREATE USER " <> user)
-      -- check if the database exists, create if not
-      dbExists <- checkDBExists postgresDB dbName
-      unless dbExists $ void $ PSQL.execute_ postgresDB (fromString $ "CREATE DATABASE " <> dbName <> " OWNER " <> user)
+  bracket (PSQL.connect defaultConnectInfo {connectUser = "postgres", connectDatabase = "postgres"}) PSQL.close $ \db -> do
+    execSQL db "SET client_min_messages TO WARNING"
+    -- check if the user exists, create if not
+    [Only userExists] <-
+      PSQL.query
+        db
+        [sql|
+          SELECT EXISTS (
+            SELECT 1 FROM pg_catalog.pg_roles
+            WHERE rolname = ?
+          )
+        |]
+        (Only user)
+    unless userExists $ execSQL db $ "CREATE USER " <> user
+    -- check if the database exists, create if not
+    dbExists <- checkDBExists db dbName
+    unless dbExists $ do
+      execSQL db $ "CREATE DATABASE " <> dbName <> " OWNER " <> user
+      bracket (PSQL.connect defaultConnectInfo {connectUser = "postgres", connectDatabase = dbName}) PSQL.close $
+        (`execSQL` "CREATE EXTENSION IF NOT EXISTS pgcrypto")
 
 checkDBExists :: PSQL.Connection -> String -> IO Bool
-checkDBExists postgresDB dbName = do
+checkDBExists db dbName = do
   [Only dbExists] <-
     PSQL.query
-      postgresDB
+      db
       [sql|
         SELECT EXISTS (
           SELECT 1 FROM pg_catalog.pg_database
@@ -56,45 +58,45 @@ checkDBExists postgresDB dbName = do
 
 dropSchema :: ConnectInfo -> String -> IO ()
 dropSchema connectInfo schema =
-  bracket (PSQL.connect connectInfo) PSQL.close $
-    \db -> do
-      void $ PSQL.execute_ db "SET client_min_messages TO WARNING"
-      void $ PSQL.execute_ db (fromString $ "DROP SCHEMA IF EXISTS " <> schema <> " CASCADE")
+  bracket (PSQL.connect connectInfo) PSQL.close $ \db -> do
+    execSQL db "SET client_min_messages TO WARNING"
+    execSQL db $ "DROP SCHEMA IF EXISTS " <> schema <> " CASCADE"
 
 dropAllSchemasExceptSystem :: ConnectInfo -> IO ()
 dropAllSchemasExceptSystem connectInfo =
-  bracket (PSQL.connect connectInfo) PSQL.close $
-    \db -> do
-      void $ PSQL.execute_ db "SET client_min_messages TO WARNING"
-      schemaNames :: [Only String] <-
-        PSQL.query_
-          db
-          [sql|
-            SELECT schema_name
-            FROM information_schema.schemata
-            WHERE schema_name NOT IN ('public', 'pg_catalog', 'information_schema')
-          |]
-      forM_ schemaNames $ \(Only schema) ->
-        PSQL.execute_ db (fromString $ "DROP SCHEMA " <> schema <> " CASCADE")
+  bracket (PSQL.connect connectInfo) PSQL.close $ \db -> do
+    execSQL db "SET client_min_messages TO WARNING"
+    schemaNames :: [Only String] <-
+      PSQL.query_
+        db
+        [sql|
+          SELECT schema_name
+          FROM information_schema.schemata
+          WHERE schema_name NOT IN ('public', 'pg_catalog', 'information_schema')
+        |]
+    forM_ schemaNames $ \(Only schema) ->
+      execSQL db $ "DROP SCHEMA " <> schema <> " CASCADE"
 
 dropDatabaseAndUser :: ConnectInfo -> IO ()
 dropDatabaseAndUser ConnectInfo {connectUser = user, connectDatabase = dbName} =
-  bracket (PSQL.connect defaultConnectInfo {connectUser = "postgres", connectDatabase = "postgres"}) PSQL.close $
-    \postgresDB -> do
-      void $ PSQL.execute_ postgresDB "SET client_min_messages TO WARNING"
-      dbExists <- checkDBExists postgresDB dbName
-      when dbExists $ do
-        void $ PSQL.execute_ postgresDB (fromString $ "ALTER DATABASE " <> dbName <> " WITH ALLOW_CONNECTIONS false")
-        -- terminate all connections to the database
-        _r :: [Only Bool] <-
-          PSQL.query
-            postgresDB
-            [sql|
-              SELECT pg_terminate_backend(pg_stat_activity.pid)
-              FROM pg_stat_activity
-              WHERE datname = ?
-                AND pid <> pg_backend_pid()
-            |]
-            (Only dbName)
-        void $ PSQL.execute_ postgresDB (fromString $ "DROP DATABASE " <> dbName)
-      void $ PSQL.execute_ postgresDB (fromString $ "DROP USER IF EXISTS " <> user)
+  bracket (PSQL.connect defaultConnectInfo {connectUser = "postgres", connectDatabase = "postgres"}) PSQL.close $ \db -> do
+    execSQL db "SET client_min_messages TO WARNING"
+    dbExists <- checkDBExists db dbName
+    when dbExists $ do
+      execSQL db $ "ALTER DATABASE " <> dbName <> " WITH ALLOW_CONNECTIONS false"
+      -- terminate all connections to the database
+      _r :: [Only Bool] <-
+        PSQL.query
+          db
+          [sql|
+            SELECT pg_terminate_backend(pg_stat_activity.pid)
+            FROM pg_stat_activity
+            WHERE datname = ?
+              AND pid <> pg_backend_pid()
+          |]
+          (Only dbName)
+      execSQL db $ "DROP DATABASE " <> dbName
+    execSQL db $ "DROP USER IF EXISTS " <> user
+
+execSQL :: PSQL.Connection -> String -> IO ()
+execSQL db = void . PSQL.execute_ db . fromString

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql

@@ -446,6 +446,7 @@ CREATE TABLE client_services(
   service_cert_hash BLOB NOT NULL,
   service_priv_key BLOB NOT NULL,
   rcv_service_id BLOB,
+  rcv_queues_hash BLOB NOT NULL DEFAULT x'',
   FOREIGN KEY(host, port) REFERENCES servers ON UPDATE CASCADE ON DELETE RESTRICT
 );
 CREATE UNIQUE INDEX idx_rcv_queues_ntf ON rcv_queues(host, port, ntf_id);

src/Simplex/Messaging/Notifications/Server/Store/Migrations.hs

@@ -117,7 +117,7 @@ ALTER TABLE smp_servers ADD COLUMN smp_notifier_ids_hash BYTEA NOT NULL DEFAULT
 WITH hashes AS (
   SELECT
     s.smp_server_id,
-    xor_aggregate(digest(s.smp_notifier_id, 'md5')) AS notifier_hash
+    xor_aggregate(public.digest(s.smp_notifier_id, 'md5')) AS notifier_hash
   FROM subscriptions s
   WHERE s.ntf_service_assoc = true
   GROUP BY s.smp_server_id
@@ -132,7 +132,7 @@ LANGUAGE plpgsql
 AS $$
 BEGIN
   UPDATE smp_servers
-  SET smp_notifier_ids_hash = xor_combine(smp_notifier_ids_hash, digest(p_notifier_id, 'md5'))
+  SET smp_notifier_ids_hash = xor_combine(smp_notifier_ids_hash, public.digest(p_notifier_id, 'md5'))
   WHERE smp_server_id = p_server_id;
 END;
 $$;

src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql

@@ -60,7 +60,7 @@ CREATE FUNCTION ntf_server.update_ids_hash(p_server_id bigint, p_notifier_id byt
     AS $$
 BEGIN
   UPDATE smp_servers
-  SET smp_notifier_ids_hash = xor_combine(smp_notifier_ids_hash, digest(p_notifier_id, 'md5'))
+  SET smp_notifier_ids_hash = xor_combine(smp_notifier_ids_hash, public.digest(p_notifier_id, 'md5'))
   WHERE smp_server_id = p_server_id;
 END;
 $$;

src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs

@@ -166,12 +166,12 @@ m20250830_queue_ids_hash =
   createXorHashFuncs
     <> T.pack
       [r|
-ALTER TABLE services ADD COLUMN queue_ids_hash BYTEA;
+ALTER TABLE services ADD COLUMN queue_ids_hash BYTEA NOT NULL DEFAULT '\x00000000000000000000000000000000';
 
 WITH hashes AS (
   SELECT
     s.service_id,
-    xor_aggregate(digest(CASE WHEN s.service_role = 'M' THEN q.recipient_id ELSE q.notifier_id END, 'md5')) AS computed_hash
+    xor_aggregate(public.digest(CASE WHEN s.service_role = 'M' THEN q.recipient_id ELSE COALESCE(q.notifier_id, '\x00000000000000000000000000000000') END, 'md5')) AS computed_hash
   FROM services s
   JOIN msg_queues q ON (s.service_id = q.rcv_service_id AND s.service_role = 'M') OR (s.service_id = q.ntf_service_id AND s.service_role = 'N')
   GROUP BY s.service_id
@@ -181,12 +181,12 @@ SET queue_ids_hash = COALESCE(h.computed_hash, '\x000000000000000000000000000000
 FROM hashes h
 WHERE s.service_id = h.service_id;
 
-CREATE OR REPLACE FUNCTION update_ids_hash(p_service_id bytea, p_role text, p_queue_id bytea) RETURNS void
+CREATE OR REPLACE FUNCTION update_ids_hash(p_service_id BYTEA, p_role text, p_queue_id BYTEA) RETURNS void
 LANGUAGE plpgsql
 AS $$
 BEGIN
   UPDATE services
-  SET queue_ids_hash = xor_combine(queue_ids_hash, digest(p_queue_id, 'md5'))
+  SET queue_ids_hash = xor_combine(queue_ids_hash, public.digest(p_queue_id, 'md5'))
   WHERE service_id = p_service_id AND service_role = p_role;
 END;
 $$;
@@ -282,8 +282,6 @@ createXorHashFuncs :: Text
 createXorHashFuncs =
   T.pack
     [r|
-CREATE EXTENSION IF NOT EXISTS pgcrypto;
-
 CREATE OR REPLACE FUNCTION xor_combine(state BYTEA, value BYTEA) RETURNS BYTEA
 LANGUAGE plpgsql IMMUTABLE STRICT
 AS $$
@@ -315,5 +313,4 @@ dropXorHashFuncs =
     [r|
 DROP AGGREGATE xor_aggregate(BYTEA);
 DROP FUNCTION xor_combine;
-DROP EXTENSION pgcrypto;
     |]

src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql

@@ -21,10 +21,10 @@ CREATE FUNCTION smp_server.on_queue_delete() RETURNS trigger
     AS $$
 BEGIN
   IF OLD.rcv_service_id IS NOT NULL THEN
-    PERFORM update_queue_hash(OLD.rcv_service_id, 'M', OLD.recipient_id);
+    PERFORM update_ids_hash(OLD.rcv_service_id, 'M', OLD.recipient_id);
   END IF;
   IF OLD.ntf_service_id IS NOT NULL THEN
-    PERFORM update_queue_hash(OLD.ntf_service_id, 'N', OLD.notifier_id);
+    PERFORM update_ids_hash(OLD.ntf_service_id, 'N', OLD.notifier_id);
   END IF;
   RETURN OLD;
 END;
@@ -37,10 +37,10 @@ CREATE FUNCTION smp_server.on_queue_insert() RETURNS trigger
     AS $$
 BEGIN
   IF NEW.rcv_service_id IS NOT NULL THEN
-    PERFORM update_queue_hash(NEW.rcv_service_id, 'M', NEW.recipient_id);
+    PERFORM update_ids_hash(NEW.rcv_service_id, 'M', NEW.recipient_id);
   END IF;
   IF NEW.ntf_service_id IS NOT NULL THEN
-    PERFORM update_queue_hash(NEW.ntf_service_id, 'N', NEW.notifier_id);
+    PERFORM update_ids_hash(NEW.ntf_service_id, 'N', NEW.notifier_id);
   END IF;
   RETURN NEW;
 END;
@@ -53,12 +53,20 @@ CREATE FUNCTION smp_server.on_queue_update() RETURNS trigger
     AS $$
 BEGIN
   IF OLD.rcv_service_id IS DISTINCT FROM NEW.rcv_service_id THEN
-    PERFORM update_queue_hash(OLD.rcv_service_id, 'M', OLD.recipient_id);
-    PERFORM update_queue_hash(NEW.rcv_service_id, 'M', NEW.recipient_id);
+    IF OLD.rcv_service_id IS NOT NULL THEN
+      PERFORM update_ids_hash(OLD.rcv_service_id, 'M', OLD.recipient_id);
+    END IF;
+    IF NEW.rcv_service_id IS NOT NULL THEN
+      PERFORM update_ids_hash(NEW.rcv_service_id, 'M', NEW.recipient_id);
+    END IF;
   END IF;
   IF OLD.ntf_service_id IS DISTINCT FROM NEW.ntf_service_id THEN
-    PERFORM update_queue_hash(OLD.ntf_service_id, 'N', OLD.notifier_id);
-    PERFORM update_queue_hash(NEW.ntf_service_id, 'N', NEW.notifier_id);
+    IF OLD.ntf_service_id IS NOT NULL THEN
+      PERFORM update_ids_hash(OLD.ntf_service_id, 'N', OLD.notifier_id);
+    END IF;
+    IF NEW.ntf_service_id IS NOT NULL THEN
+      PERFORM update_ids_hash(NEW.ntf_service_id, 'N', NEW.notifier_id);
+    END IF;
   END IF;
   RETURN NEW;
 END;
@@ -71,7 +79,7 @@ CREATE FUNCTION smp_server.update_ids_hash(p_service_id bytea, p_role text, p_qu
     AS $$
 BEGIN
   UPDATE services
-  SET queue_ids_hash = xor_combine(queue_ids_hash, digest(p_queue_id, 'md5'))
+  SET queue_ids_hash = xor_combine(queue_ids_hash, public.digest(p_queue_id, 'md5'))
   WHERE service_id = p_service_id AND service_role = p_role;
 END;
 $$;
@@ -144,7 +152,7 @@ CREATE TABLE smp_server.services (
     service_cert bytea NOT NULL,
     service_cert_hash bytea NOT NULL,
     created_at bigint NOT NULL,
-    queue_ids_hash bytea
+    queue_ids_hash bytea DEFAULT '\x00000000000000000000000000000000'::bytea NOT NULL
 );