diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 825d124bf..04872ce64 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/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))) diff --git a/src/Simplex/Messaging/Notifications/Server/Store/Migrations.hs b/src/Simplex/Messaging/Notifications/Server/Store/Migrations.hs index 55456ac4d..ca22deab2 100644 --- a/src/Simplex/Messaging/Notifications/Server/Store/Migrations.hs +++ b/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 diff --git a/src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs b/src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs index e8469d1cc..009e78b9a 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs +++ b/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; + |] diff --git a/src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql b/src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql index 6c0501d8b..744e3a1f5 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql +++ b/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; diff --git a/tests/ServerTests.hs b/tests/ServerTests.hs index 2eacc78f0..cad598a0e 100644 --- a/tests/ServerTests.hs +++ b/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