agent migrations with functions/triggers

simplexmq.cabal

@@ -163,11 +163,13 @@ library
           Simplex.Messaging.Agent.Store.Postgres.Migrations.M20250203_msg_bodies
           Simplex.Messaging.Agent.Store.Postgres.Migrations.M20250322_short_links
           Simplex.Messaging.Agent.Store.Postgres.Migrations.M20250702_conn_invitations_remove_cascade_delete
+          Simplex.Messaging.Agent.Store.Postgres.Migrations.M20250815_service_certs
   else
       exposed-modules:
           Simplex.Messaging.Agent.Store.SQLite
           Simplex.Messaging.Agent.Store.SQLite.Common
           Simplex.Messaging.Agent.Store.SQLite.DB
+          Simplex.Messaging.Agent.Store.SQLite.Functions
           Simplex.Messaging.Agent.Store.SQLite.Migrations
           Simplex.Messaging.Agent.Store.SQLite.Migrations.App
           Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
@@ -217,6 +219,7 @@ library
           Simplex.Messaging.Agent.Store.Postgres.Common
           Simplex.Messaging.Agent.Store.Postgres.DB
           Simplex.Messaging.Agent.Store.Postgres.Migrations
+          Simplex.Messaging.Agent.Store.Postgres.Migrations.Util
           Simplex.Messaging.Agent.Store.Postgres.Util
   if !flag(client_library)
       exposed-modules:

src/Simplex/Messaging/Agent/Store/AgentStore.hs

@@ -2056,7 +2056,7 @@ insertRcvQueue_ db connId' rq@RcvQueue {..} serverKeyHash_ = do
           ntf_public_key, ntf_private_key, ntf_id, rcv_ntf_dh_secret
         ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?);
     |]
-    ( (host server, port server, rcvId, rcvServiceAssoc, connId', rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret)
+    ( (host server, port server, rcvId, BI rcvServiceAssoc, connId', rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret)
         :. (sndId, queueMode, status, qId, BI primary, dbReplaceQueueId, smpClientVersion, serverKeyHash_)
         :. (shortLinkId <$> shortLink, shortLinkKey <$> shortLink, linkPrivSigKey <$> shortLink, linkEncFixedData <$> shortLink)
         :. ntfCredsFields
@@ -2242,13 +2242,13 @@ rcvQueueQuery =
 
 toRcvQueue ::
   (UserId, C.KeyHash, ConnId, NonEmpty TransportHost, ServiceName, SMP.RecipientId, SMP.RcvPrivateAuthKey, SMP.RcvDhSecret, C.PrivateKeyX25519, Maybe C.DhSecretX25519, SMP.SenderId, Maybe QueueMode)
-    :. (QueueStatus, DBEntityId, BoolInt, Maybe Int64, Maybe RcvSwitchStatus, Maybe VersionSMPC, Int, ServiceAssoc)
+    :. (QueueStatus, DBEntityId, BoolInt, Maybe Int64, Maybe RcvSwitchStatus, Maybe VersionSMPC, Int, BoolInt)
     :. (Maybe SMP.NtfPublicAuthKey, Maybe SMP.NtfPrivateAuthKey, Maybe SMP.NotifierId, Maybe RcvNtfDhSecret)
     :. (Maybe SMP.LinkId, Maybe LinkKey, Maybe C.PrivateKeyEd25519, Maybe EncDataBytes) ->
   RcvQueue
 toRcvQueue
   ( (userId, keyHash, connId, host, port, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, queueMode)
-      :. (status, dbQueueId, BI primary, dbReplaceQueueId, rcvSwchStatus, smpClientVersion_, deleteErrors, rcvServiceAssoc)
+      :. (status, dbQueueId, BI primary, dbReplaceQueueId, rcvSwchStatus, smpClientVersion_, deleteErrors, BI rcvServiceAssoc)
       :. (ntfPublicKey_, ntfPrivateKey_, notifierId_, rcvNtfDhSecret_)
       :. (shortLinkId_, shortLinkKey_, linkPrivSigKey_, linkEncFixedData_)
   ) =

src/Simplex/Messaging/Agent/Store/Postgres/Migrations/App.hs

@@ -8,6 +8,7 @@ import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20241210_initial
 import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20250203_msg_bodies
 import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20250322_short_links
 import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20250702_conn_invitations_remove_cascade_delete
+import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20250815_service_certs
 import Simplex.Messaging.Agent.Store.Shared (Migration (..))
 
 schemaMigrations :: [(String, Text, Maybe Text)]
@@ -15,7 +16,8 @@ schemaMigrations =
   [ ("20241210_initial", m20241210_initial, Nothing),
     ("20250203_msg_bodies", m20250203_msg_bodies, Just down_m20250203_msg_bodies),
     ("20250322_short_links", m20250322_short_links, Just down_m20250322_short_links),
-    ("20250702_conn_invitations_remove_cascade_delete", m20250702_conn_invitations_remove_cascade_delete, Just down_m20250702_conn_invitations_remove_cascade_delete)
+    ("20250702_conn_invitations_remove_cascade_delete", m20250702_conn_invitations_remove_cascade_delete, Just down_m20250702_conn_invitations_remove_cascade_delete),
+    ("20250815_service_certs", m20250815_service_certs, Just down_m20250815_service_certs)
   ]
 
 -- | The list of migrations in ascending order by date

src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20250815_service_certs.hs

@@ -0,0 +1,114 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.Postgres.Migrations.M20250815_service_certs where
+
+import Data.Text (Text)
+import qualified Data.Text as T
+import Simplex.Messaging.Agent.Store.Postgres.Migrations.Util
+import Text.RawString.QQ (r)
+
+m20250815_service_certs :: Text
+m20250815_service_certs =
+  createXorHashFuncs
+    <> T.pack
+      [r|
+CREATE TABLE client_services(
+  user_id BIGINT NOT NULL REFERENCES users ON UPDATE RESTRICT ON DELETE CASCADE,
+  host TEXT NOT NULL,
+  port TEXT NOT NULL,
+  service_cert BYTEA NOT NULL,
+  service_cert_hash BYTEA NOT NULL,
+  service_priv_key BYTEA NOT NULL,
+  service_id BYTEA,
+  service_queue_count BIGINT NOT NULL DEFAULT 0,
+  service_queue_ids_hash BYTEA NOT NULL DEFAULT '\x00000000000000000000000000000000',
+  FOREIGN KEY(host, port) REFERENCES servers ON UPDATE CASCADE ON DELETE RESTRICT
+);
+
+CREATE UNIQUE INDEX idx_server_certs_user_id_host_port ON client_services(user_id, host, port);
+CREATE INDEX idx_server_certs_host_port ON client_services(host, port);
+
+ALTER TABLE rcv_queues ADD COLUMN rcv_service_assoc SMALLINT NOT NULL DEFAULT 0;
+
+CREATE FUNCTION update_aggregates(p_user_id BIGINT, p_host TEXT, p_port TEXT, p_change BIGINT, p_rcv_id BYTEA) RETURNS VOID
+LANGUAGE plpgsql
+AS $$
+BEGIN
+  UPDATE client_services
+  SET service_queue_count = service_queue_count + p_change,
+      service_queue_ids_hash = xor_combine(service_queue_ids_hash, public.digest(p_rcv_id, 'md5'))
+  WHERE user_id = p_user_id AND host = p_host AND port = p_port;
+END;
+$$;
+
+CREATE FUNCTION on_rcv_queue_insert() RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+BEGIN
+  IF NEW.rcv_service_assoc != 0 AND NEW.deleted = 0 THEN
+    PERFORM update_aggregates(NEW.user_id, NEW.host, NEW.port, 1, NEW.rcv_id);
+  END IF;
+  RETURN NEW;
+END;
+$$;
+
+CREATE FUNCTION on_rcv_queue_delete() RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+BEGIN
+  IF OLD.rcv_service_assoc != 0 AND OLD.deleted = 0 THEN
+    PERFORM update_aggregates(OLD.user_id, OLD.host, OLD.port, -1, OLD.rcv_id);
+  END IF;
+  RETURN OLD;
+END;
+$$;
+
+CREATE FUNCTION on_rcv_queue_update() RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+BEGIN
+  IF OLD.rcv_service_assoc != 0 AND OLD.deleted = 0 THEN
+    IF NOT (NEW.rcv_service_assoc != 0 AND NEW.deleted = 0) THEN
+      PERFORM update_aggregates(OLD.user_id, OLD.host, OLD.port, -1, OLD.rcv_id);
+    END IF;
+  ELSIF NEW.rcv_service_assoc != 0 AND NEW.deleted = 0 THEN
+    PERFORM update_aggregates(NEW.user_id, NEW.host, NEW.port, 1, NEW.rcv_id);
+  END IF;
+  RETURN NEW;
+END;
+$$;
+
+CREATE TRIGGER tr_rcv_queue_insert
+AFTER INSERT ON rcv_queues
+FOR EACH ROW EXECUTE PROCEDURE on_rcv_queue_insert();
+
+CREATE TRIGGER tr_rcv_queue_delete
+AFTER DELETE ON rcv_queues
+FOR EACH ROW EXECUTE PROCEDURE on_rcv_queue_delete();
+
+CREATE TRIGGER tr_rcv_queue_update
+AFTER UPDATE ON rcv_queues
+FOR EACH ROW EXECUTE PROCEDURE on_rcv_queue_update();
+      |]
+
+down_m20250815_service_certs :: Text
+down_m20250815_service_certs =
+  T.pack
+    [r|
+DROP TRIGGER tr_rcv_queue_insert ON rcv_queues;
+DROP TRIGGER tr_rcv_queue_delete ON rcv_queues;
+DROP TRIGGER tr_rcv_queue_update ON rcv_queues;
+
+DROP FUNCTION on_rcv_queue_insert;
+DROP FUNCTION on_rcv_queue_delete;
+DROP FUNCTION on_rcv_queue_update;
+
+DROP FUNCTION update_aggregates;
+
+ALTER TABLE rcv_queues DROP COLUMN rcv_service_assoc;
+
+DROP INDEX idx_server_certs_host_port;
+DROP INDEX idx_server_certs_user_id_host_port;
+DROP TABLE client_services;
+    |]
+    <> dropXorHashFuncs

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

@@ -0,0 +1,44 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.Postgres.Migrations.Util where
+
+import Data.Text (Text)
+import qualified Data.Text as T
+import Text.RawString.QQ (r)
+
+createXorHashFuncs :: Text
+createXorHashFuncs =
+  T.pack
+    [r|
+CREATE OR REPLACE FUNCTION xor_combine(state BYTEA, value BYTEA) RETURNS BYTEA
+LANGUAGE plpgsql IMMUTABLE STRICT
+AS $$
+DECLARE
+  result BYTEA := state;
+  i INTEGER;
+  len INTEGER := octet_length(value);
+BEGIN
+  IF octet_length(state) != len THEN
+    RAISE EXCEPTION 'Inputs must be equal length (% != %)', octet_length(state), len;
+  END IF;
+  FOR i IN 0..len-1 LOOP
+    result := set_byte(result, i, get_byte(state, i) # get_byte(value, i));
+  END LOOP;
+  RETURN result;
+END;
+$$;
+
+CREATE OR REPLACE AGGREGATE xor_aggregate(BYTEA) (
+  SFUNC = xor_combine,
+  STYPE = BYTEA,
+  INITCOND = '\x00000000000000000000000000000000' -- 16 bytes
+);
+    |]
+
+dropXorHashFuncs :: Text
+dropXorHashFuncs =
+  T.pack
+    [r|
+DROP AGGREGATE xor_aggregate(BYTEA);
+DROP FUNCTION xor_combine;
+    |]

src/Simplex/Messaging/Agent/Store/SQLite.hs

@@ -56,12 +56,13 @@ import qualified Database.SQLite3 as SQLite3
 import Simplex.Messaging.Agent.Store.Migrations (DBMigrate (..), sharedMigrateSchema)
 import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
 import Simplex.Messaging.Agent.Store.SQLite.Common
+import Simplex.Messaging.Agent.Store.SQLite.Functions (registerFunctions)
 import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
 import Simplex.Messaging.Agent.Store.Shared (Migration (..), MigrationConfirmation (..), MigrationError (..))
 import Simplex.Messaging.Util (ifM, safeDecodeUtf8)
 import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist)
 import System.FilePath (takeDirectory)
-import UnliftIO.Exception (bracketOnError, onException)
+import UnliftIO.Exception (bracketOnError, onException, throwIO)
 import UnliftIO.MVar
 import UnliftIO.STM
 
@@ -113,6 +114,7 @@ connectDB path key track = do
           PRAGMA secure_delete = ON;
           PRAGMA auto_vacuum = FULL;
         |]
+      registerFunctions db' >>= either (\e -> putStrLn ("functions: " <> show e) >> throwIO (userError $ show e)) pure
 
 closeDBStore :: DBStore -> IO ()
 closeDBStore st@DBStore {dbClosed} =

src/Simplex/Messaging/Agent/Store/SQLite/Functions.hs

@@ -0,0 +1,37 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Functions where
+
+import Control.Monad.Trans.Except
+import Data.Bits (xor)
+import Database.SQLite3 (Database)
+import Database.SQLite3.Direct (Error (..), createFunction, funcArgBlob, funcResultBlob)
+import qualified Data.ByteString as B
+import qualified Simplex.Messaging.Crypto as C
+
+md5Func :: Database -> IO (Either Error ())
+md5Func db =
+  createFunction db "chat_md5hash" (Just 1) True $ \cxt args -> do
+    print "in chat_md5hash"
+    funcResultBlob cxt . C.md5Hash =<< funcArgBlob args 0
+
+-- Bitwise XOR
+xorCombineFunc :: Database -> IO (Either Error ())
+xorCombineFunc db =
+  createFunction db "chat_xor_combine" (Just 2) True $ \cxt args -> do
+    print "in chat_xor_combine"
+    s1 <- funcArgBlob args 0
+    s2 <- funcArgBlob args 1
+    funcResultBlob cxt $ B.pack $ B.zipWith xor s1 s2
+
+-- xorAggregate :: Database -> IO (Either Error ())
+-- xorAggregate db = createAggregate db "chat_xor_aggregate" (Just 1) (B.replicate 16 0) step funcResultBlob
+--   where
+--     step _ args st = B.pack . B.zipWith xor st <$> funcArgBlob args 0
+
+registerFunctions :: Database -> IO (Either Error ())
+registerFunctions db = runExceptT $ do
+  ExceptT $ md5Func db
+  ExceptT $ xorCombineFunc db
+  -- ExceptT $ xorAggregate db

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20250815_service_certs.hs

@@ -5,7 +5,6 @@ module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20250815_service_certs w
 import Database.SQLite.Simple (Query)
 import Database.SQLite.Simple.QQ (sql)
 
--- TODO move date forward, create migration for postgres
 m20250815_service_certs :: Query
 m20250815_service_certs =
   [sql|
@@ -16,26 +15,73 @@ CREATE TABLE client_services(
   service_cert BLOB NOT NULL,
   service_cert_hash BLOB NOT NULL,
   service_priv_key BLOB NOT NULL,
-  rcv_service_id BLOB,
-  rcv_service_queue_count INTEGER NOT NULL DEFAULT 0,
-  rcv_service_queue_ids_hash BLOB NOT NULL DEFAULT x'00000000000000000000000000000000',
+  service_id BLOB,
+  service_queue_count INTEGER NOT NULL DEFAULT 0,
+  service_queue_ids_hash BLOB NOT NULL DEFAULT x'00000000000000000000000000000000',
   FOREIGN KEY(host, port) REFERENCES servers ON UPDATE CASCADE ON DELETE RESTRICT
 );
 
 CREATE UNIQUE INDEX idx_server_certs_user_id_host_port ON client_services(user_id, host, port);
-
 CREATE INDEX idx_server_certs_host_port ON client_services(host, port);
 
 ALTER TABLE rcv_queues ADD COLUMN rcv_service_assoc INTEGER NOT NULL DEFAULT 0;
+
+CREATE TRIGGER tr_rcv_queue_insert
+AFTER INSERT ON rcv_queues
+FOR EACH ROW
+WHEN NEW.rcv_service_assoc != 0 AND NEW.deleted = 0
+BEGIN
+  UPDATE client_services
+  SET service_queue_count = service_queue_count + 1,
+      service_queue_ids_hash = chat_xor_combine(service_queue_ids_hash, chat_md5hash(NEW.rcv_id))
+  WHERE user_id = NEW.user_id AND host = NEW.host AND port = NEW.port;
+END;
+
+CREATE TRIGGER tr_rcv_queue_delete
+AFTER DELETE ON rcv_queues
+FOR EACH ROW
+WHEN OLD.rcv_service_assoc != 0 AND OLD.deleted = 0
+BEGIN
+  UPDATE client_services
+  SET service_queue_count = service_queue_count - 1,
+      service_queue_ids_hash = chat_xor_combine(service_queue_ids_hash, chat_md5hash(OLD.rcv_id))
+  WHERE user_id = OLD.user_id AND host = OLD.host AND port = OLD.port;
+END;
+
+CREATE TRIGGER tr_rcv_queue_update_remove
+AFTER UPDATE ON rcv_queues
+FOR EACH ROW
+WHEN OLD.rcv_service_assoc != 0 AND OLD.deleted = 0 AND NOT (NEW.rcv_service_assoc != 0 AND NEW.deleted = 0)
+BEGIN
+  UPDATE client_services
+  SET service_queue_count = service_queue_count - 1,
+      service_queue_ids_hash = chat_xor_combine(service_queue_ids_hash, chat_md5hash(OLD.rcv_id))
+  WHERE user_id = OLD.user_id AND host = OLD.host AND port = OLD.port;
+END;
+
+CREATE TRIGGER tr_rcv_queue_update_add
+AFTER UPDATE ON rcv_queues
+FOR EACH ROW
+WHEN NEW.rcv_service_assoc != 0 AND NEW.deleted = 0 AND NOT (OLD.rcv_service_assoc != 0 AND OLD.deleted = 0)
+BEGIN
+  UPDATE client_services
+  SET service_queue_count = service_queue_count + 1,
+      service_queue_ids_hash = chat_xor_combine(service_queue_ids_hash, chat_md5hash(NEW.rcv_id))
+  WHERE user_id = NEW.user_id AND host = NEW.host AND port = NEW.port;
+END;
   |]
 
 down_m20250815_service_certs :: Query
 down_m20250815_service_certs =
   [sql|
+DROP TRIGGER tr_rcv_queue_insert;
+DROP TRIGGER tr_rcv_queue_delete;
+DROP TRIGGER tr_rcv_queue_update_remove;
+DROP TRIGGER tr_rcv_queue_update_add;
+
 ALTER TABLE rcv_queues DROP COLUMN rcv_service_assoc;
 
 DROP INDEX idx_server_certs_host_port;
-
 DROP INDEX idx_server_certs_user_id_host_port;
 
 DROP TABLE client_services;

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

@@ -445,9 +445,9 @@ CREATE TABLE client_services(
   service_cert BLOB NOT NULL,
   service_cert_hash BLOB NOT NULL,
   service_priv_key BLOB NOT NULL,
-  rcv_service_id BLOB,
-  rcv_service_queue_count INTEGER NOT NULL DEFAULT 0,
-  rcv_service_queue_ids_hash BLOB NOT NULL DEFAULT x'00000000000000000000000000000000',
+  service_id BLOB,
+  service_queue_count INTEGER NOT NULL DEFAULT 0,
+  service_queue_ids_hash BLOB NOT NULL DEFAULT x'00000000000000000000000000000000',
   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);
@@ -591,3 +591,43 @@ CREATE UNIQUE INDEX idx_server_certs_user_id_host_port ON client_services(
   port
 );
 CREATE INDEX idx_server_certs_host_port ON client_services(host, port);
+CREATE TRIGGER tr_rcv_queue_insert
+AFTER INSERT ON rcv_queues
+FOR EACH ROW
+WHEN NEW.rcv_service_assoc != 0 AND NEW.deleted = 0
+BEGIN
+  UPDATE client_services
+  SET service_queue_count = service_queue_count + 1,
+      service_queue_ids_hash = chat_xor_combine(service_queue_ids_hash, chat_md5hash(NEW.rcv_id))
+  WHERE user_id = NEW.user_id AND host = NEW.host AND port = NEW.port;
+END;
+CREATE TRIGGER tr_rcv_queue_delete
+AFTER DELETE ON rcv_queues
+FOR EACH ROW
+WHEN OLD.rcv_service_assoc != 0 AND OLD.deleted = 0
+BEGIN
+  UPDATE client_services
+  SET service_queue_count = service_queue_count - 1,
+      service_queue_ids_hash = chat_xor_combine(service_queue_ids_hash, chat_md5hash(OLD.rcv_id))
+  WHERE user_id = OLD.user_id AND host = OLD.host AND port = OLD.port;
+END;
+CREATE TRIGGER tr_rcv_queue_update_remove
+AFTER UPDATE ON rcv_queues
+FOR EACH ROW
+WHEN OLD.rcv_service_assoc != 0 AND OLD.deleted = 0 AND NOT (NEW.rcv_service_assoc != 0 AND NEW.deleted = 0)
+BEGIN
+  UPDATE client_services
+  SET service_queue_count = service_queue_count - 1,
+      service_queue_ids_hash = chat_xor_combine(service_queue_ids_hash, chat_md5hash(OLD.rcv_id))
+  WHERE user_id = OLD.user_id AND host = OLD.host AND port = OLD.port;
+END;
+CREATE TRIGGER tr_rcv_queue_update_add
+AFTER UPDATE ON rcv_queues
+FOR EACH ROW
+WHEN NEW.rcv_service_assoc != 0 AND NEW.deleted = 0 AND NOT (OLD.rcv_service_assoc != 0 AND OLD.deleted = 0)
+BEGIN
+  UPDATE client_services
+  SET service_queue_count = service_queue_count + 1,
+      service_queue_ids_hash = chat_xor_combine(service_queue_ids_hash, chat_md5hash(NEW.rcv_id))
+  WHERE user_id = NEW.user_id AND host = NEW.host AND port = NEW.port;
+END;

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

@@ -6,8 +6,8 @@ module Simplex.Messaging.Notifications.Server.Store.Migrations where
 import Data.List (sortOn)
 import Data.Text (Text)
 import qualified Data.Text as T
+import Simplex.Messaging.Agent.Store.Postgres.Migrations.Util
 import Simplex.Messaging.Agent.Store.Shared
-import Simplex.Messaging.Server.QueueStore.Postgres.Migrations (createXorHashFuncs, dropXorHashFuncs)
 import Text.RawString.QQ (r)
 
 ntfServerSchemaMigrations :: [(String, Text, Maybe Text)]
@@ -116,7 +116,7 @@ ALTER TABLE smp_servers
   ADD COLUMN smp_notifier_count BIGINT NOT NULL DEFAULT 0,
   ADD COLUMN smp_notifier_ids_hash BYTEA NOT NULL DEFAULT '\x00000000000000000000000000000000';
 
-CREATE OR REPLACE FUNCTION should_subscribe_status(p_status TEXT) RETURNS BOOLEAN
+CREATE FUNCTION should_subscribe_status(p_status TEXT) RETURNS BOOLEAN
 LANGUAGE plpgsql IMMUTABLE STRICT
 AS $$
 BEGIN
@@ -124,7 +124,7 @@ BEGIN
 END;
 $$;
 
-CREATE OR REPLACE FUNCTION update_all_aggregates() RETURNS void
+CREATE FUNCTION update_all_aggregates() RETURNS VOID
 LANGUAGE plpgsql
 AS $$
 BEGIN
@@ -147,7 +147,7 @@ $$;
 
 SELECT update_all_aggregates();
 
-CREATE OR REPLACE FUNCTION update_aggregates(p_server_id BIGINT, p_change BIGINT, p_notifier_id BYTEA) RETURNS void
+CREATE FUNCTION update_aggregates(p_server_id BIGINT, p_change BIGINT, p_notifier_id BYTEA) RETURNS VOID
 LANGUAGE plpgsql
 AS $$
 BEGIN
@@ -158,7 +158,7 @@ BEGIN
 END;
 $$;
 
-CREATE OR REPLACE FUNCTION on_subscription_insert() RETURNS TRIGGER
+CREATE FUNCTION on_subscription_insert() RETURNS TRIGGER
 LANGUAGE plpgsql
 AS $$
 BEGIN
@@ -169,7 +169,7 @@ BEGIN
 END;
 $$;
 
-CREATE OR REPLACE FUNCTION on_subscription_delete() RETURNS TRIGGER
+CREATE FUNCTION on_subscription_delete() RETURNS TRIGGER
 LANGUAGE plpgsql
 AS $$
 BEGIN
@@ -180,7 +180,7 @@ BEGIN
 END;
 $$;
 
-CREATE OR REPLACE FUNCTION on_subscription_update() RETURNS TRIGGER
+CREATE FUNCTION on_subscription_update() RETURNS TRIGGER
 LANGUAGE plpgsql
 AS $$
 BEGIN
@@ -195,10 +195,6 @@ BEGIN
 END;
 $$;
 
-DROP TRIGGER IF EXISTS tr_subscriptions_insert ON subscriptions;
-DROP TRIGGER IF EXISTS tr_subscriptions_delete ON subscriptions;
-DROP TRIGGER IF EXISTS tr_subscriptions_update ON subscriptions;
-
 CREATE TRIGGER tr_subscriptions_insert
 AFTER INSERT ON subscriptions
 FOR EACH ROW EXECUTE PROCEDURE on_subscription_insert();

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

@@ -54,7 +54,7 @@ import Network.Socket (ServiceName)
 import Simplex.Messaging.Agent.Store.AgentStore ()
 import Simplex.Messaging.Agent.Store.Postgres (closeDBStore, createDBStore)
 import Simplex.Messaging.Agent.Store.Postgres.Common
-import Simplex.Messaging.Agent.Store.Postgres.DB (blobFieldDecoder, fromTextField_)
+import Simplex.Messaging.Agent.Store.Postgres.DB (fromTextField_)
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import qualified Simplex.Messaging.Crypto as C
@@ -63,7 +63,6 @@ import Simplex.Messaging.Notifications.Server.Store (NtfSTMStore (..), NtfSubDat
 import Simplex.Messaging.Notifications.Server.Store.Migrations
 import Simplex.Messaging.Notifications.Server.Store.Types
 import Simplex.Messaging.Notifications.Server.StoreLog
-import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Protocol (EntityId (..), EncNMsgMeta, ErrorType (..), IdsHash (..), NotifierId, NtfPrivateAuthKey, NtfPublicAuthKey, SMPServer, ServiceId, ServiceSub (..), pattern SMPServer)
 import Simplex.Messaging.Server.QueueStore (RoundedSystemTime, getSystemDate)
 import Simplex.Messaging.Server.QueueStore.Postgres (handleDuplicate, withLog_)
@@ -76,6 +75,8 @@ import System.IO (IOMode (..), hFlush, stdout, withFile)
 import Text.Hex (decodeHex)
 
 #if !defined(dbPostgres)
+import Simplex.Messaging.Agent.Store.Postgres.DB (blobFieldDecoder)
+import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Util (eitherToMaybe)
 #endif
 

src/Simplex/Messaging/Protocol.hs

@@ -1479,9 +1479,7 @@ instance ToField IdsHash where
   {-# INLINE toField #-}
 
 instance Semigroup IdsHash where
-  (IdsHash s1) <> (IdsHash s2) =
-    let !s' = BS.pack $ zipWith xor' (BS.unpack s1) (BS.unpack s2)
-     in IdsHash s'
+  (IdsHash s1) <> (IdsHash s2) = IdsHash $! BS.pack $ BS.zipWith xor s1 s2
 
 instance Monoid IdsHash where
   mempty = IdsHash $ BS.replicate 16 0

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

@@ -6,6 +6,7 @@ module Simplex.Messaging.Server.QueueStore.Postgres.Migrations where
 import Data.List (sortOn)
 import Data.Text (Text)
 import qualified Data.Text as T
+import Simplex.Messaging.Agent.Store.Postgres.Migrations.Util
 import Simplex.Messaging.Agent.Store.Shared
 import Text.RawString.QQ (r)
 
@@ -170,23 +171,31 @@ ALTER TABLE services
   ADD COLUMN queue_count BIGINT NOT NULL DEFAULT 0,
   ADD COLUMN queue_ids_hash BYTEA NOT NULL DEFAULT '\x00000000000000000000000000000000';
 
-WITH acc AS (
-  SELECT
-    s.service_id,
-    count(1) as q_count,
-    xor_aggregate(public.digest(CASE WHEN s.service_role = 'M' THEN q.recipient_id ELSE COALESCE(q.notifier_id, '\x00000000000000000000000000000000') END, 'md5')) AS q_ids_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')
-  WHERE q.deleted_at IS NULL
-  GROUP BY s.service_id
-)
-UPDATE services s
-SET queue_count = COALESCE(acc.q_count, 0),
-    queue_ids_hash = COALESCE(acc.q_ids_hash, '\x00000000000000000000000000000000')
-FROM acc
-WHERE s.service_id = acc.service_id;
+CREATE FUNCTION update_all_aggregates() RETURNS VOID
+LANGUAGE plpgsql
+AS $$
+BEGIN
+  WITH acc AS (
+    SELECT
+      s.service_id,
+      count(1) as q_count,
+      xor_aggregate(public.digest(CASE WHEN s.service_role = 'M' THEN q.recipient_id ELSE COALESCE(q.notifier_id, '\x00000000000000000000000000000000') END, 'md5')) AS q_ids_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')
+    WHERE q.deleted_at IS NULL
+    GROUP BY s.service_id
+  )
+  UPDATE services s
+  SET queue_count = COALESCE(acc.q_count, 0),
+      queue_ids_hash = COALESCE(acc.q_ids_hash, '\x00000000000000000000000000000000')
+  FROM acc
+  WHERE s.service_id = acc.service_id;
+END;
+$$;
 
-CREATE OR REPLACE FUNCTION update_ids_hash(p_service_id BYTEA, p_role TEXT, p_queue_id BYTEA, p_change BIGINT) RETURNS void
+SELECT update_all_aggregates();
+
+CREATE FUNCTION update_aggregates(p_service_id BYTEA, p_role TEXT, p_queue_id BYTEA, p_change BIGINT) RETURNS VOID
 LANGUAGE plpgsql
 AS $$
 BEGIN
@@ -197,69 +206,65 @@ BEGIN
 END;
 $$;
 
-CREATE OR REPLACE FUNCTION on_queue_insert() RETURNS TRIGGER
+CREATE FUNCTION on_queue_insert() RETURNS TRIGGER
 LANGUAGE plpgsql
 AS $$
 BEGIN
   IF NEW.rcv_service_id IS NOT NULL THEN
-    PERFORM update_ids_hash(NEW.rcv_service_id, 'M', NEW.recipient_id, 1);
+    PERFORM update_aggregates(NEW.rcv_service_id, 'M', NEW.recipient_id, 1);
   END IF;
   IF NEW.ntf_service_id IS NOT NULL AND NEW.notifier_id IS NOT NULL THEN
-    PERFORM update_ids_hash(NEW.ntf_service_id, 'N', NEW.notifier_id, 1);
+    PERFORM update_aggregates(NEW.ntf_service_id, 'N', NEW.notifier_id, 1);
   END IF;
   RETURN NEW;
 END;
 $$;
 
-CREATE OR REPLACE FUNCTION on_queue_delete() RETURNS TRIGGER
+CREATE FUNCTION on_queue_delete() RETURNS TRIGGER
 LANGUAGE plpgsql
 AS $$
 BEGIN
   IF OLD.deleted_at IS NULL THEN
     IF OLD.rcv_service_id IS NOT NULL THEN
-      PERFORM update_ids_hash(OLD.rcv_service_id, 'M', OLD.recipient_id, -1);
+      PERFORM update_aggregates(OLD.rcv_service_id, 'M', OLD.recipient_id, -1);
     END IF;
     IF OLD.ntf_service_id IS NOT NULL AND OLD.notifier_id IS NOT NULL THEN
-      PERFORM update_ids_hash(OLD.ntf_service_id, 'N', OLD.notifier_id, -1);
+      PERFORM update_aggregates(OLD.ntf_service_id, 'N', OLD.notifier_id, -1);
     END IF;
   END IF;
   RETURN OLD;
 END;
 $$;
 
-CREATE OR REPLACE FUNCTION on_queue_update() RETURNS TRIGGER
+CREATE FUNCTION on_queue_update() RETURNS TRIGGER
 LANGUAGE plpgsql
 AS $$
 BEGIN
   IF OLD.deleted_at IS NULL AND OLD.rcv_service_id IS NOT NULL THEN
     IF NOT (NEW.deleted_at IS NULL AND NEW.rcv_service_id IS NOT NULL) THEN
-      PERFORM update_ids_hash(OLD.rcv_service_id, 'M', OLD.recipient_id, -1);
+      PERFORM update_aggregates(OLD.rcv_service_id, 'M', OLD.recipient_id, -1);
     ELSIF OLD.rcv_service_id IS DISTINCT FROM NEW.rcv_service_id THEN
-      PERFORM update_ids_hash(OLD.rcv_service_id, 'M', OLD.recipient_id, -1);
-      PERFORM update_ids_hash(NEW.rcv_service_id, 'M', NEW.recipient_id, 1);
+      PERFORM update_aggregates(OLD.rcv_service_id, 'M', OLD.recipient_id, -1);
+      PERFORM update_aggregates(NEW.rcv_service_id, 'M', NEW.recipient_id, 1);
     END IF;
   ELSIF NEW.deleted_at IS NULL AND NEW.rcv_service_id IS NOT NULL THEN
-    PERFORM update_ids_hash(NEW.rcv_service_id, 'M', NEW.recipient_id, 1);
+    PERFORM update_aggregates(NEW.rcv_service_id, 'M', NEW.recipient_id, 1);
   END IF;
 
   IF OLD.deleted_at IS NULL AND OLD.ntf_service_id IS NOT NULL AND OLD.notifier_id IS NOT NULL THEN
     IF NOT (NEW.deleted_at IS NULL AND NEW.ntf_service_id IS NOT NULL AND NEW.notifier_id IS NOT NULL) THEN
-      PERFORM update_ids_hash(OLD.ntf_service_id, 'N', OLD.notifier_id, -1);
+      PERFORM update_aggregates(OLD.ntf_service_id, 'N', OLD.notifier_id, -1);
     ELSIF OLD.ntf_service_id IS DISTINCT FROM NEW.ntf_service_id OR OLD.notifier_id IS DISTINCT FROM NEW.notifier_id THEN
-      PERFORM update_ids_hash(OLD.ntf_service_id, 'N', OLD.notifier_id, -1);
-      PERFORM update_ids_hash(NEW.ntf_service_id, 'N', NEW.notifier_id, 1);
+      PERFORM update_aggregates(OLD.ntf_service_id, 'N', OLD.notifier_id, -1);
+      PERFORM update_aggregates(NEW.ntf_service_id, 'N', NEW.notifier_id, 1);
     END IF;
   ELSIF NEW.deleted_at IS NULL AND NEW.ntf_service_id IS NOT NULL AND NEW.notifier_id IS NOT NULL THEN
-    PERFORM update_ids_hash(NEW.ntf_service_id, 'N', NEW.notifier_id, 1);
+    PERFORM update_aggregates(NEW.ntf_service_id, 'N', NEW.notifier_id, 1);
   END IF;
   RETURN NEW;
 END;
 $$;
 
-DROP TRIGGER IF EXISTS tr_queue_insert ON msg_queues;
-DROP TRIGGER IF EXISTS tr_queue_delete ON msg_queues;
-DROP TRIGGER IF EXISTS tr_queue_update ON msg_queues;
-
 CREATE TRIGGER tr_queue_insert
 AFTER INSERT ON msg_queues
 FOR EACH ROW EXECUTE PROCEDURE on_queue_insert();
@@ -285,47 +290,12 @@ DROP FUNCTION on_queue_insert;
 DROP FUNCTION on_queue_delete;
 DROP FUNCTION on_queue_update;
 
-DROP FUNCTION update_ids_hash;
+DROP FUNCTION update_aggregates;
+
+DROP FUNCTION should_subscribe_status;
 
 ALTER TABLE services
   DROP COLUMN queue_count,
   DROP COLUMN queue_ids_hash;
     |]
     <> dropXorHashFuncs
-
-createXorHashFuncs :: Text
-createXorHashFuncs =
-  T.pack
-    [r|
-CREATE OR REPLACE FUNCTION xor_combine(state BYTEA, value BYTEA) RETURNS BYTEA
-LANGUAGE plpgsql IMMUTABLE STRICT
-AS $$
-DECLARE
-  result BYTEA := state;
-  i INTEGER;
-  len INTEGER := octet_length(value);
-BEGIN
-  IF octet_length(state) != len THEN
-    RAISE EXCEPTION 'Inputs must be equal length (% != %)', octet_length(state), len;
-  END IF;
-  FOR i IN 0..len-1 LOOP
-    result := set_byte(result, i, get_byte(state, i) # get_byte(value, i));
-  END LOOP;
-  RETURN result;
-END;
-$$;
-
-CREATE OR REPLACE AGGREGATE xor_aggregate(BYTEA) (
-  SFUNC = xor_combine,
-  STYPE = BYTEA,
-  INITCOND = '\x00000000000000000000000000000000' -- 16 bytes
-);
-    |]
-
-dropXorHashFuncs :: Text
-dropXorHashFuncs =
-  T.pack
-    [r|
-DROP AGGREGATE xor_aggregate(BYTEA);
-DROP FUNCTION xor_combine;
-    |]

tests/Fixtures.hs

@@ -3,7 +3,9 @@
 module Fixtures where
 
 import Data.ByteString (ByteString)
+import qualified Data.ByteString.Char8 as B
 import Database.PostgreSQL.Simple (ConnectInfo (..), defaultConnectInfo)
+import Simplex.Messaging.Agent.Store.Postgres.Options
 
 testDBConnstr :: ByteString
 testDBConnstr = "postgresql://test_agent_user@/test_agent_db"
@@ -14,3 +16,6 @@ testDBConnectInfo =
     connectUser = "test_agent_user",
     connectDatabase = "test_agent_db"
   }
+
+testDBOpts :: String -> DBOpts
+testDBOpts schema' = DBOpts testDBConnstr (B.pack schema') 1 True

tests/ServerTests.hs

@@ -1376,18 +1376,19 @@ testServiceNotificationsTwoRestarts =
     (nPub, nKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
     serviceKeys@(_, servicePK) <- atomically $ C.generateKeyPair g
     (rcvNtfPubDhKey, _) <- atomically $ C.generateKeyPair g
-    (rId, rKey, sId, dec, serviceId) <- withSmpServerStoreLogOn ps testPort $ runTest2 t $ \sh rh -> do
+    (rId, rKey, sId, dec, nId, serviceId) <- withSmpServerStoreLogOn ps testPort $ runTest2 t $ \sh rh -> do
       (sId, rId, rKey, dhShared) <- createAndSecureQueue rh sPub
       let dec = decryptMsgV3 dhShared
       Resp "0" _ (NID nId _) <- signSendRecv rh rKey ("0", rId, NKEY nPub rcvNtfPubDhKey)
       testNtfServiceClient t serviceKeys $ \nh -> do
         Resp "1" _ (SOK (Just serviceId)) <- serviceSignSendRecv nh nKey servicePK ("1", nId, NSUB)
         deliverMessage rh rId rKey sh sId sKey nh "hello" dec
-        pure (rId, rKey, sId, dec, serviceId)
+        pure (rId, rKey, sId, dec, nId, serviceId)
+    let idsHash = queueIdsHash [nId]
     threadDelay 250000
     withSmpServerStoreLogOn ps testPort $ runTest2 t $ \sh rh ->
       testNtfServiceClient t serviceKeys $ \nh -> do
-        Resp "2.1" serviceId' (SOKS n _) <- signSendRecv nh (C.APrivateAuthKey C.SEd25519 servicePK) ("2.1", serviceId, NSUBS)
+        Resp "2.1" serviceId' (SOKS n _) <- signSendRecv nh (C.APrivateAuthKey C.SEd25519 servicePK) ("2.1", serviceId, NSUBS 1 idsHash)
         n `shouldBe` 1
         Resp "2.2" _ (SOK Nothing) <- signSendRecv rh rKey ("2.2", rId, SUB)
         serviceId' `shouldBe` serviceId
@@ -1395,7 +1396,7 @@ testServiceNotificationsTwoRestarts =
     threadDelay 250000
     withSmpServerStoreLogOn ps testPort $ runTest2 t $ \sh rh ->
       testNtfServiceClient t serviceKeys $ \nh -> do
-        Resp "3.1" _ (SOKS n _) <- signSendRecv nh (C.APrivateAuthKey C.SEd25519 servicePK) ("3.1", serviceId, NSUBS)
+        Resp "3.1" _ (SOKS n _) <- signSendRecv nh (C.APrivateAuthKey C.SEd25519 servicePK) ("3.1", serviceId, NSUBS 1 idsHash)
         n `shouldBe` 1
         Resp "3.2" _ (SOK Nothing) <- signSendRecv rh rKey ("3.2", rId, SUB)
         deliverMessage rh rId rKey sh sId sKey nh "hello 3" dec

tests/Test.hs

@@ -38,6 +38,8 @@ import XFTPServerTests (xftpServerTests)
 
 #if defined(dbPostgres)
 import Fixtures
+import SMPAgentClient (testDB)
+import Simplex.Messaging.Agent.Store.Postgres.Migrations.App
 #else
 import AgentTests.SchemaDump (schemaDumpTest)
 #endif
@@ -45,13 +47,13 @@ import AgentTests.SchemaDump (schemaDumpTest)
 #if defined(dbServerPostgres)
 import NtfServerTests (ntfServerTests)
 import NtfClient (ntfTestServerDBConnectInfo, ntfTestStoreDBOpts)
-import PostgresSchemaDump (postgresSchemaDumpTest)
 import SMPClient (testServerDBConnectInfo, testStoreDBOpts)
 import Simplex.Messaging.Notifications.Server.Store.Migrations (ntfServerMigrations)
 import Simplex.Messaging.Server.QueueStore.Postgres.Migrations (serverMigrations)
 #endif
 
 #if defined(dbPostgres) || defined(dbServerPostgres)
+import PostgresSchemaDump (postgresSchemaDumpTest)
 import SMPClient (postgressBracket)
 #endif
 
@@ -71,8 +73,15 @@ main = do
       . before_ (createDirectoryIfMissing False "tests/tmp")
       . after_ (eventuallyRemove "tests/tmp" 3)
       $ do
--- TODO [postgres] schema dump for postgres
-#if !defined(dbPostgres)
+#if defined(dbPostgres)
+        around_ (postgressBracket testDBConnectInfo) $
+          describe "Agent PostgreSQL schema dump" $
+            postgresSchemaDumpTest
+              appMigrations
+              ["20250322_short_links"] -- snd_secure and last_broker_ts columns swap order on down migration
+              (testDBOpts testDB)
+              "src/Simplex/Messaging/Agent/Store/Postgres/Migrations/agent_postgres_schema.sql"
+#else
         describe "Agent SQLite schema dump" schemaDumpTest
 #endif
         describe "Core tests" $ do