From 1dbc15b2e6225c0e254564747bc8412970273e85 Mon Sep 17 00:00:00 2001 From: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> Date: Fri, 10 Oct 2025 05:41:42 +0000 Subject: [PATCH] agent: sync connections (#1654) * agent: sync subscriptions * remove comment * add shouldDelete flag * compare api * remove instance * query * rename * refactor * functor * JSON instances --------- Co-authored-by: Evgeny Poberezkin --- src/Simplex/Messaging/Agent.hs | 56 ++++++++++++++++++- .../Messaging/Agent/Store/AgentStore.hs | 9 +++ 2 files changed, 64 insertions(+), 1 deletion(-) diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index cd0d092a4..929c4031d 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -13,6 +13,7 @@ {-# LANGUAGE PatternSynonyms #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TupleSections #-} {-# LANGUAGE TypeApplications #-} {-# OPTIONS_GHC -fno-warn-ambiguous-fields #-} @@ -67,6 +68,9 @@ module Simplex.Messaging.Agent allowConnection, acceptContact, rejectContact, + DatabaseDiff (..), + compareConnections, + syncConnections, subscribeConnection, subscribeConnections, subscribeAllConnections, @@ -140,7 +144,9 @@ import Control.Monad.Except import Control.Monad.Reader import Control.Monad.Trans.Except import Crypto.Random (ChaChaDRG) +import Data.Aeson (FromJSON (..), ToJSON (..)) import qualified Data.Aeson as J +import qualified Data.Aeson.TH as JQ import Data.Bifunctor (bimap, first) import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B @@ -195,7 +201,7 @@ import Simplex.Messaging.Encoding import Simplex.Messaging.Encoding.String import Simplex.Messaging.Notifications.Protocol (DeviceToken, NtfRegCode (NtfRegCode), NtfTknStatus (..), NtfTokenId, PNMessageData (..), pnMessagesP) import Simplex.Messaging.Notifications.Types -import Simplex.Messaging.Parsers (parse) +import Simplex.Messaging.Parsers (defaultJSON, parse) import Simplex.Messaging.Protocol ( BrokerMsg, Cmd (..), @@ -434,6 +440,24 @@ rejectContact :: AgentClient -> ConfirmationId -> AE () rejectContact c = withAgentEnv c . rejectContact' c {-# INLINE rejectContact #-} +data DatabaseDiff a = DatabaseDiff + { missingIds :: [a], + extraIds :: [a] + } + deriving (Show) + +instance Functor DatabaseDiff where + fmap f DatabaseDiff {missingIds, extraIds} = + DatabaseDiff {missingIds = map f missingIds, extraIds = map f extraIds} + +compareConnections :: AgentClient -> [UserId] -> [ConnId] -> AE (DatabaseDiff UserId, DatabaseDiff ConnId) +compareConnections c = withAgentEnv c .: compareConnections' c +{-# INLINE compareConnections #-} + +syncConnections :: AgentClient -> [UserId] -> [ConnId] -> AE (DatabaseDiff UserId, DatabaseDiff ConnId) +syncConnections c = withAgentEnv c .: syncConnections' c +{-# INLINE syncConnections #-} + -- | Subscribe to receive connection messages (SUB command) subscribeConnection :: AgentClient -> ConnId -> AE (Maybe ClientServiceId) subscribeConnection c = withAgentEnv c . subscribeConnection' c @@ -1253,6 +1277,27 @@ rejectContact' c invId = withStore' c $ \db -> deleteInvitation db invId {-# INLINE rejectContact' #-} +syncConnections' :: AgentClient -> [UserId] -> [ConnId] -> AM (DatabaseDiff UserId, DatabaseDiff ConnId) +syncConnections' c userIds connIds = do + r@(DatabaseDiff {extraIds = uIds}, DatabaseDiff {extraIds = cIds}) <- compareConnections' c userIds connIds + forM_ uIds $ \uid -> deleteUser' c uid False + deleteConnectionsAsync' c False cIds + pure r + +compareConnections' :: AgentClient -> [UserId] -> [ConnId] -> AM (DatabaseDiff UserId, DatabaseDiff ConnId) +compareConnections' c userIds connIds = do + knownUserIds <- withStore' c getUserIds + knownConnIds <- withStore' c getConnIds + pure (databaseDiff userIds knownUserIds, databaseDiff connIds knownConnIds) + +databaseDiff :: Ord a => [a] -> [a] -> DatabaseDiff a +databaseDiff passed known = + let passedSet = S.fromList passed + knownSet = S.fromList known + missingIds = S.toList $ passedSet `S.difference` knownSet + extraIds = S.toList $ knownSet `S.difference` passedSet + in DatabaseDiff {missingIds, extraIds} + -- | Subscribe to receive connection messages (SUB command) in Reader monad subscribeConnection' :: AgentClient -> ConnId -> AM (Maybe ClientServiceId) subscribeConnection' c connId = toConnResult connId =<< subscribeConnections' c [connId] @@ -3478,3 +3523,12 @@ newSndQueue userId connId (Compatible (SMPQueueInfo smpClientVersion SMPQueueAdd smpClientVersion } pure (sq, e2ePubKey) + +$(pure []) + +instance FromJSON a => FromJSON (DatabaseDiff a) where + parseJSON = $(JQ.mkParseJSON defaultJSON ''DatabaseDiff) + +instance ToJSON a => ToJSON (DatabaseDiff a) where + toEncoding = $(JQ.mkToEncoding defaultJSON ''DatabaseDiff) + toJSON = $(JQ.mkToJSON defaultJSON ''DatabaseDiff) diff --git a/src/Simplex/Messaging/Agent/Store/AgentStore.hs b/src/Simplex/Messaging/Agent/Store/AgentStore.hs index a9334c331..7d408f1ee 100644 --- a/src/Simplex/Messaging/Agent/Store/AgentStore.hs +++ b/src/Simplex/Messaging/Agent/Store/AgentStore.hs @@ -28,6 +28,7 @@ module Simplex.Messaging.Agent.Store.AgentStore ( -- * Users createUserRecord, + getUserIds, deleteUserRecord, setUserDeleted, deleteUserWithoutConns, @@ -42,6 +43,7 @@ module Simplex.Messaging.Agent.Store.AgentStore getSubscriptionServers, getUserServerRcvQueueSubs, unsetQueuesToSubscribe, + getConnIds, getConn, getDeletedConn, getConns, @@ -330,6 +332,10 @@ createUserRecord db = do DB.execute_ db "INSERT INTO users DEFAULT VALUES" insertedRowId db +getUserIds :: DB.Connection -> IO [UserId] +getUserIds db = + map fromOnly <$> DB.query_ db "SELECT user_id FROM users WHERE deleted = 0" + checkUser :: DB.Connection -> UserId -> IO (Either StoreError ()) checkUser db userId = firstRow (\(_ :: Only Int64) -> ()) SEUserNotFound $ @@ -2089,6 +2095,9 @@ unsetQueuesToSubscribe db = DB.execute_ db "UPDATE rcv_queues SET to_subscribe = -- * getConn helpers +getConnIds :: DB.Connection -> IO [ConnId] +getConnIds db = map fromOnly <$> DB.query_ db "SELECT conn_id FROM connections WHERE deleted = 0" + getConn :: DB.Connection -> ConnId -> IO (Either StoreError SomeConn) getConn = getAnyConn False {-# INLINE getConn #-}