diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20250815_service_certs.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20250815_service_certs.hs index e24869606..fdbfc9e25 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20250815_service_certs.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20250815_service_certs.hs @@ -17,6 +17,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 ); diff --git a/src/Simplex/Messaging/Notifications/Server/Store/Migrations.hs b/src/Simplex/Messaging/Notifications/Server/Store/Migrations.hs index 226a02dc6..55456ac4d 100644 --- a/src/Simplex/Messaging/Notifications/Server/Store/Migrations.hs +++ b/src/Simplex/Messaging/Notifications/Server/Store/Migrations.hs @@ -12,7 +12,8 @@ import Text.RawString.QQ (r) ntfServerSchemaMigrations :: [(String, Text, Maybe Text)] ntfServerSchemaMigrations = [ ("20250417_initial", m20250417_initial, Nothing), - ("20250517_service_cert", m20250517_service_cert, Just down_m20250517_service_cert) + ("20250517_service_cert", m20250517_service_cert, Just down_m20250517_service_cert), + ("20250830_queue_ids_hash", m20250830_queue_ids_hash, Just down_m20250830_queue_ids_hash) ] -- | The list of migrations in ascending order by date @@ -104,3 +105,129 @@ ALTER TABLE smp_servers DROP COLUMN ntf_service_id; ALTER TABLE subscriptions DROP COLUMN ntf_service_assoc; |] + +m20250830_queue_ids_hash :: Text +m20250830_queue_ids_hash = + 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, + xor_aggregate(digest(s.smp_notifier_id, 'md5')) AS notifier_hash + FROM subscriptions s + WHERE s.ntf_service_assoc = true + GROUP BY s.smp_server_id +) +UPDATE smp_servers srv +SET smp_notifier_ids_hash = COALESCE(h.notifier_hash, '\x00000000000000000000000000000000') +FROM hashes h +WHERE srv.smp_server_id = h.smp_server_id; + +CREATE OR REPLACE FUNCTION update_ids_hash(p_server_id BIGINT, p_notifier_id BYTEA) RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + UPDATE smp_servers + SET smp_notifier_ids_hash = xor_combine(smp_notifier_ids_hash, digest(p_notifier_id, 'md5')) + WHERE smp_server_id = p_server_id; +END; +$$; + +CREATE OR REPLACE FUNCTION on_subscription_insert() RETURNS TRIGGER +LANGUAGE plpgsql +AS $$ +BEGIN + IF NEW.ntf_service_assoc = true THEN + PERFORM update_ids_hash(NEW.smp_server_id, NEW.smp_notifier_id); + END IF; + RETURN NEW; +END; +$$; + +CREATE OR REPLACE FUNCTION on_subscription_delete() RETURNS TRIGGER +LANGUAGE plpgsql +AS $$ +BEGIN + IF OLD.ntf_service_assoc = true THEN + PERFORM update_ids_hash(OLD.smp_server_id, OLD.smp_notifier_id); + END IF; + RETURN OLD; +END; +$$; + +CREATE OR REPLACE FUNCTION on_subscription_update() RETURNS TRIGGER +LANGUAGE plpgsql +AS $$ +BEGIN + IF OLD.ntf_service_assoc != NEW.ntf_service_assoc THEN + PERFORM update_ids_hash(NEW.smp_server_id, NEW.smp_notifier_id); + END IF; + RETURN NEW; +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(); + +CREATE TRIGGER tr_subscriptions_delete +AFTER DELETE ON subscriptions +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 = + T.pack + [r| +DROP TRIGGER tr_subscriptions_insert ON subscriptions; +DROP TRIGGER tr_subscriptions_delete ON subscriptions; +DROP TRIGGER tr_subscriptions_update ON subscriptions; + +DROP FUNCTION on_subscription_insert; +DROP FUNCTION on_subscription_delete; +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; + |] diff --git a/src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql b/src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql index 3b155fa1a..5b2ac57de 100644 --- a/src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql +++ b/src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql @@ -15,6 +15,85 @@ SET row_security = off; CREATE SCHEMA ntf_server; + +CREATE FUNCTION ntf_server.on_subscription_delete() RETURNS trigger + LANGUAGE plpgsql + AS $$ +BEGIN + IF OLD.ntf_service_assoc = true THEN + PERFORM update_ids_hash(OLD.smp_server_id, OLD.smp_notifier_id); + END IF; + RETURN OLD; +END; +$$; + + + +CREATE FUNCTION ntf_server.on_subscription_insert() RETURNS trigger + LANGUAGE plpgsql + AS $$ +BEGIN + IF NEW.ntf_service_assoc = true THEN + PERFORM update_ids_hash(NEW.smp_server_id, NEW.smp_notifier_id); + END IF; + RETURN NEW; +END; +$$; + + + +CREATE FUNCTION ntf_server.on_subscription_update() RETURNS trigger + LANGUAGE plpgsql + AS $$ +BEGIN + IF OLD.ntf_service_assoc != NEW.ntf_service_assoc THEN + PERFORM update_ids_hash(NEW.smp_server_id, NEW.smp_notifier_id); + END IF; + RETURN NEW; +END; +$$; + + + +CREATE FUNCTION ntf_server.update_ids_hash(p_server_id bigint, p_notifier_id bytea) RETURNS void + LANGUAGE plpgsql + AS $$ +BEGIN + UPDATE smp_servers + SET smp_notifier_ids_hash = xor_combine(smp_notifier_ids_hash, digest(p_notifier_id, 'md5')) + WHERE smp_server_id = p_server_id; +END; +$$; + + + +CREATE FUNCTION ntf_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 ntf_server.xor_aggregate(bytea) ( + SFUNC = ntf_server.xor_combine, + STYPE = bytea, + INITCOND = '\x00000000000000000000000000000000' +); + + SET default_table_access_method = heap; @@ -53,7 +132,8 @@ CREATE TABLE ntf_server.smp_servers ( smp_host text NOT NULL, smp_port text NOT NULL, smp_keyhash bytea NOT NULL, - ntf_service_id bytea + ntf_service_id bytea, + smp_notifier_ids_hash bytea DEFAULT '\x00000000000000000000000000000000'::bytea NOT NULL ); @@ -158,6 +238,18 @@ CREATE INDEX idx_tokens_status_cron_interval_sent_at ON ntf_server.tokens USING +CREATE TRIGGER tr_subscriptions_delete AFTER DELETE ON ntf_server.subscriptions FOR EACH ROW EXECUTE FUNCTION ntf_server.on_subscription_delete(); + + + +CREATE TRIGGER tr_subscriptions_insert AFTER INSERT ON ntf_server.subscriptions FOR EACH ROW EXECUTE FUNCTION ntf_server.on_subscription_insert(); + + + +CREATE TRIGGER tr_subscriptions_update AFTER UPDATE ON ntf_server.subscriptions FOR EACH ROW EXECUTE FUNCTION ntf_server.on_subscription_update(); + + + ALTER TABLE ONLY ntf_server.last_notifications ADD CONSTRAINT last_notifications_subscription_id_fkey FOREIGN KEY (subscription_id) REFERENCES ntf_server.subscriptions(subscription_id) ON UPDATE RESTRICT ON DELETE CASCADE; diff --git a/src/Simplex/Messaging/Server/QueueStore.hs b/src/Simplex/Messaging/Server/QueueStore.hs index 9395f5bac..ccc44e92c 100644 --- a/src/Simplex/Messaging/Server/QueueStore.hs +++ b/src/Simplex/Messaging/Server/QueueStore.hs @@ -68,6 +68,7 @@ data ServiceRec = ServiceRec serviceCert :: X.CertificateChain, serviceCertHash :: XV.Fingerprint, -- SHA512 hash of long-term service client certificate. See comment for ClientHandshake. serviceCreatedAt :: RoundedSystemTime + -- entitiesHash :: IdsHash -- a xor-hash of all associated entities } deriving (Show)