smp server: xor hash with triggers

src/Simplex/Messaging/Agent.hs

@@ -1370,7 +1370,7 @@ subscribeClientServices' c userId =
     useService = liftIO $ (Just True ==) <$> TM.lookupIO userId (useClientServices c)
     subscribe = do
       srvs <- withStore' c (`getClientServiceServers` userId)
-      lift $ M.fromList . zip srvs <$> mapConcurrently (tryAgentError' . subscribeClientService c userId) srvs
+      lift $ M.fromList . zip srvs <$> mapConcurrently (tryAllErrors' . subscribeClientService c userId) srvs
 
 -- requesting messages sequentially, to reduce memory usage
 getConnectionMessages' :: AgentClient -> NonEmpty ConnMsgReq -> AM' (NonEmpty (Either AgentErrorType (Maybe SMPMsgMeta)))

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

@@ -7,6 +7,7 @@ import Data.List (sortOn)
 import Data.Text (Text)
 import qualified Data.Text as T
 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)]
@@ -108,36 +109,11 @@ ALTER TABLE subscriptions DROP COLUMN ntf_service_assoc;
 
 m20250830_queue_ids_hash :: Text
 m20250830_queue_ids_hash =
-  T.pack
-    [r|
+  createXorHashFuncs
+    <> T.pack
+      [r|
 ALTER TABLE smp_servers ADD COLUMN smp_notifier_ids_hash BYTEA NOT NULL DEFAULT '\x00000000000000000000000000000000';
 
-CREATE EXTENSION IF NOT EXISTS pgcrypto;
-
-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
-);
-
 WITH hashes AS (
   SELECT
     s.smp_server_id,
@@ -209,7 +185,7 @@ FOR EACH ROW EXECUTE PROCEDURE on_subscription_delete();
 CREATE TRIGGER tr_subscriptions_update
 AFTER UPDATE ON subscriptions
 FOR EACH ROW EXECUTE PROCEDURE on_subscription_update();
-    |]
+      |]
 
 down_m20250830_queue_ids_hash :: Text
 down_m20250830_queue_ids_hash =
@@ -225,9 +201,6 @@ DROP FUNCTION on_subscription_update;
 
 DROP FUNCTION update_ids_hash;
 
-DROP AGGREGATE xor_aggregate(BYTEA);
-DROP FUNCTION xor_combine;
-DROP EXTENSION pgcrypto;
-
 ALTER TABLE smp_servers DROP COLUMN smp_notifier_ids_hash;
     |]
+    <> dropXorHashFuncs

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

@@ -14,7 +14,8 @@ serverSchemaMigrations =
   [ ("20250207_initial", m20250207_initial, Nothing),
     ("20250319_updated_index", m20250319_updated_index, Just down_m20250319_updated_index),
     ("20250320_short_links", m20250320_short_links, Just down_m20250320_short_links),
-    ("20250514_service_certs", m20250514_service_certs, Just down_m20250514_service_certs)
+    ("20250514_service_certs", m20250514_service_certs, Just down_m20250514_service_certs),
+    ("20250830_queue_ids_hash", m20250830_queue_ids_hash, Just down_m20250830_queue_ids_hash)
   ]
 
 -- | The list of migrations in ascending order by date
@@ -159,3 +160,160 @@ DROP INDEX idx_services_service_role;
 
 DROP TABLE services;
     |]
+
+m20250830_queue_ids_hash :: Text
+m20250830_queue_ids_hash =
+  createXorHashFuncs
+    <> T.pack
+      [r|
+ALTER TABLE services ADD COLUMN queue_ids_hash BYTEA;
+
+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
+  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
+)
+UPDATE services s
+SET queue_ids_hash = COALESCE(h.computed_hash, '\x00000000000000000000000000000000')
+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
+LANGUAGE plpgsql
+AS $$
+BEGIN
+  UPDATE services
+  SET queue_ids_hash = xor_combine(queue_ids_hash, digest(p_queue_id, 'md5'))
+  WHERE service_id = p_service_id AND service_role = p_role;
+END;
+$$;
+
+CREATE OR REPLACE 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);
+  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;
+  RETURN NEW;
+END;
+$$;
+
+CREATE OR REPLACE FUNCTION on_queue_delete() RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+BEGIN
+  IF OLD.rcv_service_id IS NOT NULL THEN
+    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_ids_hash(OLD.ntf_service_id, 'N', OLD.notifier_id);
+  END IF;
+  RETURN OLD;
+END;
+$$;
+
+CREATE OR REPLACE FUNCTION on_queue_update() RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+BEGIN
+  IF OLD.rcv_service_id IS DISTINCT FROM NEW.rcv_service_id THEN
+    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
+    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;
+$$;
+
+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();
+
+CREATE TRIGGER tr_queue_delete
+AFTER DELETE ON msg_queues
+FOR EACH ROW EXECUTE PROCEDURE on_queue_delete();
+
+CREATE TRIGGER tr_queue_update
+AFTER UPDATE ON msg_queues
+FOR EACH ROW EXECUTE PROCEDURE on_queue_update();
+      |]
+
+down_m20250830_queue_ids_hash :: Text
+down_m20250830_queue_ids_hash =
+  T.pack
+    [r|
+DROP TRIGGER tr_queue_insert ON msg_queues;
+DROP TRIGGER tr_queue_delete ON msg_queues;
+DROP TRIGGER tr_queue_update ON msg_queues;
+
+DROP FUNCTION on_queue_insert;
+DROP FUNCTION on_queue_delete;
+DROP FUNCTION on_queue_update;
+
+DROP FUNCTION update_ids_hash;
+
+ALTER TABLE services DROP COLUMN queue_ids_hash;
+    |]
+    <> dropXorHashFuncs
+
+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 $$
+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;
+DROP EXTENSION pgcrypto;
+    |]

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

@@ -15,6 +15,96 @@ SET row_security = off;
 CREATE SCHEMA smp_server;
 
 
+
+CREATE FUNCTION smp_server.on_queue_delete() RETURNS trigger
+    LANGUAGE plpgsql
+    AS $$
+BEGIN
+  IF OLD.rcv_service_id IS NOT NULL THEN
+    PERFORM update_queue_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);
+  END IF;
+  RETURN OLD;
+END;
+$$;
+
+
+
+CREATE FUNCTION smp_server.on_queue_insert() RETURNS trigger
+    LANGUAGE plpgsql
+    AS $$
+BEGIN
+  IF NEW.rcv_service_id IS NOT NULL THEN
+    PERFORM update_queue_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);
+  END IF;
+  RETURN NEW;
+END;
+$$;
+
+
+
+CREATE FUNCTION smp_server.on_queue_update() RETURNS trigger
+    LANGUAGE plpgsql
+    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);
+  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);
+  END IF;
+  RETURN NEW;
+END;
+$$;
+
+
+
+CREATE FUNCTION smp_server.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'))
+  WHERE service_id = p_service_id AND service_role = p_role;
+END;
+$$;
+
+
+
+CREATE FUNCTION smp_server.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 AGGREGATE smp_server.xor_aggregate(bytea) (
+    SFUNC = smp_server.xor_combine,
+    STYPE = bytea,
+    INITCOND = '\x00000000000000000000000000000000'
+);
+
+
 SET default_table_access_method = heap;
 
 
@@ -53,7 +143,8 @@ CREATE TABLE smp_server.services (
     service_role text NOT NULL,
     service_cert bytea NOT NULL,
     service_cert_hash bytea NOT NULL,
-    created_at bigint NOT NULL
+    created_at bigint NOT NULL,
+    queue_ids_hash bytea
 );
 
 
@@ -106,6 +197,18 @@ CREATE INDEX idx_services_service_role ON smp_server.services USING btree (servi
 
 
 
+CREATE TRIGGER tr_queue_delete AFTER DELETE ON smp_server.msg_queues FOR EACH ROW EXECUTE FUNCTION smp_server.on_queue_delete();
+
+
+
+CREATE TRIGGER tr_queue_insert AFTER INSERT ON smp_server.msg_queues FOR EACH ROW EXECUTE FUNCTION smp_server.on_queue_insert();
+
+
+
+CREATE TRIGGER tr_queue_update AFTER UPDATE ON smp_server.msg_queues FOR EACH ROW EXECUTE FUNCTION smp_server.on_queue_update();
+
+
+
 ALTER TABLE ONLY smp_server.msg_queues
     ADD CONSTRAINT msg_queues_ntf_service_id_fkey FOREIGN KEY (ntf_service_id) REFERENCES smp_server.services(service_id) ON UPDATE RESTRICT ON DELETE SET NULL;
 

tests/ServerTests.hs

@@ -724,6 +724,7 @@ testServiceDeliverSubscribe =
                     pure $ Just $ Just mId3
                   _ -> pure Nothing
               ]
+        Resp "" NoEntity SALL <- tGet1 sh
         Resp "12" _ OK <- signSendRecv sh rKey ("12", rId, ACK mId3)
         Resp "14" _ OK <- signSendRecv h sKey ("14", sId, _SEND "hello 4")
         Resp "" _ (Msg mId4 msg4) <- tGet1 sh