events sent by the agent when queue, connection and user are deleted asynchronously (#619)

* events sent by the agent when queue, connection and user are deleted asynchronously

* update test

* more tests

* add user tests

* queries

Co-authored-by: JRoberts <8711996+jr-simplex@users.noreply.github.com>

Co-authored-by: JRoberts <8711996+jr-simplex@users.noreply.github.com>

src/Simplex/Messaging/Agent.hs

@@ -129,7 +129,7 @@ import Simplex.Messaging.Notifications.Protocol (DeviceToken, NtfRegCode (NtfReg
 import Simplex.Messaging.Notifications.Server.Push.APNS (PNMessageData (..))
 import Simplex.Messaging.Notifications.Types
 import Simplex.Messaging.Parsers (parse)
-import Simplex.Messaging.Protocol (BrokerMsg, ErrorType (AUTH), MsgBody, MsgFlags, NtfServer, SMPMsgMeta, SndPublicVerifyKey, protoServer, sameSrvAddr, sameSrvAddr')
+import Simplex.Messaging.Protocol (BrokerMsg, ErrorType (AUTH), MsgBody, MsgFlags, NtfServer, SMPMsgMeta, SndPublicVerifyKey, protoServer, sameSrvAddr')
 import qualified Simplex.Messaging.Protocol as SMP
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Util
@@ -393,9 +393,13 @@ createUser' c srvs = do
 deleteUser' :: AgentMonad m => AgentClient -> UserId -> Bool -> m ()
 deleteUser' c userId delSMPQueues = do
   if delSMPQueues
-    then withStore c (`setUserDeleted` userId) >>= deleteConnectionsAsync_ (Just userId) c
+    then withStore c (`setUserDeleted` userId) >>= deleteConnectionsAsync_ delUser c
     else withStore c (`deleteUserRecord` userId)
   atomically $ TM.delete userId $ smpServers c
+  where
+    delUser =
+      whenM (withStore' c (`deleteUserWithoutConns` userId)) $
+        atomically $ writeTBQueue (subQ c) ("", "", DEL_USER userId)
 
 newConnAsync :: forall m c. (AgentMonad m, ConnectionModeI c) => AgentClient -> UserId -> ACorrId -> Bool -> SConnectionMode c -> m ConnId
 newConnAsync c userId corrId enableNtfs cMode = do
@@ -467,27 +471,25 @@ deleteConnectionAsync' c connId = withConnLock c connId "deleteConnectionAsync"
       withStore' c (`setConnDeleted` connId)
       disableConn c connId
       void . forkIO $ do
-        forM_ rqs $ deleteConnQueue c
+        forM_ rqs $ deleteConnQueue c True
         delete
   where
-    delete = withStore' c (`deleteConn` connId)
+    delete = do
+      withStore' c (`deleteConn` connId)
+      atomically $ writeTBQueue (subQ c) ("", connId, DEL_CONN)
 
 deleteConnectionsAsync' :: AgentMonad m => AgentClient -> [ConnId] -> m ()
-deleteConnectionsAsync' = deleteConnectionsAsync_ Nothing
+deleteConnectionsAsync' = deleteConnectionsAsync_ $ pure ()
 
-deleteConnectionsAsync_ :: forall m. AgentMonad m => Maybe UserId -> AgentClient -> [ConnId] -> m ()
-deleteConnectionsAsync_ userId_ c connIds = case connIds of
-  [] -> delUser
+deleteConnectionsAsync_ :: forall m. AgentMonad m => m () -> AgentClient -> [ConnId] -> m ()
+deleteConnectionsAsync_ onSuccess c connIds = case connIds of
+  [] -> onSuccess
   _ -> do
     (_, rqs, connIds') <- prepareDeleteConnections_ getConn c connIds
     withStore' c $ forM_ connIds' . setConnDeleted
     void . forkIO $
-      withLock (deleteLock c) "deleteConnectionsAsync" $ do
-        void $ deleteConnQueues c rqs
-        delUser
-  where
-    delUser :: m ()
-    delUser = forM_ userId_ $ \userId -> withStore' c (`deleteUserWithoutConns` userId)
+      withLock (deleteLock c) "deleteConnectionsAsync" $
+        deleteConnQueues c True rqs >> onSuccess
 
 -- | Add connection to the new receive queue
 switchConnectionAsync' :: AgentMonad m => AgentClient -> ACorrId -> ConnId -> m ()
@@ -870,23 +872,8 @@ runCommandProcessing c@AgentClient {subQ} server_ = do
           secure rq senderKey
           when (duplexHandshake cData == Just True) . void $
             enqueueMessage c cData sq SMP.MsgFlags {notification = True} HELLO
-        ICDeleteConn ->
-          withServer $ \srv -> tryWithLock "ICDeleteConn" $ do
-            SomeConn _ conn <- withStore c $ \db -> getDeletedConn db connId
-            case conn of
-              DuplexConnection _ (rq :| rqs) _ -> delete srv rq $ case rqs of
-                [] -> notify OK
-                RcvQueue {server = srv'} : _ -> enqueue srv'
-              RcvConnection _ rq -> delete srv rq $ notify OK
-              ContactConnection _ rq -> delete srv rq $ notify OK
-              _ -> internalErr "command requires connection with rcv queue"
-          where
-            delete :: SMPServer -> RcvQueue -> m () -> m ()
-            delete srv rq@RcvQueue {server} next
-              | sameSrvAddr srv server = deleteConnQueue c rq >> next
-              | otherwise = enqueue server
-            enqueue :: SMPServer -> m ()
-            enqueue srv = enqueueCommand c corrId connId (Just srv) $ AInternalCommand ICDeleteConn
+        -- ICDeleteConn is no longer used, but it can be present in old client databases
+        ICDeleteConn -> withStore' c (`deleteCommand` cmdId)
         ICQSecure rId senderKey ->
           withServer $ \srv -> tryWithLock "ICQSecure" . withDuplexConn $ \(DuplexConnection cData rqs sqs) ->
             case find (sameQueue (srv, rId)) rqs of
@@ -1225,7 +1212,7 @@ deleteConnection' c connId = withConnLock c connId "deleteConnection" $ do
   let rqs = connRcvQueues conn
   unless (null rqs) $ do
     disableConn c connId
-    forM_ rqs $ deleteConnQueue c
+    forM_ rqs $ deleteConnQueue c False
   withStore' c (`deleteConn` connId)
 
 connRcvQueues :: Connection d -> [RcvQueue]
@@ -1236,17 +1223,18 @@ connRcvQueues = \case
   SndConnection _ _ -> []
   NewConnection _ -> []
 
-deleteConnQueue :: AgentMonad m => AgentClient -> RcvQueue -> m ()
-deleteConnQueue c rq = do
+deleteConnQueue :: AgentMonad m => AgentClient -> Bool -> RcvQueue -> m ()
+deleteConnQueue c ntf rq = do
   maxErrs <- asks $ deleteErrorCount . config
-  deleteQueue c rq `catchError` handleError maxErrs
+  (deleteQueue c rq >> notify Nothing) `catchError` handleError maxErrs
   withStore' c (`deleteConnRcvQueue` rq)
   where
     handleError maxErrs e
       | temporaryOrHostError e && deleteErrors rq + 1 < maxErrs = do
         withStore' c (`incRcvDeleteErrors` rq)
         throwError e
-      | otherwise = pure ()
+      | otherwise = notify $ Just e
+    notify e_ = when ntf $ atomically $ writeTBQueue (subQ c) ("", qConnId rq, DEL_RCVQ (qServer rq) (queueId rq) e_)
 
 disableConn :: AgentMonad m => AgentClient -> ConnId -> m ()
 disableConn c connId = do
@@ -1256,10 +1244,10 @@ disableConn c connId = do
 
 -- Unlike deleteConnectionsAsync, this function does not mark connections as deleted in case of deletion failure.
 deleteConnections' :: forall m. AgentMonad m => AgentClient -> [ConnId] -> m (Map ConnId (Either AgentErrorType ()))
-deleteConnections' = deleteConnections_ getConn
+deleteConnections' = deleteConnections_ getConn False
 
 deleteDeletedConns :: forall m. AgentMonad m => AgentClient -> [ConnId] -> m (Map ConnId (Either AgentErrorType ()))
-deleteDeletedConns = deleteConnections_ getDeletedConn
+deleteDeletedConns = deleteConnections_ getDeletedConn True
 
 prepareDeleteConnections_ ::
   forall m.
@@ -1284,11 +1272,11 @@ prepareDeleteConnections_ getConnection c connIds = do
       [] -> Left $ Right ()
       rqs -> Right rqs
 
-deleteConnQueues :: forall m. AgentMonad m => AgentClient -> [RcvQueue] -> m (Map ConnId (Either AgentErrorType ()))
-deleteConnQueues c rqs = do
+deleteConnQueues :: forall m. AgentMonad m => AgentClient -> Bool -> [RcvQueue] -> m (Map ConnId (Either AgentErrorType ()))
+deleteConnQueues c ntf rqs = do
   rs <- connResults <$> (deleteQueueRecs =<< deleteQueues c rqs)
   forM_ (M.assocs rs) $ \case
-    (connId, Right _) -> withStore' c (`deleteConn` connId)
+    (connId, Right _) -> withStore' c (`deleteConn` connId) >> notify ("", connId, DEL_CONN)
     _ -> pure ()
   pure rs
   where
@@ -1297,11 +1285,13 @@ deleteConnQueues c rqs = do
       maxErrs <- asks $ deleteErrorCount . config
       forM rs $ \(rq, r) -> do
         r' <- case r of
-          Right _ -> withStore' c (`deleteConnRcvQueue` rq) $> r
+          Right _ -> withStore' c (`deleteConnRcvQueue` rq) >> notifyRQ rq Nothing $> r
           Left e
             | temporaryOrHostError e && deleteErrors rq + 1 < maxErrs -> withStore' c (`incRcvDeleteErrors` rq) $> r
-            | otherwise -> withStore' c (`deleteConnRcvQueue` rq) $> Right ()
+            | otherwise -> withStore' c (`deleteConnRcvQueue` rq) >> notifyRQ rq (Just e) $> Right ()
         pure (rq, r')
+    notifyRQ rq e_ = notify ("", qConnId rq, DEL_RCVQ (qServer rq) (queueId rq) e_)
+    notify = when ntf . atomically . writeTBQueue (subQ c)
     connResults :: [(RcvQueue, Either AgentErrorType ())] -> Map ConnId (Either AgentErrorType ())
     connResults = M.map snd . foldl' addResult M.empty
       where
@@ -1321,13 +1311,14 @@ deleteConnections_ ::
   forall m.
   AgentMonad m =>
   (DB.Connection -> ConnId -> IO (Either StoreError SomeConn)) ->
+  Bool ->
   AgentClient ->
   [ConnId] ->
   m (Map ConnId (Either AgentErrorType ()))
-deleteConnections_ _ _ [] = pure M.empty
-deleteConnections_ getConnection c connIds = do
+deleteConnections_ _ _ _ [] = pure M.empty
+deleteConnections_ getConnection ntf c connIds = do
   (rs, rqs, _) <- prepareDeleteConnections_ getConnection c connIds
-  rcvRs <- deleteConnQueues c rqs
+  rcvRs <- deleteConnQueues c ntf rqs
   let rs' = M.union rs rcvRs
   notifyResultError rs'
   pure rs'
@@ -1637,8 +1628,10 @@ cleanupManager c = do
     void . runExceptT $
       withLock (deleteLock c) "cleanupManager" $ do
         void $ withStore' c getDeletedConns >>= deleteDeletedConns c
-        withStore' c deleteUsersWithoutConns
+        withStore' c deleteUsersWithoutConns >>= mapM_ notifyUserDeleted
     threadDelay int
+  where
+    notifyUserDeleted userId = atomically $ writeTBQueue (subQ c) ("", "", DEL_USER userId)
 
 processSMPTransmission :: forall m. AgentMonad m => AgentClient -> ServerTransmission BrokerMsg -> m ()
 processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, sessId, rId, cmd) = do

src/Simplex/Messaging/Agent/Protocol.hs

@@ -272,6 +272,9 @@ data ACommand (p :: AParty) where
   SWCH :: ACommand Client
   OFF :: ACommand Client
   DEL :: ACommand Client
+  DEL_RCVQ :: SMPServer -> SMP.SenderId -> Maybe AgentErrorType -> ACommand Agent
+  DEL_CONN :: ACommand Agent
+  DEL_USER :: Int64 -> ACommand Agent
   CHK :: ACommand Client
   STAT :: ConnectionStats -> ACommand Agent
   OK :: ACommand Agent
@@ -311,6 +314,9 @@ data ACommandTag (p :: AParty) where
   SWCH_ :: ACommandTag Client
   OFF_ :: ACommandTag Client
   DEL_ :: ACommandTag Client
+  DEL_RCVQ_ :: ACommandTag Agent
+  DEL_CONN_ :: ACommandTag Agent
+  DEL_USER_ :: ACommandTag Agent
   CHK_ :: ACommandTag Client
   STAT_ :: ACommandTag Agent
   OK_ :: ACommandTag Agent
@@ -349,6 +355,9 @@ aCommandTag = \case
   SWCH -> SWCH_
   OFF -> OFF_
   DEL -> DEL_
+  DEL_RCVQ {} -> DEL_RCVQ_
+  DEL_CONN -> DEL_CONN_
+  DEL_USER _ -> DEL_USER_
   CHK -> CHK_
   STAT _ -> STAT_
   OK -> OK_
@@ -1225,6 +1234,9 @@ instance StrEncoding ACmdTag where
       "SWCH" -> pure $ ACmdTag SClient SWCH_
       "OFF" -> pure $ ACmdTag SClient OFF_
       "DEL" -> pure $ ACmdTag SClient DEL_
+      "DEL_RCVQ" -> pure $ ACmdTag SAgent DEL_RCVQ_
+      "DEL_CONN" -> pure $ ACmdTag SAgent DEL_CONN_
+      "DEL_USER" -> pure $ ACmdTag SAgent DEL_USER_
       "CHK" -> pure $ ACmdTag SClient CHK_
       "STAT" -> pure $ ACmdTag SAgent STAT_
       "OK" -> pure $ ACmdTag SAgent OK_
@@ -1260,6 +1272,9 @@ instance APartyI p => StrEncoding (ACommandTag p) where
     SWCH_ -> "SWCH"
     OFF_ -> "OFF"
     DEL_ -> "DEL"
+    DEL_RCVQ_ -> "DEL_RCVQ"
+    DEL_CONN_ -> "DEL_CONN"
+    DEL_USER_ -> "DEL_USER"
     CHK_ -> "CHK"
     STAT_ -> "STAT"
     OK_ -> "OK"
@@ -1308,6 +1323,9 @@ commandP binaryP =
           SENT_ -> s (SENT <$> A.decimal)
           MERR_ -> s (MERR <$> A.decimal <* A.space <*> strP)
           MSG_ -> s (MSG <$> msgMetaP <* A.space <*> smpP <* A.space <*> binaryP)
+          DEL_RCVQ_ -> s (DEL_RCVQ <$> strP_ <*> strP_ <*> strP)
+          DEL_CONN_ -> pure DEL_CONN
+          DEL_USER_ -> s (DEL_USER <$> strP)
           STAT_ -> s (STAT <$> strP)
           OK_ -> pure OK
           ERR_ -> s (ERR <$> strP)
@@ -1356,6 +1374,9 @@ serializeCommand = \case
   SWCH -> s SWCH_
   OFF -> s OFF_
   DEL -> s DEL_
+  DEL_RCVQ srv rcvId err_ -> s (DEL_RCVQ_, srv, rcvId, err_)
+  DEL_CONN -> s DEL_CONN_
+  DEL_USER userId -> s (DEL_USER_, userId)
   CHK -> s CHK_
   STAT srvs -> s (STAT_, srvs)
   CON -> s CON_

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

@@ -329,28 +329,37 @@ setUserDeleted db userId = runExceptT $ do
     DB.execute db "UPDATE users SET deleted = ? WHERE user_id = ?" (True, userId)
     map fromOnly <$> DB.query db "SELECT conn_id FROM connections WHERE user_id = ?" (Only userId)
 
-deleteUserWithoutConns :: DB.Connection -> UserId -> IO ()
-deleteUserWithoutConns db userId =
-  DB.execute
-    db
-    [sql|
-      DELETE FROM users u
-      WHERE user_id = ?
-        AND u.deleted = ?
-        AND NOT EXISTS (SELECT * FROM connections WHERE user_id = u.user_id)
-    |]
-    (userId, True)
+deleteUserWithoutConns :: DB.Connection -> UserId -> IO Bool
+deleteUserWithoutConns db userId = do
+  userId_ :: Maybe Int64 <-
+    maybeFirstRow fromOnly $
+      DB.query
+        db
+        [sql|
+          SELECT user_id FROM users u
+          WHERE u.user_id = ?
+            AND u.deleted = ?
+            AND NOT EXISTS (SELECT c.conn_id FROM connections c WHERE c.user_id = u.user_id)
+        |]
+        (userId, True)
+  case userId_ of
+    Just _ -> DB.execute db "DELETE FROM users WHERE user_id = ?" (Only userId) $> True
+    _ -> pure False
 
-deleteUsersWithoutConns :: DB.Connection -> IO ()
-deleteUsersWithoutConns db =
-  DB.execute
-    db
-    [sql|
-      DELETE FROM users u 
-      WHERE u.deleted = ?
-        AND NOT EXISTS (SELECT * FROM connections WHERE user_id = u.user_id)
-    |]
-    (Only True)
+deleteUsersWithoutConns :: DB.Connection -> IO [Int64]
+deleteUsersWithoutConns db = do
+  userIds <-
+    map fromOnly
+      <$> DB.query
+        db
+        [sql|
+          SELECT user_id FROM users u
+          WHERE u.deleted = ?
+            AND NOT EXISTS (SELECT c.conn_id FROM connections c WHERE c.user_id = u.user_id)
+        |]
+        (Only True)
+  forM_ userIds $ DB.execute db "DELETE FROM users WHERE user_id = ?" . Only
+  pure userIds
 
 createConn_ ::
   TVar ChaChaDRG ->