diff --git a/simplexmq.cabal b/simplexmq.cabal index d4d3993a6..c5bf914f5 100644 --- a/simplexmq.cabal +++ b/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: diff --git a/src/Simplex/Messaging/Agent/Store/AgentStore.hs b/src/Simplex/Messaging/Agent/Store/AgentStore.hs index 41a65b04f..cde6b56c8 100644 --- a/src/Simplex/Messaging/Agent/Store/AgentStore.hs +++ b/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_) ) = diff --git a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/App.hs b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/App.hs index 565a06760..e77b2382b 100644 --- a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/App.hs +++ b/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 diff --git a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20250815_service_certs.hs b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20250815_service_certs.hs new file mode 100644 index 000000000..139f94e26 --- /dev/null +++ b/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 diff --git a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/Util.hs b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/Util.hs new file mode 100644 index 000000000..7b05cd890 --- /dev/null +++ b/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; + |] diff --git a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/agent_postgres_schema.sql b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/agent_postgres_schema.sql new file mode 100644 index 000000000..8c3e2955f --- /dev/null +++ b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/agent_postgres_schema.sql @@ -0,0 +1,1417 @@ + + +SET statement_timeout = 0; +SET lock_timeout = 0; +SET idle_in_transaction_session_timeout = 0; +SET client_encoding = 'UTF8'; +SET standard_conforming_strings = on; +SELECT pg_catalog.set_config('search_path', '', false); +SET check_function_bodies = false; +SET xmloption = content; +SET client_min_messages = warning; +SET row_security = off; + + +CREATE SCHEMA smp_agent_test_protocol_schema; + + + +CREATE FUNCTION smp_agent_test_protocol_schema.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 smp_agent_test_protocol_schema.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 smp_agent_test_protocol_schema.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 FUNCTION smp_agent_test_protocol_schema.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 smp_agent_test_protocol_schema.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_agent_test_protocol_schema.xor_aggregate(bytea) ( + SFUNC = smp_agent_test_protocol_schema.xor_combine, + STYPE = bytea, + INITCOND = '\x00000000000000000000000000000000' +); + + +SET default_table_access_method = heap; + + +CREATE TABLE smp_agent_test_protocol_schema.client_services ( + user_id bigint NOT NULL, + 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 DEFAULT 0 NOT NULL, + service_queue_ids_hash bytea DEFAULT '\x00000000000000000000000000000000'::bytea NOT NULL +); + + + +CREATE TABLE smp_agent_test_protocol_schema.commands ( + command_id bigint NOT NULL, + conn_id bytea NOT NULL, + host text, + port text, + corr_id bytea NOT NULL, + command_tag bytea NOT NULL, + command bytea NOT NULL, + agent_version integer DEFAULT 1 NOT NULL, + server_key_hash bytea, + created_at timestamp with time zone DEFAULT '1970-01-01 00:00:00+01'::timestamp with time zone NOT NULL, + failed smallint DEFAULT 0 +); + + + +ALTER TABLE smp_agent_test_protocol_schema.commands ALTER COLUMN command_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.commands_command_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.conn_confirmations ( + confirmation_id bytea NOT NULL, + conn_id bytea NOT NULL, + e2e_snd_pub_key bytea NOT NULL, + sender_key bytea, + ratchet_state bytea NOT NULL, + sender_conn_info bytea NOT NULL, + accepted smallint NOT NULL, + own_conn_info bytea, + created_at timestamp with time zone DEFAULT now() NOT NULL, + smp_reply_queues bytea, + smp_client_version integer +); + + + +CREATE TABLE smp_agent_test_protocol_schema.conn_invitations ( + invitation_id bytea NOT NULL, + contact_conn_id bytea, + cr_invitation bytea NOT NULL, + recipient_conn_info bytea NOT NULL, + accepted smallint DEFAULT 0 NOT NULL, + own_conn_info bytea, + created_at timestamp with time zone DEFAULT now() NOT NULL +); + + + +CREATE TABLE smp_agent_test_protocol_schema.connections ( + conn_id bytea NOT NULL, + conn_mode text NOT NULL, + last_internal_msg_id bigint DEFAULT 0 NOT NULL, + last_internal_rcv_msg_id bigint DEFAULT 0 NOT NULL, + last_internal_snd_msg_id bigint DEFAULT 0 NOT NULL, + last_external_snd_msg_id bigint DEFAULT 0 NOT NULL, + last_rcv_msg_hash bytea DEFAULT '\x'::bytea NOT NULL, + last_snd_msg_hash bytea DEFAULT '\x'::bytea NOT NULL, + smp_agent_version integer DEFAULT 1 NOT NULL, + duplex_handshake smallint DEFAULT 0, + enable_ntfs smallint, + deleted smallint DEFAULT 0 NOT NULL, + user_id bigint NOT NULL, + ratchet_sync_state text DEFAULT 'ok'::text NOT NULL, + deleted_at_wait_delivery timestamp with time zone, + pq_support smallint DEFAULT 0 NOT NULL +); + + + +CREATE TABLE smp_agent_test_protocol_schema.deleted_snd_chunk_replicas ( + deleted_snd_chunk_replica_id bigint NOT NULL, + user_id bigint NOT NULL, + xftp_server_id bigint NOT NULL, + replica_id bytea NOT NULL, + replica_key bytea NOT NULL, + chunk_digest bytea NOT NULL, + delay bigint, + retries bigint DEFAULT 0 NOT NULL, + created_at timestamp with time zone DEFAULT now() NOT NULL, + updated_at timestamp with time zone DEFAULT now() NOT NULL, + failed smallint DEFAULT 0 +); + + + +ALTER TABLE smp_agent_test_protocol_schema.deleted_snd_chunk_replicas ALTER COLUMN deleted_snd_chunk_replica_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.deleted_snd_chunk_replicas_deleted_snd_chunk_replica_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.encrypted_rcv_message_hashes ( + encrypted_rcv_message_hash_id bigint NOT NULL, + conn_id bytea NOT NULL, + hash bytea NOT NULL, + created_at timestamp with time zone DEFAULT now() NOT NULL, + updated_at timestamp with time zone DEFAULT now() NOT NULL +); + + + +ALTER TABLE smp_agent_test_protocol_schema.encrypted_rcv_message_hashes ALTER COLUMN encrypted_rcv_message_hash_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.encrypted_rcv_message_hashes_encrypted_rcv_message_hash_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.inv_short_links ( + inv_short_link_id bigint NOT NULL, + host text NOT NULL, + port text NOT NULL, + server_key_hash bytea, + link_id bytea NOT NULL, + link_key bytea NOT NULL, + snd_private_key bytea NOT NULL, + snd_id bytea +); + + + +ALTER TABLE smp_agent_test_protocol_schema.inv_short_links ALTER COLUMN inv_short_link_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.inv_short_links_inv_short_link_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.messages ( + conn_id bytea NOT NULL, + internal_id bigint NOT NULL, + internal_ts timestamp with time zone NOT NULL, + internal_rcv_id bigint, + internal_snd_id bigint, + msg_type bytea NOT NULL, + msg_body bytea DEFAULT '\x'::bytea NOT NULL, + msg_flags text, + pq_encryption smallint DEFAULT 0 NOT NULL +); + + + +CREATE TABLE smp_agent_test_protocol_schema.migrations ( + name text NOT NULL, + ts timestamp without time zone NOT NULL, + down text +); + + + +CREATE TABLE smp_agent_test_protocol_schema.ntf_servers ( + ntf_host text NOT NULL, + ntf_port text NOT NULL, + ntf_key_hash bytea NOT NULL, + created_at timestamp with time zone DEFAULT now() NOT NULL, + updated_at timestamp with time zone DEFAULT now() NOT NULL +); + + + +CREATE TABLE smp_agent_test_protocol_schema.ntf_subscriptions ( + conn_id bytea NOT NULL, + smp_host text, + smp_port text, + smp_ntf_id bytea, + ntf_host text NOT NULL, + ntf_port text NOT NULL, + ntf_sub_id bytea, + ntf_sub_status text NOT NULL, + ntf_sub_action bytea, + ntf_sub_smp_action bytea, + ntf_sub_action_ts timestamp with time zone, + updated_by_supervisor smallint DEFAULT 0 NOT NULL, + created_at timestamp with time zone DEFAULT now() NOT NULL, + updated_at timestamp with time zone DEFAULT now() NOT NULL, + smp_server_key_hash bytea, + ntf_failed smallint DEFAULT 0, + smp_failed smallint DEFAULT 0 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.ntf_tokens ( + provider text NOT NULL, + device_token text NOT NULL, + ntf_host text NOT NULL, + ntf_port text NOT NULL, + tkn_id bytea, + tkn_pub_key bytea NOT NULL, + tkn_priv_key bytea NOT NULL, + tkn_pub_dh_key bytea NOT NULL, + tkn_priv_dh_key bytea NOT NULL, + tkn_dh_secret bytea, + tkn_status text NOT NULL, + tkn_action bytea, + created_at timestamp with time zone DEFAULT now() NOT NULL, + updated_at timestamp with time zone DEFAULT now() NOT NULL, + ntf_mode bytea +); + + + +CREATE TABLE smp_agent_test_protocol_schema.ntf_tokens_to_delete ( + ntf_token_to_delete_id bigint NOT NULL, + ntf_host text NOT NULL, + ntf_port text NOT NULL, + ntf_key_hash bytea NOT NULL, + tkn_id bytea NOT NULL, + tkn_priv_key bytea NOT NULL, + del_failed smallint DEFAULT 0, + created_at timestamp with time zone DEFAULT now() NOT NULL +); + + + +ALTER TABLE smp_agent_test_protocol_schema.ntf_tokens_to_delete ALTER COLUMN ntf_token_to_delete_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.ntf_tokens_to_delete_ntf_token_to_delete_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.processed_ratchet_key_hashes ( + processed_ratchet_key_hash_id bigint NOT NULL, + conn_id bytea NOT NULL, + hash bytea NOT NULL, + created_at timestamp with time zone DEFAULT now() NOT NULL, + updated_at timestamp with time zone DEFAULT now() NOT NULL +); + + + +ALTER TABLE smp_agent_test_protocol_schema.processed_ratchet_key_hashes ALTER COLUMN processed_ratchet_key_hash_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.processed_ratchet_key_hashes_processed_ratchet_key_hash_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.ratchets ( + conn_id bytea NOT NULL, + x3dh_priv_key_1 bytea, + x3dh_priv_key_2 bytea, + ratchet_state bytea, + e2e_version integer DEFAULT 1 NOT NULL, + x3dh_pub_key_1 bytea, + x3dh_pub_key_2 bytea, + pq_priv_kem bytea, + pq_pub_kem bytea +); + + + +CREATE TABLE smp_agent_test_protocol_schema.rcv_file_chunk_replicas ( + rcv_file_chunk_replica_id bigint NOT NULL, + rcv_file_chunk_id bigint NOT NULL, + replica_number bigint NOT NULL, + xftp_server_id bigint NOT NULL, + replica_id bytea NOT NULL, + replica_key bytea NOT NULL, + received smallint DEFAULT 0 NOT NULL, + delay bigint, + retries bigint DEFAULT 0 NOT NULL, + created_at timestamp with time zone DEFAULT now() NOT NULL, + updated_at timestamp with time zone DEFAULT now() NOT NULL +); + + + +ALTER TABLE smp_agent_test_protocol_schema.rcv_file_chunk_replicas ALTER COLUMN rcv_file_chunk_replica_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.rcv_file_chunk_replicas_rcv_file_chunk_replica_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.rcv_file_chunks ( + rcv_file_chunk_id bigint NOT NULL, + rcv_file_id bigint NOT NULL, + chunk_no bigint NOT NULL, + chunk_size bigint NOT NULL, + digest bytea NOT NULL, + tmp_path text, + created_at timestamp with time zone DEFAULT now() NOT NULL, + updated_at timestamp with time zone DEFAULT now() NOT NULL +); + + + +ALTER TABLE smp_agent_test_protocol_schema.rcv_file_chunks ALTER COLUMN rcv_file_chunk_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.rcv_file_chunks_rcv_file_chunk_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.rcv_files ( + rcv_file_id bigint NOT NULL, + rcv_file_entity_id bytea NOT NULL, + user_id bigint NOT NULL, + size bigint NOT NULL, + digest bytea NOT NULL, + key bytea NOT NULL, + nonce bytea NOT NULL, + chunk_size bigint NOT NULL, + prefix_path text NOT NULL, + tmp_path text, + save_path text NOT NULL, + status text NOT NULL, + deleted smallint DEFAULT 0 NOT NULL, + error text, + created_at timestamp with time zone DEFAULT now() NOT NULL, + updated_at timestamp with time zone DEFAULT now() NOT NULL, + save_file_key bytea, + save_file_nonce bytea, + failed smallint DEFAULT 0, + redirect_id bigint, + redirect_entity_id bytea, + redirect_size bigint, + redirect_digest bytea, + approved_relays smallint DEFAULT 0 NOT NULL +); + + + +ALTER TABLE smp_agent_test_protocol_schema.rcv_files ALTER COLUMN rcv_file_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.rcv_files_rcv_file_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.rcv_messages ( + conn_id bytea NOT NULL, + internal_rcv_id bigint NOT NULL, + internal_id bigint NOT NULL, + external_snd_id bigint NOT NULL, + broker_id bytea NOT NULL, + broker_ts timestamp with time zone NOT NULL, + internal_hash bytea NOT NULL, + external_prev_snd_hash bytea NOT NULL, + integrity bytea NOT NULL, + user_ack smallint DEFAULT 0, + rcv_queue_id bigint NOT NULL +); + + + +CREATE TABLE smp_agent_test_protocol_schema.rcv_queues ( + host text NOT NULL, + port text NOT NULL, + rcv_id bytea NOT NULL, + conn_id bytea NOT NULL, + rcv_private_key bytea NOT NULL, + rcv_dh_secret bytea NOT NULL, + e2e_priv_key bytea NOT NULL, + e2e_dh_secret bytea, + snd_id bytea NOT NULL, + snd_key bytea, + status text NOT NULL, + smp_server_version integer DEFAULT 1 NOT NULL, + smp_client_version integer, + ntf_public_key bytea, + ntf_private_key bytea, + ntf_id bytea, + rcv_ntf_dh_secret bytea, + rcv_queue_id bigint NOT NULL, + rcv_primary smallint NOT NULL, + replace_rcv_queue_id bigint, + delete_errors bigint DEFAULT 0 NOT NULL, + server_key_hash bytea, + switch_status text, + deleted smallint DEFAULT 0 NOT NULL, + last_broker_ts timestamp with time zone, + link_id bytea, + link_key bytea, + link_priv_sig_key bytea, + link_enc_fixed_data bytea, + queue_mode text, + rcv_service_assoc smallint DEFAULT 0 NOT NULL +); + + + +CREATE TABLE smp_agent_test_protocol_schema.servers ( + host text NOT NULL, + port text NOT NULL, + key_hash bytea NOT NULL +); + + + +CREATE TABLE smp_agent_test_protocol_schema.servers_stats ( + servers_stats_id bigint NOT NULL, + servers_stats text, + started_at timestamp with time zone DEFAULT now() NOT NULL, + created_at timestamp with time zone DEFAULT now() NOT NULL, + updated_at timestamp with time zone DEFAULT now() NOT NULL +); + + + +ALTER TABLE smp_agent_test_protocol_schema.servers_stats ALTER COLUMN servers_stats_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.servers_stats_servers_stats_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.skipped_messages ( + skipped_message_id bigint NOT NULL, + conn_id bytea NOT NULL, + header_key bytea NOT NULL, + msg_n bigint NOT NULL, + msg_key bytea NOT NULL +); + + + +ALTER TABLE smp_agent_test_protocol_schema.skipped_messages ALTER COLUMN skipped_message_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.skipped_messages_skipped_message_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.snd_file_chunk_replica_recipients ( + snd_file_chunk_replica_recipient_id bigint NOT NULL, + snd_file_chunk_replica_id bigint NOT NULL, + rcv_replica_id bytea NOT NULL, + rcv_replica_key bytea NOT NULL, + created_at timestamp with time zone DEFAULT now() NOT NULL, + updated_at timestamp with time zone DEFAULT now() NOT NULL +); + + + +ALTER TABLE smp_agent_test_protocol_schema.snd_file_chunk_replica_recipients ALTER COLUMN snd_file_chunk_replica_recipient_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.snd_file_chunk_replica_recipi_snd_file_chunk_replica_recipi_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.snd_file_chunk_replicas ( + snd_file_chunk_replica_id bigint NOT NULL, + snd_file_chunk_id bigint NOT NULL, + replica_number bigint NOT NULL, + xftp_server_id bigint NOT NULL, + replica_id bytea NOT NULL, + replica_key bytea NOT NULL, + replica_status text NOT NULL, + delay bigint, + retries bigint DEFAULT 0 NOT NULL, + created_at timestamp with time zone DEFAULT now() NOT NULL, + updated_at timestamp with time zone DEFAULT now() NOT NULL +); + + + +ALTER TABLE smp_agent_test_protocol_schema.snd_file_chunk_replicas ALTER COLUMN snd_file_chunk_replica_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.snd_file_chunk_replicas_snd_file_chunk_replica_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.snd_file_chunks ( + snd_file_chunk_id bigint NOT NULL, + snd_file_id bigint NOT NULL, + chunk_no bigint NOT NULL, + chunk_offset bigint NOT NULL, + chunk_size bigint NOT NULL, + digest bytea NOT NULL, + created_at timestamp with time zone DEFAULT now() NOT NULL, + updated_at timestamp with time zone DEFAULT now() NOT NULL +); + + + +ALTER TABLE smp_agent_test_protocol_schema.snd_file_chunks ALTER COLUMN snd_file_chunk_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.snd_file_chunks_snd_file_chunk_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.snd_files ( + snd_file_id bigint NOT NULL, + snd_file_entity_id bytea NOT NULL, + user_id bigint NOT NULL, + num_recipients bigint NOT NULL, + digest bytea, + key bytea NOT NULL, + nonce bytea NOT NULL, + path text NOT NULL, + prefix_path text, + status text NOT NULL, + deleted smallint DEFAULT 0 NOT NULL, + error text, + created_at timestamp with time zone DEFAULT now() NOT NULL, + updated_at timestamp with time zone DEFAULT now() NOT NULL, + src_file_key bytea, + src_file_nonce bytea, + failed smallint DEFAULT 0, + redirect_size bigint, + redirect_digest bytea +); + + + +ALTER TABLE smp_agent_test_protocol_schema.snd_files ALTER COLUMN snd_file_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.snd_files_snd_file_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.snd_message_bodies ( + snd_message_body_id bigint NOT NULL, + agent_msg bytea DEFAULT '\x'::bytea NOT NULL +); + + + +ALTER TABLE smp_agent_test_protocol_schema.snd_message_bodies ALTER COLUMN snd_message_body_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.snd_message_bodies_snd_message_body_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.snd_message_deliveries ( + snd_message_delivery_id bigint NOT NULL, + conn_id bytea NOT NULL, + snd_queue_id bigint NOT NULL, + internal_id bigint NOT NULL, + failed smallint DEFAULT 0 +); + + + +ALTER TABLE smp_agent_test_protocol_schema.snd_message_deliveries ALTER COLUMN snd_message_delivery_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.snd_message_deliveries_snd_message_delivery_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.snd_messages ( + conn_id bytea NOT NULL, + internal_snd_id bigint NOT NULL, + internal_id bigint NOT NULL, + internal_hash bytea NOT NULL, + previous_msg_hash bytea DEFAULT '\x'::bytea NOT NULL, + retry_int_slow bigint, + retry_int_fast bigint, + rcpt_internal_id bigint, + rcpt_status text, + msg_encrypt_key bytea, + padded_msg_len bigint, + snd_message_body_id bigint +); + + + +CREATE TABLE smp_agent_test_protocol_schema.snd_queues ( + host text NOT NULL, + port text NOT NULL, + snd_id bytea NOT NULL, + conn_id bytea NOT NULL, + snd_private_key bytea NOT NULL, + e2e_dh_secret bytea NOT NULL, + status text NOT NULL, + smp_server_version integer DEFAULT 1 NOT NULL, + smp_client_version integer DEFAULT 1 NOT NULL, + snd_public_key bytea, + e2e_pub_key bytea, + snd_queue_id bigint NOT NULL, + snd_primary smallint NOT NULL, + replace_snd_queue_id bigint, + server_key_hash bytea, + switch_status text, + queue_mode text +); + + + +CREATE TABLE smp_agent_test_protocol_schema.users ( + user_id bigint NOT NULL, + deleted smallint DEFAULT 0 NOT NULL +); + + + +ALTER TABLE smp_agent_test_protocol_schema.users ALTER COLUMN user_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.users_user_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE smp_agent_test_protocol_schema.xftp_servers ( + xftp_server_id bigint NOT NULL, + xftp_host text NOT NULL, + xftp_port text NOT NULL, + xftp_key_hash bytea NOT NULL, + created_at timestamp with time zone DEFAULT now() NOT NULL, + updated_at timestamp with time zone DEFAULT now() NOT NULL +); + + + +ALTER TABLE smp_agent_test_protocol_schema.xftp_servers ALTER COLUMN xftp_server_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME smp_agent_test_protocol_schema.xftp_servers_xftp_server_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.commands + ADD CONSTRAINT commands_pkey PRIMARY KEY (command_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.conn_confirmations + ADD CONSTRAINT conn_confirmations_pkey PRIMARY KEY (confirmation_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.conn_invitations + ADD CONSTRAINT conn_invitations_pkey PRIMARY KEY (invitation_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.connections + ADD CONSTRAINT connections_pkey PRIMARY KEY (conn_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.deleted_snd_chunk_replicas + ADD CONSTRAINT deleted_snd_chunk_replicas_pkey PRIMARY KEY (deleted_snd_chunk_replica_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.encrypted_rcv_message_hashes + ADD CONSTRAINT encrypted_rcv_message_hashes_pkey PRIMARY KEY (encrypted_rcv_message_hash_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.inv_short_links + ADD CONSTRAINT inv_short_links_pkey PRIMARY KEY (inv_short_link_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.messages + ADD CONSTRAINT messages_pkey PRIMARY KEY (conn_id, internal_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.migrations + ADD CONSTRAINT migrations_pkey PRIMARY KEY (name); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.ntf_servers + ADD CONSTRAINT ntf_servers_pkey PRIMARY KEY (ntf_host, ntf_port); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.ntf_subscriptions + ADD CONSTRAINT ntf_subscriptions_pkey PRIMARY KEY (conn_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.ntf_tokens + ADD CONSTRAINT ntf_tokens_pkey PRIMARY KEY (provider, device_token, ntf_host, ntf_port); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.ntf_tokens_to_delete + ADD CONSTRAINT ntf_tokens_to_delete_pkey PRIMARY KEY (ntf_token_to_delete_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.processed_ratchet_key_hashes + ADD CONSTRAINT processed_ratchet_key_hashes_pkey PRIMARY KEY (processed_ratchet_key_hash_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.ratchets + ADD CONSTRAINT ratchets_pkey PRIMARY KEY (conn_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.rcv_file_chunk_replicas + ADD CONSTRAINT rcv_file_chunk_replicas_pkey PRIMARY KEY (rcv_file_chunk_replica_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.rcv_file_chunks + ADD CONSTRAINT rcv_file_chunks_pkey PRIMARY KEY (rcv_file_chunk_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.rcv_files + ADD CONSTRAINT rcv_files_pkey PRIMARY KEY (rcv_file_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.rcv_files + ADD CONSTRAINT rcv_files_rcv_file_entity_id_key UNIQUE (rcv_file_entity_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.rcv_messages + ADD CONSTRAINT rcv_messages_pkey PRIMARY KEY (conn_id, internal_rcv_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.rcv_queues + ADD CONSTRAINT rcv_queues_host_port_snd_id_key UNIQUE (host, port, snd_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.rcv_queues + ADD CONSTRAINT rcv_queues_pkey PRIMARY KEY (host, port, rcv_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.servers + ADD CONSTRAINT servers_pkey PRIMARY KEY (host, port); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.servers_stats + ADD CONSTRAINT servers_stats_pkey PRIMARY KEY (servers_stats_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.skipped_messages + ADD CONSTRAINT skipped_messages_pkey PRIMARY KEY (skipped_message_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_file_chunk_replica_recipients + ADD CONSTRAINT snd_file_chunk_replica_recipients_pkey PRIMARY KEY (snd_file_chunk_replica_recipient_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_file_chunk_replicas + ADD CONSTRAINT snd_file_chunk_replicas_pkey PRIMARY KEY (snd_file_chunk_replica_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_file_chunks + ADD CONSTRAINT snd_file_chunks_pkey PRIMARY KEY (snd_file_chunk_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_files + ADD CONSTRAINT snd_files_pkey PRIMARY KEY (snd_file_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_message_bodies + ADD CONSTRAINT snd_message_bodies_pkey PRIMARY KEY (snd_message_body_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_message_deliveries + ADD CONSTRAINT snd_message_deliveries_pkey PRIMARY KEY (snd_message_delivery_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_messages + ADD CONSTRAINT snd_messages_pkey PRIMARY KEY (conn_id, internal_snd_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_queues + ADD CONSTRAINT snd_queues_pkey PRIMARY KEY (host, port, snd_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.users + ADD CONSTRAINT users_pkey PRIMARY KEY (user_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.xftp_servers + ADD CONSTRAINT xftp_servers_pkey PRIMARY KEY (xftp_server_id); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.xftp_servers + ADD CONSTRAINT xftp_servers_xftp_host_xftp_port_xftp_key_hash_key UNIQUE (xftp_host, xftp_port, xftp_key_hash); + + + +CREATE INDEX idx_commands_conn_id ON smp_agent_test_protocol_schema.commands USING btree (conn_id); + + + +CREATE INDEX idx_commands_host_port ON smp_agent_test_protocol_schema.commands USING btree (host, port); + + + +CREATE INDEX idx_commands_server_commands ON smp_agent_test_protocol_schema.commands USING btree (host, port, created_at, command_id); + + + +CREATE INDEX idx_conn_confirmations_conn_id ON smp_agent_test_protocol_schema.conn_confirmations USING btree (conn_id); + + + +CREATE INDEX idx_conn_invitations_contact_conn_id ON smp_agent_test_protocol_schema.conn_invitations USING btree (contact_conn_id); + + + +CREATE INDEX idx_connections_user ON smp_agent_test_protocol_schema.connections USING btree (user_id); + + + +CREATE INDEX idx_deleted_snd_chunk_replicas_pending ON smp_agent_test_protocol_schema.deleted_snd_chunk_replicas USING btree (created_at); + + + +CREATE INDEX idx_deleted_snd_chunk_replicas_user_id ON smp_agent_test_protocol_schema.deleted_snd_chunk_replicas USING btree (user_id); + + + +CREATE INDEX idx_deleted_snd_chunk_replicas_xftp_server_id ON smp_agent_test_protocol_schema.deleted_snd_chunk_replicas USING btree (xftp_server_id); + + + +CREATE INDEX idx_encrypted_rcv_message_hashes_created_at ON smp_agent_test_protocol_schema.encrypted_rcv_message_hashes USING btree (created_at); + + + +CREATE INDEX idx_encrypted_rcv_message_hashes_hash ON smp_agent_test_protocol_schema.encrypted_rcv_message_hashes USING btree (conn_id, hash); + + + +CREATE UNIQUE INDEX idx_inv_short_links_link_id ON smp_agent_test_protocol_schema.inv_short_links USING btree (host, port, link_id); + + + +CREATE INDEX idx_messages_conn_id ON smp_agent_test_protocol_schema.messages USING btree (conn_id); + + + +CREATE INDEX idx_messages_conn_id_internal_rcv_id ON smp_agent_test_protocol_schema.messages USING btree (conn_id, internal_rcv_id); + + + +CREATE INDEX idx_messages_conn_id_internal_snd_id ON smp_agent_test_protocol_schema.messages USING btree (conn_id, internal_snd_id); + + + +CREATE INDEX idx_messages_internal_ts ON smp_agent_test_protocol_schema.messages USING btree (internal_ts); + + + +CREATE INDEX idx_messages_snd_expired ON smp_agent_test_protocol_schema.messages USING btree (conn_id, internal_snd_id, internal_ts); + + + +CREATE INDEX idx_ntf_subscriptions_ntf_host_ntf_port ON smp_agent_test_protocol_schema.ntf_subscriptions USING btree (ntf_host, ntf_port); + + + +CREATE INDEX idx_ntf_subscriptions_smp_host_smp_port ON smp_agent_test_protocol_schema.ntf_subscriptions USING btree (smp_host, smp_port); + + + +CREATE INDEX idx_ntf_tokens_ntf_host_ntf_port ON smp_agent_test_protocol_schema.ntf_tokens USING btree (ntf_host, ntf_port); + + + +CREATE INDEX idx_processed_ratchet_key_hashes_created_at ON smp_agent_test_protocol_schema.processed_ratchet_key_hashes USING btree (created_at); + + + +CREATE INDEX idx_processed_ratchet_key_hashes_hash ON smp_agent_test_protocol_schema.processed_ratchet_key_hashes USING btree (conn_id, hash); + + + +CREATE INDEX idx_ratchets_conn_id ON smp_agent_test_protocol_schema.ratchets USING btree (conn_id); + + + +CREATE INDEX idx_rcv_file_chunk_replicas_pending ON smp_agent_test_protocol_schema.rcv_file_chunk_replicas USING btree (received, replica_number); + + + +CREATE INDEX idx_rcv_file_chunk_replicas_rcv_file_chunk_id ON smp_agent_test_protocol_schema.rcv_file_chunk_replicas USING btree (rcv_file_chunk_id); + + + +CREATE INDEX idx_rcv_file_chunk_replicas_xftp_server_id ON smp_agent_test_protocol_schema.rcv_file_chunk_replicas USING btree (xftp_server_id); + + + +CREATE INDEX idx_rcv_file_chunks_rcv_file_id ON smp_agent_test_protocol_schema.rcv_file_chunks USING btree (rcv_file_id); + + + +CREATE INDEX idx_rcv_files_redirect_id ON smp_agent_test_protocol_schema.rcv_files USING btree (redirect_id); + + + +CREATE INDEX idx_rcv_files_status_created_at ON smp_agent_test_protocol_schema.rcv_files USING btree (status, created_at); + + + +CREATE INDEX idx_rcv_files_user_id ON smp_agent_test_protocol_schema.rcv_files USING btree (user_id); + + + +CREATE INDEX idx_rcv_messages_conn_id_internal_id ON smp_agent_test_protocol_schema.rcv_messages USING btree (conn_id, internal_id); + + + +CREATE UNIQUE INDEX idx_rcv_queue_id ON smp_agent_test_protocol_schema.rcv_queues USING btree (conn_id, rcv_queue_id); + + + +CREATE UNIQUE INDEX idx_rcv_queues_link_id ON smp_agent_test_protocol_schema.rcv_queues USING btree (host, port, link_id); + + + +CREATE UNIQUE INDEX idx_rcv_queues_ntf ON smp_agent_test_protocol_schema.rcv_queues USING btree (host, port, ntf_id); + + + +CREATE INDEX idx_server_certs_host_port ON smp_agent_test_protocol_schema.client_services USING btree (host, port); + + + +CREATE UNIQUE INDEX idx_server_certs_user_id_host_port ON smp_agent_test_protocol_schema.client_services USING btree (user_id, host, port); + + + +CREATE INDEX idx_skipped_messages_conn_id ON smp_agent_test_protocol_schema.skipped_messages USING btree (conn_id); + + + +CREATE INDEX idx_snd_file_chunk_replica_recipients_snd_file_chunk_replica_id ON smp_agent_test_protocol_schema.snd_file_chunk_replica_recipients USING btree (snd_file_chunk_replica_id); + + + +CREATE INDEX idx_snd_file_chunk_replicas_pending ON smp_agent_test_protocol_schema.snd_file_chunk_replicas USING btree (replica_status, replica_number); + + + +CREATE INDEX idx_snd_file_chunk_replicas_snd_file_chunk_id ON smp_agent_test_protocol_schema.snd_file_chunk_replicas USING btree (snd_file_chunk_id); + + + +CREATE INDEX idx_snd_file_chunk_replicas_xftp_server_id ON smp_agent_test_protocol_schema.snd_file_chunk_replicas USING btree (xftp_server_id); + + + +CREATE INDEX idx_snd_file_chunks_snd_file_id ON smp_agent_test_protocol_schema.snd_file_chunks USING btree (snd_file_id); + + + +CREATE INDEX idx_snd_files_snd_file_entity_id ON smp_agent_test_protocol_schema.snd_files USING btree (snd_file_entity_id); + + + +CREATE INDEX idx_snd_files_status_created_at ON smp_agent_test_protocol_schema.snd_files USING btree (status, created_at); + + + +CREATE INDEX idx_snd_files_user_id ON smp_agent_test_protocol_schema.snd_files USING btree (user_id); + + + +CREATE INDEX idx_snd_message_deliveries ON smp_agent_test_protocol_schema.snd_message_deliveries USING btree (conn_id, snd_queue_id); + + + +CREATE INDEX idx_snd_message_deliveries_conn_id_internal_id ON smp_agent_test_protocol_schema.snd_message_deliveries USING btree (conn_id, internal_id); + + + +CREATE INDEX idx_snd_message_deliveries_expired ON smp_agent_test_protocol_schema.snd_message_deliveries USING btree (conn_id, snd_queue_id, failed, internal_id); + + + +CREATE INDEX idx_snd_messages_conn_id_internal_id ON smp_agent_test_protocol_schema.snd_messages USING btree (conn_id, internal_id); + + + +CREATE INDEX idx_snd_messages_rcpt_internal_id ON smp_agent_test_protocol_schema.snd_messages USING btree (conn_id, rcpt_internal_id); + + + +CREATE INDEX idx_snd_messages_snd_message_body_id ON smp_agent_test_protocol_schema.snd_messages USING btree (snd_message_body_id); + + + +CREATE UNIQUE INDEX idx_snd_queue_id ON smp_agent_test_protocol_schema.snd_queues USING btree (conn_id, snd_queue_id); + + + +CREATE INDEX idx_snd_queues_host_port ON smp_agent_test_protocol_schema.snd_queues USING btree (host, port); + + + +CREATE TRIGGER tr_rcv_queue_delete AFTER DELETE ON smp_agent_test_protocol_schema.rcv_queues FOR EACH ROW EXECUTE FUNCTION smp_agent_test_protocol_schema.on_rcv_queue_delete(); + + + +CREATE TRIGGER tr_rcv_queue_insert AFTER INSERT ON smp_agent_test_protocol_schema.rcv_queues FOR EACH ROW EXECUTE FUNCTION smp_agent_test_protocol_schema.on_rcv_queue_insert(); + + + +CREATE TRIGGER tr_rcv_queue_update AFTER UPDATE ON smp_agent_test_protocol_schema.rcv_queues FOR EACH ROW EXECUTE FUNCTION smp_agent_test_protocol_schema.on_rcv_queue_update(); + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.client_services + ADD CONSTRAINT client_services_host_port_fkey FOREIGN KEY (host, port) REFERENCES smp_agent_test_protocol_schema.servers(host, port) ON UPDATE CASCADE ON DELETE RESTRICT; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.client_services + ADD CONSTRAINT client_services_user_id_fkey FOREIGN KEY (user_id) REFERENCES smp_agent_test_protocol_schema.users(user_id) ON UPDATE RESTRICT ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.commands + ADD CONSTRAINT commands_conn_id_fkey FOREIGN KEY (conn_id) REFERENCES smp_agent_test_protocol_schema.connections(conn_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.commands + ADD CONSTRAINT commands_host_port_fkey FOREIGN KEY (host, port) REFERENCES smp_agent_test_protocol_schema.servers(host, port) ON UPDATE CASCADE ON DELETE RESTRICT; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.conn_confirmations + ADD CONSTRAINT conn_confirmations_conn_id_fkey FOREIGN KEY (conn_id) REFERENCES smp_agent_test_protocol_schema.connections(conn_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.conn_invitations + ADD CONSTRAINT conn_invitations_contact_conn_id_fkey FOREIGN KEY (contact_conn_id) REFERENCES smp_agent_test_protocol_schema.connections(conn_id) ON DELETE SET NULL; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.connections + ADD CONSTRAINT connections_user_id_fkey FOREIGN KEY (user_id) REFERENCES smp_agent_test_protocol_schema.users(user_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.deleted_snd_chunk_replicas + ADD CONSTRAINT deleted_snd_chunk_replicas_user_id_fkey FOREIGN KEY (user_id) REFERENCES smp_agent_test_protocol_schema.users(user_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.deleted_snd_chunk_replicas + ADD CONSTRAINT deleted_snd_chunk_replicas_xftp_server_id_fkey FOREIGN KEY (xftp_server_id) REFERENCES smp_agent_test_protocol_schema.xftp_servers(xftp_server_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.encrypted_rcv_message_hashes + ADD CONSTRAINT encrypted_rcv_message_hashes_conn_id_fkey FOREIGN KEY (conn_id) REFERENCES smp_agent_test_protocol_schema.connections(conn_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.messages + ADD CONSTRAINT fk_messages_rcv_messages FOREIGN KEY (conn_id, internal_rcv_id) REFERENCES smp_agent_test_protocol_schema.rcv_messages(conn_id, internal_rcv_id) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.messages + ADD CONSTRAINT fk_messages_snd_messages FOREIGN KEY (conn_id, internal_snd_id) REFERENCES smp_agent_test_protocol_schema.snd_messages(conn_id, internal_snd_id) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.inv_short_links + ADD CONSTRAINT inv_short_links_host_port_fkey FOREIGN KEY (host, port) REFERENCES smp_agent_test_protocol_schema.servers(host, port) ON UPDATE CASCADE ON DELETE RESTRICT; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.messages + ADD CONSTRAINT messages_conn_id_fkey FOREIGN KEY (conn_id) REFERENCES smp_agent_test_protocol_schema.connections(conn_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.ntf_subscriptions + ADD CONSTRAINT ntf_subscriptions_ntf_host_ntf_port_fkey FOREIGN KEY (ntf_host, ntf_port) REFERENCES smp_agent_test_protocol_schema.ntf_servers(ntf_host, ntf_port) ON UPDATE CASCADE ON DELETE RESTRICT; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.ntf_subscriptions + ADD CONSTRAINT ntf_subscriptions_smp_host_smp_port_fkey FOREIGN KEY (smp_host, smp_port) REFERENCES smp_agent_test_protocol_schema.servers(host, port) ON UPDATE CASCADE ON DELETE SET NULL; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.ntf_tokens + ADD CONSTRAINT ntf_tokens_ntf_host_ntf_port_fkey FOREIGN KEY (ntf_host, ntf_port) REFERENCES smp_agent_test_protocol_schema.ntf_servers(ntf_host, ntf_port) ON UPDATE CASCADE ON DELETE RESTRICT; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.processed_ratchet_key_hashes + ADD CONSTRAINT processed_ratchet_key_hashes_conn_id_fkey FOREIGN KEY (conn_id) REFERENCES smp_agent_test_protocol_schema.connections(conn_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.ratchets + ADD CONSTRAINT ratchets_conn_id_fkey FOREIGN KEY (conn_id) REFERENCES smp_agent_test_protocol_schema.connections(conn_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.rcv_file_chunk_replicas + ADD CONSTRAINT rcv_file_chunk_replicas_rcv_file_chunk_id_fkey FOREIGN KEY (rcv_file_chunk_id) REFERENCES smp_agent_test_protocol_schema.rcv_file_chunks(rcv_file_chunk_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.rcv_file_chunk_replicas + ADD CONSTRAINT rcv_file_chunk_replicas_xftp_server_id_fkey FOREIGN KEY (xftp_server_id) REFERENCES smp_agent_test_protocol_schema.xftp_servers(xftp_server_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.rcv_file_chunks + ADD CONSTRAINT rcv_file_chunks_rcv_file_id_fkey FOREIGN KEY (rcv_file_id) REFERENCES smp_agent_test_protocol_schema.rcv_files(rcv_file_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.rcv_files + ADD CONSTRAINT rcv_files_redirect_id_fkey FOREIGN KEY (redirect_id) REFERENCES smp_agent_test_protocol_schema.rcv_files(rcv_file_id) ON DELETE SET NULL; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.rcv_files + ADD CONSTRAINT rcv_files_user_id_fkey FOREIGN KEY (user_id) REFERENCES smp_agent_test_protocol_schema.users(user_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.rcv_messages + ADD CONSTRAINT rcv_messages_conn_id_internal_id_fkey FOREIGN KEY (conn_id, internal_id) REFERENCES smp_agent_test_protocol_schema.messages(conn_id, internal_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.rcv_queues + ADD CONSTRAINT rcv_queues_conn_id_fkey FOREIGN KEY (conn_id) REFERENCES smp_agent_test_protocol_schema.connections(conn_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.rcv_queues + ADD CONSTRAINT rcv_queues_host_port_fkey FOREIGN KEY (host, port) REFERENCES smp_agent_test_protocol_schema.servers(host, port) ON UPDATE CASCADE ON DELETE RESTRICT; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.skipped_messages + ADD CONSTRAINT skipped_messages_conn_id_fkey FOREIGN KEY (conn_id) REFERENCES smp_agent_test_protocol_schema.ratchets(conn_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_file_chunk_replica_recipients + ADD CONSTRAINT snd_file_chunk_replica_recipient_snd_file_chunk_replica_id_fkey FOREIGN KEY (snd_file_chunk_replica_id) REFERENCES smp_agent_test_protocol_schema.snd_file_chunk_replicas(snd_file_chunk_replica_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_file_chunk_replicas + ADD CONSTRAINT snd_file_chunk_replicas_snd_file_chunk_id_fkey FOREIGN KEY (snd_file_chunk_id) REFERENCES smp_agent_test_protocol_schema.snd_file_chunks(snd_file_chunk_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_file_chunk_replicas + ADD CONSTRAINT snd_file_chunk_replicas_xftp_server_id_fkey FOREIGN KEY (xftp_server_id) REFERENCES smp_agent_test_protocol_schema.xftp_servers(xftp_server_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_file_chunks + ADD CONSTRAINT snd_file_chunks_snd_file_id_fkey FOREIGN KEY (snd_file_id) REFERENCES smp_agent_test_protocol_schema.snd_files(snd_file_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_files + ADD CONSTRAINT snd_files_user_id_fkey FOREIGN KEY (user_id) REFERENCES smp_agent_test_protocol_schema.users(user_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_message_deliveries + ADD CONSTRAINT snd_message_deliveries_conn_id_fkey FOREIGN KEY (conn_id) REFERENCES smp_agent_test_protocol_schema.connections(conn_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_message_deliveries + ADD CONSTRAINT snd_message_deliveries_conn_id_internal_id_fkey FOREIGN KEY (conn_id, internal_id) REFERENCES smp_agent_test_protocol_schema.messages(conn_id, internal_id) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_messages + ADD CONSTRAINT snd_messages_conn_id_internal_id_fkey FOREIGN KEY (conn_id, internal_id) REFERENCES smp_agent_test_protocol_schema.messages(conn_id, internal_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_messages + ADD CONSTRAINT snd_messages_snd_message_body_id_fkey FOREIGN KEY (snd_message_body_id) REFERENCES smp_agent_test_protocol_schema.snd_message_bodies(snd_message_body_id) ON DELETE SET NULL; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_queues + ADD CONSTRAINT snd_queues_conn_id_fkey FOREIGN KEY (conn_id) REFERENCES smp_agent_test_protocol_schema.connections(conn_id) ON DELETE CASCADE; + + + +ALTER TABLE ONLY smp_agent_test_protocol_schema.snd_queues + ADD CONSTRAINT snd_queues_host_port_fkey FOREIGN KEY (host, port) REFERENCES smp_agent_test_protocol_schema.servers(host, port) ON UPDATE CASCADE ON DELETE RESTRICT; + + + diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index c724c031b..e1ae81995 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/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} = diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Functions.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Functions.hs new file mode 100644 index 000000000..904269804 --- /dev/null +++ b/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 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 fb76d0f26..8caa58f7e 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 @@ -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; 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 25063a33e..784b0f0a7 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql +++ b/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; diff --git a/src/Simplex/Messaging/Notifications/Server/Store/Migrations.hs b/src/Simplex/Messaging/Notifications/Server/Store/Migrations.hs index 1051cc4fb..7d77d39ab 100644 --- a/src/Simplex/Messaging/Notifications/Server/Store/Migrations.hs +++ b/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(); diff --git a/src/Simplex/Messaging/Notifications/Server/Store/Postgres.hs b/src/Simplex/Messaging/Notifications/Server/Store/Postgres.hs index 7f10fc59f..d900097ee 100644 --- a/src/Simplex/Messaging/Notifications/Server/Store/Postgres.hs +++ b/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 diff --git a/src/Simplex/Messaging/Protocol.hs b/src/Simplex/Messaging/Protocol.hs index cfc2c0ffd..f4d4288b5 100644 --- a/src/Simplex/Messaging/Protocol.hs +++ b/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 diff --git a/src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs b/src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs index 2401279ee..426dbb0d5 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs +++ b/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; - |] diff --git a/tests/Fixtures.hs b/tests/Fixtures.hs index 2360a7ba6..f2f314fed 100644 --- a/tests/Fixtures.hs +++ b/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 diff --git a/tests/ServerTests.hs b/tests/ServerTests.hs index 1c0e609d1..2c8632ca9 100644 --- a/tests/ServerTests.hs +++ b/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 diff --git a/tests/Test.hs b/tests/Test.hs index b88eb3bf5..3a3b53c2d 100644 --- a/tests/Test.hs +++ b/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