diff --git a/src/Simplex/Messaging/Agent/Store/Postgres/Util.hs b/src/Simplex/Messaging/Agent/Store/Postgres/Util.hs index 0913c76e3..bcbb0e281 100644 --- a/src/Simplex/Messaging/Agent/Store/Postgres/Util.hs +++ b/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 diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql index 5ae592dec..5338c7d04 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql +++ b/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); diff --git a/src/Simplex/Messaging/Notifications/Server/Store/Migrations.hs b/src/Simplex/Messaging/Notifications/Server/Store/Migrations.hs index ca22deab2..155969bbe 100644 --- a/src/Simplex/Messaging/Notifications/Server/Store/Migrations.hs +++ b/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; $$; 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 5b2ac57de..030792905 100644 --- a/src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql +++ b/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; $$; diff --git a/src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs b/src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs index 009e78b9a..0a4fec8c3 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs +++ b/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; |] diff --git a/src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql b/src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql index 744e3a1f5..76f9ded2f 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql +++ b/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 );