allow different queue ID sizes

Author: epoberezkin

src/Simplex/Messaging/Agent/Store/Postgres/Migrations/Util.hs

@@ -16,13 +16,17 @@ AS $$
 DECLARE
   result BYTEA := state;
   i INTEGER;
-  len INTEGER := octet_length(value);
+  state_len INTEGER := octet_length(state);
+  value_len INTEGER := octet_length(value);
+  max_len INTEGER := GREATEST(state_len, value_len);
+  byte_val INTEGER;
 BEGIN
-  IF octet_length(state) != len THEN
-    RAISE EXCEPTION 'Inputs must be equal length (% != %)', octet_length(state), len;
+  IF value_len > state_len THEN
+    result := result || repeat('\x00'::BYTEA, value_len - state_len);
   END IF;
-  FOR i IN 0..len-1 LOOP
-    result := set_byte(result, i, get_byte(state, i) # get_byte(value, i));
+  FOR i IN 0..max_len-1 LOOP
+    byte_val := CASE WHEN i < value_len THEN get_byte(value, i) ELSE 0 END;
+    result := set_byte(result, i, get_byte(result, i) # byte_val);
   END LOOP;
   RETURN result;
 END;

src/Simplex/Messaging/Agent/Store/Postgres/Migrations/agent_postgres_schema.sql

@@ -15,91 +15,32 @@ SET row_security = off;
 CREATE SCHEMA smp_agent_test_protocol_schema;
 
 
+SET default_table_access_method = heap;
+
 
-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'
+CREATE TABLE smp_agent_test_protocol_schema.client_notices (
+    client_notice_id bigint NOT NULL,
+    protocol text NOT NULL,
+    host text NOT NULL,
+    port text NOT NULL,
+    entity_id bytea NOT NULL,
+    server_key_hash bytea,
+    notice_ttl bigint,
+    created_at bigint NOT NULL,
+    updated_at bigint NOT NULL
 );
 
 
-SET default_table_access_method = heap;
+
+ALTER TABLE smp_agent_test_protocol_schema.client_notices ALTER COLUMN client_notice_id ADD GENERATED ALWAYS AS IDENTITY (
+    SEQUENCE NAME smp_agent_test_protocol_schema.client_notices_client_notice_id_seq
+    START WITH 1
+    INCREMENT BY 1
+    NO MINVALUE
+    NO MAXVALUE
+    CACHE 1
+);
+
 
 
 CREATE TABLE smp_agent_test_protocol_schema.client_services (
@@ -535,6 +476,8 @@ CREATE TABLE smp_agent_test_protocol_schema.rcv_queues (
     link_priv_sig_key bytea,
     link_enc_fixed_data bytea,
     queue_mode text,
+    to_subscribe smallint DEFAULT 0 NOT NULL,
+    client_notice_id bigint,
     rcv_service_assoc smallint DEFAULT 0 NOT NULL
 );
 
@@ -816,6 +759,11 @@ ALTER TABLE smp_agent_test_protocol_schema.xftp_servers ALTER COLUMN xftp_server
 
 
 
+ALTER TABLE ONLY smp_agent_test_protocol_schema.client_notices
+    ADD CONSTRAINT client_notices_pkey PRIMARY KEY (client_notice_id);
+
+
+
 ALTER TABLE ONLY smp_agent_test_protocol_schema.commands
     ADD CONSTRAINT commands_pkey PRIMARY KEY (command_id);
 
@@ -996,6 +944,10 @@ ALTER TABLE ONLY smp_agent_test_protocol_schema.xftp_servers
 
 
 
+CREATE UNIQUE INDEX idx_client_notices_entity ON smp_agent_test_protocol_schema.client_notices USING btree (protocol, host, port, entity_id);
+
+
+
 CREATE INDEX idx_commands_conn_id ON smp_agent_test_protocol_schema.commands USING btree (conn_id);
 
 
@@ -1124,6 +1076,10 @@ CREATE UNIQUE INDEX idx_rcv_queue_id ON smp_agent_test_protocol_schema.rcv_queue
 
 
 
+CREATE INDEX idx_rcv_queues_client_notice_id ON smp_agent_test_protocol_schema.rcv_queues USING btree (client_notice_id);
+
+
+
 CREATE UNIQUE INDEX idx_rcv_queues_link_id ON smp_agent_test_protocol_schema.rcv_queues USING btree (host, port, link_id);
 
 
@@ -1132,6 +1088,10 @@ CREATE UNIQUE INDEX idx_rcv_queues_ntf ON smp_agent_test_protocol_schema.rcv_que
 
 
 
+CREATE INDEX idx_rcv_queues_to_subscribe ON smp_agent_test_protocol_schema.rcv_queues USING btree (to_subscribe);
+
+
+
 CREATE INDEX idx_server_certs_host_port ON smp_agent_test_protocol_schema.client_services USING btree (host, port);
 
 
@@ -1208,18 +1168,6 @@ CREATE INDEX idx_snd_queues_host_port ON smp_agent_test_protocol_schema.snd_queu
 
 
 
-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;
 
@@ -1345,6 +1293,11 @@ ALTER TABLE ONLY smp_agent_test_protocol_schema.rcv_messages
 
 
 
+ALTER TABLE ONLY smp_agent_test_protocol_schema.rcv_queues
+    ADD CONSTRAINT rcv_queues_client_notice_id_fkey FOREIGN KEY (client_notice_id) REFERENCES smp_agent_test_protocol_schema.client_notices(client_notice_id) ON UPDATE RESTRICT ON DELETE SET NULL;
+
+
+
 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;
 

src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql

@@ -111,13 +111,17 @@ CREATE FUNCTION ntf_server.xor_combine(state bytea, value bytea) RETURNS bytea
 DECLARE
   result BYTEA := state;
   i INTEGER;
-  len INTEGER := octet_length(value);
+  state_len INTEGER := octet_length(state);
+  value_len INTEGER := octet_length(value);
+  max_len INTEGER := GREATEST(state_len, value_len);
+  byte_val INTEGER;
 BEGIN
-  IF octet_length(state) != len THEN
-    RAISE EXCEPTION 'Inputs must be equal length (% != %)', octet_length(state), len;
+  IF value_len > state_len THEN
+    result := result || repeat('\x00'::BYTEA, value_len - state_len);
   END IF;
-  FOR i IN 0..len-1 LOOP
-    result := set_byte(result, i, get_byte(state, i) # get_byte(value, i));
+  FOR i IN 0..max_len-1 LOOP
+    byte_val := CASE WHEN i < value_len THEN get_byte(value, i) ELSE 0 END;
+    result := set_byte(result, i, get_byte(result, i) # byte_val);
   END LOOP;
   RETURN result;
 END;

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

@@ -365,13 +365,17 @@ CREATE FUNCTION smp_server.xor_combine(state bytea, value bytea) RETURNS bytea
 DECLARE
   result BYTEA := state;
   i INTEGER;
-  len INTEGER := octet_length(value);
+  state_len INTEGER := octet_length(state);
+  value_len INTEGER := octet_length(value);
+  max_len INTEGER := GREATEST(state_len, value_len);
+  byte_val INTEGER;
 BEGIN
-  IF octet_length(state) != len THEN
-    RAISE EXCEPTION 'Inputs must be equal length (% != %)', octet_length(state), len;
+  IF value_len > state_len THEN
+    result := result || repeat('\x00'::BYTEA, value_len - state_len);
   END IF;
-  FOR i IN 0..len-1 LOOP
-    result := set_byte(result, i, get_byte(state, i) # get_byte(value, i));
+  FOR i IN 0..max_len-1 LOOP
+    byte_val := CASE WHEN i < value_len THEN get_byte(value, i) ELSE 0 END;
+    result := set_byte(result, i, get_byte(result, i) # byte_val);
   END LOOP;
   RETURN result;
 END;