agent database triggers

Author: epoberezkin

src/Simplex/Messaging/Agent/Client.hs

@@ -2302,7 +2302,8 @@ withStore c action = do
       [ E.Handler $ \(e :: SQL.SQLError) ->
           let se = SQL.sqlError e
               busy = se == SQL.ErrorBusy || se == SQL.ErrorLocked
-           in pure . Left . (if busy then SEDatabaseBusy else SEInternal) $ bshow se,
+              err = tshow se <> ": " <> SQL.sqlErrorDetails e <> ", " <> SQL.sqlErrorContext e
+           in pure . Left . (if busy then SEDatabaseBusy else SEInternal) $ encodeUtf8 err,
         E.Handler $ \(E.SomeException e) -> pure . Left $ SEInternal $ bshow e
       ]
 #endif

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

@@ -27,12 +27,84 @@ CREATE UNIQUE INDEX idx_server_certs_user_id_host_port ON client_services(user_i
 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_conn_id BYTEA, p_host TEXT, p_port TEXT, p_change BIGINT, p_rcv_id BYTEA) RETURNS VOID
+LANGUAGE plpgsql
+AS $$
+DECLARE q_user_id BIGINT;
+BEGIN
+  SELECT user_id INTO q_user_id FROM connections WHERE conn_id = p_conn_id;
+  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 = q_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.conn_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.conn_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.conn_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.conn_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_m20251020_service_certs :: Text
 down_m20251020_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;

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

@@ -21,7 +21,7 @@ CREATE FUNCTION smp_agent_test_protocol_schema.on_rcv_queue_delete() RETURNS tri
     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);
+    PERFORM update_aggregates(OLD.conn_id, OLD.host, OLD.port, -1, OLD.rcv_id);
   END IF;
   RETURN OLD;
 END;
@@ -34,7 +34,7 @@ CREATE FUNCTION smp_agent_test_protocol_schema.on_rcv_queue_insert() RETURNS tri
     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);
+    PERFORM update_aggregates(NEW.conn_id, NEW.host, NEW.port, 1, NEW.rcv_id);
   END IF;
   RETURN NEW;
 END;
@@ -48,58 +48,57 @@ CREATE FUNCTION smp_agent_test_protocol_schema.on_rcv_queue_update() RETURNS tri
 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);
+      PERFORM update_aggregates(OLD.conn_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);
+    PERFORM update_aggregates(NEW.conn_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
+CREATE FUNCTION smp_agent_test_protocol_schema.update_aggregates(p_conn_id bytea, p_host text, p_port text, p_change bigint, p_rcv_id bytea) RETURNS void
     LANGUAGE plpgsql
     AS $$
+DECLARE q_user_id BIGINT;
 BEGIN
+  SELECT user_id INTO q_user_id FROM connections WHERE conn_id = p_conn_id;
   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;
+  WHERE user_id = q_user_id AND host = p_host AND port = p_port;
 END;
 $$;
 
 
+SET default_table_access_method = heap;
 
-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 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
+);
 
 
-CREATE AGGREGATE smp_agent_test_protocol_schema.xor_aggregate(bytea) (
-    SFUNC = smp_agent_test_protocol_schema.xor_combine,
-    STYPE = bytea,
-    INITCOND = '\x00000000000000000000000000000000'
-);
 
+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
+);
 
-SET default_table_access_method = heap;
 
 
 CREATE TABLE smp_agent_test_protocol_schema.client_services (
@@ -535,6 +534,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 +817,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 +1002,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 +1134,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 +1146,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);
 
 
@@ -1345,6 +1363,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/Agent/Store/SQLite.hs

@@ -43,26 +43,31 @@ module Simplex.Messaging.Agent.Store.SQLite
 where
 
 import Control.Monad
+import Data.Bits (xor)
 import Data.ByteArray (ScrubbedBytes)
 import qualified Data.ByteArray as BA
+import Data.ByteString (ByteString)
+import qualified Data.ByteString as B
 import Data.Functor (($>))
 import Data.IORef
 import Data.Maybe (fromMaybe)
 import Data.Text (Text)
 import qualified Data.Text as T
 import Database.SQLite.Simple (Query (..))
 import qualified Database.SQLite.Simple as SQL
+import Database.SQLite.Simple.Function
 import Database.SQLite.Simple.QQ (sql)
 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 qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
 import Simplex.Messaging.Agent.Store.Shared (Migration (..), MigrationConfig (..), MigrationError (..))
+import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Util (ifM, safeDecodeUtf8)
 import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist)
 import System.FilePath (takeDirectory, takeFileName, (</>))
-import UnliftIO.Exception (bracketOnError, onException)
+import UnliftIO.Exception (bracketOnError, onException, throwIO)
 import UnliftIO.MVar
 import UnliftIO.STM
 
@@ -119,6 +124,11 @@ connectDB path key track = do
           PRAGMA secure_delete = ON;
           PRAGMA auto_vacuum = FULL;
         |]
+      createFunction (DB.conn db) "simplex_xor_md5_combine" xorMd5Combine
+        >>= either (throwIO . userError . show) pure
+
+xorMd5Combine :: ByteString -> ByteString -> ByteString
+xorMd5Combine idsHash rId = B.packZipWith xor idsHash $ C.md5Hash rId
 
 closeDBStore :: DBStore -> IO ()
 closeDBStore st@DBStore {dbClosed} =

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20251020_service_certs.hs

@@ -25,11 +25,64 @@ CREATE UNIQUE INDEX idx_server_certs_user_id_host_port ON client_services(user_i
 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 = simplex_xor_md5_combine(service_queue_ids_hash, NEW.rcv_id)
+  WHERE user_id = (SELECT user_id FROM connections WHERE conn_id = NEW.conn_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 = simplex_xor_md5_combine(service_queue_ids_hash, OLD.rcv_id)
+  WHERE user_id = (SELECT user_id FROM connections WHERE conn_id = OLD.conn_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 = simplex_xor_md5_combine(service_queue_ids_hash, OLD.rcv_id)
+  WHERE user_id = (SELECT user_id FROM connections WHERE conn_id = OLD.conn_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 = simplex_xor_md5_combine(service_queue_ids_hash, NEW.rcv_id)
+  WHERE user_id = (SELECT user_id FROM connections WHERE conn_id = NEW.conn_id)
+    AND host = NEW.host AND port = NEW.port;
+END;
   |]
 
 down_m20251020_service_certs :: Query
 down_m20251020_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;