dandri/simplexmq
1
0
Fork 0
mirror of https://github.com/simplex-chat/simplexmq.git synced 2026-08-28 05:04:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

remove agent triggers

Browse Source
This commit is contained in:
Evgeny Poberezkin
2025-09-02 22:59:03 +01:00
parent f7126a34ae
commit 7fba70acc1
5 changed files with 4 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
simplexmq.cabal
-1
View File
@@ -169,7 +169,6 @@ library
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
src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20250815_service_certs.hs
+3 -76
View File
@@ -4,14 +4,12 @@ module Simplex.Messaging.Agent.Store.Postgres.Migrations.M20250815_service_certs
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|
T.pack
[r|
CREATE TABLE client_services(
user_id BIGINT NOT NULL REFERENCES users ON UPDATE RESTRICT ON DELETE CASCADE,
host TEXT NOT NULL,
@@ -29,86 +27,15 @@ 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_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
src/Simplex/Messaging/Agent/Store/SQLite.hs
+1 -3
View File
@@ -56,13 +56,12 @@ 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, throwIO)
import UnliftIO.Exception (bracketOnError, onException)
import UnliftIO.MVar
import UnliftIO.STM
@@ -114,7 +113,6 @@ 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} =
src/Simplex/Messaging/Agent/Store/SQLite/Functions.hs
-37
View File
@@ -1,37 +0,0 @@
{-# 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
src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20250815_service_certs.hs
-49
View File
@@ -25,60 +25,11 @@ 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 = 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;